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

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

import io.debezium.config.Configuration;
import io.debezium.config.Field;
import io.debezium.embedded.async.AsyncEmbeddedEngine;
import io.debezium.engine.spi.OffsetCommitPolicy;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.component.debezium.DebeziumConstants;
import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriParams;
import org.apache.camel.spi.UriPath;
import org.apache.camel.util.ObjectHelper;
import org.apache.kafka.connect.json.JsonConverter;

@UriParams
public abstract class EmbeddedDebeziumConfiguration implements Cloneable {

    private static final String LABEL_NAME = "consumer";

    private Class<?> connectorClass;
    // name
    @UriPath(label = LABEL_NAME, description = "Unique name for the connector. "
                                               + "Attempting to register again with the same name will fail.")
    @Metadata(required = true)
    private String name;
    // offset.storage
    @UriParam(label = LABEL_NAME, defaultValue = "org.apache.kafka.connect.storage.FileOffsetBackingStore",
              description = "The name of the Java class that is responsible for persistence of connector offsets.")
    private String offsetStorage = DebeziumConstants.DEFAULT_OFFSET_STORAGE;
    // offset.storage.file.filename
    @UriParam(label = LABEL_NAME, description = "Path to file where offsets are to be stored. "
                                                + "Required when offset.storage is set to the FileOffsetBackingStore.")
    private String offsetStorageFileName;
    // offset.storage.topic
    @UriParam(label = LABEL_NAME, description = "The name of the Kafka topic where offsets are "
                                                + "to be stored. Required when offset.storage is set to the KafkaOffsetBackingStore.")
    private String offsetStorageTopic;
    // offset.storage.partitions
    @UriParam(label = LABEL_NAME, description = "The number of partitions used when creating the "
                                                + "offset storage topic. Required when offset.storage is set to the 'KafkaOffsetBackingStore'.")
    private int offsetStoragePartitions;
    // offset.storage.replication.factor
    @UriParam(label = LABEL_NAME, description = "Replication factor used when creating the offset "
                                                + "storage topic. Required when offset.storage is set to the KafkaOffsetBackingStore")
    private int offsetStorageReplicationFactor;
    // offset.commit.policy
    @UriParam(label = LABEL_NAME,
              description = "The name of the Java class of the commit policy. It defines when offsets "
                            + "commit has to be triggered based on the number of events processed and the "
                            + "time elapsed since the last commit. This class must implement the interface "
                            + "'OffsetCommitPolicy'. The default is a periodic commit policy based upon "
                            + "time intervals.")
    private String offsetCommitPolicy = OffsetCommitPolicy.PeriodicCommitOffsetPolicy.class.getName();
    // offset.flush.interval.ms
    @UriParam(label = LABEL_NAME, defaultValue = "60000", description = "Interval at which to try committing "
                                                                        + "offsets. The default is 1 minute.",
              javaType = "java.time.Duration")
    private long offsetFlushIntervalMs = 60000;
    // offset.commit.timeout.ms
    @UriParam(label = LABEL_NAME, defaultValue = "5000", description = "Maximum number of milliseconds "
                                                                       + "to wait for records to flush and partition offset data to be committed to offset storage "
                                                                       + "before cancelling the process and restoring the offset data to be committed in a future "
                                                                       + "attempt. The default is 5 seconds.",
              javaType = "java.time.Duration")
    private long offsetCommitTimeoutMs = 5000;
    // internal.key.converter
    @UriParam(label = LABEL_NAME, defaultValue = "org.apache.kafka.connect.json.JsonConverter",
              description = "The Converter class that should be used to serialize and deserialize key data "
                            + "for offsets. The default is JSON converter.")
    private String internalKeyConverter = JsonConverter.class.getName();
    // internal.value.converter
    @UriParam(label = LABEL_NAME, defaultValue = "org.apache.kafka.connect.json.JsonConverter",
              description = "The Converter class that should be used to serialize and deserialize value "
                            + "data for offsets. The default is JSON converter.")
    private String internalValueConverter = JsonConverter.class.getName();
    // Additional properties
    @UriParam(label = "common", prefix = "additionalProperties.", multiValue = true,
              description = "Additional properties for debezium components in case they can't be set directly "
                            + "on the camel configurations (e.g: setting Kafka Connect properties needed by Debezium engine, "
                            + "for example setting KafkaOffsetBackingStore), the properties have to be prefixed with "
                            + "`additionalProperties.`. E.g: `additionalProperties.transactional.id=12345&additionalProperties.schema.registry.url=http://localhost:8811/avro`")
    private Map<String, Object> additionalProperties = new HashMap<>();

    protected EmbeddedDebeziumConfiguration() {
        ObjectHelper.notNull(configureConnectorClass(), "connectorClass");
        this.connectorClass = configureConnectorClass();
    }

    public Object copy() {
        try {
            EmbeddedDebeziumConfiguration answer = (EmbeddedDebeziumConfiguration) clone();
            // make sure the map is copied in its own instance
            Map<String, Object> additionalPropertiesCopy = new HashMap<>(additionalProperties);
            answer.setAdditionalProperties(additionalPropertiesCopy);
            return answer;
        } catch (CloneNotSupportedException e) {
            throw new RuntimeCamelException(e);
        }
    }

