package io.aiven.klaw.service;

import static io.aiven.klaw.error.KlawErrorMessages.*;
import static io.aiven.klaw.helpers.KwConstants.*;
import static io.aiven.klaw.helpers.UtilMethods.updateEnvStatus;

import io.aiven.klaw.config.ManageDatabase;
import io.aiven.klaw.dao.AclRequests;
import io.aiven.klaw.dao.Env;
import io.aiven.klaw.dao.KwClusters;
import io.aiven.klaw.dao.SchemaRequest;
import io.aiven.klaw.error.KlawException;
import io.aiven.klaw.error.KlawRestException;
import io.aiven.klaw.error.RestErrorResponse;
import io.aiven.klaw.model.ApiResponse;
import io.aiven.klaw.model.cluster.ClusterAclRequest;
import io.aiven.klaw.model.cluster.ClusterConnectorRequest;
import io.aiven.klaw.model.cluster.ClusterSchemaRequest;
import io.aiven.klaw.model.cluster.ClusterTopicRequest;
import io.aiven.klaw.model.cluster.ConnectorsStatus;
import io.aiven.klaw.model.cluster.LoadTopicsResponse;
import io.aiven.klaw.model.cluster.SchemasInfoOfClusterResponse;
import io.aiven.klaw.model.cluster.consumergroup.ResetConsumerGroupOffsetsRequest;
import io.aiven.klaw.model.enums.AclPatternType;
import io.aiven.klaw.model.enums.AclType;
import io.aiven.klaw.model.enums.AclsNativeType;
import io.aiven.klaw.model.enums.ApiResultStatus;
import io.aiven.klaw.model.enums.ClusterStatus;
import io.aiven.klaw.model.enums.KafkaClustersType;
import io.aiven.klaw.model.enums.KafkaFlavors;
import io.aiven.klaw.model.enums.KafkaSupportedProtocol;
import io.aiven.klaw.model.enums.RequestOperationType;
import io.aiven.klaw.model.enums.SchemaType;
import io.aiven.klaw.model.requests.KafkaConnectorRestartModel;
import io.aiven.klaw.model.response.OffsetDetails;
import io.aiven.klaw.model.response.ServiceAccountDetails;
import io.jsonwebtoken.Jwts;
import io.jsonwebtoken.SignatureAlgorithm;
import jakarta.annotation.PostConstruct;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.security.Key;
import java.security.KeyManagementException;
import java.security.KeyStore;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.security.UnrecoverableKeyException;
import java.security.cert.CertificateException;
import java.security.cert.X509Certificate;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.TreeMap;
import java.util.UUID;
import javax.crypto.spec.SecretKeySpec;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.binary.Base64;
import org.apache.commons.lang3.StringUtils;
import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
import org.apache.hc.client5.http.impl.classic.HttpClients;
import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
import org.apache.hc.client5.http.io.HttpClientConnectionManager;
import org.apache.hc.client5.http.socket.ConnectionSocketFactory;
import org.apache.hc.client5.http.socket.PlainConnectionSocketFactory;
import org.apache.hc.client5.http.ssl.NoopHostnameVerifier;
import org.apache.hc.client5.http.ssl.SSLConnectionSocketFactory;
import org.apache.hc.core5.http.config.Registry;
import org.apache.hc.core5.http.config.RegistryBuilder;
import org.apache.hc.core5.ssl.SSLContextBuilder;
import org.apache.hc.core5.ssl.TrustStrategy;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
import org.springframework.stereotype.Service;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.util.ResourceUtils;
import org.springframework.web.client.HttpClientErrorException;
import org.springframework.web.client.HttpServerErrorException;
import org.springframework.web.client.HttpStatusCodeException;
import org.springframework.web.client.RestTemplate;

@Service
@Slf4j
public class ClusterApiService {
  private static final String URL_DELIMITER = "/";
  public static final String URI_CONNECTOR_STATUS = "?connectorStatus=";
  public static final String TOPICS_NATIVE_TYPE = "topicsNativeType";
  public static final String RESET_CACHE = "resetCache";
  public static final String PARTITION_ID = "partitionId";
  public static final String RANGE_OFFSETS = "rangeOffsets";
  public static final String SELECTED_NUMBER_OF_OFFSETS = "selectedNumberOfOffsets";

