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

import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.streams.state.DslStoreSuppliers;
import org.apache.kafka.streams.state.WindowBytesStoreSupplier;

import java.util.HashMap;
import java.util.Map;

/**
 * Class used to configure the name of the join processor, the repartition topic name,
 * state stores or state store names in  Stream-Stream join.
 * @param <K>   the key type
 * @param <V1>  this value type
 * @param <V2>  other value type
 */
public class StreamJoined<K, V1, V2> implements NamedOperation<StreamJoined<K, V1, V2>> {

    protected final Serde<K> keySerde;
    protected final Serde<V1> valueSerde;
    protected final Serde<V2> otherValueSerde;
    protected final WindowBytesStoreSupplier thisStoreSupplier;
    protected final WindowBytesStoreSupplier otherStoreSupplier;
    protected final String name;
    protected final String storeName;
    protected final boolean loggingEnabled;
    protected final Map<String, String> topicConfig;

    // not final because it is potentially overridden by TopologyConfig
    protected DslStoreSuppliers dslStoreSuppliers;

    protected StreamJoined(final StreamJoined<K, V1, V2> streamJoined) {
        this(streamJoined.keySerde,
            streamJoined.valueSerde,
            streamJoined.otherValueSerde,
            streamJoined.dslStoreSuppliers,
            streamJoined.thisStoreSupplier,
            streamJoined.otherStoreSupplier,
            streamJoined.name,
            streamJoined.storeName,
            streamJoined.loggingEnabled,
            streamJoined.topicConfig);
    }

    private StreamJoined(final Serde<K> keySerde,
                         final Serde<V1> valueSerde,
                         final Serde<V2> otherValueSerde,
                         final DslStoreSuppliers dslStoreSuppliers,
                         final WindowBytesStoreSupplier thisStoreSupplier,
                         final WindowBytesStoreSupplier otherStoreSupplier,
                         final String name,
                         final String storeName,
                         final boolean loggingEnabled,
                         final Map<String, String> topicConfig) {
        this.keySerde = keySerde;
        this.valueSerde = valueSerde;
        this.otherValueSerde = otherValueSerde;
        this.dslStoreSuppliers = dslStoreSuppliers;
        this.thisStoreSupplier = thisStoreSupplier;
        this.otherStoreSupplier = otherStoreSupplier;
        this.name = name;
        this.storeName = storeName;
        this.loggingEnabled = loggingEnabled;
        this.topicConfig = topicConfig;
    }

    /**
     * Creates a StreamJoined instance with the provided store suppliers. The store suppliers must implement
     * the {@link WindowBytesStoreSupplier} interface.  The store suppliers must provide unique names or a
     * {@link org.apache.kafka.streams.errors.StreamsException} is thrown.
     *
     * @param storeSupplier       this store supplier
     * @param otherStoreSupplier  other store supplier
     * @param <K>                 the key type
     * @param <V1>                this value type
     * @param <V2>                other value type
     * @return                    {@link StreamJoined} instance
     */
    public static <K, V1, V2> StreamJoined<K, V1, V2> with(final WindowBytesStoreSupplier storeSupplier,
                                                           final WindowBytesStoreSupplier otherStoreSupplier) {
        return new StreamJoined<>(
            null,
            null,
            null,
            null,
            storeSupplier,
            otherStoreSupplier,
            null,
            null,
            true,
            new HashMap<>()
        );
    }

    /**
     * Creates a StreamJoined instance with the given {@link DslStoreSuppliers}. The store plugin
     * will be used to get all the state stores in this operation that do not otherwise have an
     * explicitly configured {@link org.apache.kafka.streams.state.DslStoreSuppliers}.
     *
     * @param storeSuppliers  the store plugin that will be used for state stores
     * @param <K>             the key type
     * @param <V1>            this value type
     * @param <V2>            other value type
     * @return                {@link StreamJoined} instance
     */
    public static <K, V1, V2> StreamJoined<K, V1, V2> with(final DslStoreSuppliers storeSuppliers) {
        return new StreamJoined<>(
                null,
                null,
                null,
                storeSuppliers,
                null,
                null,
                null,
                null,
                true,
                new HashMap<>()
        );
    }

