content: /*
   * 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.maven;
  
  import java.io.BufferedReader;
  import java.io.FileInputStream;
  import java.io.IOException;
  import java.io.InputStream;
  import java.io.InputStreamReader;
  import java.nio.charset.StandardCharsets;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Properties;
  import java.util.Set;
  import java.util.stream.Collectors;
  import java.util.stream.Stream;
  
  import io.github.classgraph.ClassGraph;
  import io.github.classgraph.ScanResult;
  import org.apache.camel.catalog.CamelCatalog;
  import org.apache.camel.catalog.DefaultCamelCatalog;
  import org.apache.camel.kafkaconnector.maven.model.KameletModel;
  import org.apache.camel.kafkaconnector.maven.utils.MavenUtils;
  import org.apache.camel.kafkaconnector.maven.utils.YamlKameletMapper;
  import org.apache.camel.kamelets.catalog.KameletsCatalog;
  import org.apache.maven.ProjectDependenciesResolver;
  import org.apache.maven.execution.MavenSession;
  import org.apache.maven.plugin.BuildPluginManager;
  import org.apache.maven.plugin.MojoExecutionException;
  import org.apache.maven.plugins.annotations.Component;
  import org.apache.maven.plugins.annotations.LifecyclePhase;
  import org.apache.maven.plugins.annotations.Mojo;
  import org.apache.maven.plugins.annotations.Parameter;
  import org.apache.maven.plugins.annotations.ResolutionScope;
  import org.apache.maven.project.ProjectBuilder;
  import org.codehaus.plexus.resource.loader.FileResourceCreationException;
  import org.codehaus.plexus.resource.loader.ResourceNotFoundException;
  
  import static org.twdata.maven.mojoexecutor.MojoExecutor.artifactId;
  import static org.twdata.maven.mojoexecutor.MojoExecutor.configuration;
  import static org.twdata.maven.mojoexecutor.MojoExecutor.element;
  import static org.twdata.maven.mojoexecutor.MojoExecutor.executeMojo;
  import static org.twdata.maven.mojoexecutor.MojoExecutor.executionEnvironment;
  import static org.twdata.maven.mojoexecutor.MojoExecutor.goal;
  import static org.twdata.maven.mojoexecutor.MojoExecutor.groupId;
  import static org.twdata.maven.mojoexecutor.MojoExecutor.name;
  import static org.twdata.maven.mojoexecutor.MojoExecutor.plugin;
  import static org.twdata.maven.mojoexecutor.MojoExecutor.version;
  
  /**
   * Generate Camel Kafka Connector for the component
   */
  @Mojo(name = "generate-camel-kafka-connectors", threadSafe = true,
          requiresDependencyCollection = ResolutionScope.COMPILE_PLUS_RUNTIME,
          requiresDependencyResolution = ResolutionScope.COMPILE_PLUS_RUNTIME,
          defaultPhase = LifecyclePhase.GENERATE_RESOURCES)
  public class GenerateCamelKafkaConnectorsMojo extends AbstractCamelKafkaConnectorMojo {
  
      private static final String GENERATED_SECTION_START = "START OF GENERATED CODE";
      private static final String GENERATED_SECTION_START_COMMENT = "<!--" + GENERATED_SECTION_START + "-->";
      private static final String GENERATED_SECTION_END = "END OF GENERATED CODE";
      private static final String GENERATED_SECTION_END_COMMENT = "<!--" + GENERATED_SECTION_END + "-->";
  
      private static final String KAMELETS_DIR = "kamelets";
      private static final String KAMELETS_FILE_SUFFIX = ".kamelet.yaml";
  
      @Parameter(property = "overridePomFile", required = false, defaultValue = "false")
      protected Boolean overridePomFile;
  
      //components
      /**
       * The initial pom template file.
       */
      @Parameter(defaultValue = "camel-kafka-connector-template-pom.template")
      protected String initialPomTemplate;
  
      /**
       * Properties file to configure additional dependencies.
       */
      @Parameter(defaultValue = "camel-kafka-connector-fix-dependencies.properties")
      protected String fixDependenciesProperties;
  
      /**
       * A comma separated list of column separated GAV to include as dependencies
       * to the generated camel kafka connector. (i.e.
       * groupId:ArtifactId:version,groupId_2:ArtifactId_2:version_2)
       */
      @Parameter(defaultValue = "", readonly = true)
      protected String additionalComponentDependencies;
  
      //Kamelets
      /**
       * The initial kamelet pom template file.
       */
      @Parameter(defaultValue = "camel-kafka-connector-kamelet-template-pom.template")
      protected String initialKameletPomTemplate;
  
      /**
       * Properties kamelet file to configure additional dependencies.
       */
      @Parameter(defaultValue = "camel-kafka-connector-kamelet-fix-dependencies.properties")
      protected String fixKameletDependenciesProperties;
  
      /**
       * A comma separated list of column separated GAV to include as dependencies
       * to the generated camel kafka connector. (i.e.
       * groupId:ArtifactId:version,groupId_2:ArtifactId_2:version_2)
       */
      @Parameter(defaultValue = "", readonly = true)
      private String additionalKameletDependencies;
  
      /**
       * The maven session.
       */
      @Parameter(defaultValue = "${session}", readonly = true)
      private MavenSession session;
  
      @Component
      private BuildPluginManager pluginManager;
  
      /**
       * The Camel Component Filter to select for which components generate the corresponding camel kafka connector.
       */
      @Parameter(defaultValue = "", readonly = true)
      private String filter;
  
      /**
       * The Camel Component Exclusion List to select for which components must be skipped while generating kafka connector.
       */
      @Parameter(defaultValue = "", readonly = true)
      private List excludedComponents = Collections.EMPTY_LIST;
  
      /**
       * The Camel Kamelets Exclusion List to select for which kamelet must be skipped while generating kafka connector.
       */
      @Parameter(defaultValue = "", readonly = true)
      private List excludedKamelets = Collections.EMPTY_LIST;
  
      /**
       * The Exclusion List of connectors that must be skipped while deleting kafka connector.
       */
      @Parameter(defaultValue = "", readonly = true)
      private List excludedConnectorsFromDeletion = Collections.EMPTY_LIST;
  
      @Component
      private ProjectDependenciesResolver projectDependenciesResolver;
  
      @Component
      private ProjectBuilder projectBuilder;
  
      @Override
      protected String getMainDepArtifactId() {
          throw new UnsupportedOperationException("getMainDepArtifactId() in GenerateCamelKafkaConnectorsMojo");
      }
  
      @Override
      protected void executeAll() throws MojoExecutionException, IOException, ResourceNotFoundException, FileResourceCreationException {
          // load some project version properties
          final Properties properties = new Properties();
          try (InputStream stream = new FileInputStream(rm.getResourceAsFile("project.properties"))) {
              properties.load(stream);
          }
  
          Map<String, String> kameletsResources = new HashMap<>();
          Set<String> camelComponentsUsedInKamelets = new HashSet<>();
          List<String> resourceNames;
          try (ScanResult scanResult = new ClassGraph().acceptPaths("/" + KAMELETS_DIR + "/").scan()) {
              resourceNames = scanResult.getAllResources().getPaths();
          }
          for (String fileName: resourceNames) {
              String pathInJar = "/" + fileName;
              String kamelet = new BufferedReader(
                      new InputStreamReader(KameletsCatalog.class.getResourceAsStream(pathInJar), StandardCharsets.UTF_8))
                      .lines()
                      .collect(Collectors.joining("\n"));
              KameletModel kameletModel = YamlKameletMapper.parseKameletYaml(kamelet);
  
              // filter all kamelets with type not in {source,sink} and not excluded
              if (("source".equals(kameletModel.getType()) || "sink".equals(kameletModel.getType())) && !excludedKamelets.contains(kameletModel.getName())) {
                  kameletsResources.put(kameletModel.getName(), kamelet);
                  camelComponentsUsedInKamelets.addAll(
                          kameletModel.getDependencies().stream()
                                  .filter(d -> d.startsWith("camel:"))
                                  .map(d -> d.replaceFirst("camel:", ""))
                                  .collect(Collectors.toSet())
                  );
              }
              //TODO: add include (filter)
          }
          getLog().info("Kamelets found to be used to generate/update a kafka connector: " + kameletsResources.keySet());
  
          for (String kamelet : kameletsResources.keySet()) {
              executeMojo(
                      plugin(
                              groupId(properties.getProperty("groupId")),
                              artifactId(properties.getProperty("artifactId")),
                              version(properties.getProperty("version"))
                      ),
                      goal("camel-kafka-connector-kamelet-create"),
                      configuration(
                              element(name("name"), kamelet),
                              element(name("initialKameletPomTemplate"), initialKameletPomTemplate),
                              element(name("noticeTemplate"), noticeTemplate),
                              element(name("licenseTemplate"), licenseTemplate),
                              element(name("fixKameletDependenciesProperties"), fixKameletDependenciesProperties),
                              element(name("packageFileTemplate"), packageFileTemplate),
                              element(name("overridePomFile"), overridePomFile.toString()),
                              element(name("connectorsProjectName"), connectorsProjectName)
                      ),
                      executionEnvironment(
                              project,
                              session,
                              pluginManager
                      )
              );
  
              executeMojo(
                      plugin(
                              groupId(properties.getProperty("groupId")),
                              artifactId(properties.getProperty("artifactId")),
                              version(properties.getProperty("version"))
                      ),
                      goal("camel-kafka-connector-kamelet-update"),
                      configuration(
                              element(name("additionalDependencies"), additionalComponentDependencies),
                              element(name("name"), kamelet),
                              element(name("kameletYaml"), kameletsResources.get(kamelet)),
                              element(name("initialKameletPomTemplate"), initialKameletPomTemplate),
                              element(name("noticeTemplate"), noticeTemplate),
                              element(name("licenseTemplate"), licenseTemplate),
                              element(name("fixKameletDependenciesProperties"), fixKameletDependenciesProperties),
                              element(name("packageFileTemplate"), packageFileTemplate),
                              element(name("connectorsProjectName"), connectorsProjectName)
                      ),
                      executionEnvironment(
                              project,
                              session,
                              pluginManager
                      )
              );
          }
  
          CamelCatalog cc = new DefaultCamelCatalog();
          List<String> components;
          List<String> filteredComponents;
          if (filter == null || filter.isEmpty()) {
              components = cc.findComponentNames();
          } else {
              Set<String> filterComponentNames = new HashSet<>(Arrays.asList(filter.split(",")));
              components = cc.findComponentNames().stream().filter(componentName -> filterComponentNames.contains(componentName)).collect(Collectors.toList());
          }
          // exclude all components used in a kamelet
          camelComponentsUsedInKamelets.addAll(excludedComponents);
          excludedComponents = camelComponentsUsedInKamelets.stream().collect(Collectors.toList());
          if (excludedComponents == null || excludedComponents.isEmpty()) {
              filteredComponents = components;
          } else {
              filteredComponents = components.stream().filter(component -> !excludedComponents.contains(component)).collect(Collectors.toList());
          }
          if (filter != null && !filter.isEmpty()) {
              getLog().info("Filtered Components that will be used to generate a kafka connector: " + filter);
          }
          if (excludedComponents != null && !excludedComponents.isEmpty()) {
              getLog().info("Excluded Components that won't be used to generate a kafka connector: " + excludedComponents);
          }
          getLog().info("Components found to be used to generate/update a kafka connector: " + filteredComponents);
  
          for (String component : filteredComponents) {
              String cJson = cc.componentJSonSchema(component);
  
              executeMojo(
                      plugin(
                              groupId(properties.getProperty("groupId")),
                              artifactId(properties.getProperty("artifactId")),
                              version(properties.getProperty("version"))
                      ),
                      goal("camel-kafka-connector-create"),
                      configuration(
                              element(name("name"), component),
                              element(name("componentJson"), cJson),
                              element(name("initialPomTemplate"), initialPomTemplate),
                              element(name("noticeTemplate"), noticeTemplate),
                              element(name("licenseTemplate"), licenseTemplate),
                              element(name("fixDependenciesProperties"), fixDependenciesProperties),
                              element(name("packageFileTemplate"), packageFileTemplate),
                              element(name("overridePomFile"), overridePomFile.toString()),
                              element(name("connectorsProjectName"), connectorsProjectName)
                      ),
                      executionEnvironment(
                              project,
                              session,
                              pluginManager
                      )
              );
  
              executeMojo(
                      plugin(
                              groupId(properties.getProperty("groupId")),
                              artifactId(properties.getProperty("artifactId")),
                              version(properties.getProperty("version"))
                      ),
                      goal("camel-kafka-connector-update"),
                      configuration(
                              element(name("additionalDependencies"), additionalComponentDependencies),
                              element(name("name"), component),
                              element(name("componentJson"), cJson),
                              element(name("initialPomTemplate"), initialPomTemplate),
                              element(name("noticeTemplate"), noticeTemplate),
                              element(name("licenseTemplate"), licenseTemplate),
                              element(name("fixDependenciesProperties"), fixDependenciesProperties),
                              element(name("packageFileTemplate"), packageFileTemplate),
                              element(name("connectorsProjectName"), connectorsProjectName)
                      ),
                      executionEnvironment(
                              project,
                              session,
                              pluginManager
                      )
              );
          }
  
          if (removeMissingComponents) {
              if (projectDir != null && projectDir.isDirectory()) {
                  // sanitize names, as there are some camel components with + signal which are sanitized when creating the kafka connector
                  List<String> sanitizedGeneratedFromComponentsConnectorsNames = filteredComponents.stream().map(MavenUtils::sanitizeMavenArtifactId).collect(Collectors.toList());
                  List<String> sanitizedGeneratedFromKameletsConnectorsNames = kameletsResources.keySet().stream().map(MavenUtils::sanitizeMavenArtifactId).collect(Collectors.toList());
                  // retrieve the list of existing camel kafka connectors
                  String[] existingConnectorNames = projectDir.list((dir, filename) -> filename.endsWith(KAFKA_CONNECTORS_SUFFIX));
                  if (existingConnectorNames != null) {
                      List<String> connectorsToRemove = Stream.of(existingConnectorNames).sorted().filter(filename -> {
                          String componentName = extractComponentName(filename);
                          // set to remove connectors that are not generated from camel components or a kamelet and are not excluded in excludedConnectorsFromDeletion
                          return !sanitizedGeneratedFromComponentsConnectorsNames.contains(componentName)
                                  && !sanitizedGeneratedFromKameletsConnectorsNames.contains(componentName)
                                  && !excludedConnectorsFromDeletion.contains(componentName);
                      }).collect(Collectors.toList());
  
                      getLog().info("Connectors previously generated found to be removed: " + connectorsToRemove);
  
                      for (String connectorToBeRemoved: connectorsToRemove) {
  
                          executeMojo(
                                  plugin(
                                          groupId(properties.getProperty("groupId")),
                                          artifactId(properties.getProperty("artifactId")),
                                          version(properties.getProperty("version"))
                                  ),
                                  goal("camel-kafka-connector-delete"),
                                  configuration(
                                          element(name("name"), connectorToBeRemoved),
                                          element(name("connectorsProjectName"), connectorsProjectName)
                                  ),
                                  executionEnvironment(
                                          project,
                                          session,
                                          pluginManager
                                  )
                          );
                      }
                  }
              }
          }
      }
  
      private String extractComponentName(String connectorName) {
          // remove starting "camel-"
          String name = connectorName.substring("camel-".length());
          // remove final KAFKA_CONNECTORS_SUFFIX
          return name.substring(0, name.length() - KAFKA_CONNECTORS_SUFFIX.length());
      }
  }
  