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

import java.net.URI;
import java.net.URISyntaxException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

import com.couchbase.client.java.Bucket;
import com.couchbase.client.java.Cluster;
import com.couchbase.client.java.ClusterOptions;
import com.couchbase.client.java.codec.DefaultJsonSerializer;
import com.couchbase.client.java.env.ClusterEnvironment;
import org.apache.camel.CamelException;
import org.apache.camel.Category;
import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.spi.EndpointServiceLocation;
import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.UriEndpoint;
import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriPath;
import org.apache.camel.support.ScheduledPollEndpoint;

import static org.apache.camel.component.couchbase.CouchbaseConstants.COUCHBASE_PUT;
import static org.apache.camel.component.couchbase.CouchbaseConstants.COUCHBASE_URI_ERROR;
import static org.apache.camel.component.couchbase.CouchbaseConstants.DEFAULT_CONNECT_TIMEOUT;
import static org.apache.camel.component.couchbase.CouchbaseConstants.DEFAULT_CONSUME_PROCESSED_STRATEGY;
import static org.apache.camel.component.couchbase.CouchbaseConstants.DEFAULT_COUCHBASE_PORT;
import static org.apache.camel.component.couchbase.CouchbaseConstants.DEFAULT_DESIGN_DOCUMENT_NAME;
import static org.apache.camel.component.couchbase.CouchbaseConstants.DEFAULT_PAUSE_BETWEEN_RETRIES;
import static org.apache.camel.component.couchbase.CouchbaseConstants.DEFAULT_PRODUCER_RETRIES;
import static org.apache.camel.component.couchbase.CouchbaseConstants.DEFAULT_QUERY_TIMEOUT;
import static org.apache.camel.component.couchbase.CouchbaseConstants.DEFAULT_VIEWNAME;

/**
 * Query Couchbase databases using SQL++ (N1QL) queries or MapReduce Views with a poll strategy and/or perform various
 * operations against Couchbase databases.
 */
@UriEndpoint(firstVersion = "2.19.0", scheme = "couchbase", title = "Couchbase", syntax = "couchbase:protocol://hostname:port",
             category = { Category.DATABASE }, headersClass = CouchbaseConstants.class)
public class CouchbaseEndpoint extends ScheduledPollEndpoint implements EndpointServiceLocation {

    @UriPath
    @Metadata(required = true)
    private String protocol;
    @UriPath
    @Metadata(required = true)
    private String hostname;
    @UriPath(defaultValue = "8091")
    private int port;

    @UriParam(endpointIdentity = true)
    @Metadata(required = true)
    private String bucket;

    @UriParam
    private String collection;

    @UriParam
    private String scope;

    // Couchbase key
    @UriParam
    private String key;

    // Authentication
    @UriParam(label = "security", security = "secret")
    private String username;
    @UriParam(label = "security", security = "secret")
    private String password;

    // Additional hosts
    @UriParam(label = "advanced")
    private String additionalHosts;

    // Connection string
    @UriParam(label = "advanced")
    private String connectionString;

    // Persistence and replication parameters
    @UriParam(label = "producer", defaultValue = "0")
    private int persistTo;

    @UriParam(label = "producer", defaultValue = "0")
    private int replicateTo;

    // Producer parameters
    @UriParam(label = "producer", defaultValue = COUCHBASE_PUT)
    private String operation = COUCHBASE_PUT;
    @UriParam(label = "producer", defaultValue = "false")
    private boolean autoStartIdForInserts;
    @UriParam(label = "producer", defaultValue = "2")
    private int producerRetryAttempts = DEFAULT_PRODUCER_RETRIES;
    @UriParam(label = "producer", defaultValue = "5000")
    private int producerRetryPause = DEFAULT_PAUSE_BETWEEN_RETRIES;

    @UriParam(label = "producer")
    private long startingIdForInsertsFrom;

    // SQL++ query
    @UriParam(label = "consumer", defaultValue = "false",
              description = "If true, use the deprecated MapReduce Views to query documents via designDocumentName and viewName."
                            + " If false (default), use SQL++ queries instead.")
    private boolean useView;
    @UriParam(label = "consumer")
    private String statement;

