/*
 * 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.nifi.processors.elasticsearch;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
import java.io.InputStream;
import java.net.Proxy;
import java.net.URL;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import okhttp3.Authenticator;
import okhttp3.Credentials;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.RequestBody;
import okhttp3.Response;
import okhttp3.Route;
import org.apache.commons.text.StringEscapeUtils;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.ValidationContext;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.proxy.ProxyConfiguration;
import org.apache.nifi.proxy.ProxySpec;
import org.apache.nifi.ssl.SSLContextService;
import org.apache.nifi.util.StringUtils;

import javax.net.ssl.SSLContext;
import javax.net.ssl.X509TrustManager;

/**
 * A base class for Elasticsearch processors that use the HTTP API
 */
@Deprecated
public abstract class AbstractElasticsearchHttpProcessor extends AbstractElasticsearchProcessor {
    static final String SOURCE_QUERY_PARAM = "_source";
    static final String QUERY_QUERY_PARAM = "q";
    static final String SORT_QUERY_PARAM = "sort";
    static final String SIZE_QUERY_PARAM = "size";


    public static final PropertyDescriptor ES_URL = new PropertyDescriptor.Builder()
            .name("elasticsearch-http-url")
            .displayName("Elasticsearch URL")
            .description("Elasticsearch URL which will be connected to, including scheme (http, e.g.), host, and port. The default port for the REST API is 9200.")
            .required(true)
            .addValidator(StandardValidators.URL_VALIDATOR)
            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
            .build();

    public static final PropertyDescriptor PROXY_HOST = new PropertyDescriptor.Builder()
            .name("elasticsearch-http-proxy-host")
            .displayName("Proxy Host")
            .description("The fully qualified hostname or IP address of the proxy server")
            .required(false)
            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();

    public static final PropertyDescriptor PROXY_PORT = new PropertyDescriptor.Builder()
            .name("elasticsearch-http-proxy-port")
            .displayName("Proxy Port")
            .description("The port of the proxy server")
            .required(false)
            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
            .addValidator(StandardValidators.PORT_VALIDATOR)
            .build();

    public static final PropertyDescriptor PROXY_USERNAME = new PropertyDescriptor.Builder()
            .name("proxy-username")
            .displayName("Proxy Username")
            .description("Proxy Username")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .required(false)
            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
            .build();
    public static final PropertyDescriptor PROXY_PASSWORD = new PropertyDescriptor.Builder()
            .name("proxy-password")
            .displayName("Proxy Password")
            .description("Proxy Password")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .required(false)
            .sensitive(true)
            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
            .build();

    public static final PropertyDescriptor CONNECT_TIMEOUT = new PropertyDescriptor.Builder()
            .name("elasticsearch-http-connect-timeout")
            .displayName("Connection Timeout")
            .description("Max wait time for the connection to the Elasticsearch REST API.")
            .required(true)
            .defaultValue("5 secs")
            .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
            .build();

    public static final PropertyDescriptor RESPONSE_TIMEOUT = new PropertyDescriptor.Builder()
            .name("elasticsearch-http-response-timeout")
            .displayName("Response Timeout")
            .description("Max wait time for a response from the Elasticsearch REST API.")
            .required(true)
            .defaultValue("15 secs")
            .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
            .build();

    private final AtomicReference<OkHttpClient> okHttpClientAtomicReference = new AtomicReference<>();

    final ObjectMapper mapper = new ObjectMapper();

    @Override
    protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(String propertyDescriptorName) {
        return new PropertyDescriptor.Builder()
                .name(propertyDescriptorName)
                .required(false)
                .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
                .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
                .dynamic(true)
                .build();
    }

    private static final ProxySpec[] PROXY_SPECS = {ProxySpec.HTTP_AUTH, ProxySpec.SOCKS};
    public static final PropertyDescriptor PROXY_CONFIGURATION_SERVICE
            = ProxyConfiguration.createProxyConfigPropertyDescriptor(true, PROXY_SPECS);
    static final List<PropertyDescriptor> COMMON_PROPERTY_DESCRIPTORS;

