/*
 * 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.kafka.streams.kstream.internals.foreignkeyjoin;

import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.kstream.internals.Change;
import org.apache.kafka.streams.processor.StateStoreContext;
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.Record;
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.streams.state.Stores;
import org.apache.kafka.streams.state.TimestampedKeyValueStore;
import org.apache.kafka.streams.state.ValueAndTimestamp;
import org.apache.kafka.test.MockInternalNewProcessorContext;
import org.apache.kafka.test.StreamsTestUtils;
import org.apache.kafka.test.TestUtils;

import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.io.File;
import java.io.IOException;
import java.util.Properties;
import java.util.function.Supplier;

import static org.apache.kafka.streams.kstream.internals.foreignkeyjoin.ResponseJoinProcessorSupplierTest.getDroppedRecordsRateMetric;
import static org.apache.kafka.streams.kstream.internals.foreignkeyjoin.ResponseJoinProcessorSupplierTest.getDroppedRecordsTotalMetric;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.is;

public class ForeignTableJoinProcessorSupplierTests {

    private static final Supplier<String> PK_SERDE_TOPIC_SUPPLIER = () -> "pk-topic";
    private static final CombinedKeySchema<String, String> COMBINED_KEY_SCHEMA = new CombinedKeySchema<>(
        () -> "fk-topic",
        Serdes.String(),
        PK_SERDE_TOPIC_SUPPLIER,
        Serdes.String()
    );

    private MockInternalNewProcessorContext<String, SubscriptionResponseWrapper<String>> context = null;
    private TimestampedKeyValueStore<Bytes, SubscriptionWrapper<String>> stateStore = null;
    private Processor<String, Change<String>, String, SubscriptionResponseWrapper<String>> processor = null;
    private File stateDir;

    @BeforeEach
    public void setUp() {
        stateDir = TestUtils.tempDirectory();
        final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.String());
        context = new MockInternalNewProcessorContext<>(props, new TaskId(0, 0), stateDir);

        final StoreBuilder<TimestampedKeyValueStore<Bytes, SubscriptionWrapper<String>>> storeBuilder = storeBuilder();
        processor = new ForeignTableJoinProcessorSupplier<String, String, String>(storeBuilder().name(), COMBINED_KEY_SCHEMA).get();
        stateStore = storeBuilder.build();
        context.addStateStore(stateStore);
        stateStore.init((StateStoreContext) context, stateStore);
        processor.init(context);
    }

    @AfterEach
    public void cleanUp() throws IOException {
        if (stateStore != null) {
            stateStore.close();
        }
        Utils.delete(stateDir);
    }

    private final String pk1 = "pk1";
    private final String pk2 = "pk2";
    private final String fk1 = "fk1";

    private final long[] hash = new long[]{1L, 2L};

    @Test
    public void shouldPropagateRightRecordForEachMatchingPrimaryKey() {
        putInStore(fk1, pk1);
        putInStore(fk1, pk2);

        final Record<String, Change<String>> record = new Record<>(fk1, new Change<>("new_value", null), 0);

        processor.process(record);

        assertThat(context.forwarded().size(), is(2));
        assertThat(
            context.forwarded().get(0).record(),
            is(new Record<>(pk1, new SubscriptionResponseWrapper<>(hash, "new_value", null), 0))
        );
        assertThat(
            context.forwarded().get(1).record(),
            is(new Record<>(pk2, new SubscriptionResponseWrapper<>(hash, "new_value", null), 0))
        );
    }

    @Test
    public void shouldPropagateNothingIfNoMatchingPrimaryKey() {
        final String fk2 = "fk2";
        putInStore(fk1, pk1);

        final Record<String, Change<String>> record = new Record<>(fk2, new Change<>("new_value", null), 0);

        processor.process(record);

        assertThat(context.forwarded(), empty());
    }

    @Test
    public void shouldPropagateTombstoneRightRecordForEachMatchingPrimaryKey() {
        putInStore(fk1, pk1);
        putInStore(fk1, pk2);

        final Record<String, Change<String>> record = new Record<>(fk1, new Change<>(null, "new_value"), 0);

        processor.process(record);

        assertThat(context.forwarded().size(), is(2));
        assertThat(
            context.forwarded().get(0).record(),
            is(new Record<>(pk1, new SubscriptionResponseWrapper<>(hash, null, null), 0))
        );
        assertThat(
            context.forwarded().get(1).record(),
            is(new Record<>(pk2, new SubscriptionResponseWrapper<>(hash, null, null), 0))
        );
    }

    @Test
    public void shouldNotMatchForeignKeysHavingThisFKAsPrefix() {
        final String fk = "fk";
        putInStore(fk1, pk1);
        putInStore(fk, pk2);

        final Record<String, Change<String>> record = new Record<>(fk, new Change<>("new_value", null), 0);

        processor.process(record);

        assertThat(context.forwarded().size(), is(1));
        assertThat(
            context.forwarded().get(0).record(),
            is(new Record<>(pk2, new SubscriptionResponseWrapper<>(hash, "new_value", null), 0))
        );
    }

    @Test
    public void shouldIgnoreRecordWithNullKey() {
        putInStore(fk1, pk1);

        final Record<String, Change<String>> record = new Record<>(null, new Change<>("new_value", null), 0);

        processor.process(record);

        assertThat(context.forwarded(), empty());

        // test dropped-records sensors
        Assertions.assertEquals(1.0, getDroppedRecordsTotalMetric(context));
        Assertions.assertNotEquals(0.0, getDroppedRecordsRateMetric(context));
    }

    private void putInStore(final String fk, final String pk) {
        final SubscriptionWrapper<String> oldWrapper = new SubscriptionWrapper<>(
            hash,
            SubscriptionWrapper.Instruction.PROPAGATE_ONLY_IF_FK_VAL_AVAILABLE,
            pk,
            SubscriptionWrapper.VERSION_0,
            null
        );
        final ValueAndTimestamp<SubscriptionWrapper<String>> oldValue = ValueAndTimestamp.make(oldWrapper, 0);

        final Bytes key = COMBINED_KEY_SCHEMA.toBytes(fk, pk);
        stateStore.put(key, oldValue);
    }

    private StoreBuilder<TimestampedKeyValueStore<Bytes, SubscriptionWrapper<String>>> storeBuilder() {
        final Serde<SubscriptionWrapper<String>> subscriptionWrapperSerde = new SubscriptionWrapperSerde<>(
            PK_SERDE_TOPIC_SUPPLIER, Serdes.String());
        return Stores.timestampedKeyValueStoreBuilder(
            Stores.persistentTimestampedKeyValueStore(
                "Store"
            ),
            new Serdes.BytesSerde(),
            subscriptionWrapperSerde
        );
    }
}