    // View control
    @UriParam(label = "consumer", defaultValue = DEFAULT_DESIGN_DOCUMENT_NAME,
              description = "The design document name to use. Deprecated: use the statement option with SQL++ queries instead.")
    @Deprecated
    private String designDocumentName = DEFAULT_DESIGN_DOCUMENT_NAME;
    @UriParam(label = "consumer", defaultValue = DEFAULT_VIEWNAME,
              description = "The view name to use. Deprecated: use the statement option with SQL++ queries instead.")
    @Deprecated
    private String viewName = DEFAULT_VIEWNAME;
    @UriParam(label = "consumer", defaultValue = "-1")
    private int limit = -1;
    @UriParam(label = "consumer", defaultValue = "false")
    private boolean descending;
    @UriParam(label = "consumer", defaultValue = "-1")
    private int skip = -1;
    @UriParam(label = "consumer")
    private String rangeStartKey;
    @UriParam(label = "consumer")
    private String rangeEndKey = "";
    @UriParam(label = "consumer", defaultValue = "false")
    private boolean fullDocument = true;
    @UriParam(label = "consumer", defaultValue = "5000")
    private int consumerRetryPause = DEFAULT_PAUSE_BETWEEN_RETRIES;

    // Consumer strategy
    @UriParam(label = "consumer", defaultValue = DEFAULT_CONSUME_PROCESSED_STRATEGY)
    private String consumerProcessedStrategy = DEFAULT_CONSUME_PROCESSED_STRATEGY;

    // Connection fine tuning parameters
    @UriParam(label = "advanced", defaultValue = "2500", javaType = "java.time.Duration")
    private long queryTimeout = DEFAULT_QUERY_TIMEOUT;

    // Connection fine tuning parameters
    @UriParam(label = "advanced", defaultValue = "30000", javaType = "java.time.Duration")
    private long connectTimeout = DEFAULT_CONNECT_TIMEOUT;

    public CouchbaseEndpoint() {
    }

    public CouchbaseEndpoint(String uri, String remaining, CouchbaseComponent component) throws URISyntaxException {
        super(uri, component);
        URI remainingUri = new URI(remaining);

        protocol = remainingUri.getScheme();
        if (protocol == null) {
            throw new IllegalArgumentException(COUCHBASE_URI_ERROR);
        }

        port = remainingUri.getPort() == -1 ? DEFAULT_COUCHBASE_PORT : remainingUri.getPort();

        hostname = remainingUri.getHost();
        if (hostname == null) {
            throw new IllegalArgumentException(COUCHBASE_URI_ERROR);
        }
    }

    public CouchbaseEndpoint(String endpointUri, CouchbaseComponent component) {
        super(endpointUri, component);
    }

    @Override
    public String getServiceUrl() {
        return protocol + ":" + hostname + ":" + port;
    }

    @Override
    public String getServiceProtocol() {
        return protocol;
    }

    @Override
    public Map<String, String> getServiceMetadata() {
        if (username != null) {
            return Map.of("username", username);
        }
        return null;
    }

    @Override
    public Producer createProducer() throws Exception {
        return new CouchbaseProducer(this, createClient(), persistTo, replicateTo);
    }

    @Override
    public Consumer createConsumer(Processor processor) throws Exception {
        CouchbaseConsumer consumer = new CouchbaseConsumer(this, createClient(), processor);
        setPollStrategy(consumer.getPollStrategy());
        configureConsumer(consumer);
        return consumer;
    }

    public String getProtocol() {
        return protocol;
    }

    /**
     * The protocol to use
     */
    public void setProtocol(String protocol) {
        this.protocol = protocol;
    }

    public String getBucket() {
        return bucket;
    }

    /**
     * The bucket to use
     */
    public void setBucket(String bucket) {
        this.bucket = bucket;
    }

    public String getHostname() {
        return hostname;
    }

    /**
     * The hostname to use
     */
    public void setHostname(String hostname) {
        this.hostname = hostname;
    }

    public int getPort() {
        return port;
    }