  @Autowired private ManageDatabase manageDatabase;

  @Value("${server.ssl.key-store:null}")
  private String keyStore;

  @Value("${server.ssl.key-store-password:null}")
  private String keyStorePwd;

  @Value("${server.ssl.key-store-type:JKS}")
  private String keyStoreType;

  @Value("${klaw.clusterapi.access.username}")
  private String clusterApiUser;

  @Value("${klaw.clusterapi.access.base64.secret:#{''}}")
  private String clusterApiAccessBase64Secret;

  private static String clusterConnUrl;
  protected static HttpComponentsClientHttpRequestFactory requestFactory;
  RestTemplate httpRestTemplate, httpsRestTemplate;

  public ClusterApiService(ManageDatabase manageDatabase) {
    this.manageDatabase = manageDatabase;
  }

  private RestTemplate getRestTemplate(String clusterConnectivityUrl) {
    if (StringUtils.isEmpty(clusterConnectivityUrl)) {
      clusterConnectivityUrl = clusterConnUrl;
    }
    if (clusterConnectivityUrl.toLowerCase().startsWith("https")) {
      if (this.httpsRestTemplate == null) {
        this.httpsRestTemplate = new RestTemplate(requestFactory);
      }
      return this.httpsRestTemplate;
    } else {
      if (this.httpRestTemplate == null) {
        this.httpRestTemplate = new RestTemplate();
      }
      return this.httpRestTemplate;
    }
  }

  private void getClusterApiProperties(int tenantId) {
    clusterConnUrl = manageDatabase.getKwPropertyValue(CLUSTER_CONN_URL_KEY, tenantId);
    if (clusterApiAccessBase64Secret.isBlank()) {
      log.info(
          "CONFIGURE CLUSTER API SECRET FOR CLUSTER OPERATIONS. klaw.clusterapi.access.base64.secret");
    }
  }

  public ClusterStatus getClusterApiStatus(
      String clusterApiUrl, boolean testConnection, int tenantId) {
    log.info(
        "getClusterApiStatus clusterApiUrl {} testConnection{}", clusterApiUrl, testConnection);
    getClusterApiProperties(tenantId);
    try {
      String uriClusterApiStatus = URI_CLUSTER_API;
      String uri;
      RestTemplate restTemplate;
      if (testConnection) {
        uri = clusterApiUrl + uriClusterApiStatus;
        restTemplate = getRestTemplate(clusterApiUrl);
      } else {
        uri = clusterConnUrl + uriClusterApiStatus; // from stored kw props
        restTemplate = getRestTemplate(clusterConnUrl);
      }

      ResponseEntity<ClusterStatus> resultBody =
          restTemplate.exchange(uri, HttpMethod.GET, getHttpEntity(), ClusterStatus.class);
      return Objects.requireNonNull(resultBody.getBody());
    } catch (Exception e) {
      log.error("Error from getClusterApiStatus ", e);
      return ClusterStatus.OFFLINE;
    }
  }

  ClusterStatus getKafkaClusterStatus(
      String bootstrapHost,
      KafkaSupportedProtocol protocol,
      String clusterIdentification,
      String clusterType,
      String kafkaFlavor,
      int tenantId) {
    log.debug("getKafkaClusterStatus {} {}", bootstrapHost, protocol);
    getClusterApiProperties(tenantId);

    try {
      String uri =
          clusterConnUrl
              + URI_KAFKA_SR_CONN_STATUS
              + bootstrapHost
              + URL_DELIMITER
              + String.join(
                  URL_DELIMITER,
                  protocol.getName(),
                  clusterIdentification,
                  clusterType,
                  "kafkaFlavor",
                  kafkaFlavor);

      ResponseEntity<ClusterStatus> resultBody =
          getRestTemplate(null).exchange(uri, HttpMethod.GET, getHttpEntity(), ClusterStatus.class);
      return Objects.requireNonNull(resultBody.getBody());
    } catch (Exception e) {
      log.error("Error from getKafkaClusterStatus ", e);
      return ClusterStatus.NOT_KNOWN;
    }
  }