    static {
        final List<PropertyDescriptor> properties = new ArrayList<>();
        properties.add(ES_URL);
        properties.add(PROP_SSL_CONTEXT_SERVICE);
        properties.add(CHARSET);
        properties.add(USERNAME);
        properties.add(PASSWORD);
        properties.add(CONNECT_TIMEOUT);
        properties.add(RESPONSE_TIMEOUT);
        properties.add(PROXY_CONFIGURATION_SERVICE);
        properties.add(PROXY_HOST);
        properties.add(PROXY_PORT);
        properties.add(PROXY_USERNAME);
        properties.add(PROXY_PASSWORD);

        COMMON_PROPERTY_DESCRIPTORS = Collections.unmodifiableList(properties);
    }

    @Override
    protected void createElasticsearchClient(ProcessContext context) throws ProcessException {
        okHttpClientAtomicReference.set(null);

        OkHttpClient.Builder okHttpClient = new OkHttpClient.Builder();

        // Add a proxy if set
        final ProxyConfiguration proxyConfig = ProxyConfiguration.getConfiguration(context, () -> {
            final String proxyHost = context.getProperty(PROXY_HOST).evaluateAttributeExpressions().getValue();
            final Integer proxyPort = context.getProperty(PROXY_PORT).evaluateAttributeExpressions().asInteger();
            if (proxyHost != null && proxyPort != null) {
                final ProxyConfiguration componentProxyConfig = new ProxyConfiguration();
                componentProxyConfig.setProxyType(Proxy.Type.HTTP);
                componentProxyConfig.setProxyServerHost(proxyHost);
                componentProxyConfig.setProxyServerPort(proxyPort);
                componentProxyConfig.setProxyUserName(context.getProperty(PROXY_USERNAME).evaluateAttributeExpressions().getValue());
                componentProxyConfig.setProxyUserPassword(context.getProperty(PROXY_PASSWORD).evaluateAttributeExpressions().getValue());
                return componentProxyConfig;
            }
            return ProxyConfiguration.DIRECT_CONFIGURATION;
        });

        if (!Proxy.Type.DIRECT.equals(proxyConfig.getProxyType())) {
            final Proxy proxy = proxyConfig.createProxy();
            okHttpClient.proxy(proxy);

            if (proxyConfig.hasCredential()) {
                okHttpClient.proxyAuthenticator(new Authenticator() {
                    @Override
                    public Request authenticate(Route route, Response response) throws IOException {
                        final String credential = Credentials.basic(proxyConfig.getProxyUserName(), proxyConfig.getProxyUserPassword());
                        return response.request().newBuilder()
                                .header("Proxy-Authorization", credential)
                                .build();
                    }
                });
            }
        }

        // Set timeouts
        okHttpClient.connectTimeout((context.getProperty(CONNECT_TIMEOUT).evaluateAttributeExpressions().asTimePeriod(TimeUnit.MILLISECONDS).intValue()), TimeUnit.MILLISECONDS);
        okHttpClient.readTimeout(context.getProperty(RESPONSE_TIMEOUT).evaluateAttributeExpressions().asTimePeriod(TimeUnit.MILLISECONDS).intValue(), TimeUnit.MILLISECONDS);

        // Apply the TLS configuration if present
        final SSLContextService sslService = context.getProperty(PROP_SSL_CONTEXT_SERVICE).asControllerService(SSLContextService.class);
        if (sslService != null) {
            final SSLContext sslContext = sslService.createContext();
            final X509TrustManager trustManager = sslService.createTrustManager();
            okHttpClient.sslSocketFactory(sslContext.getSocketFactory(), trustManager);
        }

        okHttpClientAtomicReference.set(okHttpClient.build());
    }

    @Override
    protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
        List<ValidationResult> results = new ArrayList<>(super.customValidate(validationContext));
        if (validationContext.getProperty(PROXY_HOST).isSet() != validationContext.getProperty(PROXY_PORT).isSet()) {
            results.add(new ValidationResult.Builder()
                    .valid(false)
                    .explanation("Proxy Host and Proxy Port must be both set or empty")
                    .subject("Proxy server configuration")
                    .build());
        }

