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

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalLong;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import javax.net.ssl.SSLContext;

import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.annotation.behavior.DynamicProperty;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnEnabled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.PropertyValue;
import org.apache.nifi.components.ValidationContext;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.components.Validator;
import org.apache.nifi.confluent.schemaregistry.client.AuthenticationType;
import org.apache.nifi.confluent.schemaregistry.client.CachingSchemaRegistryClient;
import org.apache.nifi.confluent.schemaregistry.client.RestSchemaRegistryClient;
import org.apache.nifi.confluent.schemaregistry.client.SchemaRegistryClient;
import org.apache.nifi.context.PropertyContext;
import org.apache.nifi.controller.AbstractControllerService;
import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.schema.access.SchemaField;
import org.apache.nifi.schema.access.SchemaNotFoundException;
import org.apache.nifi.schemaregistry.services.SchemaRegistry;
import org.apache.nifi.serialization.record.RecordSchema;
import org.apache.nifi.serialization.record.SchemaIdentifier;
import org.apache.nifi.ssl.SSLContextService;

@Tags({"schema", "registry", "confluent", "avro", "kafka"})
@CapabilityDescription("Provides a Schema Registry that interacts with the Confluent Schema Registry so that those Schemas that are stored in the Confluent Schema "
    + "Registry can be used in NiFi. The Confluent Schema Registry has a notion of a \"subject\" for schemas, which is their terminology for a schema name. When a Schema "
    + "is looked up by name by this registry, it will find a Schema in the Confluent Schema Registry with that subject.")
@DynamicProperty(name = "request.header.*", value = "String literal, may not be empty", description = "Properties that begin with 'request.header.' " +
        "are populated into a map and passed as http headers in REST requests to the Confluent Schema Registry")
public class ConfluentSchemaRegistry extends AbstractControllerService implements SchemaRegistry {

    private static final Set<SchemaField> schemaFields = EnumSet.of(SchemaField.SCHEMA_NAME, SchemaField.SCHEMA_TEXT,
        SchemaField.SCHEMA_TEXT_FORMAT, SchemaField.SCHEMA_IDENTIFIER, SchemaField.SCHEMA_VERSION);

    private static final String REQUEST_HEADER_PREFIX = "request.header.";


    static final PropertyDescriptor SCHEMA_REGISTRY_URLS = new PropertyDescriptor.Builder()
        .name("url")
        .displayName("Schema Registry URLs")
        .description("A comma-separated list of URLs of the Schema Registry to interact with")
        .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
        .defaultValue("http://localhost:8081")
        .required(true)
        .addValidator(new MultipleURLValidator())
        .build();

    static final PropertyDescriptor SSL_CONTEXT = new PropertyDescriptor.Builder()
        .name("ssl-context")
        .displayName("SSL Context Service")
        .description("Specifies the SSL Context Service to use for interacting with the Confluent Schema Registry")
        .identifiesControllerService(SSLContextService.class)
        .required(false)
        .build();

    static final PropertyDescriptor CACHE_SIZE = new PropertyDescriptor.Builder()
        .name("cache-size")
        .displayName("Cache Size")
        .description("Specifies how many Schemas should be cached from the Schema Registry")
        .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
        .defaultValue("1000")
        .required(true)
        .build();

    static final PropertyDescriptor CACHE_EXPIRATION = new PropertyDescriptor.Builder()
        .name("cache-expiration")
        .displayName("Cache Expiration")
        .description("Specifies how long a Schema that is cached should remain in the cache. Once this time period elapses, a "
            + "cached version of a schema will no longer be used, and the service will have to communicate with the "
            + "Schema Registry again in order to obtain the schema.")
        .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
        .defaultValue("1 hour")
        .required(true)
        .build();

    static final PropertyDescriptor TIMEOUT = new PropertyDescriptor.Builder()
        .name("timeout")
        .displayName("Communications Timeout")
        .description("Specifies how long to wait to receive data from the Schema Registry before considering the communications a failure")
        .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
        .expressionLanguageSupported(ExpressionLanguageScope.NONE)
        .defaultValue("30 secs")
        .required(true)
        .build();