  public List<OffsetDetails> getConsumerOffsets(
      String bootstrapHost,
      KafkaSupportedProtocol protocol,
      String clusterIdentification,
      String topic,
      String consumerGroupId,
      int tenantId)
      throws KlawException {
    log.info("getConsumerOffsets {} {} {} {}", bootstrapHost, protocol, topic, consumerGroupId);
    getClusterApiProperties(tenantId);
    List<OffsetDetails> offsetDetailsList;
    try {
      String url = URI_GET_CONSUMER_OFFSETS;
      url =
          clusterConnUrl
              + url
              + bootstrapHost
              + URL_DELIMITER
              + String.join(
                  URL_DELIMITER, protocol.getName(), clusterIdentification, consumerGroupId, topic);

      ResponseEntity<List<OffsetDetails>> resultBody =
          getRestTemplate(null)
              .exchange(
                  url, HttpMethod.GET, getHttpEntity(), new ParameterizedTypeReference<>() {});

      offsetDetailsList = new ArrayList<>(Objects.requireNonNull(resultBody.getBody()));
    } catch (Exception e) {
      log.error("Error from getConsumerOffsets ", e);
      throw new KlawException(CLUSTER_API_ERR_101);
    }
    return offsetDetailsList;
  }

  public Map<String, String> getTopicEvents(
      String bootstrapHost,
      KafkaSupportedProtocol protocol,
      String clusterIdentification,
      String topic,
      String offsetId,
      Integer selectedPartitionId,
      Integer selectedNumberOfOffsets,
      Integer selectedOffsetRangeStart,
      Integer selectedOffsetRangeEnd,
      String consumerGroupId,
      int tenantId)
      throws KlawException {
    getClusterApiProperties(tenantId);
    Map<String, String> eventsMap;
    try {
      String url = URI_GET_TOPIC_CONTENTS;
      url =
          clusterConnUrl
              + url
              + bootstrapHost
              + URL_DELIMITER
              + String.join(
                  URL_DELIMITER,
                  protocol.getName(),
                  consumerGroupId,
                  topic,
                  offsetId,
                  PARTITION_ID,
                  String.valueOf(selectedPartitionId),
                  SELECTED_NUMBER_OF_OFFSETS,
                  String.valueOf(selectedNumberOfOffsets),
                  clusterIdentification,
                  RANGE_OFFSETS,
                  String.valueOf(selectedOffsetRangeStart),
                  String.valueOf(selectedOffsetRangeEnd));

      ResponseEntity<Map<String, String>> resultBody =
          getRestTemplate(null)
              .exchange(
                  url, HttpMethod.GET, getHttpEntity(), new ParameterizedTypeReference<>() {});

      eventsMap = new TreeMap<>(Objects.requireNonNull(resultBody.getBody()));
    } catch (Exception e) {
      log.error("Error from getTopicEvents {} ", topic, e);
      throw new KlawException(String.format(CLUSTER_API_ERR_102, topic));
    }
    return eventsMap;
  }

