package com.airlock.iam.servicecontainer.app.infrastructure.event.kafka;

import com.airlock.iam.base.api.domain.model.exception.FailureException;
import com.airlock.iam.plugin.api.config.ConfigurationContext;
import com.airlock.iam.plugin.api.config.ConfigurationEnvironment;
import com.airlock.iam.plugin.api.config.PluginConfiguration;
import com.airlock.iam.plugin.api.exception.InitializeException;
import com.airlock.iam.plugin.api.exception.PropertyConfigurationException;
import com.airlock.iam.plugin.api.exception.PropertyConstraintViolationException;
import com.airlock.iam.plugin.api.exception.PropertyTypeMismatchException;
import com.airlock.iam.plugin.api.testlet.Testlet;
import com.airlock.iam.plugin.api.testlet.Testlet.TestletResult;
import com.airlock.iam.plugin.api.testlet.Testlet.TestletType;
import com.airlock.iam.plugin.api.testlet.Testlet.TestletResult.TestResult;
import com.airlock.iam.plugin.api.valuereader.EnumPropertyValueReader;
import com.airlock.iam.plugin.api.valuereader.FilePropertyValueReader;
import com.airlock.iam.plugin.api.valuereader.IntegerPropertyValueReader;
import com.airlock.iam.plugin.api.valuereader.PasswordPropertyValueReader;
import com.airlock.iam.plugin.api.valuereader.PluginMapPropertyValueReader;
import com.airlock.iam.plugin.api.valuereader.PluginPropertyValueReader;
import com.airlock.iam.plugin.api.valuereader.StringListPropertyValueReader;
import com.airlock.iam.plugin.api.valuereader.StringPropertyValueReader;
import com.airlock.iam.servicecontainer.api.application.configuration.event.kafka.KafkaSettings;
import com.airlock.iam.servicecontainer.app.application.configuration.event.kafka.KafkaConfigValue;
import com.airlock.iam.servicecontainer.app.application.configuration.event.kafka.KafkaMessageBrokerConnectorSettings;
import com.airlock.iam.servicecontainer.app.application.configuration.event.kafka.KafkaNoAuthentication;
import com.airlock.iam.servicecontainer.app.application.configuration.event.kafka.KafkaNoAuthenticationConfig;
import com.airlock.iam.servicecontainer.app.application.configuration.event.kafka.KafkaSaslOAuthBearerAuthentication;
import com.airlock.iam.servicecontainer.app.application.configuration.event.kafka.KafkaSaslOAuthBearerAuthenticationConfig;
import com.airlock.iam.servicecontainer.app.application.configuration.event.kafka.KafkaSaslPlainAuthentication;
import com.airlock.iam.servicecontainer.app.application.configuration.event.kafka.KafkaSaslPlainAuthenticationConfig;
import com.airlock.iam.servicecontainer.app.application.configuration.event.kafka.KafkaSaslScramAuthentication;
import com.airlock.iam.servicecontainer.app.application.configuration.event.kafka.KafkaSaslScramAuthenticationConfig;
import com.airlock.iam.servicecontainer.app.application.configuration.event.kafka.ScramHashAlgorithm;
import com.airlock.iam.servicecontainer.app.application.configuration.event.kafka.SensitiveKafkaConfigValue;
import com.google.common.annotations.VisibleForTesting;
import jakarta.annotation.Nonnull;
import java.io.File;
import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.function.Function;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.common.errors.TimeoutException;

public final class KafkaConnectionTestlet implements Testlet {
   private static final String TESTLET_TITLE = "Kafka Connection Testlet";
   private static final String MAX_BLOCK_TIMEOUT_MS = "10000";
   private Optional<Producer<String, byte[]>> producer = Optional.empty();
   private final Function<KafkaConnectionTestlet.TestletInput, Producer<String, byte[]>> producerSupplier;

   public KafkaConnectionTestlet() {
      this.producerSupplier = this.defaultProducerSupplier();
   }

   @VisibleForTesting
   KafkaConnectionTestlet(Producer<String, byte[]> producer) {
      this.producer = Optional.of(producer);
      this.producerSupplier = var1x -> producer;
   }