    static final PropertyDescriptor AUTHENTICATION_TYPE = new PropertyDescriptor.Builder()
            .name("authentication-type")
            .displayName("Authentication Type")
            .description("HTTP Client Authentication Type for Confluent Schema Registry")
            .required(false)
            .allowableValues(AuthenticationType.values())
            .defaultValue(AuthenticationType.NONE.toString())
            .build();

    static final PropertyDescriptor USERNAME = new PropertyDescriptor.Builder()
            .name("username")
            .displayName("Username")
            .description("Username for authentication to Confluent Schema Registry")
            .addValidator(StandardValidators.createRegexMatchingValidator(Pattern.compile("^[\\x20-\\x39\\x3b-\\x7e\\x80-\\xff]+$")))
            .required(false)
            .dependsOn(AUTHENTICATION_TYPE, AuthenticationType.BASIC.toString())
            .build();

    static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder()
            .name("password")
            .displayName("Password")
            .description("Password for authentication to Confluent Schema Registry")
            .addValidator(StandardValidators.createRegexMatchingValidator(Pattern.compile("^[\\x20-\\x7e\\x80-\\xff]+$")))
            .required(false)
            .dependsOn(AUTHENTICATION_TYPE, AuthenticationType.BASIC.toString())
            .sensitive(true)
            .build();

    private volatile SchemaRegistryClient client;


