/*
 * Copyright Debezium Authors.
 *
 * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
 */
package io.debezium.connector.mongodb.transforms;

import static org.fest.assertions.Assertions.assertThat;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;
import org.bson.Document;
import org.bson.RawBsonDocument;
import org.junit.Test;

import io.debezium.data.Envelope;
import io.debezium.data.SchemaUtil;
import io.debezium.doc.FixFor;

/**
 * Integration test for {@link ExtractNewDocumentState}. It sends operations into
 * MongoDB and listens on messages that are generated by Debezium plug-in. The messages
 * are then run through the SMT itself.
 *
 * @author Jiri Pechanec
 */
public class ExtractNewDocumentStateTestIT extends AbstractExtractNewDocumentStateTestIT {

    private static final String CONFIG_DROP_TOMBSTONES = "drop.tombstones";
    private static final String HANDLE_DELETES = "delete.handling.mode";

    @Override
    protected String getCollectionName() {
        return "functional";
    }

    @Test
    @FixFor("DBZ-563")
    public void shouldDropTombstoneByDefault() throws InterruptedException {
        // First insert
        primary().execute("insert", client -> {
            client.getDatabase(DB_NAME).getCollection(this.getCollectionName())
                    .insertOne(Document.parse("{'_id': 1, 'dataStr': 'hello', 'dataInt': 123, 'dataLong': 80000000000}"));
        });

        SourceRecords records = consumeRecordsByTopic(1);

        assertThat(records.recordsForTopic(this.topicName()).size()).isEqualTo(1);

        // Test Delete
        primary().execute("delete", client -> {
            client.getDatabase(DB_NAME).getCollection(this.getCollectionName()).deleteOne(RawBsonDocument.parse("{'_id' : 1}"));
        });

        // First delete record to arrive is coming from the oplog
        SourceRecord firstRecord = getRecordByOperation(Envelope.Operation.DELETE);
        final SourceRecord transformedDelete = transformation.apply(firstRecord);
        assertThat(transformedDelete).isNull();

        // Second record is the tombstone
        SourceRecord tombstoneRecord = getNextRecord();
        assertThat(tombstoneRecord).isNotNull();

        // Test tombstone record is dropped
        final SourceRecord transformedTombstone = transformation.apply(tombstoneRecord);
        assertThat(transformedTombstone).isNull();
    }