    /**
     * The port number to use
     */
    public void setPort(int port) {
        this.port = port;
    }

    /**
     * The collection to use
     */
    public String getCollection() {
        return this.collection;
    }

    public void setCollection(String collection) {
        this.collection = collection;
    }

    public String getScope() {
        return this.scope;
    }

    /**
     * The scope to use
     */
    public void setScope(String scope) {
        this.scope = scope;
    }

    public String getKey() {
        return key;
    }

    /**
     * The key to use
     */
    public void setKey(String key) {
        this.key = key;
    }

    public String getUsername() {
        return username;
    }

    /**
     * The username to use
     */
    public void setUsername(String username) {
        this.username = username;
    }

    public String getPassword() {
        return password;
    }

    /**
     * The password to use
     */
    public void setPassword(String password) {
        this.password = password;
    }

    public String getAdditionalHosts() {
        return additionalHosts;
    }

    /**
     * The additional hosts
     */
    public void setAdditionalHosts(String additionalHosts) {
        this.additionalHosts = additionalHosts;
    }

    public String getConnectionString() {
        return connectionString;
    }

    /**
     * The Couchbase SDK connection string to use (e.g., couchbase://hostname:11210). When set, this takes precedence
     * over the hostname and port options for the SDK connection. This is useful when the KV port is not the default
     * 11210, for example when connecting to a container with dynamic port mappings.
     */
    public void setConnectionString(String connectionString) {
        this.connectionString = connectionString;
    }

    public int getPersistTo() {
        return persistTo;
    }

    /**
     * Where to persist the data
     */
    public void setPersistTo(int persistTo) {
        this.persistTo = persistTo;
    }

    public int getReplicateTo() {
        return replicateTo;
    }

    /**
     * Where to replicate the data
     */
    public void setReplicateTo(int replicateTo) {
        this.replicateTo = replicateTo;
    }

    public String getOperation() {
        return operation;
    }

    /**
     * The operation to do
     */
    public void setOperation(String operation) {
        this.operation = operation;
    }

    public boolean isAutoStartIdForInserts() {
        return autoStartIdForInserts;
    }

    /**
     * Define if we want an autostart Id when we are doing an insert operation
     */
    public void setAutoStartIdForInserts(boolean autoStartIdForInserts) {
        this.autoStartIdForInserts = autoStartIdForInserts;
    }

    public long getStartingIdForInsertsFrom() {
        return startingIdForInsertsFrom;
    }

    /**
     * Define the starting Id where we are doing an insert operation
     */
    public void setStartingIdForInsertsFrom(long startingIdForInsertsFrom) {
        this.startingIdForInsertsFrom = startingIdForInsertsFrom;
    }

    public int getProducerRetryAttempts() {
        return producerRetryAttempts;
    }

    /**
     * Define the number of retry attempts
     */
    public void setProducerRetryAttempts(int producerRetryAttempts) {
        this.producerRetryAttempts = producerRetryAttempts;
    }

    public int getProducerRetryPause() {
        return producerRetryPause;
    }

    /**
     * Define the producer retry pause between different attempts
     */
    public void setProducerRetryPause(int producerRetryPause) {
        this.producerRetryPause = producerRetryPause;
    }

    public int getConsumerRetryPause() {
        return consumerRetryPause;
    }

    /**
     * Define the consumer retry pause between different attempts
     */
    public void setConsumerRetryPause(int consumerRetryPause) {
        this.consumerRetryPause = consumerRetryPause;
    }

    public String getStatement() {
        return statement;
    }

    /**
     * A SQL++ (N1QL) query statement for consuming documents. When set, the consumer uses SQL++ queries instead of
     * MapReduce views. The query should select META().id AS __id to identify documents. Example: SELECT META().id AS
     * __id, * FROM `myCollection` WHERE type = 'order' LIMIT 100. If not set and useView is false, a SQL++ query is
     * auto-generated from the bucket, scope, collection, limit, skip, descending, rangeStartKey, and rangeEndKey
     * options.
     */
    public void setStatement(String statement) {
        this.statement = statement;
    }

