content: package io.aiven.klaw.clusterapi.services;
  
  import com.google.common.base.Strings;
  import io.aiven.klaw.clusterapi.models.ApiResponse;
  import io.aiven.klaw.clusterapi.models.ClusterAclRequest;
  import io.aiven.klaw.clusterapi.models.ClusterTopicRequest;
  import io.aiven.klaw.clusterapi.models.LoadTopicsResponse;
  import io.aiven.klaw.clusterapi.models.TopicConfig;
  import io.aiven.klaw.clusterapi.models.confluentcloud.AclObject;
  import io.aiven.klaw.clusterapi.models.confluentcloud.Config;
  import io.aiven.klaw.clusterapi.models.confluentcloud.ListAclsResponse;
  import io.aiven.klaw.clusterapi.models.confluentcloud.ListTopicsResponse;
  import io.aiven.klaw.clusterapi.models.confluentcloud.TopicCreateRequest;
  import io.aiven.klaw.clusterapi.models.confluentcloud.TopicObject;
  import io.aiven.klaw.clusterapi.models.enums.AclIPPrincipleType;
  import io.aiven.klaw.clusterapi.models.enums.AclPatternType;
  import io.aiven.klaw.clusterapi.models.enums.AclType;
  import io.aiven.klaw.clusterapi.models.enums.ApiResultStatus;
  import io.aiven.klaw.clusterapi.models.enums.KafkaClustersType;
  import io.aiven.klaw.clusterapi.models.enums.KafkaSupportedProtocol;
  import io.aiven.klaw.clusterapi.models.enums.RequestOperationType;
  import io.aiven.klaw.clusterapi.utils.ClusterApiUtils;
  import jakarta.validation.Valid;
  import java.util.ArrayList;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Set;
  import lombok.extern.slf4j.Slf4j;
  import org.apache.kafka.common.acl.AclOperation;
  import org.apache.kafka.common.acl.AclPermissionType;
  import org.apache.kafka.common.resource.ResourceType;
  import org.springframework.core.ParameterizedTypeReference;
  import org.springframework.core.env.Environment;
  import org.springframework.http.HttpEntity;
  import org.springframework.http.HttpHeaders;
  import org.springframework.http.HttpMethod;
  import org.springframework.http.ResponseEntity;
  import org.springframework.stereotype.Service;
  import org.springframework.web.client.RestClientException;
  import org.springframework.web.client.RestTemplate;
  import org.springframework.web.util.DefaultUriBuilderFactory;
  
  /*
  Confluent cloud API Reference
  https://docs.confluent.io/cloud/current/api.html#tag/Topic-(v3)
  https://docs.confluent.io/cloud/current/api.html#tag/ACL-(v3)
   */
  @Service
  @Slf4j
  public class ConfluentCloudApiService {
  
    public static final String HTTPS_PREFIX = "https://";
    public static final String TOPIC_API_URI_KEY = ".klaw.clusters.counfluentcloud.topics.api";
    public static final String ACLS_API_URI_KEY = ".klaw.clusters.counfluentcloud.acls.api";
    private final Environment env;
    final ClusterApiUtils clusterApiUtils;
  
    private static boolean topicsLoadingStatus;
  
    public ConfluentCloudApiService(Environment env, ClusterApiUtils clusterApiUtils) {
      this.env = env;
      this.clusterApiUtils = clusterApiUtils;
    }
  
    public LoadTopicsResponse listTopics(
        String restApiHost, KafkaSupportedProtocol protocol, String clusterIdentification)
        throws Exception {
      RestTemplate restTemplate = getRestTemplate();
      log.info("loadTopics {} {} {}", restApiHost, protocol, clusterIdentification);
  
      String listTopicsUri =
          getResourceUri(
              clusterIdentification, restApiHost, "createTopic", ResourceType.TOPIC.name());
  
      HttpHeaders headers =
          clusterApiUtils.createHeaders(clusterIdentification, KafkaClustersType.KAFKA);
      HttpEntity<Map<String, String>> request = new HttpEntity<>(headers);
  
      try {
        ResponseEntity<ListTopicsResponse> responseEntity =
            restTemplate.exchange(
                listTopicsUri, HttpMethod.GET, request, new ParameterizedTypeReference<>() {});
  
        List<TopicConfig> topicsListUpdated = processListTopicsResponse(responseEntity);
  
        return LoadTopicsResponse.builder()
            .loadingInProgress(topicsLoadingStatus)
            .topicConfigSet(new HashSet<>(topicsListUpdated))
            .build();
      } catch (RestClientException e) {
        log.error("Exception:", e);
        throw new Exception("Error in listing topics : " + e.getMessage());
      }
    }
  
    public Set<Map<String, String>> listAcls(
        String restApiHost, @Valid KafkaSupportedProtocol protocol, String clusterIdentification)
        throws Exception {
      RestTemplate restTemplate = getRestTemplate();
      log.info("listAcls {} {} {}", restApiHost, protocol, clusterIdentification);
  
      String listAclsUri = getResourceUri(clusterIdentification, restApiHost, "listAclsUri", "ACLS");
  
      HttpHeaders headers =
          clusterApiUtils.createHeaders(clusterIdentification, KafkaClustersType.KAFKA);
      HttpEntity<Map<String, String>> request = new HttpEntity<>(headers);
  
      try {
        ResponseEntity<ListAclsResponse> responseEntity =
            restTemplate.exchange(
                listAclsUri, HttpMethod.GET, request, new ParameterizedTypeReference<>() {});
  
        List<Map<String, String>> aclsListUpdated = processListAclsResponse(responseEntity);
  
        return new HashSet<>(aclsListUpdated);
      } catch (RestClientException e) {
        log.error("Exception:", e);
        throw new Exception("Error in listing acls : " + e.getMessage());
      }
    }
  
    public Map<String, String> createAcls(@Valid ClusterAclRequest clusterAclRequest)
        throws Exception {
      Map<String, String> resultMap = new HashMap<>();
      RestTemplate restTemplate = getRestTemplate();
      String createAclsUri =
          getResourceUri(
              clusterAclRequest.getClusterName(), clusterAclRequest.getEnv(), "createAcls", "ACLS");
  
      HttpHeaders headers =
          clusterApiUtils.createHeaders(clusterAclRequest.getClusterName(), KafkaClustersType.KAFKA);
  
      if (AclType.PRODUCER.value.equals(clusterAclRequest.getAclType())) {
        // Write on Topic
        applyOperation(
            clusterAclRequest,
            restTemplate,
            createAclsUri,
            headers,
            AclOperation.WRITE.name(),
            ResourceType.TOPIC.name(),
            clusterAclRequest.getTopicName());
        // Describe on Topic
        applyOperation(
            clusterAclRequest,
            restTemplate,
            createAclsUri,
            headers,
            AclOperation.DESCRIBE.name(),
            ResourceType.TOPIC.name(),
            clusterAclRequest.getTopicName());
        // Txn id access
        if (clusterAclRequest.getTransactionalId() != null
            && clusterAclRequest.getTransactionalId().length() > 0) {
          applyOperation(
              clusterAclRequest,
              restTemplate,
              createAclsUri,
              headers,
              AclOperation.WRITE.name(),
              ResourceType.TRANSACTIONAL_ID.name(),
              clusterAclRequest.getTransactionalId());
        }
      } else {
        // Read on Group, Describe on Group, Read on Topic
        applyOperation(
            clusterAclRequest,
            restTemplate,
            createAclsUri,
            headers,
            AclOperation.READ.name(),
            ResourceType.GROUP.name(),
            clusterAclRequest.getConsumerGroup());
        applyOperation(
            clusterAclRequest,
            restTemplate,
            createAclsUri,
            headers,
            AclOperation.DESCRIBE.name(),
            ResourceType.GROUP.name(),
            clusterAclRequest.getConsumerGroup());
        applyOperation(
            clusterAclRequest,
            restTemplate,
            createAclsUri,
            headers,
            AclOperation.READ.name(),
            ResourceType.TOPIC.name(),
            clusterAclRequest.getTopicName());
      }
      resultMap.put("result", ApiResultStatus.SUCCESS.value);
      return resultMap;
    }
  
    public String deleteAcls(ClusterAclRequest clusterAclRequest) throws Exception {
      RestTemplate restTemplate = getRestTemplate();
      log.info("deleteAcls {}", clusterAclRequest);
  
      String baseAclsUri =
          getResourceUri(
              clusterAclRequest.getClusterName(), clusterAclRequest.getEnv(), "deleteAcls", "ACLS");
  
      HttpHeaders headers =
          clusterApiUtils.createHeaders(clusterAclRequest.getClusterName(), KafkaClustersType.KAFKA);
      HttpEntity<String> request = new HttpEntity<>(headers);
  
      if (clusterAclRequest.getAclType().equals(AclType.PRODUCER.value)) {
        // delete WRITE on Topic acls
        String deleteAclsUri =
            updateQueryParams(
                clusterAclRequest,
                ResourceType.TOPIC.name(),
                clusterAclRequest.getTopicName(),
                AclOperation.WRITE.name(),
                baseAclsUri);
        deleteAclsRestCall(restTemplate, deleteAclsUri, clusterAclRequest, request);
  
        // delete DESCRIBE on Topic acls
        deleteAclsUri =
            updateQueryParams(
                clusterAclRequest,
                ResourceType.TOPIC.name(),
                clusterAclRequest.getTopicName(),
                AclOperation.DESCRIBE.name(),
                baseAclsUri);
        deleteAclsRestCall(restTemplate, deleteAclsUri, clusterAclRequest, request);
      } else {
        // delete consumer group read acls
        String deleteAclsUri =
            updateQueryParams(
                clusterAclRequest,
                ResourceType.GROUP.name(),
                clusterAclRequest.getConsumerGroup(),
                AclOperation.READ.name(),
                baseAclsUri);
        deleteAclsRestCall(restTemplate, deleteAclsUri, clusterAclRequest, request);
  
        // delete consumer group DESCRIBE acls
        deleteAclsUri =
            updateQueryParams(
                clusterAclRequest,
                ResourceType.GROUP.name(),
                clusterAclRequest.getConsumerGroup(),
                AclOperation.DESCRIBE.name(),
                baseAclsUri);
        deleteAclsRestCall(restTemplate, deleteAclsUri, clusterAclRequest, request);
  
        // delete read topic acls
        deleteAclsUri =
            updateQueryParams(
                clusterAclRequest,
                ResourceType.TOPIC.name(),
                clusterAclRequest.getTopicName(),
                AclOperation.READ.name(),
                baseAclsUri);
        deleteAclsRestCall(restTemplate, deleteAclsUri, clusterAclRequest, request);
      }
  
      return ApiResultStatus.SUCCESS.value;
    }
  
    public ApiResponse createTopic(ClusterTopicRequest clusterTopicRequest) throws Exception {
      RestTemplate restTemplate = getRestTemplate();
      log.info(
          "createTopic {} {} {}",
          clusterTopicRequest.getEnv(),
          clusterTopicRequest.getProtocol(),
          clusterTopicRequest.getClusterName());
  
      String createTopicsUri =
          getResourceUri(
              clusterTopicRequest.getClusterName(),
              clusterTopicRequest.getEnv(),
              "createTopic",
              ResourceType.TOPIC.name());
  
      TopicCreateRequest topicCreateReq = getTopicCreateObj(clusterTopicRequest);
      HttpHeaders headers =
          clusterApiUtils.createHeaders(
              clusterTopicRequest.getClusterName(), KafkaClustersType.KAFKA);
      HttpEntity<TopicCreateRequest> request = new HttpEntity<>(topicCreateReq, headers);
  
      try {
        restTemplate.postForEntity(createTopicsUri, request, String.class);
      } catch (Exception e) {
        log.error(
            "Unable to create topic {}, {}", clusterTopicRequest.getTopicName(), e.getMessage());
        if (e.getMessage().contains("already exists")) {
          log.warn(
              "Topic: {} already exists in {}",
              clusterTopicRequest.getTopicName(),
              clusterTopicRequest.getEnv());
          return ApiResponse.builder().success(false).message(e.getMessage()).build();
        }
        throw e;
      }
  
      return ApiResponse.builder().success(true).message(ApiResultStatus.SUCCESS.value).build();
    }
  
    public ApiResponse deleteTopic(ClusterTopicRequest clusterTopicRequest) throws Exception {
      RestTemplate restTemplate = getRestTemplate();
      log.info("createTopic {}", clusterTopicRequest);
  
      String deleteTopicsUri =
          getResourceUri(
              clusterTopicRequest.getClusterName(),
              clusterTopicRequest.getEnv(),
              "deleteTopic",
              ResourceType.TOPIC.name());
      deleteTopicsUri = deleteTopicsUri + "/" + clusterTopicRequest.getTopicName();
  
      HttpHeaders headers =
          clusterApiUtils.createHeaders(
              clusterTopicRequest.getClusterName(), KafkaClustersType.KAFKA);
      HttpEntity<String> request = new HttpEntity<>(headers);
  
      try {
        restTemplate.exchange(deleteTopicsUri, HttpMethod.DELETE, request, String.class);
      } catch (Exception e) {
        log.error(
            "Unable to delete topic {}, {}", clusterTopicRequest.getTopicName(), e.getMessage());
        if (e.getMessage().contains("This server does not host this topic")) {
          log.warn(
              "Topic: {} do not exist in {}",
              clusterTopicRequest.getTopicName(),
              clusterTopicRequest.getEnv());
          return ApiResponse.builder().success(false).message(e.getMessage()).build();
        }
        throw e;
      }
  
      return ApiResponse.builder().success(true).message(ApiResultStatus.SUCCESS.value).build();
    }
  
    // Confluent cloud doesn't provide api to update partitions/config of a topic. so either delete
    // the topic and recreate. Or this functionality is not provided by klaw.
    public ApiResponse updateTopic(ClusterTopicRequest clusterTopicRequest) throws Exception {
      //    deleteTopic(clusterTopicRequest);
      //    createTopic(clusterTopicRequest);
      return ApiResponse.builder().success(false).message(ApiResultStatus.FAILURE.value).build();
    }
  
    String updateQueryParams(
        ClusterAclRequest clusterAclRequest,
        String resourceType,
        String resourceName,
        String operation,
        String baseAclsUri) {
      DefaultUriBuilderFactory uriBuilderFactory = new DefaultUriBuilderFactory(baseAclsUri);
      uriBuilderFactory.setEncodingMode(DefaultUriBuilderFactory.EncodingMode.NONE);
      return uriBuilderFactory
          .builder()
          .queryParam("resource_type", resourceType)
          .queryParam("resource_name", resourceName)
          .queryParam(
              "pattern_type",
              (clusterAclRequest.isPrefixAcl() ? AclPatternType.PREFIXED : AclPatternType.LITERAL))
          .queryParam(
              "principal",
              (clusterAclRequest.getAclIpPrincipleType().equals(AclIPPrincipleType.PRINCIPAL.name())
                  ? "User:" + clusterAclRequest.getAclSsl()
                  : "User:*"))
          .queryParam(
              "host",
              (clusterAclRequest.getAclIpPrincipleType().equals(AclIPPrincipleType.IP_ADDRESS.name())
                  ? clusterAclRequest.getAclIp()
                  : "*"))
          .queryParam("operation", operation)
          .queryParam("permission", AclPermissionType.ALLOW.name())
          .build()
          .toString();
    }
  
    private String getResourceUri(
        String clusterIdentification, String host, String requestMethod, String requestType)
        throws Exception {
      String relevantUri;
      if (ResourceType.TOPIC.name().equals(requestType)) {
        relevantUri = env.getProperty(clusterIdentification.toLowerCase() + TOPIC_API_URI_KEY);
      } else {
        relevantUri = env.getProperty(clusterIdentification.toLowerCase() + ACLS_API_URI_KEY);
      }
  
      if (Strings.isNullOrEmpty(relevantUri)) {
        log.error("Exception: {}", requestMethod + " api/uri for confluent cloud is not configured");
        throw new Exception(requestMethod + " api/uri for confluent cloud is not configured");
      }
      relevantUri = HTTPS_PREFIX + host + relevantUri;
  
      return relevantUri;
    }
  
    private String deleteAclsRestCall(
        RestTemplate restTemplate,
        String deleteAclsUri,
        ClusterAclRequest clusterAclRequest,
        HttpEntity<String> request) {
      try {
        restTemplate.exchange(deleteAclsUri, HttpMethod.DELETE, request, String.class);
      } catch (Exception e) {
        log.error("Unable to delete topic {}, {}", clusterAclRequest.getTopicName(), e.getMessage());
        if (e.getMessage().contains("This server does not host this topic")) {
          log.warn("Acl: {} does not exist", clusterAclRequest);
          return e.getMessage();
        }
        throw e;
      }
  
      return ApiResultStatus.SUCCESS.value;
    }
  
    private void applyOperation(
        ClusterAclRequest clusterAclRequest,
        RestTemplate restTemplate,
        String createAclsUri,
        HttpHeaders headers,
        String operation,
        String resourceType,
        String resourceName) {
      Map<String, String> aclMap;
      aclMap = new HashMap<>();
      aclMap.put("operation", operation);
      aclMap.put("resource_type", resourceType);
      aclMap.put("resource_name", resourceName);
  
      updateAclMap(clusterAclRequest, aclMap);
      HttpEntity<Map<String, String>> request = new HttpEntity<>(aclMap, headers);
  
      // WRITE operation
      createAclsPostEntity(restTemplate, createAclsUri, request);
    }
  
    private void createAclsPostEntity(
        RestTemplate restTemplate, String createAclsUri, HttpEntity<Map<String, String>> request) {
      try {
        restTemplate.postForEntity(createAclsUri, request, String.class);
      } catch (Exception e) {
        log.error("Exception in creating acls : ", e);
        throw e;
      }
    }
  
    TopicCreateRequest getTopicCreateObj(ClusterTopicRequest clusterTopicRequest) {
      TopicCreateRequest topicObject = new TopicCreateRequest();
      topicObject.setTopic_name(clusterTopicRequest.getTopicName());
      topicObject.setPartitions_count(clusterTopicRequest.getPartitions());
      topicObject.setReplication_factor(clusterTopicRequest.getReplicationFactor());
      ArrayList<Config> configs = new ArrayList<>();
      if (clusterTopicRequest.getAdvancedTopicConfiguration() != null) {
        for (String key : clusterTopicRequest.getAdvancedTopicConfiguration().keySet()) {
          Config config = new Config();
          config.setName(key);
          config.setValue(clusterTopicRequest.getAdvancedTopicConfiguration().get(key));
          configs.add(config);
        }
      }
      topicObject.setConfigs(configs);
  
      return topicObject;
    }
  
    Map<String, String> updateAclMap(
        ClusterAclRequest clusterAclRequest, Map<String, String> aclMap) {
      if (clusterAclRequest.isPrefixAcl()) {
        aclMap.put("pattern_type", AclPatternType.PREFIXED.value);
      } else {
        aclMap.put("pattern_type", AclPatternType.LITERAL.value);
      }
  
      if (AclIPPrincipleType.PRINCIPAL.name().equals(clusterAclRequest.getAclIpPrincipleType())) {
        aclMap.put("principal", "User:" + clusterAclRequest.getAclSsl());
        aclMap.put("host", "*");
      } else {
        aclMap.put("principal", "User:*");
        aclMap.put("host", clusterAclRequest.getAclIp());
      }
  
      if (clusterAclRequest.getRequestOperationType().equals(RequestOperationType.CREATE)) {
        aclMap.put("permission", AclPermissionType.ALLOW.name());
      }
      return aclMap;
    }
  
    private List<TopicConfig> processListTopicsResponse(
        ResponseEntity<ListTopicsResponse> responseEntity) {
      ListTopicsResponse topicsList = Objects.requireNonNull(responseEntity.getBody());
      List<TopicConfig> topicsListUpdated = new ArrayList<>();
      for (TopicObject topicObject : topicsList.data) {
        TopicConfig topicsMapUpdated = new TopicConfig();
        topicsMapUpdated.setTopicName(topicObject.topic_name);
        topicsMapUpdated.setReplicationFactor("" + topicObject.replication_factor);
        topicsMapUpdated.setPartitions("" + topicObject.partitions_count);
  
        topicsListUpdated.add(topicsMapUpdated);
      }
      return topicsListUpdated;
    }
  
    private List<Map<String, String>> processListAclsResponse(
        ResponseEntity<ListAclsResponse> responseEntity) {
      ListAclsResponse aclsList = Objects.requireNonNull(responseEntity.getBody());
      List<Map<String, String>> aclsListUpdated = new ArrayList<>();
      for (AclObject aclObject : aclsList.data) {
        Map<String, String> aclsMapUpdated = new HashMap<>();
        aclsMapUpdated.put("operation", aclObject.operation.toUpperCase()); // DESCRIBE/READ/ALL..
        aclsMapUpdated.put("resourceType", aclObject.resource_type); // TOPIC/GROUP/CLUSTER..
        aclsMapUpdated.put(
            "resourceName", aclObject.resource_name); // topic-name, consumergroupname..
        aclsMapUpdated.put("principle", aclObject.principal); // User:*/username/ssldn..
        aclsMapUpdated.put("host", aclObject.host); // ipaddress/*..
        aclsMapUpdated.put("permissionType", aclObject.permission); // ALLOW/DENY..
  
        aclsListUpdated.add(aclsMapUpdated);
      }
      return aclsListUpdated;
    }
  
    private RestTemplate getRestTemplate() {
      return clusterApiUtils.getRequestDetails("", KafkaSupportedProtocol.SSL).getRight();
    }
  }
  