    @Test
    public void shouldTransformEvents() throws InterruptedException, IOException {
        final Map<String, String> transformationConfig = new HashMap<>();
        transformationConfig.put(CONFIG_DROP_TOMBSTONES, "false");
        transformationConfig.put(HANDLE_DELETES, "none");
        transformation.configure(transformationConfig);

        // Test insert
        primary().execute("insert", client -> {
            client.getDatabase(DB_NAME).getCollection(this.getCollectionName())
                    .insertOne(Document.parse("{'_id': 1, 'dataStr': 'hello', 'dataInt': 123, 'dataLong': 80000000000}"));
        });

        SourceRecords records = consumeRecordsByTopic(1);

        assertThat(records.recordsForTopic(this.topicName()).size()).isEqualTo(1);
        final SourceRecord insertRecord = records.recordsForTopic(this.topicName()).get(0);
        final SourceRecord transformedInsert = transformation.apply(insertRecord);
        final Struct transformedInsertValue = (Struct) transformedInsert.value();

        assertThat(transformedInsert.valueSchema().field("id").schema()).isEqualTo(Schema.OPTIONAL_INT32_SCHEMA);
        assertThat(transformedInsert.valueSchema().field("dataStr").schema()).isEqualTo(Schema.OPTIONAL_STRING_SCHEMA);
        assertThat(transformedInsert.valueSchema().field("dataInt").schema()).isEqualTo(Schema.OPTIONAL_INT32_SCHEMA);
        assertThat(transformedInsert.valueSchema().field("dataLong").schema()).isEqualTo(Schema.OPTIONAL_INT64_SCHEMA);
        assertThat(transformedInsertValue.get("id")).isEqualTo(1);
        assertThat(transformedInsertValue.get("dataStr")).isEqualTo("hello");
        assertThat(transformedInsertValue.get("dataInt")).isEqualTo(123);
        assertThat(transformedInsertValue.get("dataLong")).isEqualTo(80_000_000_000l);

        // Test update
        primary().execute("update", client -> {
            client.getDatabase(DB_NAME).getCollection(this.getCollectionName()).updateOne(RawBsonDocument.parse("{'_id' : 1}"),
                    RawBsonDocument.parse("{'$set': {'dataStr': 'bye'}}"));
        });

        records = consumeRecordsByTopic(1);
        final SourceRecord candidateUpdateRecord = records.recordsForTopic(this.topicName()).get(0);
        if (((Struct) candidateUpdateRecord.value()).get("op").equals("c")) {
            // MongoDB is not providing really consistent snapshot, so the initial insert
            // can arrive both in initial sync snapshot and in oplog
            records = consumeRecordsByTopic(1);
        }

        assertThat(records.recordsForTopic(this.topicName()).size()).isEqualTo(1);
        final SourceRecord updateRecord = records.recordsForTopic(this.topicName()).get(0);
        final SourceRecord transformedUpdate = transformation.apply(updateRecord);
        final Struct transformedUpdateValue = (Struct) transformedUpdate.value();

        assertThat(transformedUpdate.valueSchema().field("id").schema()).isEqualTo(Schema.OPTIONAL_INT32_SCHEMA);
        assertThat(transformedUpdate.valueSchema().field("dataStr").schema()).isEqualTo(Schema.OPTIONAL_STRING_SCHEMA);
        assertThat(transformedUpdateValue.get("id")).isEqualTo(1);
        assertThat(transformedUpdateValue.get("dataStr")).isEqualTo("bye");

        // Test Update Multiple Fields
        primary().execute("update", client -> {
            client.getDatabase(DB_NAME).getCollection(this.getCollectionName()).updateOne(RawBsonDocument.parse("{'_id' : 1}"),
                    RawBsonDocument.parse("{'$set': {'newStr': 'hello', 'dataInt': 456}}"));
        });

        records = consumeRecordsByTopic(1);

        assertThat(records.recordsForTopic(this.topicName()).size()).isEqualTo(1);
        final SourceRecord updateMultipleRecord = records.recordsForTopic(this.topicName()).get(0);
        final SourceRecord transformedMultipleUpdate = transformation.apply(updateMultipleRecord);
        final Struct transformedMultipleUpdateValue = (Struct) transformedMultipleUpdate.value();

        assertThat(transformedMultipleUpdate.valueSchema().field("id").schema()).isEqualTo(Schema.OPTIONAL_INT32_SCHEMA);
        assertThat(transformedMultipleUpdate.valueSchema().field("newStr").schema()).isEqualTo(Schema.OPTIONAL_STRING_SCHEMA);
        assertThat(transformedMultipleUpdate.valueSchema().field("dataInt").schema()).isEqualTo(Schema.OPTIONAL_INT32_SCHEMA);
        assertThat(transformedMultipleUpdateValue.get("id")).isEqualTo(1);
        assertThat(transformedMultipleUpdateValue.get("newStr")).isEqualTo("hello");
        assertThat(transformedMultipleUpdateValue.get("dataInt")).isEqualTo(456);

        // Test Update with $unset operation
        primary().execute("update", client -> {
            client.getDatabase(DB_NAME).getCollection(this.getCollectionName()).updateOne(RawBsonDocument.parse("{'_id' : 1}"),
                    RawBsonDocument.parse("{'$unset': {'newStr': ''}}"));
        });

        records = consumeRecordsByTopic(1);

        assertThat(records.recordsForTopic(this.topicName()).size()).isEqualTo(1);
        final SourceRecord updateUnsetRecord = records.recordsForTopic(this.topicName()).get(0);
        final SourceRecord transformedUnsetUpdate = transformation.apply(updateUnsetRecord);
        final Struct transformedUnsetUpdateValue = (Struct) transformedUnsetUpdate.value();

        assertThat(transformedUnsetUpdate.valueSchema().field("id").schema()).isEqualTo(Schema.OPTIONAL_INT32_SCHEMA);
        assertThat(transformedUnsetUpdate.valueSchema().field("newStr").schema()).isEqualTo(Schema.OPTIONAL_STRING_SCHEMA);
        assertThat(transformedUnsetUpdateValue.get("id")).isEqualTo(1);
        assertThat(transformedUnsetUpdateValue.get("newStr")).isEqualTo(null);

        // Test FullUpdate
        primary().execute("update", client -> {
            client.getDatabase(DB_NAME).getCollection(this.getCollectionName()).updateOne(RawBsonDocument.parse("{'_id' : 1}"),
                    RawBsonDocument.parse("{'dataStr': 'Hi again'}"));
        });

        records = consumeRecordsByTopic(1);
        final SourceRecord candidateFullUpdateRecord = records.recordsForTopic(this.topicName()).get(0);
        if (((Struct) candidateFullUpdateRecord.value()).get("op").equals("c")) {
            // MongoDB is not providing really consistent snapshot, so the initial insert
            // can arrive both in initial sync snapshot and in oplog
            records = consumeRecordsByTopic(1);
        }

        assertThat(records.recordsForTopic(this.topicName()).size()).isEqualTo(1);
        final SourceRecord FullUpdateRecord = records.recordsForTopic(this.topicName()).get(0);
        final SourceRecord transformedFullUpdate = transformation.apply(FullUpdateRecord);
        final Struct transformedFullUpdateValue = (Struct) transformedFullUpdate.value();

        assertThat(transformedFullUpdate.valueSchema().field("id").schema()).isEqualTo(Schema.OPTIONAL_INT32_SCHEMA);
        assertThat(transformedFullUpdate.valueSchema().field("dataStr").schema()).isEqualTo(Schema.OPTIONAL_STRING_SCHEMA);
        assertThat(transformedFullUpdateValue.get("id")).isEqualTo(1);
        assertThat(transformedFullUpdateValue.get("dataStr")).isEqualTo("Hi again");

        // Test Delete
        primary().execute("delete", client -> {
            client.getDatabase(DB_NAME).getCollection(this.getCollectionName()).deleteOne(RawBsonDocument.parse("{'_id' : 1}"));
        });

        records = consumeRecordsByTopic(2);
        assertThat(records.recordsForTopic(this.topicName()).size()).isEqualTo(2);

        // Test mongo Deletion operation
        final SourceRecord deleteRecord = records.recordsForTopic(this.topicName()).get(0);
        final SourceRecord transformedDelete = transformation.apply(deleteRecord);
        final Struct transformedDeleteValue = (Struct) transformedDelete.value();

        assertThat(transformedDeleteValue).isNull();

        // Test tombstone record
        final SourceRecord tombstoneRecord = records.recordsForTopic(this.topicName()).get(1);
        final SourceRecord transformedTombstone = transformation.apply(tombstoneRecord);

        assertThat(transformedTombstone.value()).isNull();

        // Assert deletion preserves key
        assertThat(SchemaUtil.asString(transformedDelete.keySchema())).isEqualTo(SchemaUtil.asString(transformedTombstone.keySchema()));
        assertThat(transformedDelete.key().toString()).isEqualTo(transformedTombstone.key().toString());
    }
}