    /**
     * Configure the Debezium connector class supported by Debezium
     *
     * @return {@link Class}
     */
    protected abstract Class<?> configureConnectorClass();

    /**
     * Create a specific {@link Configuration} for a concrete configuration
     *
     * @return {@link Configuration}
     */
    protected abstract Configuration createConnectorConfiguration();

    /**
     * Validate a concrete configuration
     *
     * @return {@link ConfigurationValidation}
     */
    protected abstract ConfigurationValidation validateConnectorConfiguration();

    /**
     * The Debezium connector type that is supported by Camel Debezium component.
     *
     * @return {@link String}
     */
    public abstract String getConnectorDatabaseType();

    /**
     * Creates a Debezium configuration of type {@link Configuration} in order to be used in the engine.
     *
     * @return {@link Configuration}
     */
    public Configuration createDebeziumConfiguration() {
        final Configuration connectorConfiguration = createConnectorConfiguration();

        ObjectHelper.notNull(connectorConfiguration, "createConnectorConfiguration");

        return Configuration.create().with(createDebeziumEmbeddedEngineConfiguration())
                .with(createConnectorConfiguration()).build();
    }

    private Configuration createDebeziumEmbeddedEngineConfiguration() {
        final Configuration.Builder configBuilder = Configuration.create();

        addPropertyIfNotNull(configBuilder, AsyncEmbeddedEngine.ENGINE_NAME, name);
        addPropertyIfNotNull(configBuilder, AsyncEmbeddedEngine.CONNECTOR_CLASS, connectorClass.getName());
        addPropertyIfNotNull(configBuilder, AsyncEmbeddedEngine.OFFSET_STORAGE, offsetStorage);
        addPropertyIfNotNull(configBuilder, AsyncEmbeddedEngine.OFFSET_STORAGE_FILE_FILENAME,
                offsetStorageFileName);
        addPropertyIfNotNull(configBuilder, AsyncEmbeddedEngine.OFFSET_STORAGE_KAFKA_TOPIC, offsetStorageTopic);
        addPropertyIfNotNull(configBuilder, AsyncEmbeddedEngine.OFFSET_STORAGE_KAFKA_PARTITIONS,
                offsetStoragePartitions);
        addPropertyIfNotNull(configBuilder, AsyncEmbeddedEngine.OFFSET_STORAGE_KAFKA_REPLICATION_FACTOR,
                offsetStorageReplicationFactor);
        addPropertyIfNotNull(configBuilder, AsyncEmbeddedEngine.OFFSET_COMMIT_POLICY, offsetCommitPolicy);
        addPropertyIfNotNull(configBuilder, AsyncEmbeddedEngine.OFFSET_FLUSH_INTERVAL_MS, offsetFlushIntervalMs);
        addPropertyIfNotNull(configBuilder, AsyncEmbeddedEngine.OFFSET_COMMIT_TIMEOUT_MS, offsetCommitTimeoutMs);

        if (internalKeyConverter != null && internalValueConverter != null) {
            configBuilder.with("internal.key.converter", internalKeyConverter);
            configBuilder.with("internal.value.converter", internalValueConverter);
        }

        // additional properties
        applyAdditionalProperties(configBuilder, getAdditionalProperties());

        return configBuilder.build();
    }

    protected static <T> void addPropertyIfNotNull(
            final Configuration.Builder configBuilder,
            final Field field, final T value) {
        if (value != null) {
            configBuilder.with(field, value);
        }
    }

    protected static <T> void addPropertyIfNotNull(
            final Configuration.Builder configBuilder,
            final String key, final T value) {
        if (value != null) {
            configBuilder.with(key, value);
        }
    }

    private void applyAdditionalProperties(
            final Configuration.Builder configBuilder, final Map<String, Object> additionalProperties) {
        if (!ObjectHelper.isEmpty(getAdditionalProperties())) {
            additionalProperties.forEach((property, value) -> addPropertyIfNotNull(configBuilder, property, value));
        }
    }

    /**
     * Validate all configurations defined and return {@link ConfigurationValidation} instance which contains the
     * validation results
     *
     * @return {@link ConfigurationValidation}
     */
    public ConfigurationValidation validateConfiguration() {
        final ConfigurationValidation embeddedEngineValidation = validateDebeziumEmbeddedEngineConfiguration();
        // only if embeddedEngineValidation is true, we check the connector validation
        if (embeddedEngineValidation.isValid()) {
            final ConfigurationValidation connectorValidation = validateConnectorConfiguration();

            ObjectHelper.notNull(connectorValidation, "validateConnectorConfiguration");

            return connectorValidation;
        }
        return embeddedEngineValidation;
    }

    private ConfigurationValidation validateDebeziumEmbeddedEngineConfiguration() {
        if (isFieldValueNotSet(name)) {
            return ConfigurationValidation.notValid("Required field 'name' must be set.");
        }
        // check for offsetStorageFileName
        if (offsetStorage.equals(DebeziumConstants.DEFAULT_OFFSET_STORAGE)
                && isFieldValueNotSet(offsetStorageFileName)) {
            return ConfigurationValidation.notValid(String
                    .format("Required field 'offsetStorageFileName' must be set since 'offsetStorage' is set to '%s'",
                            DebeziumConstants.DEFAULT_OFFSET_STORAGE));
        }
        return ConfigurationValidation.valid();
    }