  public List<Map<String, String>> getAcls(
      String bootstrapHost, Env envSelected, KafkaSupportedProtocol protocol, int tenantId)
      throws KlawException {
    log.info("getAcls {} {} {}", bootstrapHost, protocol, tenantId);
    getClusterApiProperties(tenantId);

    List<Map<String, String>> aclListOriginal;
    try {
      String uriGetAcls = URI_GET_ACLS;
      KwClusters kwClusters =
          manageDatabase
              .getClusters(KafkaClustersType.KAFKA, tenantId)
              .get(envSelected.getClusterId());

      String uri;
      // aiven config
      if (KafkaFlavors.AIVEN_FOR_APACHE_KAFKA.value.equals(kwClusters.getKafkaFlavor())) {
        uri =
            clusterConnUrl
                + uriGetAcls
                + bootstrapHost
                + URL_DELIMITER
                + String.join(
                    URL_DELIMITER,
                    AclsNativeType.AIVEN.name(),
                    protocol.getName(),
                    kwClusters.getClusterName() + kwClusters.getClusterId(),
                    kwClusters.getProjectName(),
                    kwClusters.getServiceName());
      } // confluent cloud config
      else if (KafkaFlavors.CONFLUENT_CLOUD.value.equals(kwClusters.getKafkaFlavor())) {
        uri =
            clusterConnUrl
                + uriGetAcls
                + bootstrapHost
                + URL_DELIMITER
                + String.join(
                    URL_DELIMITER,
                    AclsNativeType.CONFLUENT_CLOUD.name(),
                    protocol.getName(),
                    kwClusters.getClusterName() + kwClusters.getClusterId(),
                    kwClusters.getProjectName(),
                    kwClusters.getServiceName());
      } else {
        uri =
            clusterConnUrl
                + uriGetAcls
                + bootstrapHost
                + URL_DELIMITER
                + String.join(
                    URL_DELIMITER,
                    AclsNativeType.NATIVE.name(),
                    protocol.getName(),
                    kwClusters.getClusterName() + kwClusters.getClusterId(),
                    "na",
                    "na");
      }

      ResponseEntity<Set<Map<String, String>>> resultBody =
          getRestTemplate(null)
              .exchange(
                  uri, HttpMethod.GET, getHttpEntity(), new ParameterizedTypeReference<>() {});
      aclListOriginal = new ArrayList<>(Objects.requireNonNull(resultBody.getBody()));
    } catch (Exception e) {
      log.error("Error from getAcls", e);
      throw new KlawException(CLUSTER_API_ERR_103);
    }
    return aclListOriginal;
  }

  public LoadTopicsResponse getAllTopics(
      String bootstrapHost,
      KafkaSupportedProtocol protocol,
      String clusterIdentification,
      String kafkaFlavors,
      int tenantId,
      boolean resetTopicsCache)
      throws Exception {
    log.info("getAllTopics {} {}", bootstrapHost, protocol);
    getClusterApiProperties(tenantId);

    LoadTopicsResponse loadTopicsResponse;
    String aclsNativeType = AclsNativeType.NATIVE.value;

    if (KafkaFlavors.CONFLUENT_CLOUD.value.equals(kafkaFlavors)) {
      aclsNativeType = AclsNativeType.CONFLUENT_CLOUD.value;
    }
    try {
      String uriGetTopicsFull =
          clusterConnUrl
              + URI_GET_TOPICS
              + bootstrapHost
              + URL_DELIMITER
              + String.join(
                  URL_DELIMITER,
                  protocol.getName(),
                  clusterIdentification,
                  TOPICS_NATIVE_TYPE,
                  aclsNativeType,
                  RESET_CACHE,
                  String.valueOf(resetTopicsCache));

      HttpEntity<String> entity = getHttpEntity();
      ResponseEntity<LoadTopicsResponse> s =
          getRestTemplate(null)
              .exchange(
                  uriGetTopicsFull, HttpMethod.GET, entity, new ParameterizedTypeReference<>() {});
      loadTopicsResponse = Objects.requireNonNull(s.getBody());
    } catch (Exception e) {
      log.error("Error from getAllTopics", e);
      throw new KlawException(CLUSTER_API_ERR_104);
    }

    return loadTopicsResponse;
  }

