/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 org.apache.camel.component.debezium;

import java.time.Instant;
import java.util.HashMap;
import java.util.Map;

import io.debezium.data.Envelope;
import org.apache.camel.Consumer;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.component.debezium.configuration.FileConnectorEmbeddedDebeziumConfiguration;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;

@ExtendWith(MockitoExtension.class)
public class DebeziumEndpointTest {

    private DebeziumEndpoint debeziumEndpoint;

    @Mock
    private Processor processor;

    @BeforeEach
    public void setUp() {
        debeziumEndpoint = new DebeziumTestEndpoint(
                "", new DebeziumTestComponent(new DefaultCamelContext()),
                new FileConnectorEmbeddedDebeziumConfiguration());
    }

    @Test
    void testIfCreatesConsumer() throws Exception {
        final Consumer debeziumConsumer = debeziumEndpoint.createConsumer(processor);

        assertNotNull(debeziumConsumer);
    }

    @Test
    void testIfFailsToCreateProducer() {
        assertThrows(UnsupportedOperationException.class, () -> {
            debeziumEndpoint.createProducer();
        });
    }

    @Test
    void testIfCreatesExchangeFromSourceCreateRecord() {
        final SourceRecord sourceRecord = createCreateRecord();

        final Exchange exchange = debeziumEndpoint.createDbzExchange(null, sourceRecord);
        final Message inMessage = exchange.getIn();

        assertNotNull(exchange);
        // assert headers
        assertEquals("dummy", inMessage.getHeader(DebeziumConstants.HEADER_IDENTIFIER));
        assertEquals(Envelope.Operation.CREATE.code(),
                inMessage.getHeader(DebeziumConstants.HEADER_OPERATION));
        final Struct key = (Struct) inMessage.getHeader(DebeziumConstants.HEADER_KEY);
        assertEquals(12345, key.getInt32("id").intValue());
        assertSourceMetadata(inMessage);
        assertNotNull(inMessage.getHeader(DebeziumConstants.HEADER_TIMESTAMP));

        // assert value
        final Struct body = (Struct) inMessage.getBody();
        assertNotNull(body);
        assertEquals((byte) 1, body.getInt8("id").byteValue());

        // assert schema
        assertSchema(body.schema());
    }

    @Test
    void testIfCreatesExchangeFromSourceDeleteRecord() {
        final SourceRecord sourceRecord = createDeleteRecord();

        final Exchange exchange = debeziumEndpoint.createDbzExchange(null, sourceRecord);
        final Message inMessage = exchange.getIn();

        assertNotNull(exchange);
        // assert headers
        assertEquals("dummy", inMessage.getHeader(DebeziumConstants.HEADER_IDENTIFIER));
        assertEquals(Envelope.Operation.DELETE.code(),
                inMessage.getHeader(DebeziumConstants.HEADER_OPERATION));
        final Struct key = (Struct) inMessage.getHeader(DebeziumConstants.HEADER_KEY);
        assertEquals(12345, key.getInt32("id").intValue());
        assertNotNull(inMessage.getHeader(DebeziumConstants.HEADER_BEFORE));

        // assert value
        final Struct body = (Struct) inMessage.getBody();
        assertNull(body); // we expect body to be null since is a delete
    }

    @Test
    void testIfCreatesExchangeFromSourceDeleteRecordWithNull() {
        final SourceRecord sourceRecord = createDeleteRecordWithNull();

        final Exchange exchange = debeziumEndpoint.createDbzExchange(null, sourceRecord);
        final Message inMessage = exchange.getIn();

        assertNotNull(exchange);
        // assert headers
        assertEquals("dummy", inMessage.getHeader(DebeziumConstants.HEADER_IDENTIFIER));
        final Struct key = (Struct) inMessage.getHeader(DebeziumConstants.HEADER_KEY);
        assertEquals(12345, key.getInt32("id").intValue());

        // assert value
        final Struct body = (Struct) inMessage.getBody();
        assertNull(body);
    }

