content: package io.aiven.klaw.clusterapi.controller;
  
  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.ServiceAccountDetails;
  import io.aiven.klaw.clusterapi.models.enums.AclType;
  import io.aiven.klaw.clusterapi.models.enums.AclsNativeType;
  import io.aiven.klaw.clusterapi.models.enums.ApiResultStatus;
  import io.aiven.klaw.clusterapi.models.enums.ClusterStatus;
  import io.aiven.klaw.clusterapi.models.enums.KafkaSupportedProtocol;
  import io.aiven.klaw.clusterapi.services.AivenApiService;
  import io.aiven.klaw.clusterapi.services.ApacheKafkaAclService;
  import io.aiven.klaw.clusterapi.services.ApacheKafkaTopicService;
  import io.aiven.klaw.clusterapi.services.ConfluentCloudApiService;
  import io.aiven.klaw.clusterapi.services.UtilComponentsService;
  import jakarta.validation.Valid;
  import java.util.HashMap;
  import java.util.Map;
  import java.util.Set;
  import lombok.AllArgsConstructor;
  import lombok.extern.slf4j.Slf4j;
  import org.springframework.http.HttpStatus;
  import org.springframework.http.MediaType;
  import org.springframework.http.ResponseEntity;
  import org.springframework.web.bind.annotation.PathVariable;
  import org.springframework.web.bind.annotation.PostMapping;
  import org.springframework.web.bind.annotation.RequestBody;
  import org.springframework.web.bind.annotation.RequestMapping;
  import org.springframework.web.bind.annotation.RequestMethod;
  import org.springframework.web.bind.annotation.RestController;
  
  @RestController
  @RequestMapping("/topics")
  @Slf4j
  @AllArgsConstructor
  public class ClusterApiController {
  
    UtilComponentsService utilComponentsService;
  
    ApacheKafkaAclService apacheKafkaAclService;
  
    ApacheKafkaTopicService apacheKafkaTopicService;
  
    AivenApiService aivenApiService;
  
    ConfluentCloudApiService confluentCloudApiService;
  
    @RequestMapping(
        value = "/getApiStatus",
        method = RequestMethod.GET,
        produces = {MediaType.APPLICATION_JSON_VALUE})
    public ResponseEntity<ClusterStatus> getApiStatus() {
      return new ResponseEntity<>(ClusterStatus.ONLINE, HttpStatus.OK);
    }
  
    @RequestMapping(
        value =
            "/getStatus/{bootstrapServers}/{protocol}/{clusterName}/{clusterType}/kafkaFlavor/{kafkaFlavor}",
        method = RequestMethod.GET,
        produces = {MediaType.APPLICATION_JSON_VALUE})
    public ResponseEntity<ClusterStatus> getStatus(
        @PathVariable String bootstrapServers,
        @Valid @PathVariable KafkaSupportedProtocol protocol,
        @PathVariable String clusterName,
        @PathVariable String clusterType,
        @PathVariable String kafkaFlavor) {
      return new ResponseEntity<>(
          utilComponentsService.getStatus(
              bootstrapServers, protocol, clusterName, clusterType, kafkaFlavor),
          HttpStatus.OK);
    }
  
    @RequestMapping(
        value =
            "/getTopics/{bootstrapServers}/{protocol}/{clusterName}/topicsNativeType/{aclsNativeType}/resetCache/{resetCache}",
        method = RequestMethod.GET,
        produces = {MediaType.APPLICATION_JSON_VALUE})
    public ResponseEntity<LoadTopicsResponse> getTopics(
        @PathVariable String bootstrapServers,
        @Valid @PathVariable KafkaSupportedProtocol protocol,
        @PathVariable String clusterName,
        @PathVariable String aclsNativeType,
        @PathVariable boolean resetCache)
        throws Exception {
      LoadTopicsResponse loadTopicsResponse;
      if (AclsNativeType.CONFLUENT_CLOUD.name().equals(aclsNativeType)) {
        loadTopicsResponse =
            confluentCloudApiService.listTopics(bootstrapServers, protocol, clusterName);
      } else {
        loadTopicsResponse =
            apacheKafkaTopicService.loadTopics(bootstrapServers, protocol, clusterName, resetCache);
      }
      return new ResponseEntity<>(loadTopicsResponse, HttpStatus.OK);
    }
  
    @RequestMapping(
        value =
            "/getAcls/{bootstrapServers}/{aclsNativeType}/{protocol}/{clusterName}/{projectName}/{serviceName}",
        method = RequestMethod.GET,
        produces = {MediaType.APPLICATION_JSON_VALUE})
    public ResponseEntity<Set<Map<String, String>>> getAcls(
        @PathVariable String bootstrapServers,
        @Valid @PathVariable KafkaSupportedProtocol protocol,
        @PathVariable String clusterName,
        @PathVariable String aclsNativeType,
        @PathVariable String projectName,
        @PathVariable String serviceName)
        throws Exception {
      Set<Map<String, String>> acls;
      if (AclsNativeType.NATIVE.name().equals(aclsNativeType)) {
        acls = apacheKafkaAclService.loadAcls(bootstrapServers, protocol, clusterName);
      } else if (AclsNativeType.CONFLUENT_CLOUD.name().equals(aclsNativeType)) {
        acls = confluentCloudApiService.listAcls(bootstrapServers, protocol, clusterName);
      } else {
        acls = aivenApiService.listAcls(projectName, serviceName);
      }
      return new ResponseEntity<>(acls, HttpStatus.OK);
    }
  
    /*
    Based on the project, service and username, service account details (in Aiven system) are retrieved.
     */
    @RequestMapping(
        value = "/serviceAccountDetails/project/{projectName}/service/{serviceName}/user/{userName}",
        method = RequestMethod.GET,
        produces = {MediaType.APPLICATION_JSON_VALUE})
    public ResponseEntity<ServiceAccountDetails> getServiceAccountCredentials(
        @PathVariable String projectName,
        @PathVariable String serviceName,
        @PathVariable String userName) {
      return new ResponseEntity<>(
          aivenApiService.getServiceAccountDetails(projectName, serviceName, userName),
          HttpStatus.OK);
    }
  
    /*
    Based on the project, service, user service accounts (in Aiven system) are retrieved.
     */
    @RequestMapping(
        value = "/serviceAccounts/project/{projectName}/service/{serviceName}",
        method = RequestMethod.GET,
        produces = {MediaType.APPLICATION_JSON_VALUE})
    public ResponseEntity<ApiResponse> getServiceAccounts(
        @PathVariable String projectName, @PathVariable String serviceName) {
      ApiResponse apiResponse =
          ApiResponse.builder()
              .data(aivenApiService.getServiceAccountUsers(projectName, serviceName))
              .message(ApiResultStatus.SUCCESS.value)
              .success(true)
              .build();
      return new ResponseEntity<>(apiResponse, HttpStatus.OK);
    }
  
    @PostMapping(
        value = "/createTopics",
        produces = {MediaType.APPLICATION_JSON_VALUE})
    public ResponseEntity<ApiResponse> createTopics(
        @RequestBody @Valid ClusterTopicRequest clusterTopicRequest) {
      try {
        log.info("createTopics clusterTopicRequest {}", clusterTopicRequest);
        if (AclsNativeType.CONFLUENT_CLOUD == clusterTopicRequest.getAclsNativeType()) {
          return new ResponseEntity<>(
              confluentCloudApiService.createTopic(clusterTopicRequest), HttpStatus.OK);
        } else {
          return new ResponseEntity<>(
              apacheKafkaTopicService.createTopic(clusterTopicRequest), HttpStatus.OK);
        }
  
      } catch (Exception e) {
        return handleException(e);
      }
    }
  
    @PostMapping(
        value = "/updateTopics",
        produces = {MediaType.APPLICATION_JSON_VALUE})
    public ResponseEntity<ApiResponse> updateTopics(
        @RequestBody @Valid ClusterTopicRequest clusterTopicRequest) {
      try {
        log.info("updateTopics clusterTopicRequest {}", clusterTopicRequest);
        if (AclsNativeType.CONFLUENT_CLOUD == clusterTopicRequest.getAclsNativeType()) {
          return new ResponseEntity<>(
              confluentCloudApiService.updateTopic(clusterTopicRequest), HttpStatus.OK);
        } else {
          return new ResponseEntity<>(
              apacheKafkaTopicService.updateTopic(clusterTopicRequest), HttpStatus.OK);
        }
      } catch (Exception e) {
        return handleException(e);
      }
    }
  
    @PostMapping(
        value = "/deleteTopics",
        produces = {MediaType.APPLICATION_JSON_VALUE})
    public ResponseEntity<ApiResponse> deleteTopics(
        @RequestBody @Valid ClusterTopicRequest clusterTopicRequest) {
      try {
        if (AclsNativeType.CONFLUENT_CLOUD == clusterTopicRequest.getAclsNativeType()) {
          return new ResponseEntity<>(
              confluentCloudApiService.deleteTopic(clusterTopicRequest), HttpStatus.OK);
        } else {
          return new ResponseEntity<>(
              apacheKafkaTopicService.deleteTopic(clusterTopicRequest), HttpStatus.OK);
        }
      } catch (Exception e) {
        return handleException(e);
      }
    }
  
    @PostMapping(
        value = "/createAcls",
        produces = {MediaType.APPLICATION_JSON_VALUE})
    public ResponseEntity<ApiResponse> createAcls(
        @RequestBody @Valid ClusterAclRequest clusterAclRequest) {
  
      Map<String, String> resultMap = new HashMap<>();
      String result;
      try {
        if (AclsNativeType.NATIVE.name().equals(clusterAclRequest.getAclNativeType())) {
          if (AclType.PRODUCER.value.equals(clusterAclRequest.getAclType())) {
            result = apacheKafkaAclService.updateProducerAcl(clusterAclRequest);
          } else {
            result = apacheKafkaAclService.updateConsumerAcl(clusterAclRequest);
          }
          return new ResponseEntity<>(
              ApiResponse.builder()
                  .success(result.equals(ApiResultStatus.SUCCESS.value))
                  .message(result)
                  .build(),
              HttpStatus.OK);
        } else if (AclsNativeType.AIVEN.name().equals(clusterAclRequest.getAclNativeType())) {
          resultMap = aivenApiService.createAcls(clusterAclRequest);
          String resultStatus = resultMap.get("result");
          return new ResponseEntity<>(
              ApiResponse.builder()
                  .success(resultStatus.equals(ApiResultStatus.SUCCESS.value))
                  .message(resultStatus)
                  .data(resultMap)
                  .build(),
              HttpStatus.OK);
        } else if (AclsNativeType.CONFLUENT_CLOUD
            .name()
            .equals(clusterAclRequest.getAclNativeType())) {
          resultMap = confluentCloudApiService.createAcls(clusterAclRequest);
          String resultStatus = resultMap.get("result");
          return new ResponseEntity<>(
              ApiResponse.builder()
                  .success(resultStatus.equals(ApiResultStatus.SUCCESS.value))
                  .message(resultStatus)
                  .data(resultMap)
                  .build(),
              HttpStatus.OK);
        }
      } catch (Exception e) {
        resultMap.put("result", "failure " + e.getMessage());
        return new ResponseEntity<>(
            ApiResponse.builder().success(false).message("failure " + e.getMessage()).build(),
            HttpStatus.INTERNAL_SERVER_ERROR);
      }
      resultMap.put("result", "Not a valid request");
      return new ResponseEntity<>(
          ApiResponse.builder().success(false).message("Not a valid request").build(),
          HttpStatus.INTERNAL_SERVER_ERROR);
    }
  
    @PostMapping(
        value = "/deleteAcls",
        produces = {MediaType.APPLICATION_JSON_VALUE})
    public ResponseEntity<ApiResponse> deleteAcls(
        @RequestBody @Valid ClusterAclRequest clusterAclRequest) {
      String result;
      try {
        if (AclsNativeType.NATIVE.name().equals(clusterAclRequest.getAclNativeType())) {
  
          if (AclType.PRODUCER.value.equals(clusterAclRequest.getAclType())) {
            result = apacheKafkaAclService.updateProducerAcl(clusterAclRequest);
          } else {
            result = apacheKafkaAclService.updateConsumerAcl(clusterAclRequest);
          }
  
          return new ResponseEntity<>(
              ApiResponse.builder()
                  .success(result.equals(ApiResultStatus.SUCCESS.value))
                  .message(result)
                  .build(),
              HttpStatus.OK);
        } else if (AclsNativeType.AIVEN.name().equals(clusterAclRequest.getAclNativeType())) {
          ApiResponse apiResponse = aivenApiService.deleteAcls(clusterAclRequest);
          return new ResponseEntity<>(apiResponse, HttpStatus.OK);
        } else if (AclsNativeType.CONFLUENT_CLOUD
            .name()
            .equals(clusterAclRequest.getAclNativeType())) {
          result = confluentCloudApiService.deleteAcls(clusterAclRequest);
  
          return new ResponseEntity<>(
              ApiResponse.builder()
                  .success(result.equals(ApiResultStatus.SUCCESS.value))
                  .message(result)
                  .build(),
              HttpStatus.OK);
        }
  
      } catch (Exception e) {
        return handleException(e);
      }
      return new ResponseEntity<>(
          ApiResponse.builder().success(false).message("Not a valid request").build(),
          HttpStatus.INTERNAL_SERVER_ERROR);
    }
  
    private static ResponseEntity<ApiResponse> handleException(Exception e) {
      log.error("Exception:", e);
      return new ResponseEntity<>(
          ApiResponse.builder().success(false).message(e.getMessage()).build(),
          HttpStatus.INTERNAL_SERVER_ERROR);
    }
  }
  