feat: mongodb sync template improvements and tests

templates/flamingock-mongodb-sync-template/src/main/java/io/flamingock/template/mongodb/MongoChangeTemplate.java

@@ -21,9 +21,75 @@
 import io.flamingock.api.annotations.Nullable;
 import io.flamingock.api.annotations.Rollback;
 import io.flamingock.api.template.AbstractChangeTemplate;
+import io.flamingock.internal.util.log.FlamingockLoggerFactory;
+import io.flamingock.template.mongodb.model.MongoApplyPayload;
 import io.flamingock.template.mongodb.model.MongoOperation;
+import io.flamingock.template.mongodb.validation.MongoOperationValidator;
+import io.flamingock.template.mongodb.validation.MongoTemplateValidationException;
+import io.flamingock.template.mongodb.validation.ValidationError;
+import org.slf4j.Logger;
 
-public class MongoChangeTemplate extends AbstractChangeTemplate<Void, MongoOperation, MongoOperation> {
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * MongoDB Change Template for executing declarative MongoDB operations defined in YAML.
+ *
+ * <h2>Apply Behavior</h2>
+ * <p>The {@link #apply} method executes all operations defined in the payload sequentially.
+ * The behavior differs based on the transactional mode:</p>
+ *
+ * <h3>Transactional Mode ({@code transactional: true})</h3>
+ * <ul>
+ *   <li>All operations execute within a MongoDB transaction (ClientSession)</li>
+ *   <li>If any operation fails, MongoDB automatically rolls back the entire transaction</li>
+ *   <li>Per-operation rollback definitions are NOT executed (the transaction handles atomicity)</li>
+ * </ul>
+ *
+ * <h3>Non-Transactional Mode ({@code transactional: false})</h3>
+ * <ul>
+ *   <li>Each operation executes independently without a transaction</li>
+ *   <li>Successfully completed operations are tracked</li>
+ *   <li>If operation N fails, auto-rollback is triggered for operations 1 to N-1</li>
+ *   <li>Auto-rollback executes per-operation rollbacks in reverse order</li>
+ *   <li>Operations without a rollback definition are skipped during auto-rollback</li>
+ *   <li>After auto-rollback completes, the original exception is re-thrown</li>
+ * </ul>
+ *
+ * <h2>YAML Example</h2>
+ * <pre>{@code
+ * id: create-orders-collection
+ * transactional: false
+ * template: MongoChangeTemplate
+ * targetSystem:
+ *   id: "mongodb"
+ * apply:
+ *   operations:
+ *     - type: createCollection
+ *       collection: orders
+ *       rollback:
+ *         type: dropCollection
+ *         collection: orders
+ *
+ *     - type: insert
+ *       collection: orders
+ *       parameters:
+ *         documents:
+ *           - orderId: "ORD-001"
+ *             customer: "John Doe"
+ *       rollback:
+ *         type: delete
+ *         collection: orders
+ *         parameters:
+ *           filter: {}
+ * }</pre>
+ *
+ * @see MongoOperation
+ * @see MongoApplyPayload
+ */
+public class MongoChangeTemplate extends AbstractChangeTemplate<Void, MongoApplyPayload, MongoApplyPayload> {
+
+    private static final Logger logger = FlamingockLoggerFactory.getLogger(MongoChangeTemplate.class);
 
     public MongoChangeTemplate() {
         super(MongoOperation.class);
@@ -34,19 +100,89 @@ public void apply(MongoDatabase db, @Nullable ClientSession clientSession) {
         if (this.isTransactional && clientSession == null) {
             throw new IllegalArgumentException(String.format("Transactional change[%s] requires transactional ecosystem with ClientSession", changeId));
         }
-        executeOp(db, applyPayload, clientSession);
+        validatePayload(applyPayload, changeId);
+        executeOperationsWithAutoRollback(db, applyPayload, clientSession);
     }
 
     @Rollback
     public void rollback(MongoDatabase db, @Nullable ClientSession clientSession) {
         if (this.isTransactional && clientSession == null) {
             throw new IllegalArgumentException(String.format("Transactional change[%s] requires transactional ecosystem with ClientSession", changeId));
         }
-        executeOp(db, rollbackPayload, clientSession);
+        validatePayload(rollbackPayload, changeId + ".rollback");
+        executeRollbackOperations(db, applyPayload, clientSession);
+    }
+
+    private void validatePayload(MongoApplyPayload payload, String entityId) {
+        if (payload == null || payload.getOperations() == null) {
+            return;
+        }
+
+        List<ValidationError> errors = new ArrayList<>();
+        List<MongoOperation> operations = payload.getOperations();
+        for (int i = 0; i < operations.size(); i++) {
+            String opId = entityId + ".operations[" + i + "]";
+            errors.addAll(MongoOperationValidator.validate(operations.get(i), opId));
+        }
+
+        if (!errors.isEmpty()) {
+            throw new MongoTemplateValidationException(errors);
+        }
     }
 
-    private void executeOp(MongoDatabase db, MongoOperation op, ClientSession clientSession) {
-        op.getOperator(db).apply(clientSession);
+    private void executeOperationsWithAutoRollback(MongoDatabase db, MongoApplyPayload payload, ClientSession clientSession) {
+        if (payload == null) {
+            return;
+        }
+
+        List<MongoOperation> operations = payload.getOperations();
+
+        // Transactional, MongoDB handles rollback
+        if (this.isTransactional && clientSession != null) {
+            for (MongoOperation op : operations) {
+                op.getOperator(db).apply(clientSession);
+            }
+            return;
+        }
+
+        // Non-transactional: auto-rollback on failure
+        List<MongoOperation> successfulOps = new ArrayList<>();
+        for (MongoOperation op : operations) {
+            try {
+                op.getOperator(db).apply(clientSession);
+                successfulOps.add(op);
+            } catch (Exception e) {
+                rollbackSuccessfulOperations(db, successfulOps, clientSession);
+                throw e;
+            }
+        }
+    }
+
+    private void rollbackSuccessfulOperations(MongoDatabase db, List<MongoOperation> successfulOps, ClientSession clientSession) {
+        for (int i = successfulOps.size() - 1; i >= 0; i--) {
+            MongoOperation op = successfulOps.get(i);
+            if (op.getRollback() != null) {
+                try {
+                    op.getRollback().getOperator(db).apply(clientSession);
+                } catch (Exception rollbackEx) {
+                    logger.warn("Rollback failed for operation: {}", op, rollbackEx);
+                }
+            }
+        }
+    }
+
+    private void executeRollbackOperations(MongoDatabase db, MongoApplyPayload payload, ClientSession clientSession) {
+        if (payload == null) {
+            return;
+        }
+
+        List<MongoOperation> operations = payload.getOperations();
+        for (int i = operations.size() - 1; i >= 0; i--) {
+            MongoOperation op = operations.get(i);
+            if (op.getRollback() != null) {
+                op.getRollback().getOperator(db).apply(clientSession);
+            }
+        }
     }
 
 }

templates/flamingock-mongodb-sync-template/src/main/java/io/flamingock/template/mongodb/mapper/CreateViewOptionsMapper.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2025 Flamingock (https://www.flamingock.io)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.flamingock.template.mongodb.mapper;
+
+import com.mongodb.client.model.CreateViewOptions;
+
+import java.util.Map;
+
+import static io.flamingock.template.mongodb.mapper.MapperUtil.getCollation;
+
+public final class CreateViewOptionsMapper {
+
+    private CreateViewOptionsMapper() {}
+
+    public static CreateViewOptions map(Map<String, Object> options) {
+        CreateViewOptions result = new CreateViewOptions();
+
+        if (options.containsKey("collation")) {
+            result.collation(getCollation(options, "collation"));
+        }
+
+        return result;
+    }
+}

templates/flamingock-mongodb-sync-template/src/main/java/io/flamingock/template/mongodb/mapper/IndexOptionsMapper.java

@@ -30,7 +30,9 @@
 import java.util.concurrent.TimeUnit;
 
 
-public class IndexOptionsMapper {
+public final class IndexOptionsMapper {
+
+    private IndexOptionsMapper() {}
 
     public static IndexOptions mapToIndexOptions(Map<String, Object> options) {
         IndexOptions indexOptions = new IndexOptions();
@@ -98,10 +100,5 @@ public static IndexOptions mapToIndexOptions(Map<String, Object> options) {
 
         return indexOptions;
     }
-
-
-    // Utility methods for safe type checking with exception handling
-
-
 }
 

templates/flamingock-mongodb-sync-template/src/main/java/io/flamingock/template/mongodb/mapper/InsertOptionsMapper.java

@@ -19,13 +19,14 @@
 import com.mongodb.client.model.InsertOneOptions;
 
 import java.util.Map;
-import java.util.concurrent.TimeUnit;
 
 import static io.flamingock.template.mongodb.mapper.MapperUtil.getBoolean;
 
-public class InsertOptionsMapper {
+public final class InsertOptionsMapper {
 
-    public static InsertOneOptions mapToInertOneOptions(Map<String, Object> options) {
+    private InsertOptionsMapper() {}
+
+    public static InsertOneOptions mapToInsertOneOptions(Map<String, Object> options) {
         InsertOneOptions insertOneOptions = new InsertOneOptions();
 
         if (options.containsKey("bypassDocumentValidation")) {
@@ -35,18 +36,18 @@ public static InsertOneOptions mapToInertOneOptions(Map<String, Object> options)
         return insertOneOptions;
     }
 
-    public static InsertManyOptions mapToInertManyOptions(Map<String, Object> options) {
-        InsertManyOptions insertOneOptions = new InsertManyOptions();
+    public static InsertManyOptions mapToInsertManyOptions(Map<String, Object> options) {
+        InsertManyOptions insertManyOptions = new InsertManyOptions();
 
         if (options.containsKey("bypassDocumentValidation")) {
-            insertOneOptions.bypassDocumentValidation(getBoolean(options, "bypassDocumentValidation"));
+            insertManyOptions.bypassDocumentValidation(getBoolean(options, "bypassDocumentValidation"));
         }
 
         if (options.containsKey("ordered")) {
-            insertOneOptions.bypassDocumentValidation(getBoolean(options, "ordered"));
+            insertManyOptions.ordered(getBoolean(options, "ordered"));
         }
 
-        return insertOneOptions;
+        return insertManyOptions;
     }
 
 }

templates/flamingock-mongodb-sync-template/src/main/java/io/flamingock/template/mongodb/mapper/MapperUtil.java

@@ -16,10 +16,12 @@
 package io.flamingock.template.mongodb.mapper;
 
 import com.mongodb.client.model.Collation;
+import org.bson.BsonArray;
 import org.bson.BsonDocument;
 import org.bson.BsonValue;
 import org.bson.conversions.Bson;
 
+import java.util.List;
 import java.util.Map;
 
 public final class MapperUtil {
@@ -101,7 +103,9 @@ public static BsonDocument toBsonDocument(Map<String, Object> map) {
     // Converts Java types into BSON types
     @SuppressWarnings("unchecked")
     public static BsonValue toBsonValue(Object value) {
-        if (value instanceof String) {
+        if (value == null) {
+            return org.bson.BsonNull.VALUE;
+        } else if (value instanceof String) {
             return new org.bson.BsonString((String) value);
         } else if (value instanceof Integer) {
             return new org.bson.BsonInt32((Integer) value);
@@ -111,9 +115,20 @@ public static BsonValue toBsonValue(Object value) {
             return new org.bson.BsonDouble((Double) value);
         } else if (value instanceof Boolean) {
             return new org.bson.BsonBoolean((Boolean) value);
+        } else if (value instanceof List) {
+            return toBsonArray((List<?>) value);
         } else if (value instanceof Map) {
             return toBsonDocument((Map<String, Object>) value);
         }
         throw new IllegalArgumentException("Unsupported BSON type: " + value.getClass().getSimpleName());
     }
+
+    // Converts a List to BsonArray
+    public static BsonArray toBsonArray(List<?> list) {
+        BsonArray array = new BsonArray();
+        for (Object item : list) {
+            array.add(toBsonValue(item));
+        }
+        return array;
+    }
 }

templates/flamingock-mongodb-sync-template/src/main/java/io/flamingock/template/mongodb/mapper/RenameCollectionOptionsMapper.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2025 Flamingock (https://www.flamingock.io)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.flamingock.template.mongodb.mapper;
+
+import com.mongodb.client.model.RenameCollectionOptions;
+
+import java.util.Map;
+
+import static io.flamingock.template.mongodb.mapper.MapperUtil.getBoolean;
+
+public final class RenameCollectionOptionsMapper {
+
+    private RenameCollectionOptionsMapper() {}
+
+    public static RenameCollectionOptions map(Map<String, Object> options) {
+        RenameCollectionOptions result = new RenameCollectionOptions();
+
+        if (options.containsKey("dropTarget")) {
+            result.dropTarget(getBoolean(options, "dropTarget"));
+        }
+
+        return result;
+    }
+}