  public String approveConnectorRequests(
      String connectorName,
      KafkaSupportedProtocol protocol,
      String connectorType,
      String connectorConfig,
      String kafkaConnectHost,
      String clusterIdentification,
      String environemt,
      int tenantId)
      throws KlawException, KlawRestException {
    log.info("approveConnectorRequests {} {}", connectorConfig, kafkaConnectHost);
    getClusterApiProperties(tenantId);
    ResponseEntity<ApiResponse> response;
    try {
      ClusterConnectorRequest clusterConnectorRequest =
          ClusterConnectorRequest.builder()
              .env(kafkaConnectHost)
              .connectorName(connectorName)
              .connectorConfig(connectorConfig)
              .protocol(protocol)
              .clusterIdentification(clusterIdentification)
              .build();

      String uri;

      if (RequestOperationType.CREATE.value.equals(connectorType)
          || RequestOperationType.PROMOTE.value.equals(connectorType)) {
        uri = clusterConnUrl + URI_POST_CONNECTOR;
      } else if (RequestOperationType.UPDATE.value.equals(connectorType)) {
        uri = clusterConnUrl + URI_UPDATE_CONNECTOR;
      } else {
        uri = clusterConnUrl + URI_DELETE_CONNECTOR;
      }

      HttpHeaders headers = createHeaders(clusterApiUser);

      HttpEntity<ClusterConnectorRequest> request =
          new HttpEntity<>(clusterConnectorRequest, headers);
      response =
          getRestTemplate(null)
              .exchange(uri, HttpMethod.POST, request, new ParameterizedTypeReference<>() {});
      updateEnvStatus(response, manageDatabase, tenantId, environemt);
      ApiResponse apiResponse = response.getBody();
      if (apiResponse != null) {
        if (apiResponse.isSuccess()) {
          return ApiResultStatus.SUCCESS.value;
        } else {
          if (apiResponse.getMessage().contains("Connector " + connectorName + " not found")) {
            // if connector not found in cluster, delete from klaw
            return ApiResultStatus.SUCCESS.value;
          }
          return apiResponse.getMessage();
        }
      }

    } catch (HttpServerErrorException | HttpClientErrorException e) {
      log.error("approveConnectorRequests {} {}", connectorName, e.getMessage());
      if (e.getMessage().contains(CLUSTER_API_ERR_120)
          || e.getMessage().contains(CLUSTER_API_ERR_121)) {
        return CLUSTER_API_ERR_118;
      }
      String errorResponse = getRestErrorResponse(e, CLUSTER_API_ERR_118);
      throw new KlawRestException(errorResponse);
    } catch (Exception ex) {
      throw new KlawException(CLUSTER_API_ERR_105);
    }
    return ApiResultStatus.FAILURE.value;
  }

  private String getRestErrorResponse(HttpStatusCodeException e, String defaultErrorMsg) {
    RestErrorResponse errorResponse = null;
    try {
      errorResponse = e.getResponseBodyAs(RestErrorResponse.class);
    } catch (Exception ex) {
      log.error("Exception caught trying to process error message: ", ex);
      return defaultErrorMsg;
    }
    return errorResponse.getMessage();
  }

  public ResponseEntity<ApiResponse> approveTopicRequests(
      String topicName,
      String topicRequestType,
      int topicPartitions,
      String replicationFactor,
      String topicEnvId,
      Map<String, String> advancedTopicConfiguration,
      int tenantId,
      Boolean deleteAssociatedSchema)
      throws KlawException {
    log.info("approveTopicRequests {} {}", topicName, topicEnvId);
    getClusterApiProperties(tenantId);
    ResponseEntity<ApiResponse> response;
    ClusterTopicRequest clusterTopicRequest;
    try {
      Env envSelected = manageDatabase.getHandleDbRequests().getEnvDetails(topicEnvId, tenantId);
      KwClusters kwClusters =
          manageDatabase
              .getClusters(KafkaClustersType.KAFKA, tenantId)
              .get(envSelected.getClusterId());
      clusterTopicRequest =
          ClusterTopicRequest.builder()
              .env(kwClusters.getBootstrapServers())
              .protocol(kwClusters.getProtocol())
              .clusterName(kwClusters.getClusterName() + kwClusters.getClusterId())
              .topicName(topicName)
              .aclsNativeType(
                  (Objects.equals(kwClusters.getKafkaFlavor(), KafkaFlavors.CONFLUENT_CLOUD.value))
                      ? AclsNativeType.CONFLUENT_CLOUD
                      : AclsNativeType.NATIVE)
              .build();

      String uri;
      if (RequestOperationType.CREATE.value.equals(topicRequestType)
          || RequestOperationType.PROMOTE.value.equals(topicRequestType)) {
        uri = clusterConnUrl + URI_CREATE_TOPICS;
        clusterTopicRequest =
            clusterTopicRequest.toBuilder()
                .partitions(topicPartitions)
                .replicationFactor(Short.parseShort(replicationFactor))
                .advancedTopicConfiguration(advancedTopicConfiguration)
                .build();
      } else if (RequestOperationType.UPDATE.value.equals(topicRequestType)) {
        uri = clusterConnUrl + URI_UPDATE_TOPICS;
        clusterTopicRequest =
            clusterTopicRequest.toBuilder()
                .partitions(topicPartitions)
                .replicationFactor(Short.parseShort(replicationFactor))
                .advancedTopicConfiguration(advancedTopicConfiguration)
                .build();
      } else {
        uri = clusterConnUrl + URI_DELETE_TOPICS;
        if (deleteAssociatedSchema && envSelected.getAssociatedEnv() != null) {
          // get associated schema env
          Env schemaEnvSelected =
              manageDatabase
                  .getHandleDbRequests()
                  .getEnvDetails(envSelected.getAssociatedEnv().getId(), tenantId);
          KwClusters kwClustersSchemaEnv =
              manageDatabase
                  .getClusters(KafkaClustersType.SCHEMA_REGISTRY, tenantId)
                  .get(schemaEnvSelected.getClusterId());
          clusterTopicRequest =
              clusterTopicRequest.toBuilder()
                  .deleteAssociatedSchema(true)
                  .schemaClusterIdentification(
                      kwClustersSchemaEnv.getClusterName() + kwClustersSchemaEnv.getClusterId())
                  .schemaEnv(kwClustersSchemaEnv.getBootstrapServers())
                  .schemaEnvProtocol(kwClustersSchemaEnv.getProtocol())
                  .build();
        } else {
          clusterTopicRequest =
              clusterTopicRequest.toBuilder().deleteAssociatedSchema(false).build();
        }
      }

      HttpHeaders headers = createHeaders(clusterApiUser);
      HttpEntity<ClusterTopicRequest> request = new HttpEntity<>(clusterTopicRequest, headers);
      response = getRestTemplate(null).postForEntity(uri, request, ApiResponse.class);
    } catch (Exception e) {
      log.error("approveTopicRequests {}", topicName, e);
      return throwCommonErrors(e, CLUSTER_API_ERR_106);
    }
    return response;
  }

