/*
 * 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.test;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.api.MockProcessorContext;
import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.SessionBytesStoreSupplier;
import org.apache.kafka.streams.state.SessionStore;
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.streams.state.Stores;
import org.apache.kafka.streams.state.WindowBytesStoreSupplier;
import org.apache.kafka.streams.state.WindowStore;
import org.apache.kafka.test.TestUtils;

import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;

import java.io.File;
import java.io.IOException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.stream.Stream;

import static java.util.Arrays.asList;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
import static org.apache.kafka.common.utils.Utils.mkProperties;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

public class MockProcessorContextStateStoreTest {

    public static Stream<Arguments> parameters() {
        final List<Boolean> booleans = asList(true, false);

        final List<Arguments> values = new ArrayList<>();

        for (final Boolean timestamped : booleans) {
            for (final Boolean caching : booleans) {
                for (final Boolean logging : booleans) {
                    final List<KeyValueBytesStoreSupplier> keyValueBytesStoreSuppliers = asList(
                        Stores.inMemoryKeyValueStore("kv" + timestamped + caching + logging),
                        Stores.persistentKeyValueStore("kv" + timestamped + caching + logging),
                        Stores.persistentTimestampedKeyValueStore("kv" + timestamped + caching + logging)
                    );
                    for (final KeyValueBytesStoreSupplier supplier : keyValueBytesStoreSuppliers) {
                        final StoreBuilder<? extends KeyValueStore<String, ?>> builder;
                        if (timestamped) {
                            builder = Stores.timestampedKeyValueStoreBuilder(supplier, Serdes.String(), Serdes.Long());
                        } else {
                            builder = Stores.keyValueStoreBuilder(supplier, Serdes.String(), Serdes.Long());
                        }
                        if (caching) {
                            builder.withCachingEnabled();
                        } else {
                            builder.withCachingDisabled();
                        }
                        if (logging) {
                            builder.withLoggingEnabled(Collections.emptyMap());
                        } else {
                            builder.withLoggingDisabled();
                        }

                        values.add(Arguments.of(builder, timestamped, caching, logging));
                    }
                }
            }
        }

        for (final Boolean timestamped : booleans) {
            for (final Boolean caching : booleans) {
                for (final Boolean logging : booleans) {
                    final List<WindowBytesStoreSupplier> windowBytesStoreSuppliers = asList(
                        Stores.inMemoryWindowStore("w" + timestamped + caching + logging, Duration.ofSeconds(1), Duration.ofSeconds(1), false),
                        Stores.persistentWindowStore("w" + timestamped + caching + logging, Duration.ofSeconds(1), Duration.ofSeconds(1), false),
                        Stores.persistentTimestampedWindowStore("w" + timestamped + caching + logging, Duration.ofSeconds(1), Duration.ofSeconds(1), false)
                    );

                    for (final WindowBytesStoreSupplier supplier : windowBytesStoreSuppliers) {
                        final StoreBuilder<? extends WindowStore<String, ?>> builder;
                        if (timestamped) {
                            builder = Stores.timestampedWindowStoreBuilder(supplier, Serdes.String(), Serdes.Long());
                        } else {
                            builder = Stores.windowStoreBuilder(supplier, Serdes.String(), Serdes.Long());
                        }
                        if (caching) {
                            builder.withCachingEnabled();
                        } else {
                            builder.withCachingDisabled();
                        }
                        if (logging) {
                            builder.withLoggingEnabled(Collections.emptyMap());
                        } else {
                            builder.withLoggingDisabled();
                        }

                        values.add(Arguments.of(builder, timestamped, caching, logging));
                    }
                }
            }
        }

        for (final Boolean caching : booleans) {
            for (final Boolean logging : booleans) {
                final List<SessionBytesStoreSupplier> sessionBytesStoreSuppliers = asList(
                    Stores.inMemorySessionStore("s" + caching + logging, Duration.ofSeconds(1)),
                    Stores.persistentSessionStore("s" + caching + logging, Duration.ofSeconds(1))
                );

                for (final SessionBytesStoreSupplier supplier : sessionBytesStoreSuppliers) {
                    final StoreBuilder<? extends SessionStore<String, ?>> builder =
                        Stores.sessionStoreBuilder(supplier, Serdes.String(), Serdes.Long());
                    if (caching) {
                        builder.withCachingEnabled();
                    } else {
                        builder.withCachingDisabled();
                    }
                    if (logging) {
                        builder.withLoggingEnabled(Collections.emptyMap());
                    } else {
                        builder.withLoggingDisabled();
                    }

                    values.add(Arguments.of(builder, false, caching, logging));
                }
            }
        }

        return values.stream();
    }

    @ParameterizedTest
    @MethodSource(value = "parameters")
    public void shouldEitherInitOrThrow(final StoreBuilder<StateStore> builder,
                                        final boolean timestamped,
                                        final boolean caching,
                                        final boolean logging) {
        final File stateDir = TestUtils.tempDirectory();
        try {
            final MockProcessorContext<Void, Void> context = new MockProcessorContext<>(
                mkProperties(mkMap(
                    mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, ""),
                    mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "mock-localhost:9092")
                )),
                new TaskId(0, 0),
                stateDir
            );
            final StateStore store = builder.build();
            if (caching || logging) {
                assertThrows(
                    IllegalArgumentException.class,
                    () -> store.init(context.getStateStoreContext(), store)
                );
            } else {
                final InternalProcessorContext<?, ?> internalProcessorContext = mock(InternalProcessorContext.class);
                when(internalProcessorContext.taskId()).thenReturn(context.taskId());
                when(internalProcessorContext.stateDir()).thenReturn(stateDir);
                when(internalProcessorContext.metrics()).thenReturn((StreamsMetricsImpl) context.metrics());
                when(internalProcessorContext.appConfigs()).thenReturn(context.appConfigs());
                store.init(internalProcessorContext, store);
                store.close();
            }
        } finally {
            try {
                Utils.delete(stateDir);
            } catch (final IOException e) {
                // Failed to clean up the state dir. The JVM hooks will try again later.
            }
        }
    }
}