    protected static boolean isFieldValueNotSet(final Object field) {
        return ObjectHelper.isEmpty(field);
    }

    /**
     * The name of the Java class for the connector
     */
    public Class<?> getConnectorClass() {
        return connectorClass;
    }

    public void setConnectorClass(Class<?> connectorClass) {
        this.connectorClass = connectorClass;
    }

    /**
     * Unique name for the connector. Attempting to register again with the same name will fail.
     */
    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    /**
     * The name of the Java class that is responsible for persistence of connector offsets.
     */
    public String getOffsetStorage() {
        return offsetStorage;
    }

    public void setOffsetStorage(String offsetStorage) {
        this.offsetStorage = offsetStorage;
    }

    /**
     * Path to file where offsets are to be stored. Required when offset.storage is set to the FileOffsetBackingStore
     */
    public String getOffsetStorageFileName() {
        return offsetStorageFileName;
    }

    public void setOffsetStorageFileName(String offsetStorageFileName) {
        this.offsetStorageFileName = offsetStorageFileName;
    }

    /**
     * The name of the Kafka topic where offsets are to be stored. Required when offset.storage is set to the
     * KafkaOffsetBackingStore.
     */
    public String getOffsetStorageTopic() {
        return offsetStorageTopic;
    }

    public void setOffsetStorageTopic(String offsetStorageTopic) {
        this.offsetStorageTopic = offsetStorageTopic;
    }

    /**
     * Replication factor used when creating the offset storage topic. Required when offset.storage is set to the
     * KafkaOffsetBackingStore
     */
    public int getOffsetStorageReplicationFactor() {
        return offsetStorageReplicationFactor;
    }

    public void setOffsetStorageReplicationFactor(int offsetStorageReplicationFactor) {
        this.offsetStorageReplicationFactor = offsetStorageReplicationFactor;
    }

    /**
     * The name of the Java class of the commit policy. It defines when offsets commit has to be triggered based on the
     * number of events processed and the time elapsed since the last commit. This class must implement the interface
     * 'OffsetCommitPolicy'. The default is a periodic commit policy based upon time intervals.
     */
    public String getOffsetCommitPolicy() {
        return offsetCommitPolicy;
    }

    public void setOffsetCommitPolicy(String offsetCommitPolicy) {
        this.offsetCommitPolicy = offsetCommitPolicy;
    }

    /**
     * Interval at which to try committing offsets. The default is 1 minute.
     */
    public long getOffsetFlushIntervalMs() {
        return offsetFlushIntervalMs;
    }

    public void setOffsetFlushIntervalMs(long offsetFlushIntervalMs) {
        this.offsetFlushIntervalMs = offsetFlushIntervalMs;
    }

    /**
     * Maximum number of milliseconds to wait for records to flush and partition offset data to be committed to offset
     * storage before cancelling the process and restoring the offset data to be committed in a future attempt. The
     * default is 5 seconds.
     */
    public long getOffsetCommitTimeoutMs() {
        return offsetCommitTimeoutMs;
    }

    public void setOffsetCommitTimeoutMs(long offsetCommitTimeoutMs) {
        this.offsetCommitTimeoutMs = offsetCommitTimeoutMs;
    }

    /**
     * The number of partitions used when creating the offset storage topic. Required when offset.storage is set to the
     * 'KafkaOffsetBackingStore'.
     */
    public int getOffsetStoragePartitions() {
        return offsetStoragePartitions;
    }

    public void setOffsetStoragePartitions(int offsetStoragePartitions) {
        this.offsetStoragePartitions = offsetStoragePartitions;
    }

    /**
     * The Converter class that should be used to serialize and deserialize key data for offsets. The default is JSON
     * converter.
     */
    public String getInternalKeyConverter() {
        return internalKeyConverter;
    }

    public void setInternalKeyConverter(String internalKeyConverter) {
        this.internalKeyConverter = internalKeyConverter;
    }

    /**
     * The Converter class that should be used to serialize and deserialize value data for offsets. The default is JSON
     * converter.
     */
    public String getInternalValueConverter() {
        return internalValueConverter;
    }

    public void setInternalValueConverter(String internalValueConverter) {
        this.internalValueConverter = internalValueConverter;
    }

    /**
     * Sets additional properties for debezium components in case they can't be set directly on the camel configurations
     * (e.g: setting Kafka Connect properties needed by Debezium engine, for example, setting KafkaOffsetBackingStore),
     * the properties have to be prefixed with `additionalProperties.`. E.g.:
     * `additionalProperties.transactional.id=12345&additionalProperties.schema.registry.url=http://localhost:8811/avro`
     */
    public void setAdditionalProperties(Map<String, Object> additionalProperties) {
        this.additionalProperties = additionalProperties;
    }

    public Map<String, Object> getAdditionalProperties() {
        return additionalProperties;
    }
}