    /**
     * Creates a {@link StreamJoined} instance using the provided name for the state stores and hence the changelog
     * topics for the join stores.  The name for the stores will be ${applicationId}-&lt;storeName&gt;-this-join and ${applicationId}-&lt;storeName&gt;-other-join
     * or ${applicationId}-&lt;storeName&gt;-outer-this-join and ${applicationId}-&lt;storeName&gt;-outer-other-join depending if the join is an inner-join
     * or an outer join. The changelog topics will have the -changelog suffix.  The user should note that even though the join stores will have a
     * specified name, the stores will remain unavailable for querying.
     *
     * Please note that if you are using {@link StreamJoined} to replace deprecated {@link KStream#join} functions with
     * {@link Joined} parameters in order to set the name for the join processors, you would need to create the {@link StreamJoined}
     * object first and then call {@link StreamJoined#withName}
     *
     * @param storeName  The name to use for the store
     * @param <K>        The key type
     * @param <V1>       This value type
     * @param <V2>       Other value type
     * @return            {@link StreamJoined} instance
     */
    public static <K, V1, V2> StreamJoined<K, V1, V2> as(final String storeName) {
        return new StreamJoined<>(
            null,
            null,
            null,
            null,
            null,
            null,
            null,
            storeName,
            true,
            new HashMap<>()
        );
    }


    /**
     * Creates a {@link StreamJoined} instance with the provided serdes to configure the stores
     * for the join.
     * @param keySerde          The key serde
     * @param valueSerde        This value serde
     * @param otherValueSerde   Other value serde
     * @param <K>               The key type
     * @param <V1>              This value type
     * @param <V2>              Other value type
     * @return                  {@link StreamJoined} instance
     */
    public static <K, V1, V2> StreamJoined<K, V1, V2> with(final Serde<K> keySerde,
                                                           final Serde<V1> valueSerde,
                                                           final Serde<V2> otherValueSerde
    ) {
        return new StreamJoined<>(
            keySerde,
            valueSerde,
            otherValueSerde,
            null,
            null,
            null,
            null,
            null,
            true,
            new HashMap<>()
        );
    }

    /**
     * Set the name to use for the join processor and the repartition topic(s) if required.
     * @param name  the name to use
     * @return      a new {@link StreamJoined} instance
     */
    @Override
    public StreamJoined<K, V1, V2> withName(final String name) {
        return new StreamJoined<>(
            keySerde,
            valueSerde,
            otherValueSerde,
            dslStoreSuppliers,
            thisStoreSupplier,
            otherStoreSupplier,
            name,
            storeName,
            loggingEnabled,
            topicConfig
        );
    }

    /**
     * Sets the base store name to use for both sides of the join. The name for the state stores and hence the changelog
     * topics for the join stores.  The name for the stores will be ${applicationId}-&lt;storeName&gt;-this-join and ${applicationId}-&lt;storeName&gt;-other-join
     * or ${applicationId}-&lt;storeName&gt;-outer-this-join and ${applicationId}-&lt;storeName&gt;-outer-other-join depending if the join is an inner-join
     * or an outer join. The changelog topics will have the -changelog suffix.  The user should note that even though the join stores will have a
     * specified name, the stores will remain unavailable for querying.
     *
     * @param storeName the storeName to use
     * @return a new {@link StreamJoined} instance
     */
    public StreamJoined<K, V1, V2> withStoreName(final String storeName) {
        return new StreamJoined<>(
            keySerde,
            valueSerde,
            otherValueSerde,
            dslStoreSuppliers,
            thisStoreSupplier,
            otherStoreSupplier,
            name,
            storeName,
            loggingEnabled,
            topicConfig
        );
    }

    /**
     * Configure with the provided {@link Serde Serde<K>} for the key
     * @param keySerde  the serde to use for the key
     * @return          a new {@link StreamJoined} configured with the keySerde
     */
    public StreamJoined<K, V1, V2> withKeySerde(final Serde<K> keySerde) {
        return new StreamJoined<>(
            keySerde,
            valueSerde,
            otherValueSerde,
            dslStoreSuppliers,
            thisStoreSupplier,
            otherStoreSupplier,
            name,
            storeName,
            loggingEnabled,
            topicConfig
        );
    }

    /**
     * Configure with the provided {@link Serde Serde<V1>} for this value
     * @param valueSerde  the serde to use for this value (calling or left side of the join)
     * @return            a new {@link StreamJoined} configured with the valueSerde
     */
    public StreamJoined<K, V1, V2> withValueSerde(final Serde<V1> valueSerde) {
        return new StreamJoined<>(
            keySerde,
            valueSerde,
            otherValueSerde,
            dslStoreSuppliers,
            thisStoreSupplier,
            otherStoreSupplier,
            name,
            storeName,
            loggingEnabled,
            topicConfig
        );
    }