    @Test
    void testIfCreatesExchangeFromSourceUpdateRecord() {
        final SourceRecord sourceRecord = createUpdateRecord();

        final Exchange exchange = debeziumEndpoint.createDbzExchange(null, sourceRecord);
        final Message inMessage = exchange.getIn();

        assertNotNull(exchange);
        // assert headers
        assertEquals("dummy", inMessage.getHeader(DebeziumConstants.HEADER_IDENTIFIER));
        assertEquals(Envelope.Operation.UPDATE.code(),
                inMessage.getHeader(DebeziumConstants.HEADER_OPERATION));
        final Struct key = (Struct) inMessage.getHeader(DebeziumConstants.HEADER_KEY);
        assertEquals(12345, key.getInt32("id").intValue());
        assertSourceMetadata(inMessage);

        // assert value
        final Struct before = (Struct) inMessage.getHeader(DebeziumConstants.HEADER_BEFORE);
        final Struct after = (Struct) inMessage.getBody();
        assertNotNull(before);
        assertNotNull(after);
        assertEquals((byte) 1, before.getInt8("id").byteValue());
        assertEquals((byte) 2, after.getInt8("id").byteValue());
    }

    @Test
    void testIfCreatesExchangeFromSourceRecordOtherThanStruct() {
        final SourceRecord sourceRecord = createStringRecord();

        final Exchange exchange = debeziumEndpoint.createDbzExchange(null, sourceRecord);
        final Message inMessage = exchange.getIn();

        assertNotNull(exchange);

        // assert headers
        assertEquals("dummy", inMessage.getHeader(DebeziumConstants.HEADER_IDENTIFIER));
        assertNull(inMessage.getHeader(DebeziumConstants.HEADER_OPERATION));

        // assert value
        final String value = (String) inMessage.getBody();
        assertEquals(sourceRecord.value(), value);
    }

    @Test
    void testIfHandlesUnknownSchema() {
        final SourceRecord sourceRecord = createUnknownUnnamedSchemaRecord();

        final Exchange exchange = debeziumEndpoint.createDbzExchange(null, sourceRecord);
        final Message inMessage = exchange.getIn();

        assertNotNull(exchange);
        // assert headers
        assertEquals("dummy", inMessage.getHeader(DebeziumConstants.HEADER_IDENTIFIER));
        assertNull(inMessage.getHeader(DebeziumConstants.HEADER_OPERATION));
        assertNull(inMessage.getHeader(DebeziumConstants.HEADER_KEY));

        // assert value
        final Struct body = (Struct) inMessage.getBody();
        // we have to get value of after with struct, we are strict about this case
        assertNull(body);
    }

    @Test
    void testIfCreatesExchangeFromSourceDdlRecord() {
        final SourceRecord sourceRecord = createDdlSQLRecord();

        final Exchange exchange = debeziumEndpoint.createDbzExchange(null, sourceRecord);
        final Message inMessage = exchange.getIn();

        assertNotNull(exchange);
        // assert headers
        assertEquals("dummy", inMessage.getHeader(DebeziumConstants.HEADER_IDENTIFIER));
        assertEquals("SET character_set_server=utf8, collation_server=utf8_bin",
                inMessage.getHeader(DebeziumConstants.HEADER_DDL_SQL));
    }

    private SourceRecord createCreateRecord() {
        final Schema recordSchema = SchemaBuilder.struct().field("id", SchemaBuilder.int8()).build();
        final Schema sourceSchema = SchemaBuilder.struct().field("lsn", SchemaBuilder.int32()).build();
        Envelope envelope = Envelope.defineSchema().withName("dummy.Envelope").withRecord(recordSchema)
                .withSource(sourceSchema).build();
        final Struct after = new Struct(recordSchema);
        final Struct source = new Struct(sourceSchema);

        after.put("id", (byte) 1);
        source.put("lsn", 1234);
        final Struct payload = envelope.create(after, source, Instant.now());
        return new SourceRecord(
                new HashMap<>(), createSourceOffset(), "dummy", createKeySchema(),
                createKeyRecord(), envelope.schema(), payload);
    }

