/*
 * 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.net.URISyntaxException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

import org.apache.camel.CamelContext;
import org.apache.camel.ExtendedCamelContext;
import org.apache.camel.LoggingLevel;
import org.apache.camel.spi.EndpointUriFactory;
import org.apache.kafka.connect.connector.ConnectRecord;
import org.slf4j.Logger;

public final class TaskHelper {

    private TaskHelper() {
    }

    /**
     * Try to build a url of a Camel {@link org.apache.camel.Endpoint}.
     *
     * @param camelContext the {@link CamelContext} used to retrieve an instance of a EndpointUriFactory for the given component schema.
     * @param props properties used to build the url in the form of a key -> value {@link Map}.
     * @param componentSchema the schema name of the Camel {@link org.apache.camel.Component} used to build the Camel {@link org.apache.camel.Endpoint} url.
     * @param endpointPropertiesPrefix prefix of all the Camel {@link org.apache.camel.Endpoint} properties.
     * @param pathPropertiesPrefix prefix of all the properties used in the Camel {@link org.apache.camel.Endpoint} path.
     *
     * @return A String representing the built url.
     * @throws {@link URISyntaxException} in case of uri build failure.
     */
    public static String buildUrl(CamelContext camelContext, Map<String, String> props, String componentSchema, String endpointPropertiesPrefix, String pathPropertiesPrefix) throws URISyntaxException {
        Map<String, Object> filteredProps = props.entrySet().stream()
            .filter(e -> e.getKey().startsWith(endpointPropertiesPrefix) || e.getKey().startsWith(pathPropertiesPrefix))
            .collect(Collectors.toMap(
                e -> e.getKey().replace(endpointPropertiesPrefix, "").replace(pathPropertiesPrefix, ""),
                Map.Entry::getValue
            ));

        EndpointUriFactory factory = camelContext.adapt(ExtendedCamelContext.class).getEndpointUriFactory(componentSchema);
        if (factory == null) {
            throw new IllegalStateException("Unable to compute endpoint uri. Reason: uri factory for schema `" + componentSchema + "` not found");
        }
        if (!factory.isEnabled(componentSchema)) {
            throw new IllegalStateException("Unable to compute endpoint uri. Reason: uri factory for schema `" + componentSchema + "` not enabled");
        }



        return factory.buildUri(componentSchema, filteredProps, false);
    }

    /**
     * Combines default properties with loaded properties with the following logic:
     * 1) loaded properties overwrite default properties.
     * 2) default properties that starts with a loaded property are removed.
     *
     * Example:
     * default properties:
     *   defaultProperty=defaultValue
     *   overwrittenDefaultProperty=defaultValue
     *   camel.component.x.objectProperty=#class:my.package.MyClass
     *   camel.component.x.objectProperty.removedDefaultField=defaultValue
     *   camel.component.x.defaultProperty=defaultValue
     *
     * loaded properties:
     *   overwrittenDefaultProperty=loadedValue
     *   camel.component.x.objectProperty=#class:my.package.MyOtherClass
     *   camel.component.x.objectProperty.loadedField=loadedValue
     *   camel.component.x.loadedProperty=loadedValue
     *
     * will result in:
     *   defaultProperty=defaultValue
     *   overwrittenDefaultProperty=loadedValue
     *   camel.component.x.objectProperty=#class:my.package.MyOtherClass
     *   camel.component.x.objectProperty.anotherField=loadedValue
     *   camel.component.x.loadedProperty=loadedValue
     *   camel.component.x.defaultProperty=defaultValue
     *
     * See org.apache.camel.kafkaconnector.utils.TaskHelperTests for some examples.
     *
     * @param defaultProps RuntimeCamelCatalog used to build the url.
     * @param loadedProps properties used to build the url in the form of a key -> value {@link Map}.
     */
    public static Map<String, String> combineDefaultAndLoadedProperties(Map<String, String> defaultProps, Map<String, String> loadedProps) {
        if (loadedProps == null && defaultProps == null) {
            return Collections.emptyMap();
        } else if (loadedProps == null) {
            return new HashMap<>(defaultProps);
        } else if (defaultProps == null) {
            return new HashMap<>(loadedProps);
        } else {
            Map<String, String> result = new HashMap<>(loadedProps);
            defaultProps.keySet().stream()
                    .filter(k -> !stringStartWithOneOfPrefixes(k, loadedProps.keySet()))
                    .filter(k -> !result.containsKey(k))
                    .forEach(k -> result.put(k, defaultProps.get(k)));
            return result;
        }
    }

    private static Boolean stringStartWithOneOfPrefixes(String s, Set<String> prefixes) {
        if (s == null || prefixes == null) {
            return false;
        }
        for (String p : prefixes) {
            if (s.startsWith(p)) {
                return true;
            }
        }
        return false;
    }

    public static void logRecordContent(Logger logger, LoggingLevel level, ConnectRecord<?> record) {
        if (level == LoggingLevel.OFF) {
            return;
        }

        switch (level) {
            case TRACE:
                logger.trace(record.toString());
                break;
            case DEBUG:
                logger.debug(record.toString());
                break;
            case INFO:
                logger.info(record.toString());
                break;
            case WARN:
                logger.warn(record.toString());
                break;
            case ERROR:
                logger.error(record.toString());
                break;
            default:
                break;
        }
    }
}