  public ResponseEntity<ApiResponse> approveAclRequests(AclRequests aclReq, int tenantId)
      throws KlawException {
    log.info("approveAclRequests {}", aclReq);
    getClusterApiProperties(tenantId);
    ResponseEntity<ApiResponse> response;

    try {
      String env = aclReq.getEnvironment();
      String uri;

      ClusterAclRequest clusterAclRequest;
      Env envSelected = manageDatabase.getHandleDbRequests().getEnvDetails(env, tenantId);
      KwClusters kwClusters =
          manageDatabase
              .getClusters(KafkaClustersType.KAFKA, tenantId)
              .get(envSelected.getClusterId());

      // aiven config
      if (Objects.equals(KafkaFlavors.AIVEN_FOR_APACHE_KAFKA.value, kwClusters.getKafkaFlavor())) {
        clusterAclRequest =
            ClusterAclRequest.builder()
                .aclNativeType(AclsNativeType.AIVEN.name())
                .projectName(kwClusters.getProjectName())
                .serviceName(kwClusters.getServiceName())
                .topicName(aclReq.getTopicname())
                .username(aclReq.getAcl_ssl())
                .build();

        if (Objects.equals(aclReq.getAclType(), AclType.PRODUCER.value)) {
          clusterAclRequest = clusterAclRequest.toBuilder().permission("write").build();
        } else {
          clusterAclRequest = clusterAclRequest.toBuilder().permission("read").build();
        }

        if (Objects.equals(RequestOperationType.DELETE.value, aclReq.getRequestOperationType())
            && null != aclReq.getJsonParams()) {
          Map<String, String> jsonObj = aclReq.getJsonParams();
          String aivenAclKey = "aivenaclid";
          if (jsonObj.containsKey(aivenAclKey)) {
            clusterAclRequest =
                clusterAclRequest.toBuilder().aivenAclKey(jsonObj.get(aivenAclKey)).build();
          } else {
            log.error("Error from approveAclRequests : AclId - aivenaclid not found");
            throw new KlawException(CLUSTER_API_ERR_107);
          }
        }
      } else if (Objects.equals(KafkaFlavors.CONFLUENT_CLOUD.value, kwClusters.getKafkaFlavor())) {
        String aclPatternType = aclReq.getAclPatternType();
        clusterAclRequest =
            ClusterAclRequest.builder()
                .aclNativeType(AclsNativeType.CONFLUENT_CLOUD.name())
                .env(kwClusters.getBootstrapServers())
                .protocol(kwClusters.getProtocol())
                .clusterName(kwClusters.getClusterName() + kwClusters.getClusterId())
                .topicName(aclReq.getTopicname())
                .consumerGroup(aclReq.getConsumergroup())
                .aclType(aclReq.getAclType())
                .aclIp(aclReq.getAcl_ip())
                .aclSsl(aclReq.getAcl_ssl())
                .transactionalId(aclReq.getTransactionalId())
                .aclIpPrincipleType(aclReq.getAclIpPrincipleType().name())
                .isPrefixAcl(AclPatternType.PREFIXED.value.equals(aclPatternType))
                .build();
      } else {
        String aclPatternType = aclReq.getAclPatternType();
        clusterAclRequest =
            ClusterAclRequest.builder()
                .aclNativeType(AclsNativeType.NATIVE.name())
                .env(kwClusters.getBootstrapServers())
                .protocol(kwClusters.getProtocol())
                .clusterName(kwClusters.getClusterName() + kwClusters.getClusterId())
                .topicName(aclReq.getTopicname())
                .consumerGroup(aclReq.getConsumergroup())
                .aclType(aclReq.getAclType())
                .aclIp(aclReq.getAcl_ip())
                .aclSsl(aclReq.getAcl_ssl())
                .transactionalId(aclReq.getTransactionalId())
                .aclIpPrincipleType(aclReq.getAclIpPrincipleType().name())
                .isPrefixAcl(AclPatternType.PREFIXED.value.equals(aclPatternType))
                .build();
      }

      if (RequestOperationType.CREATE.value.equals(aclReq.getRequestOperationType())) {
        uri = clusterConnUrl + URI_CREATE_ACLS;
        clusterAclRequest =
            clusterAclRequest.toBuilder().requestOperationType(RequestOperationType.CREATE).build();
      } else {
        uri = clusterConnUrl + URI_DELETE_ACLS;
        clusterAclRequest =
            clusterAclRequest.toBuilder().requestOperationType(RequestOperationType.DELETE).build();
      }

      HttpHeaders headers = createHeaders(clusterApiUser);
      headers.setContentType(MediaType.APPLICATION_JSON);

      HttpEntity<ClusterAclRequest> request = new HttpEntity<>(clusterAclRequest, headers);
      response =
          getRestTemplate(null)
              .exchange(uri, HttpMethod.POST, request, new ParameterizedTypeReference<>() {});
      return response;
    } catch (Exception e) {
      log.error("Error from approveAclRequests", e);
      if (e.getMessage().contains(CLUSTER_API_ERR_120)
          || e.getMessage().contains(CLUSTER_API_ERR_121)) {
        return new ResponseEntity<>(
            ApiResponse.notOk(CLUSTER_API_ERR_118), HttpStatus.INTERNAL_SERVER_ERROR);
      }
      throw new KlawException(CLUSTER_API_ERR_108);
    }
  }