    private SourceRecord createDeleteRecord() {
        final Schema recordSchema = SchemaBuilder.struct().field("id", SchemaBuilder.int8()).build();
        Envelope envelope = Envelope.defineSchema().withName("dummy.Envelope").withRecord(recordSchema)
                .withSource(SchemaBuilder.struct().build()).build();
        final Struct before = new Struct(recordSchema);
        before.put("id", (byte) 1);
        final Struct payload = envelope.delete(before, null, Instant.now());
        return new SourceRecord(
                new HashMap<>(), createSourceOffset(), "dummy", createKeySchema(),
                createKeyRecord(), envelope.schema(), payload);
    }

    private SourceRecord createDeleteRecordWithNull() {
        final Schema recordSchema = SchemaBuilder.struct().field("id", SchemaBuilder.int8()).build();
        Envelope.defineSchema().withName("dummy.Envelope").withRecord(recordSchema).withSource(SchemaBuilder.struct().build())
                .build();
        final Struct before = new Struct(recordSchema);
        before.put("id", (byte) 1);
        return new SourceRecord(
                new HashMap<>(), createSourceOffset(), "dummy", createKeySchema(),
                createKeyRecord(), null, null);
    }

    private SourceRecord createUpdateRecord() {
        final Schema recordSchema = SchemaBuilder.struct().field("id", SchemaBuilder.int8()).build();
        final Schema sourceSchema = SchemaBuilder.struct().field("lsn", SchemaBuilder.int32()).build();
        Envelope envelope = Envelope.defineSchema().withName("dummy.Envelope").withRecord(recordSchema)
                .withSource(sourceSchema).build();
        final Struct before = new Struct(recordSchema);
        final Struct source = new Struct(sourceSchema);
        final Struct after = new Struct(recordSchema);

        before.put("id", (byte) 1);
        after.put("id", (byte) 2);
        source.put("lsn", 1234);
        final Struct payload = envelope.update(before, after, source, Instant.now());
        return new SourceRecord(
                new HashMap<>(), createSourceOffset(), "dummy", createKeySchema(),
                createKeyRecord(), envelope.schema(), payload);
    }

    private SourceRecord createDdlSQLRecord() {
        final Schema recordSchema = SchemaBuilder.struct().field("ddl", SchemaBuilder.string()).build();
        Envelope.defineSchema().withName("dummy.Envelope").withRecord(recordSchema).withSource(SchemaBuilder.struct().build())
                .build();
        final Struct recordValue = new Struct(recordSchema);
        recordValue.put("ddl", "SET character_set_server=utf8, collation_server=utf8_bin");
        return new SourceRecord(
                new HashMap<>(), createSourceOffset(), "dummy", null,
                null, recordValue.schema(), recordValue);
    }

    private SourceRecord createUnknownUnnamedSchemaRecord() {
        final Schema recordSchema = SchemaBuilder.struct().field("id", SchemaBuilder.int8()).build();
        final Struct before = new Struct(recordSchema);
        before.put("id", (byte) 1);
        return new SourceRecord(new HashMap<>(), new HashMap<>(), "dummy", recordSchema, before);
    }

    private SourceRecord createStringRecord() {
        final Schema recordSchema = Schema.STRING_SCHEMA;
        return new SourceRecord(new HashMap<>(), createSourceOffset(), "dummy", recordSchema, "test_record");
    }

    private HashMap<String, ?> createSourceOffset() {
        final HashMap<String, Integer> sourceOffsets = new HashMap<>();
        sourceOffsets.put("pos", 111);

        return sourceOffsets;
    }

    private Schema createKeySchema() {
        return SchemaBuilder.struct().field("id", SchemaBuilder.int32().build());
    }

    private Struct createKeyRecord() {
        final Struct key = new Struct(createKeySchema());
        key.put("id", 12345);

        return key;
    }

    private void assertSourceMetadata(final Message inMessage) {
        @SuppressWarnings("unchecked")
        final Map<String, Object> source = inMessage.getHeader(DebeziumConstants.HEADER_SOURCE_METADATA, Map.class);
        assertEquals(1234, source.get("lsn"));
    }

    private void assertSchema(final Schema schema) {
        assertNotNull(schema);
        assertFalse(schema.fields().isEmpty());
    }
}