    @Override
    protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        final List<PropertyDescriptor> properties = new ArrayList<>();
        properties.add(SCHEMA_REGISTRY_URLS);
        properties.add(SSL_CONTEXT);
        properties.add(TIMEOUT);
        properties.add(CACHE_SIZE);
        properties.add(CACHE_EXPIRATION);
        properties.add(AUTHENTICATION_TYPE);
        properties.add(USERNAME);
        properties.add(PASSWORD);
        return properties;
    }

    private static final Validator REQUEST_HEADER_VALIDATOR = new Validator() {
        @Override
        public ValidationResult validate(final String subject, final String value, final ValidationContext context) {
            return new ValidationResult.Builder()
                    .subject(subject)
                    .input(value)
                    .valid(subject.startsWith(REQUEST_HEADER_PREFIX)
                            && subject.length() > REQUEST_HEADER_PREFIX.length())
                    .explanation("Dynamic property names must be of format 'request.header.*'")
                    .build();
        }
    };

    @Override
    protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(String propertyDescriptionName) {
        return new PropertyDescriptor.Builder()
                .name(propertyDescriptionName)
                .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
                .addValidator(REQUEST_HEADER_VALIDATOR)
                .build();
    }

    @OnEnabled
    public void onEnabled(final ConfigurationContext context) {
        final List<String> baseUrls = getBaseURLs(context);
        final int timeoutMillis = context.getProperty(TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue();

        final SSLContext sslContext;
        final SSLContextService sslContextService = context.getProperty(SSL_CONTEXT).asControllerService(SSLContextService.class);
        if (sslContextService == null) {
            sslContext = null;
        } else {
            sslContext = sslContextService.createContext();
        }

        final String username = context.getProperty(USERNAME).getValue();
        final String password = context.getProperty(PASSWORD).getValue();

        // generate a map of http headers where the key is the remainder of the property name after
        // the request header prefix
        final Map<String, String> httpHeaders =
                context.getProperties().entrySet()
                        .stream()
                        .filter(e -> e.getKey().getName().startsWith(REQUEST_HEADER_PREFIX))
                        .collect(Collectors.toMap(
                                map -> map.getKey().getName().substring(REQUEST_HEADER_PREFIX.length()),
                                Map.Entry::getValue)
                        );

        final SchemaRegistryClient restClient = new RestSchemaRegistryClient(baseUrls, timeoutMillis,
                sslContext, username, password, getLogger(), httpHeaders);

        final int cacheSize = context.getProperty(CACHE_SIZE).asInteger();
        final long cacheExpiration = context.getProperty(CACHE_EXPIRATION).asTimePeriod(TimeUnit.NANOSECONDS).longValue();

        client = new CachingSchemaRegistryClient(restClient, cacheSize, cacheExpiration);
    }

    @Override
    protected Collection<ValidationResult> customValidate(final ValidationContext validationContext) {
        final List<ValidationResult> results = new ArrayList<>();

        final boolean sslContextSet = validationContext.getProperty(SSL_CONTEXT).isSet();
        if (sslContextSet) {
            final List<String> baseUrls = getBaseURLs(validationContext);
            final List<String> insecure = baseUrls.stream()
                .filter(url -> !url.startsWith("https"))
                .collect(Collectors.toList());

            if (!insecure.isEmpty()) {
                results.add(new ValidationResult.Builder()
                    .subject(SCHEMA_REGISTRY_URLS.getDisplayName())
                    .input(insecure.get(0))
                    .valid(false)
                    .explanation("When SSL Context is configured, all Schema Registry URL's must use HTTPS, not HTTP")
                    .build());
            }
        }

        final PropertyValue authenticationTypeProperty = validationContext.getProperty(AUTHENTICATION_TYPE);
        if (authenticationTypeProperty.isSet()) {
            final AuthenticationType authenticationType = AuthenticationType.valueOf(authenticationTypeProperty.getValue());
            if (AuthenticationType.BASIC.equals(authenticationType)) {
                final String username = validationContext.getProperty(USERNAME).getValue();
                if (StringUtils.isBlank(username)) {
                    results.add(new ValidationResult.Builder()
                            .subject(USERNAME.getDisplayName())
                            .valid(false)
                            .explanation("Username is required for Basic Authentication")
                            .build());
                }
                final String password = validationContext.getProperty(PASSWORD).getValue();
                if (StringUtils.isBlank(password)) {
                    results.add(new ValidationResult.Builder()
                            .subject(PASSWORD.getDisplayName())
                            .valid(false)
                            .explanation("Password is required for Basic Authentication")
                            .build());
                }
            }
        }

        return results;
    }

    private List<String> getBaseURLs(final PropertyContext context) {
        final String urls = context.getProperty(SCHEMA_REGISTRY_URLS).evaluateAttributeExpressions().getValue();
        final List<String> baseUrls = Stream.of(urls.split(","))
            .map(url -> url.trim())
            .collect(Collectors.toList());

        return baseUrls;
    }

    private RecordSchema retrieveSchemaByName(final SchemaIdentifier schemaIdentifier) throws IOException, SchemaNotFoundException {
        final Optional<String> schemaName = schemaIdentifier.getName();
        if (!schemaName.isPresent()) {
            throw new org.apache.nifi.schema.access.SchemaNotFoundException("Cannot retrieve schema because Schema Name is not present");
        }

        final RecordSchema schema;
        if (schemaIdentifier.getVersion().isPresent()) {
            schema = client.getSchema(schemaName.get(), schemaIdentifier.getVersion().getAsInt());
        } else {
            schema = client.getSchema(schemaName.get());
        }

        return schema;
    }

    private RecordSchema retrieveSchemaById(final SchemaIdentifier schemaIdentifier) throws IOException, SchemaNotFoundException {
        final OptionalLong schemaId = schemaIdentifier.getIdentifier();
        if (!schemaId.isPresent()) {
            throw new org.apache.nifi.schema.access.SchemaNotFoundException("Cannot retrieve schema because Schema Id is not present");
        }

        final RecordSchema schema = client.getSchema((int) schemaId.getAsLong());
        return schema;
    }

    @Override
    public RecordSchema retrieveSchema(final SchemaIdentifier schemaIdentifier) throws IOException, SchemaNotFoundException {
        if (schemaIdentifier.getName().isPresent()) {
            return retrieveSchemaByName(schemaIdentifier);
        } else {
            return retrieveSchemaById(schemaIdentifier);
        }
    }

    @Override
    public Set<SchemaField> getSuppliedSchemaFields() {
        return schemaFields;
    }
}