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

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.stream.Collectors;

import org.apache.camel.CamelContext;
import org.apache.camel.ConsumerTemplate;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.DefaultErrorHandlerBuilder;
import org.apache.camel.builder.NoErrorHandlerBuilder;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.kafkaconnector.CamelConnectorConfig;
import org.apache.camel.main.SimpleMain;
import org.apache.camel.model.ProcessorDefinition;
import org.apache.camel.processor.idempotent.kafka.KafkaIdempotentRepository;
import org.apache.camel.spi.IdempotentRepository;
import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.SensitiveUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class CamelKafkaConnectMain extends SimpleMain {
    public static final String KAMELET_MARSHAL_TEMPLATE_PARAMETERS_PREFIX = "camel.kamelet.ckcMarshal.";
    public static final String KAMELET_UNMARSHAL_TEMPLATE_PARAMETERS_PREFIX = "camel.kamelet.ckcUnMarshal.";
    public static final String KAMELET_AGGREGATORL_TEMPLATE_PARAMETERS_PREFIX = "camel.kamelet.ckcAggregator.";
    public static final String KAMELET_IDEMPOTENT_TEMPLATE_PARAMETERS_PREFIX = "camel.kamelet.ckcIdempotent.";
    public static final String KAMELET_REMOVEHEADER_TEMPLATE_PARAMETERS_PREFIX = "camel.kamelet.ckcRemoveHeader.";

    private static final Logger LOG = LoggerFactory.getLogger(CamelKafkaConnectMain.class);

    protected volatile ConsumerTemplate consumerTemplate;
    protected volatile ProducerTemplate producerTemplate;

    public CamelKafkaConnectMain(CamelContext context) {
        super(context);
    }

    @Override
    protected void doStop() throws Exception {
        ServiceHelper.stopService(consumerTemplate);
        consumerTemplate = null;

        ServiceHelper.stopService(producerTemplate);
        producerTemplate = null;

        super.doStop();
    }

    public ProducerTemplate getProducerTemplate() {
        if (this.producerTemplate == null) {
            synchronized (this) {
                if (this.producerTemplate == null) {
                    this.producerTemplate = getCamelContext().createProducerTemplate();
                }
            }
        }

        return this.producerTemplate;
    }

    public ConsumerTemplate getConsumerTemplate() {
        if (this.consumerTemplate == null) {
            synchronized (this) {
                if (this.consumerTemplate == null) {
                    this.consumerTemplate = getCamelContext().createConsumerTemplate();
                }
            }
        }

        return this.consumerTemplate;
    }

    public static Builder builder(String from, String to) {
        return new Builder(from, to);
    }

    public static final class Builder {
        private final String from;
        private final String to;
        private Map<String, String> props;
        private String marshallDataFormat;
        private String unmarshallDataFormat;
        private int aggregationSize;
        private long aggregationTimeout;
        private String errorHandler;
        private int maxRedeliveries;
        private long redeliveryDelay;
        private boolean idempotencyEnabled;
        private String expressionType;
        private String expressionHeader;
        private int memoryDimension;
        private String idempotentRepositoryType;
        private String idempotentRepositoryTopicName;
        private String idempotentRepositoryKafkaServers;
        private int idempotentRepositoryKafkaMaxCacheSize;
        private int idempotentRepositoryKafkaPollDuration;
        private String headersExcludePattern;

        public Builder(String from, String to) {
            this.from = from;
            this.to = to;
        }

        public Builder withProperties(Map<String, String> props) {
            this.props = new HashMap<>(props);
            return this;
        }

        public Builder withMarshallDataFormat(String dataformatId) {
            this.marshallDataFormat = dataformatId;
            return this;
        }

        public Builder withUnmarshallDataFormat(String dataformatId) {
            this.unmarshallDataFormat = dataformatId;
            return this;
        }

        public Builder withAggregationSize(int aggregationSize) {
            this.aggregationSize = aggregationSize;
            return this;
        }

        public Builder withAggregationTimeout(long aggregationTimeout) {
            this.aggregationTimeout = aggregationTimeout;
            return this;
        }

        public Builder withErrorHandler(String errorHandler) {
            this.errorHandler = errorHandler;
            return this;
        }

        public Builder withMaxRedeliveries(int maxRedeliveries) {
            this.maxRedeliveries = maxRedeliveries;
            return this;
        }

        public Builder withRedeliveryDelay(long redeliveryDelay) {
            this.redeliveryDelay = redeliveryDelay;
            return this;
        }

        public Builder withIdempotencyEnabled(boolean idempotencyEnabled) {
            this.idempotencyEnabled = idempotencyEnabled;
            return this;
        }

        public Builder withExpressionType(String expressionType) {
            this.expressionType = expressionType;
            return this;
        }

        public Builder withExpressionHeader(String expressionHeader) {
            this.expressionHeader = expressionHeader;
            return this;
        }

        public Builder withMemoryDimension(int memoryDimension) {
            this.memoryDimension = memoryDimension;
            return this;
        }

        public Builder withIdempotentRepositoryType(String idempotentRepositoryType) {
            this.idempotentRepositoryType = idempotentRepositoryType;
            return this;
        }

        public Builder withIdempotentRepositoryTopicName(String idempotentRepositoryTopicName) {
            this.idempotentRepositoryTopicName = idempotentRepositoryTopicName;
            return this;
        }

        public Builder withIdempotentRepositoryKafkaServers(String idempotentRepositoryKafkaServers) {
            this.idempotentRepositoryKafkaServers = idempotentRepositoryKafkaServers;
            return this;
        }

        public Builder withIdempotentRepositoryKafkaMaxCacheSize(int idempotentRepositoryKafkaMaxCacheSize) {
            this.idempotentRepositoryKafkaMaxCacheSize = idempotentRepositoryKafkaMaxCacheSize;
            return this;
        }

        public Builder withIdempotentRepositoryKafkaPollDuration(int idempotentRepositoryKafkaPollDuration) {
            this.idempotentRepositoryKafkaPollDuration = idempotentRepositoryKafkaPollDuration;
            return this;
        }

        public Builder withHeadersExcludePattern(String headersExcludePattern) {
            this.headersExcludePattern = headersExcludePattern;
            return this;
        }

        private String filterSensitive(Map.Entry<Object, Object> entry) {

            if (SensitiveUtils.containsSensitive((String) entry.getKey())) {
                return entry.getKey() + "=xxxxxxx";
            }
            return entry.getKey() + "=" + entry.getValue();
        }

        public CamelKafkaConnectMain build(CamelContext camelContext) {
            CamelKafkaConnectMain camelMain = new CamelKafkaConnectMain(camelContext);
            camelMain.configure().setAutoConfigurationLogSummary(false);
            //TODO: make it configurable
            camelMain.configure().setDumpRoutes(Boolean.TRUE.toString());

            Properties camelProperties = new Properties();
            camelProperties.putAll(props);

            //error handler
            camelMain.getCamelContext().getRegistry().bind("ckcErrorHandler", new DefaultErrorHandlerBuilder());
            if (errorHandler != null) {
                switch (errorHandler) {
                    case "no":
                        camelMain.getCamelContext().getRegistry().bind("ckcErrorHandler", new NoErrorHandlerBuilder());
                        break;
                    case "default":
                        camelMain.getCamelContext().getRegistry().bind("ckcErrorHandler", new DefaultErrorHandlerBuilder().maximumRedeliveries(maxRedeliveries).redeliveryDelay(redeliveryDelay));
                        break;
                    default:
                        break;
                }
            }

            //dataformats
            if (!ObjectHelper.isEmpty(marshallDataFormat)) {
                camelProperties.put(KAMELET_MARSHAL_TEMPLATE_PARAMETERS_PREFIX + "marshal", marshallDataFormat);
            }
            if (!ObjectHelper.isEmpty(unmarshallDataFormat)) {
                camelProperties.put(KAMELET_UNMARSHAL_TEMPLATE_PARAMETERS_PREFIX + "unmarshal", unmarshallDataFormat);
            }

            //aggregator
            if (!ObjectHelper.isEmpty(aggregationSize)) {
                camelProperties.put(KAMELET_AGGREGATORL_TEMPLATE_PARAMETERS_PREFIX + "aggregationSize", String.valueOf(aggregationSize));
            }
            if (!ObjectHelper.isEmpty(aggregationTimeout)) {
                camelProperties.put(KAMELET_AGGREGATORL_TEMPLATE_PARAMETERS_PREFIX + "aggregationTimeout", String.valueOf(aggregationTimeout));
            }

            //idempotency
            if (idempotencyEnabled) {
                switch (expressionType) {
                    case "body":
                        camelProperties.put(KAMELET_IDEMPOTENT_TEMPLATE_PARAMETERS_PREFIX + "idempotentExpression", "${body}");
                        break;
                    case "header":
                        camelProperties.put(KAMELET_IDEMPOTENT_TEMPLATE_PARAMETERS_PREFIX + "idempotentExpression", "${headers." + expressionHeader + "}");
                        break;
                    default:
                        break;
                }
                // Instantiating the idempotent Repository here and inject it in registry to be referenced
                IdempotentRepository idempotentRepo = null;
                switch (idempotentRepositoryType) {
                    case "memory":
                        idempotentRepo = MemoryIdempotentRepository.memoryIdempotentRepository(memoryDimension);
                        break;
                    case "kafka":
                        idempotentRepo = new KafkaIdempotentRepository(idempotentRepositoryTopicName, idempotentRepositoryKafkaServers, idempotentRepositoryKafkaMaxCacheSize, idempotentRepositoryKafkaPollDuration);
                        break;
                    default:
                        break;
                }
                camelMain.getCamelContext().getRegistry().bind("ckcIdempotentRepository", idempotentRepo);
            }

            //remove headers
            if (!ObjectHelper.isEmpty(headersExcludePattern)) {
                camelProperties.put(KAMELET_REMOVEHEADER_TEMPLATE_PARAMETERS_PREFIX + "headersExcludePattern", headersExcludePattern);
            }

            // log filtered properties and set initial camel properties
            List<String> filteredProps = camelProperties.entrySet().stream().map(this::filterSensitive).collect(Collectors.toList());
            LOG.info("Setting initial properties in Camel context: [{}]", filteredProps);
            camelMain.setInitialProperties(camelProperties);

            camelMain.configure().addRoutesBuilder(new RouteBuilder() {
                public void configure() {

                    //create marshal template
                    routeTemplate("ckcMarshal")
                            .templateParameter("marshal", "dummyDataformat")
                            .from("kamelet:source")
                            .marshal("{{marshal}}")
                            .to("kamelet:sink");

                    //create unmarshal template
                    routeTemplate("ckcUnMarshal")
                            .templateParameter("unmarshal", "dummyDataformat")
                            .from("kamelet:source")
                            .marshal("{{unmarshal}}")
                            .to("kamelet:sink");

                    //create aggregator template
                    routeTemplate("ckcAggregator")
                            //TODO: change CamelConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_NAME to ckcAggregationStrategy?
                            .templateParameter("aggregationStrategy", CamelConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_NAME)
                            .templateParameter("aggregationSize", "1")
                            .templateParameter("aggregationTimeout", String.valueOf(Long.MAX_VALUE))
                            .from("kamelet:source")
                            .aggregate(constant(true))
                                .aggregationStrategy("{{aggregationStrategy}}")
                                .completionSize("{{aggregationSize}}")
                                .completionTimeout("{{aggregationTimeout}}")
                                .to("kamelet:sink")
                            .end();

                    //create idempotent template
                    routeTemplate("ckcIdempotent")
                            .templateParameter("idempotentExpression", "dummyExpression")
                            .templateParameter("idempotentRepository", "ckcIdempotentRepository")
                            .from("kamelet:source")
                            .idempotentConsumer(simple("{{idempotentExpression}}")).idempotentRepository("{{idempotentRepository}}")
                            .to("kamelet:sink");

                    //create removeHeader template
                    routeTemplate("ckcRemoveHeader")
                            .templateParameter("headersExcludePattern", "(?!)")
                            .from("kamelet:source")
                            .removeHeaders("{{headersExcludePattern}}")
                            .to("kamelet:sink");

                    //creating source template
                    routeTemplate("ckcSource")
                            .templateParameter("fromUrl")
                            .templateParameter("errorHandler", "ckcErrorHandler")
                            .from("{{fromUrl}}")
                            .errorHandler("ckcErrorHandler")
                            .to("kamelet:sink");

                    //creating sink template
                    routeTemplate("ckcSink")
                            .templateParameter("toUrl")
                            .templateParameter("errorHandler", "ckcErrorHandler")
                            .from("kamelet:source")
                            .errorHandler("ckcErrorHandler")
                            .to("{{toUrl}}");

                    //creating the actual route
                    ProcessorDefinition<?> rd = from(from);
                    if (!ObjectHelper.isEmpty(marshallDataFormat)) {
                        rd = rd.kamelet("ckcMarshal");
                    }
                    if (!ObjectHelper.isEmpty(unmarshallDataFormat)) {
                        rd = rd.kamelet("ckcUnMarshal");
                    }
                    if (getContext().getRegistry().lookupByName("aggregate") != null) {
                        rd = rd.kamelet("ckcAggregator");
                    }
                    if (idempotencyEnabled) {
                        rd = rd.kamelet("ckcIdempotent");
                    }
                    rd = rd.kamelet("ckcRemoveHeader");
                    rd.toD(to);
                }
            });

            return camelMain;
        }
    }
}