    public boolean isUseView() {
        return useView;
    }

    /**
     * If true, use the deprecated MapReduce Views to query documents via designDocumentName and viewName. If false
     * (default), use SQL++ queries instead. When both useView is false and statement is not set, a SQL++ query is
     * auto-generated from the endpoint options (limit, skip, descending, etc.).
     */
    public void setUseView(boolean useView) {
        this.useView = useView;
    }

    /**
     * Builds a SQL++ query from the endpoint options.
     */
    String buildSqlQuery() {
        StringBuilder sb = new StringBuilder("SELECT META().id AS __id, * FROM `");
        // Use the collection if set, otherwise use the bucket name (default collection)
        if (collection != null && !collection.isEmpty()) {
            sb.append(collection);
        } else {
            sb.append(bucket);
        }
        sb.append('`');

        // WHERE clause from range keys
        String startKey = rangeStartKey;
        String endKey = rangeEndKey;
        boolean hasStart = startKey != null && !startKey.isEmpty();
        boolean hasEnd = endKey != null && !endKey.isEmpty();
        if (hasStart || hasEnd) {
            sb.append(" WHERE ");
            if (hasStart && hasEnd) {
                sb.append("META().id >= '").append(startKey).append("' AND META().id <= '").append(endKey).append('\'');
            } else if (hasStart) {
                sb.append("META().id >= '").append(startKey).append('\'');
            } else {
                sb.append("META().id <= '").append(endKey).append('\'');
            }
        }

        // ORDER BY
        if (descending) {
            sb.append(" ORDER BY META().id DESC");
        }

        // LIMIT
        if (limit > 0) {
            sb.append(" LIMIT ").append(limit);
        }

        // OFFSET
        if (skip > 0) {
            sb.append(" OFFSET ").append(skip);
        }

        return sb.toString();
    }

    /**
     * @deprecated Use {@link #setStatement(String)} with SQL++ queries instead.
     */
    @Deprecated
    public String getDesignDocumentName() {
        return designDocumentName;
    }

    /**
     * @deprecated Use {@link #setStatement(String)} with SQL++ queries instead.
     */
    @Deprecated
    public void setDesignDocumentName(String designDocumentName) {
        this.designDocumentName = designDocumentName;
    }

    /**
     * @deprecated Use {@link #setStatement(String)} with SQL++ queries instead.
     */
    @Deprecated
    public String getViewName() {
        return viewName;
    }

    /**
     * @deprecated Use {@link #setStatement(String)} with SQL++ queries instead.
     */
    @Deprecated
    public void setViewName(String viewName) {
        this.viewName = viewName;
    }

    public int getLimit() {
        return limit;
    }

    /**
     * The output limit to use
     */
    public void setLimit(int limit) {
        this.limit = limit;
    }

    public boolean isDescending() {
        return descending;
    }

    /**
     * Define if this operation is descending or not
     */
    public void setDescending(boolean descending) {
        this.descending = descending;
    }

    public int getSkip() {
        return skip;
    }

    /**
     * Define the skip to use
     */
    public void setSkip(int skip) {
        this.skip = skip;
    }

    public String getRangeStartKey() {
        return rangeStartKey;
    }

    /**
     * Define a range for the start key
     */
    public void setRangeStartKey(String rangeStartKey) {
        this.rangeStartKey = rangeStartKey;
    }

    public String getRangeEndKey() {
        return rangeEndKey;
    }

    /**
     * Define a range for the end key
     */
    public void setRangeEndKey(String rangeEndKey) {
        this.rangeEndKey = rangeEndKey;
    }

    public boolean isFullDocument() {
        return fullDocument;
    }

    /**
     * If true consumer will return complete document instead data defined in view
     */
    public void setFullDocument(boolean fullDocument) {
        this.fullDocument = fullDocument;
    }

    public String getConsumerProcessedStrategy() {
        return consumerProcessedStrategy;
    }

    /**
     * Define the consumer Processed strategy to use
     */
    public void setConsumerProcessedStrategy(String consumerProcessedStrategy) {
        this.consumerProcessedStrategy = consumerProcessedStrategy;
    }