   private Function<KafkaConnectionTestlet.TestletInput, Producer<String, byte[]>> defaultProducerSupplier() {
      return input -> {
         try {
            KafkaMessageBrokerConnectorSettings settings = buildSettings(input.pc, input.env, input.ctx);
            KafkaProducer<String, byte[]> newProducer = new KafkaProducer(settings.asKafkaProperties());
            this.producer = Optional.of(newProducer);
            return newProducer;
         } catch (InitializeException var4) {
            throw new FailureException("Unable to create producer for testing Kafka connection.");
         }
      };
   }

   private static KafkaMessageBrokerConnectorSettings buildSettings(PluginConfiguration pc, ConfigurationEnvironment env, ConfigurationContext ctx) throws PropertyTypeMismatchException, PropertyConstraintViolationException {
      List<String> bootstrapServers = configuredBootstrapServers(pc, env, ctx);
      String topic = StringPropertyValueReader.effectiveStringConfigValue(pc, "topic", env, ctx);
      String key = StringPropertyValueReader.effectiveStringConfigValue(pc, "key", env, ctx);
      Integer partition = IntegerPropertyValueReader.effectiveIntegerConfigValue(pc, "partition", env, ctx);
      String clientId = StringPropertyValueReader.effectiveStringConfigValue(pc, "clientId", env, ctx);
      PluginConfiguration authPlugin = PluginPropertyValueReader.effectivePluginPropertyValue(pc, "authentication", env, ctx);
      KafkaSettings authentication = authenticationSettings(authPlugin, env, ctx);
      String trustStorePath = FilePropertyValueReader.effectiveFilePropertyValue(pc, "trustStorePath", env, ctx);
      String trustStorePassword = PasswordPropertyValueReader.effectivePasswordConfigValue(pc, "trustStorePassword", env, ctx);
      String trustStoreType = StringPropertyValueReader.effectiveStringConfigValue(pc, "trustStoreType", env, ctx);
      String keyStorePath = FilePropertyValueReader.effectiveFilePropertyValue(pc, "keyStorePath", env, ctx);
      String keyStorePassword = PasswordPropertyValueReader.effectivePasswordConfigValue(pc, "keyStorePassword", env, ctx);
      String keyPassword = PasswordPropertyValueReader.effectivePasswordConfigValue(pc, "keyPassword", env, ctx);
      String keyStoreType = StringPropertyValueReader.effectiveStringConfigValue(pc, "keyStoreType", env, ctx);
      Map<String, String> customProperties = new HashMap<>();
      PluginMapPropertyValueReader.effectivePluginMapPropertyValue(pc, "customProperties", env, ctx)
         .forEach((k, v) -> customProperties.put(k, extractCustomPropertyValue(v, env, ctx)));
      customProperties.putIfAbsent("max.block.ms", "10000");
      long requestTimeout = Long.parseLong(customProperties.computeIfAbsent("request.timeout.ms", var0x -> "10000"));
      long lingerMs = Long.parseLong(customProperties.getOrDefault("linger.ms", "100"));
      customProperties.putIfAbsent("delivery.timeout.ms", String.valueOf(requestTimeout + lingerMs));
      return KafkaMessageBrokerConnectorSettings.builder()
         .bootstrapServers(bootstrapServers)
         .topic(topic)
         .key(key)
         .partition(Optional.ofNullable(partition))
         .clientId(clientId)
         .authentication(authentication)
         .trustStorePath(Optional.ofNullable(trustStorePath).map(File::new))
         .trustStorePassword(Optional.ofNullable(trustStorePassword))
         .trustStoreType(Optional.ofNullable(trustStoreType))
         .keyStorePath(Optional.ofNullable(keyStorePath).map(File::new))
         .keyStorePassword(Optional.ofNullable(keyStorePassword))
         .keyPassword(Optional.ofNullable(keyPassword))
         .keyStoreType(Optional.ofNullable(keyStoreType))
         .customKafkaProperties(customProperties)
         .build();
   }

   private static List<String> configuredBootstrapServers(PluginConfiguration pc, ConfigurationEnvironment env, ConfigurationContext ctx) {
      try {
         return StringListPropertyValueReader.effectiveStringListConfigValue(pc, "bootstrapServers", env, ctx);
      } catch (PropertyTypeMismatchException e) {
         throw new FailureException("Failed to retrieve configured bootstrap servers for testing Kafka connection.");
      }
   }

