// Path: src/main/java/at/grahsl/kafka/connect/mongodb/converter/SinkDocument.java
// public class SinkDocument {
// 
//     private final Optional<BsonDocument> keyDoc;
//     private final Optional<BsonDocument> valueDoc;
// 
//     public SinkDocument(BsonDocument keyDoc, BsonDocument valueDoc) {
//         this.keyDoc = Optional.ofNullable(keyDoc);
//         this.valueDoc = Optional.ofNullable(valueDoc);
//     }
// 
//     public Optional<BsonDocument> getKeyDoc() {
//         return keyDoc;
//     }
// 
//     public Optional<BsonDocument> getValueDoc() {
//         return valueDoc;
//     }
// 
//     public SinkDocument clone() {
//         BsonDocument kd = keyDoc.isPresent() ? keyDoc.get().clone() : null;
//         BsonDocument vd = valueDoc.isPresent() ? valueDoc.get().clone() : null;
//         return new SinkDocument(kd,vd);
//     }
// 
// }


import at.grahsl.kafka.connect.mongodb.converter.SinkDocument;
import com.mongodb.DBCollection;
import com.mongodb.client.model.UpdateOneModel;
import com.mongodb.client.model.UpdateOptions;
import com.mongodb.client.model.WriteModel;
import org.apache.kafka.connect.errors.DataException;
import org.apache.kafka.connect.sink.SinkRecord;
import org.bson.*;
import org.bson.conversions.Bson;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
package at.grahsl.kafka.connect.mongodb.writemodel.strategy;



/**
 * This WriteModelStrategy implementation adds the kafka coordinates of processed
 * records to the actual SinkDocument as meta-data before it gets written to the
 * MongoDB collection. The pre-defined and currently not(!) configurable data format
 * for this is using a sub-document with the following structure, field names
 * and value &lt;PLACEHOLDERS&gt;:
 *
 * {
 *      ...,
 *
 *      "_kafkaCoords":{
 *   "_topic": "&lt;TOPIC_NAME&gt;",
 *   "_partition": &lt;PARTITION_NUMBER&gt;,
 *   "_offset": &lt;OFFSET_NUMBER&gt;
 *      },
 *
 *      ...
 * }
 *
 * This "meta-data" is used to perform the actual staleness check, namely, that upsert operations
 * based on the corresponding document's _id field will get suppressed in case newer data has
 * already been written to the collection in question. Newer data means a document exhibiting
 * a greater than or equal offset for the same kafka topic and partition is already present in the sink.
 *
 * ! IMPORTANT NOTE !
 * This WriteModelStrategy needs MongoDB version 4.2+ and Java Driver 3.11+ since
 * lower versions of either lack the support for leveraging update pipeline syntax.
 *
 */
public class MonotonicWritesDefaultStrategy implements WriteModelStrategy {

    public static final String FIELD_KAFKA_COORDS = "_kafkaCoords";
    public static final String FIELD_TOPIC = "_topic";
    public static final String FIELD_PARTITION = "_partition";
    public static final String FIELD_OFFSET = "_offset";

    private static final UpdateOptions UPDATE_OPTIONS =
// Path: src/main/java/at/grahsl/kafka/connect/mongodb/converter/SinkDocument.java
// public class SinkDocument {
// 
//     private final Optional<BsonDocument> keyDoc;
//     private final Optional<BsonDocument> valueDoc;
// 
//     public SinkDocument(BsonDocument keyDoc, BsonDocument valueDoc) {
//         this.keyDoc = Optional.ofNullable(keyDoc);
//         this.valueDoc = Optional.ofNullable(valueDoc);
//     }
// 
//     public Optional<BsonDocument> getKeyDoc() {
//         return keyDoc;
//     }
// 
//     public Optional<BsonDocument> getValueDoc() {
//         return valueDoc;
//     }
// 
//     public SinkDocument clone() {
//         BsonDocument kd = keyDoc.isPresent() ? keyDoc.get().clone() : null;
//         BsonDocument vd = valueDoc.isPresent() ? valueDoc.get().clone() : null;
//         return new SinkDocument(kd,vd);
//     }
// 
// }


// Path: src/main/java/at/grahsl/kafka/connect/mongodb/writemodel/strategy/MonotonicWritesDefaultStrategy.java
import at.grahsl.kafka.connect.mongodb.converter.SinkDocument;
import com.mongodb.DBCollection;
import com.mongodb.client.model.UpdateOneModel;
import com.mongodb.client.model.UpdateOptions;
import com.mongodb.client.model.WriteModel;
import org.apache.kafka.connect.errors.DataException;
import org.apache.kafka.connect.sink.SinkRecord;
import org.bson.*;
import org.bson.conversions.Bson;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

package at.grahsl.kafka.connect.mongodb.writemodel.strategy;



/**
 * This WriteModelStrategy implementation adds the kafka coordinates of processed
 * records to the actual SinkDocument as meta-data before it gets written to the
 * MongoDB collection. The pre-defined and currently not(!) configurable data format
 * for this is using a sub-document with the following structure, field names
 * and value &lt;PLACEHOLDERS&gt;:
 *
 * {
 *      ...,
 *
 *      "_kafkaCoords":{
 *   "_topic": "&lt;TOPIC_NAME&gt;",
 *   "_partition": &lt;PARTITION_NUMBER&gt;,
 *   "_offset": &lt;OFFSET_NUMBER&gt;
 *      },
 *
 *      ...
 * }
 *
 * This "meta-data" is used to perform the actual staleness check, namely, that upsert operations
 * based on the corresponding document's _id field will get suppressed in case newer data has
 * already been written to the collection in question. Newer data means a document exhibiting
 * a greater than or equal offset for the same kafka topic and partition is already present in the sink.
 *
 * ! IMPORTANT NOTE !
 * This WriteModelStrategy needs MongoDB version 4.2+ and Java Driver 3.11+ since
 * lower versions of either lack the support for leveraging update pipeline syntax.
 *
 */
public class MonotonicWritesDefaultStrategy implements WriteModelStrategy {

    public static final String FIELD_KAFKA_COORDS = "_kafkaCoords";
    public static final String FIELD_TOPIC = "_topic";
    public static final String FIELD_PARTITION = "_partition";
    public static final String FIELD_OFFSET = "_offset";

    private static final UpdateOptions UPDATE_OPTIONS =
            new UpdateOptions().upsert(true);