package io.aiven.klaw.clusterapi.services;

import io.aiven.klaw.clusterapi.models.enums.ClusterStatus;
import io.aiven.klaw.clusterapi.models.enums.KafkaSupportedProtocol;
import io.aiven.klaw.clusterapi.utils.ClusterApiUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Service;

@Service
@Slf4j
public class UtilComponentsService {

  Environment env;

  ClusterApiUtils clusterApiUtils;

  SchemaService schemaService;

  KafkaConnectService kafkaConnectService;

  ConfluentCloudApiService confluentCloudApiService;

  public UtilComponentsService() {}

  public UtilComponentsService(Environment env, ClusterApiUtils clusterApiUtils) {
    this.env = env;
    this.clusterApiUtils = clusterApiUtils;
  }

  @Autowired
  public UtilComponentsService(
      Environment env,
      ClusterApiUtils clusterApiUtils,
      SchemaService schemaService,
      KafkaConnectService kafkaConnectService,
      ConfluentCloudApiService confluentCloudApiService) {
    this.env = env;
    this.clusterApiUtils = clusterApiUtils;
    this.schemaService = schemaService;
    this.kafkaConnectService = kafkaConnectService;
    this.confluentCloudApiService = confluentCloudApiService;
  }

  //    public String reloadTruststore(String protocol, String clusterName){
  //        getAdminClient.removeSSLElementFromAdminClientMap(protocol, clusterName);
  //        return ApiResultStatus.SUCCESS.value;
  //    }

  public ClusterStatus getStatus(
      String environment,
      KafkaSupportedProtocol protocol,
      String clusterIdentification,
      String clusterType,
      String kafkaFlavor) {
    log.info("getStatus {} {}", environment, protocol);
    return switch (clusterType) {
      case "kafka" -> getStatusKafka(environment, protocol, clusterIdentification, kafkaFlavor);
      case "schemaregistry" -> schemaService.getSchemaRegistryStatus(
          environment, protocol, clusterIdentification);
      case "kafkaconnect" -> kafkaConnectService.getKafkaConnectStatus(
          environment, protocol, clusterIdentification);
      default -> ClusterStatus.OFFLINE;
    };
  }

  private ClusterStatus getStatusKafka(
      String environment, KafkaSupportedProtocol protocol, String clusterName, String kafkaFlavor) {
    try {
      if (kafkaFlavor != null && kafkaFlavor.equals("Confluent Cloud")) {
        confluentCloudApiService.listTopics(environment, protocol, clusterName);
        return ClusterStatus.ONLINE;
      } else {
        AdminClient client = clusterApiUtils.getAdminClient(environment, protocol, clusterName);
        if (client != null) {
          return ClusterStatus.ONLINE;
        }
      }

      return ClusterStatus.OFFLINE;
    } catch (Exception e) {
      log.error("Exception:", e);
      return ClusterStatus.OFFLINE;
    }
  }
}