  public ServiceAccountDetails getAivenServiceAccountDetails(
      String projectName, String serviceName, String userName, int tenantId) throws KlawException {
    getClusterApiProperties(tenantId);
    try {
      String uriGetServiceAccountDetails = clusterConnUrl + URI_AIVEN_SERVICE_ACCOUNT_DETAIL;
      uriGetServiceAccountDetails =
          uriGetServiceAccountDetails
              .replace("projectName", projectName)
              .replace("serviceName", serviceName)
              .replace("userName", userName);

      HttpEntity<String> entity = getHttpEntity();
      ResponseEntity<ServiceAccountDetails> apiResponseResponseEntity =
          getRestTemplate(null)
              .exchange(
                  uriGetServiceAccountDetails,
                  HttpMethod.GET,
                  entity,
                  new ParameterizedTypeReference<>() {});
      return apiResponseResponseEntity.getBody();
    } catch (Exception e) {
      log.error("Error from getAivenServiceAccountDetails", e);
      throw new KlawException(CLUSTER_API_ERR_109);
    }
  }

  public ApiResponse getAivenServiceAccounts(String projectName, String serviceName, int tenantId)
      throws KlawException {
    getClusterApiProperties(tenantId);
    try {
      String uriGetServiceAccounts = clusterConnUrl + URI_AIVEN_SERVICE_ACCOUNTS;
      uriGetServiceAccounts =
          uriGetServiceAccounts
              .replace("projectName", projectName)
              .replace("serviceName", serviceName);

      HttpEntity<String> entity = getHttpEntity();
      ResponseEntity<ApiResponse> apiResponseResponseEntity =
          getRestTemplate(null)
              .exchange(
                  uriGetServiceAccounts,
                  HttpMethod.GET,
                  entity,
                  new ParameterizedTypeReference<>() {});
      return apiResponseResponseEntity.getBody();
    } catch (Exception e) {
      log.error("Error from getAivenServiceAccounts", e);
      throw new KlawException(CLUSTER_API_ERR_110);
    }
  }