        ProxyConfiguration.validateProxySpec(validationContext, results, PROXY_SPECS);

        return results;
    }

    protected OkHttpClient getClient() {
        return okHttpClientAtomicReference.get();
    }

    protected boolean isSuccess(int statusCode) {
        return statusCode / 100 == 2;
    }

    protected Response sendRequestToElasticsearch(OkHttpClient client, URL url, String username, String password, String verb, RequestBody body) throws IOException {

        final ComponentLog log = getLogger();
        Request.Builder requestBuilder = new Request.Builder()
                .url(url);
        if ("get".equalsIgnoreCase(verb)) {
            requestBuilder = requestBuilder.get();
        } else if ("put".equalsIgnoreCase(verb)) {
            requestBuilder = requestBuilder.put(body);
        } else if ("post".equalsIgnoreCase(verb)) {
            requestBuilder = requestBuilder.post(body);
        } else {
            throw new IllegalArgumentException("Elasticsearch REST API verb not supported by this processor: " + verb);
        }

        if (!StringUtils.isEmpty(username) && !StringUtils.isEmpty(password)) {
            String credential = Credentials.basic(username, password);
            requestBuilder = requestBuilder.header("Authorization", credential);
        }
        Request httpRequest = requestBuilder.build();
        log.debug("Sending Elasticsearch request to {}", new Object[]{url});

        Response responseHttp = client.newCall(httpRequest).execute();

        // store the status code and message
        int statusCode = responseHttp.code();

        if (statusCode == 0) {
            throw new IllegalStateException("Status code unknown, connection hasn't been attempted.");
        }

        log.debug("Received response from Elasticsearch with status code {}", new Object[]{statusCode});

        return responseHttp;
    }

    protected JsonNode parseJsonResponse(InputStream in) throws IOException {
        return mapper.readTree(in);
    }

    protected void buildBulkCommand(StringBuilder sb, String index, String docType, String indexOp, String id, String jsonString) {
        if (indexOp.equalsIgnoreCase("index") || indexOp.equalsIgnoreCase("create")) {
            sb.append("{\"");
            sb.append(indexOp.toLowerCase());
            sb.append("\": { \"_index\": \"");
            sb.append(StringEscapeUtils.escapeJson(index));
            sb.append("\"");
            if (StringUtils.isNotBlank(docType)) {
                sb.append(", \"_type\": \"");
                sb.append(StringEscapeUtils.escapeJson(docType));
                sb.append("\"");
            }
            if (StringUtils.isNotBlank(id)) {
                sb.append(", \"_id\": \"");
                sb.append(StringEscapeUtils.escapeJson(id));
                sb.append("\"");
            }
            sb.append("}}\n");
            sb.append(jsonString);
            sb.append("\n");
        } else if (indexOp.equalsIgnoreCase("upsert") || indexOp.equalsIgnoreCase("update")) {
            sb.append("{\"update\": { \"_index\": \"");
            sb.append(StringEscapeUtils.escapeJson(index));
            sb.append("\"");
            if (StringUtils.isNotBlank(docType)) {
                sb.append(", \"_type\": \"");
                sb.append(StringEscapeUtils.escapeJson(docType));
                sb.append("\"");
            }
            sb.append(", \"_id\": \"");
            sb.append(StringEscapeUtils.escapeJson(id));
            sb.append("\" } }\n");
            sb.append("{\"doc\": ");
            sb.append(jsonString);
            sb.append(", \"doc_as_upsert\": ");
            sb.append(indexOp.equalsIgnoreCase("upsert"));
            sb.append(" }\n");
        } else if (indexOp.equalsIgnoreCase("delete")) {
            sb.append("{\"delete\": { \"_index\": \"");
            sb.append(StringEscapeUtils.escapeJson(index));
            sb.append("\"");
            if (StringUtils.isNotBlank(docType)) {
                sb.append(", \"_type\": \"");
                sb.append(StringEscapeUtils.escapeJson(docType));
                sb.append("\"");
            }
            sb.append(", \"_id\": \"");
            sb.append(StringEscapeUtils.escapeJson(id));
            sb.append("\" } }\n");
        }
    }
}