   private static String extractCustomPropertyValue(PluginConfiguration plugin, ConfigurationEnvironment env, ConfigurationContext ctx) {
      try {
         Class<?> pluginClass = plugin.getPluginClass();
         if (pluginClass.equals(KafkaConfigValue.class)) {
            return StringPropertyValueReader.effectiveStringConfigValue(plugin, "value", env, ctx);
         }

         if (pluginClass.equals(SensitiveKafkaConfigValue.class)) {
            return PasswordPropertyValueReader.effectivePasswordConfigValue(plugin, "value", env, ctx);
         }
      } catch (Exception var4) {
      }

      throw new FailureException("Failed to initialize custom properties for testing Kafka connection.");
   }

   private static KafkaSettings authenticationSettings(PluginConfiguration plugin, ConfigurationEnvironment env, ConfigurationContext ctx) {
      try {
         Class<?> pluginClass = plugin.getPluginClass();
         if (pluginClass.equals(KafkaNoAuthenticationConfig.class)) {
            return new KafkaNoAuthentication();
         }

         if (pluginClass.equals(KafkaSaslPlainAuthenticationConfig.class)) {
            return kafkaSaslPlainAuthenticationSettings(env, ctx, plugin);
         }

         if (pluginClass.equals(KafkaSaslScramAuthenticationConfig.class)) {
            return kafkaSaslScramAuthenticationSettings(env, ctx, plugin);
         }

         if (pluginClass.equals(KafkaSaslOAuthBearerAuthenticationConfig.class)) {
            return kafkaSaslOAuthBearerAuthenticationSettings(env, ctx, plugin);
         }
      } catch (InitializeException var4) {
      }

      throw new FailureException("Failed to initialize authentication settings for testing Kafka connection");
   }

   private static KafkaSaslPlainAuthentication kafkaSaslPlainAuthenticationSettings(
      ConfigurationEnvironment env, ConfigurationContext ctx, PluginConfiguration plugin
   ) throws PropertyTypeMismatchException {
      return KafkaSaslPlainAuthentication.builder()
         .username(StringPropertyValueReader.effectiveStringConfigValue(plugin, "username", env, ctx))
         .password(PasswordPropertyValueReader.effectivePasswordConfigValue(plugin, "password", env, ctx))
         .build();
   }

   private static KafkaSaslScramAuthentication kafkaSaslScramAuthenticationSettings(
      ConfigurationEnvironment env, ConfigurationContext ctx, PluginConfiguration plugin
   ) throws PropertyTypeMismatchException {
      return KafkaSaslScramAuthentication.builder()
         .hashAlgorithm((ScramHashAlgorithm)EnumPropertyValueReader.effectiveConfigEnum(plugin, "hashAlgorithm", env, ctx))
         .username(StringPropertyValueReader.effectiveStringConfigValue(plugin, "username", env, ctx))
         .password(PasswordPropertyValueReader.effectivePasswordConfigValue(plugin, "password", env, ctx))
         .build();
   }

   private static KafkaSaslOAuthBearerAuthentication kafkaSaslOAuthBearerAuthenticationSettings(
      ConfigurationEnvironment env, ConfigurationContext ctx, PluginConfiguration plugin
   ) throws PropertyTypeMismatchException {
      return KafkaSaslOAuthBearerAuthentication.builder()
         .clientId(StringPropertyValueReader.effectiveStringConfigValue(plugin, "clientId", env, ctx))
         .clientSecret(PasswordPropertyValueReader.effectivePasswordConfigValue(plugin, "clientSecret", env, ctx))
         .scope(StringPropertyValueReader.effectiveStringConfigValue(plugin, "scope", env, ctx))
         .tokenEndpointUrl(StringPropertyValueReader.effectiveStringConfigValue(plugin, "tokenEndpointUrl", env, ctx))
         .build();
   }