  ResponseEntity<ApiResponse> deleteSchema(String topicName, String kafkaEnvId, int tenantId)
      throws KlawException {
    log.info("delete schema subject on cluster {} {}", topicName, kafkaEnvId);
    getClusterApiProperties(tenantId);
    ResponseEntity<ApiResponse> response;
    ClusterTopicRequest clusterTopicRequest;
    try {
      Env envSelected = manageDatabase.getHandleDbRequests().getEnvDetails(kafkaEnvId, tenantId);
      KwClusters kwClusters =
          manageDatabase
              .getClusters(KafkaClustersType.KAFKA, tenantId)
              .get(envSelected.getClusterId());
      clusterTopicRequest =
          ClusterTopicRequest.builder()
              .env(kwClusters.getBootstrapServers())
              .protocol(kwClusters.getProtocol())
              .clusterName(kwClusters.getClusterName() + kwClusters.getClusterId())
              .topicName(topicName)
              .build();

      String uri;

      uri = clusterConnUrl + URI_DELETE_SCHEMAS;
      if (envSelected.getAssociatedEnv() != null) {
        // get associated schema env
        Env schemaEnvSelected =
            manageDatabase
                .getHandleDbRequests()
                .getEnvDetails(envSelected.getAssociatedEnv().getId(), tenantId);
        KwClusters kwClustersSchemaEnv =
            manageDatabase
                .getClusters(KafkaClustersType.SCHEMA_REGISTRY, tenantId)
                .get(schemaEnvSelected.getClusterId());
        clusterTopicRequest =
            clusterTopicRequest.toBuilder()
                .deleteAssociatedSchema(true)
                .schemaClusterIdentification(
                    kwClustersSchemaEnv.getClusterName() + kwClustersSchemaEnv.getClusterId())
                .schemaEnv(kwClustersSchemaEnv.getBootstrapServers())
                .schemaEnvProtocol(kwClustersSchemaEnv.getProtocol())
                .build();
      }

      HttpHeaders headers = createHeaders(clusterApiUser);
      HttpEntity<ClusterTopicRequest> request = new HttpEntity<>(clusterTopicRequest, headers);
      response = getRestTemplate(null).postForEntity(uri, request, ApiResponse.class);
    } catch (Exception e) {
      log.error("deleteSchema {}", topicName, e);
      return throwCommonErrors(e, CLUSTER_API_ERR_123);
    }
    return response;
  }

  private ResponseEntity<ApiResponse> throwCommonErrors(Exception e, String clusterApiErr123)
      throws KlawException {
    if (e.getMessage().contains(CLUSTER_API_ERR_120)
        || e.getMessage().contains(CLUSTER_API_ERR_121)) {
      return new ResponseEntity<>(
          ApiResponse.notOk(CLUSTER_API_ERR_118), HttpStatus.INTERNAL_SERVER_ERROR);
    } else if (e.getMessage().contains("Cannot connect to cluster.")) {
      return new ResponseEntity<>(
          ApiResponse.notOk(CLUSTER_API_ERR_119), HttpStatus.INTERNAL_SERVER_ERROR);
    }
    throw new KlawException(clusterApiErr123);
  }