    /**
     * Configure with the provided {@link Serde Serde<V2>} for the other value
     * @param otherValueSerde  the serde to use for the other value (other or right side of the join)
     * @return                 a new {@link StreamJoined} configured with the otherValueSerde
     */
    public StreamJoined<K, V1, V2> withOtherValueSerde(final Serde<V2> otherValueSerde) {
        return new StreamJoined<>(
            keySerde,
            valueSerde,
            otherValueSerde,
            dslStoreSuppliers,
            thisStoreSupplier,
            otherStoreSupplier,
            name,
            storeName,
            loggingEnabled,
            topicConfig
        );
    }

    /**
     * Configure with the provided {@link DslStoreSuppliers} for store suppliers that are not provided.
     *
     * @param dslStoreSuppliers the default store suppliers to use for this StreamJoined
     * @return                  a new {@link StreamJoined} configured with dslStoreSuppliers
     */
    public StreamJoined<K, V1, V2> withDslStoreSuppliers(final DslStoreSuppliers dslStoreSuppliers) {
        return new StreamJoined<>(
                keySerde,
                valueSerde,
                otherValueSerde,
                dslStoreSuppliers,
                thisStoreSupplier,
                otherStoreSupplier,
                name,
                storeName,
                loggingEnabled,
                topicConfig
        );
    }

    /**
     * Configure with the provided {@link WindowBytesStoreSupplier} for this store supplier.  Please note
     * this method only provides the store supplier for the left side of the join.  If you wish to also provide a
     * store supplier for the right (i.e., other) side you must use the {@link StreamJoined#withOtherStoreSupplier(WindowBytesStoreSupplier)}
     * method
     * @param thisStoreSupplier  the store supplier to use for this store supplier (calling or left side of the join)
     * @return            a new {@link StreamJoined} configured with thisStoreSupplier
     */
    public StreamJoined<K, V1, V2> withThisStoreSupplier(final WindowBytesStoreSupplier thisStoreSupplier) {
        return new StreamJoined<>(
            keySerde,
            valueSerde,
            otherValueSerde,
            dslStoreSuppliers,
            thisStoreSupplier,
            otherStoreSupplier,
            name,
            storeName,
            loggingEnabled,
            topicConfig
        );
    }

    /**
     * Configure with the provided {@link WindowBytesStoreSupplier} for the other store supplier.  Please note
     * this method only provides the store supplier for the right side of the join.  If you wish to also provide a
     * store supplier for the left side you must use the {@link StreamJoined#withThisStoreSupplier(WindowBytesStoreSupplier)}
     * method
     * @param otherStoreSupplier  the store supplier to use for the other store supplier (other or right side of the join)
     * @return            a new {@link StreamJoined} configured with otherStoreSupplier
     */
    public StreamJoined<K, V1, V2> withOtherStoreSupplier(final WindowBytesStoreSupplier otherStoreSupplier) {
        return new StreamJoined<>(
            keySerde,
            valueSerde,
            otherValueSerde,
            dslStoreSuppliers,
            thisStoreSupplier,
            otherStoreSupplier,
            name,
            storeName,
            loggingEnabled,
            topicConfig
        );
    }

    /**
     * Configures logging for both state stores. The changelog will be created with the provided configs.
     * <p>
     * Note: Any unrecognized configs will be ignored
     * @param config  configs applied to the changelog topic
     * @return            a new {@link StreamJoined} configured with logging enabled
     */
    public StreamJoined<K, V1, V2> withLoggingEnabled(final Map<String, String> config) {

        return new StreamJoined<>(
            keySerde,
            valueSerde,
            otherValueSerde,
            dslStoreSuppliers,
            thisStoreSupplier,
            otherStoreSupplier,
            name,
            storeName,
            true,
            config
        );
    }

    /**
     * Disable change logging for both state stores.
     * @return            a new {@link StreamJoined} configured with logging disabled
     */
    public StreamJoined<K, V1, V2> withLoggingDisabled() {
        return new StreamJoined<>(
            keySerde,
            valueSerde,
            otherValueSerde,
            dslStoreSuppliers,
            thisStoreSupplier,
            otherStoreSupplier,
            name,
            storeName,
            false,
            new HashMap<>()
        );
    }

    @Override
    public String toString() {
        return "StreamJoin{" +
            "keySerde=" + keySerde +
            ", valueSerde=" + valueSerde +
            ", otherValueSerde=" + otherValueSerde +
            ", thisStoreSupplier=" + thisStoreSupplier +
            ", otherStoreSupplier=" + otherStoreSupplier +
            ", name='" + name + '\'' +
            ", storeName='" + storeName + '\'' +
            ", loggingEnabled=" + loggingEnabled +
            ", topicConfig=" + topicConfig +
            '}';
    }
}
