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

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.apache.commons.io.DirectoryWalker;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public final class PluginPathHelper {
    private static final Logger LOG = LoggerFactory.getLogger(PluginPathHelper.class);

    private static PluginPathHelper instance;
    private volatile List<String> connectors;

    private static class PluginWalker extends DirectoryWalker<String> {

        public PluginWalker() {
            super(null, 2);
        }

        @Override
        protected void handleFile(File file, int depth, Collection<String> results) throws IOException {
            String fileName = file.getName();

            if (fileName.endsWith(".jar")) {
                if (fileName.contains("kafka-connector") && fileName.contains("camel")) {
                    String parentDir = file.getParentFile().getCanonicalPath();
                    if (parentDir.endsWith("target")) {
                        LOG.debug("Adding directory (1): {}: {}", depth, parentDir);

                        results.add(parentDir);
                    }
                }
            }
        }


        @Override
        protected boolean handleDirectory(File directory, int depth, Collection<String> results) throws IOException {
            String directoryName = directory.getName();

            if (directoryName.equals("target")) {
                String pluginDir = directory.getCanonicalPath();
                LOG.debug("Adding directory (2): {}: {}", depth, pluginDir);

                results.add(pluginDir);
            }

            return true;
        }

        public List<String> findPlugins(File startDir) {
            List<String> results = new ArrayList<>();

            try {
                walk(startDir, results);
            } catch (IOException e) {
                LOG.error("I/O error while traversing candidate plugin dirs: {}", e.getMessage(), e);
            }
            return results;
        }
    }

    private PluginPathHelper() {

    }

    private List<String> findPluginPaths() {
        List<String> pluginPaths = new ArrayList<>();

        for (String module : connectors) {
            String path = System.getProperty("project.basedir") + File.separator + module;
            File pathFile = new File(path);
            try {
                LOG.debug("Base dir used for search: {}, with path: {}, coming from components: {}, {}, {}", pathFile.getCanonicalPath(), path, System.getProperty("project.basedir"), File.separator, module);
            } catch (IOException e) {
                LOG.error("I/O exception: {}", e.getMessage(), e);
            }
            PluginWalker pluginWalker = new PluginWalker();

            pluginPaths.addAll(pluginWalker.findPlugins(pathFile));
        }

        return pluginPaths;
    }

    /*
     * We need to construct a list of directories containing *only* the connector classes (ie.: those that
     * specialize Kafka's Connector abstract class.
     *
     * Our build generates jar files with for every connector and puts that in the target directory which the
     * build dir (ie.: ${project.root}/core/target/camel-kafka-connector-0.0.1-SNAPSHOT.jar,
     * ${project.root}/connectors/camel-sjms2-kafka-connector/target/camel-sjms2-kafka-connector-0.0.1-SNAPSHOT.jar,
     * etc).
     *
     * The code within the pluginPaths traverses the directories for the core and connectors module, filtering any
     * file that matches all of the following conditions:
     * 1) ends with jar
     * 2) is located in the target directory
     * 3) contains the strings 'camel' and 'kafka-connector' as part of their name.
     *
     * Then for every connector jar file that it finds, it configures the embedded runtime to includes the parent dir
     * into the configuration.
     *
     * Why it does this?
     *
     * 1) Because having the connector classes in the classpath could cause library conflicts causing the connectors
     * and the Kafka connect runtime to fail.
     * 2) Having the connectors on the classpath causes the following error to appear multiple times in the logs:
     * 'Plugin class loader for connector: [name] was not found'
     *
     * ref: https://docs.confluent.io/current/connect/userguide.html
     */
    public String pluginPaths() {
        String ret = String.join(",", findPluginPaths());
        LOG.info("Returning the following directories for the plugin path: {}", ret);

        return ret;
    }

    public void registerConnector(String[] connectorNames) {
        // prepend "connectors/" to all connectors name
        connectors = Stream.of(connectorNames)
                .map(connectorName -> "connectors" + File.separator + connectorName)
                .collect(Collectors.toList());
    }

    public static PluginPathHelper getInstance() {
        if (instance == null) {
            instance = new PluginPathHelper();
        }

        return instance;
    }
}