    public long getQueryTimeout() {
        return queryTimeout;
    }

    /**
     * Define the operation timeout in milliseconds
     */
    public void setQueryTimeout(long queryTimeout) {
        this.queryTimeout = queryTimeout;
    }

    public long getConnectTimeout() {
        return connectTimeout;
    }

    /**
     * Define the timeoutconnect in milliseconds
     */
    public void setConnectTimeout(long connectTimeout) {
        this.connectTimeout = connectTimeout;
    }

    public URI[] makeBootstrapURI() throws URISyntaxException {

        if (additionalHosts == null || additionalHosts.isEmpty()) {
            return new URI[] { new URI(protocol + "://" + hostname + ":" + port + "/pools") };
        }
        return getAllUris();

    }

    private URI[] getAllUris() throws URISyntaxException {

        String[] hosts = additionalHosts.split(",");

        for (int i = 0; i < hosts.length; i++) {
            hosts[i] = hosts[i].trim();
        }

        List<String> hostList = new ArrayList<>();
        hostList.add(hostname);
        hostList.addAll(Arrays.asList(hosts));
        Set<String> hostSet = new LinkedHashSet<>(hostList);
        hosts = hostSet.toArray(new String[0]);

        URI[] uriArray = new URI[hosts.length];

        for (int i = 0; i < hosts.length; i++) {
            uriArray[i] = new URI(protocol + "://" + hosts[i] + ":" + port + "/pools");
        }

        return uriArray;
    }

    //create from couchbase-client
    private Bucket createClient() throws Exception {
        if (bucket == null || bucket.isEmpty()) {
            throw new CamelException(COUCHBASE_URI_ERROR);
        }

        ClusterEnvironment env = createClusterEnvironment();

        String connStr;
        if (connectionString != null && !connectionString.isEmpty()) {
            connStr = connectionString;
        } else {
            List<URI> hosts = Arrays.asList(makeBootstrapURI());
            String addHosts = hosts.stream()
                    .map(URI::getHost)
                    .collect(Collectors.joining(","));
            connStr = !addHosts.isEmpty() ? addHosts : hostname;
        }

        Cluster cluster = Cluster.connect(connStr, ClusterOptions
                .clusterOptions(username, password)
                .environment(env));

        return cluster.bucket(bucket);
    }

    /**
     * Creates the {@link ClusterEnvironment} for connecting to Couchbase.
     * <p>
     * Explicitly configures the {@link DefaultJsonSerializer} to prevent the Couchbase SDK from auto-detecting Jackson
     * on the classpath (CAMEL-22090). Without this, when a non-shaded Jackson library is present (e.g., via
     * camel-jackson, Spring Boot, or Quarkus), the SDK would use {@code JacksonJsonSerializer} backed by the non-shaded
     * Jackson {@code ObjectMapper}. This causes deserialization failures because the SDK's internal types (such as view
     * row keys) rely on the shaded Jackson classes bundled within the Couchbase SDK.
     * <p>
     * The {@link DefaultJsonSerializer} uses the shaded Jackson {@code ObjectMapper} bundled inside the Couchbase SDK,
     * ensuring consistent serialization/deserialization regardless of what other Jackson versions are on the classpath.
     */
    ClusterEnvironment createClusterEnvironment() {
        ClusterEnvironment.Builder cfb = ClusterEnvironment.builder();
        cfb.jsonSerializer(DefaultJsonSerializer.create());
        if (queryTimeout != DEFAULT_QUERY_TIMEOUT) {
            cfb.timeoutConfig()
                    .connectTimeout(Duration.ofMillis(connectTimeout))
                    .queryTimeout(Duration.ofMillis(queryTimeout));
        }
        return cfb.build();
    }

    /**
     * Compares retry strategy with query timeout and gets the higher value : for write operations with retry
     *
     * @return
     */
    public long getWriteQueryTimeout() {
        long retryTimeout = producerRetryAttempts * (long) producerRetryPause;
        return retryTimeout > queryTimeout ? retryTimeout : queryTimeout;
    }
}