   private static boolean areRequiredAuthenticationPropertiesConfigured(PluginConfiguration plugin, ConfigurationEnvironment env, ConfigurationContext ctx) {
      try {
         Class<?> pluginClass = plugin.getPluginClass();
         if (pluginClass.equals(KafkaNoAuthenticationConfig.class)) {
            return true;
         }

         if (pluginClass.equals(KafkaSaslPlainAuthenticationConfig.class)) {
            return StringPropertyValueReader.effectiveStringConfigValue(plugin, "username", env, ctx) != null
               && PasswordPropertyValueReader.effectivePasswordConfigValue(plugin, "password", env, ctx) != null;
         }

         if (pluginClass.equals(KafkaSaslScramAuthenticationConfig.class)) {
            return StringPropertyValueReader.effectiveStringConfigValue(plugin, "username", env, ctx) != null
               && PasswordPropertyValueReader.effectivePasswordConfigValue(plugin, "password", env, ctx) != null;
         }

         if (pluginClass.equals(KafkaSaslOAuthBearerAuthenticationConfig.class)) {
            return StringPropertyValueReader.effectiveStringConfigValue(plugin, "clientId", env, ctx) != null
               && PasswordPropertyValueReader.effectivePasswordConfigValue(plugin, "clientSecret", env, ctx) != null
               && StringPropertyValueReader.effectiveStringConfigValue(plugin, "tokenEndpointUrl", env, ctx) != null;
         }
      } catch (InitializeException var4) {
      }

      return true;
   }

   @Nonnull
   private static String failureText(String errorMessage, List<String> bootstrapServers) {
      String servers = String.join(", ", bootstrapServers);
      return "Connection to Kafka broker failed.\n<p>Error message: %s</p>\n<p>Please ensure that:\n<ul>\n<li>A TCP connection can be established to at least on of the configured bootstrap servers: %s</li>\n<li>The configured TLS/mTLS settings correspond to the expected security protocol by the broker (e.g. PLAIN, SSL, SSL_CLIENT_AUTH).</li>\n<li>The configured authentication plugin corresponds to the expected SASL authentication mechanism by the broker (e.g. PLAIN, OAUTHBEARER, SCRAM-SHA-256, ...).</li>\n<li>The configured topic exists on the Kafka broker.</li>\n</ul></p>"
         .formatted(errorMessage, servers);
   }

   public TestletResult performTest(PluginConfiguration config, ConfigurationEnvironment env, ConfigurationContext context) {
      List<String> bootstrapServers = configuredBootstrapServers(config, env, context);

      try {
         Producer<String, byte[]> producer = this.producerSupplier.apply(new KafkaConnectionTestlet.TestletInput(config, env, context));

         TestletResult var13;
         try {
            String topic = StringPropertyValueReader.effectiveStringConfigValue(config, "topic", env, context);
            producer.partitionsFor(topic);
            var13 = new TestletResult(TestResult.SUCCESSFUL, "Kafka Connection Testlet", "Connection to the Kafka cluster succeeded.");
         } catch (Throwable var11) {
            if (producer != null) {
               try {
                  producer.close();
               } catch (Throwable var10) {
                  var11.addSuppressed(var10);
               }
            }

            throw var11;
         }

         if (producer != null) {
            producer.close();
         }

         return var13;
      } catch (Exception e) {
         return switch (e) {
            case TimeoutException te -> new TestletResult(
               TestResult.FAILED, "Kafka Connection Testlet", failureText("%s Various reasons possible.".formatted(te.getMessage()), bootstrapServers)
            );
            case FailureException fe -> new TestletResult(TestResult.FAILED, "Kafka Connection Testlet", fe.getMessage());
            default -> new TestletResult(TestResult.FAILED, "Kafka Connection Testlet", failureText(e.getMessage(), bootstrapServers));
         };
      }
   }

   public boolean areRequiredPropertiesConfigured(PluginConfiguration config, ConfigurationEnvironment env, ConfigurationContext context) {
      try {
         List<String> bootstrapServers = configuredBootstrapServers(config, env, context);
         String topicConfig = StringPropertyValueReader.effectiveStringConfigValue(config, "topic", env, context);
         PluginConfiguration authenticationConfig = PluginPropertyValueReader.effectivePluginPropertyValue(config, "authentication", env, context);
         return !bootstrapServers.isEmpty()
            && topicConfig != null
            && authenticationConfig != null
            && areRequiredAuthenticationPropertiesConfigured(authenticationConfig, env, context);
      } catch (PropertyConfigurationException var7) {
         return false;
      }
   }

   public String getRequiredPropertiesMissingMessage() {
      return "Please configure all mandatory properties (Bootstrap Servers, Authentication and Topic).";
   }

   public TestletType getType() {
      return TestletType.RESULT;
   }

   public void close() {
      this.producer.ifPresent(p -> p.close(Duration.ofSeconds(5L)));
   }

   private record TestletInput(PluginConfiguration pc, ConfigurationEnvironment env, ConfigurationContext ctx) {
   }
}