content: package io.aiven.klaw.clusterapi.services;
  
  import static io.aiven.klaw.clusterapi.models.error.ClusterApiErrorMessages.CLUSTER_API_ERR_1;
  import static io.aiven.klaw.clusterapi.models.error.ClusterApiErrorMessages.CLUSTER_API_ERR_2;
  import static io.aiven.klaw.clusterapi.models.error.ClusterApiErrorMessages.CLUSTER_API_ERR_3;
  
  import io.aiven.klaw.clusterapi.models.ApiResponse;
  import io.aiven.klaw.clusterapi.models.ClusterConnectorRequest;
  import io.aiven.klaw.clusterapi.models.connect.ConnectorState;
  import io.aiven.klaw.clusterapi.models.connect.ConnectorsStatus;
  import io.aiven.klaw.clusterapi.models.connect.Status;
  import io.aiven.klaw.clusterapi.models.enums.ClusterStatus;
  import io.aiven.klaw.clusterapi.models.enums.KafkaClustersType;
  import io.aiven.klaw.clusterapi.models.enums.KafkaSupportedProtocol;
  import io.aiven.klaw.clusterapi.models.error.RestErrorResponse;
  import io.aiven.klaw.clusterapi.utils.ClusterApiUtils;
  import java.util.*;
  import lombok.extern.slf4j.Slf4j;
  import org.apache.commons.lang3.StringUtils;
  import org.apache.commons.lang3.tuple.Pair;
  import org.springframework.core.ParameterizedTypeReference;
  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.HttpClientErrorException;
  import org.springframework.web.client.HttpServerErrorException;
  import org.springframework.web.client.HttpStatusCodeException;
  import org.springframework.web.client.RestClientException;
  import org.springframework.web.client.RestTemplate;
  
  @Service
  @Slf4j
  public class KafkaConnectService {
  
    private static final ParameterizedTypeReference<List<String>> GET_CONNECTORS_STR_TYPEREF =
        new ParameterizedTypeReference<>() {};
    private static final ParameterizedTypeReference<Map<String, Object>>
        GET_CONNECTOR_DETAILS_TYPEREF = new ParameterizedTypeReference<>() {};
    public static final String FAILED_STATUS = "FAILED";
    public static final String RUNNING_STATUS = "RUNNING";
    public static final String CONNECTOR_URI_EXPAND_STATUS = "?expand=status";
  
    final ClusterApiUtils clusterApiUtils;
  
    public KafkaConnectService(ClusterApiUtils clusterApiUtils) {
      this.clusterApiUtils = clusterApiUtils;
    }
  
    public ApiResponse deleteConnector(ClusterConnectorRequest clusterConnectorRequest) {
      Set<String> errMsgResponse = new HashSet<>();
      for (String envUrl : getEnvironment(clusterConnectorRequest.getEnv())) {
        try {
          log.info("Into deleteConnector {}", clusterConnectorRequest);
          String suffixUrl = envUrl + "/connectors/" + clusterConnectorRequest.getConnectorName();
          Pair<String, RestTemplate> reqDetails =
              clusterApiUtils.getRequestDetails(suffixUrl, clusterConnectorRequest.getProtocol());
          HttpHeaders headers =
              clusterApiUtils.createHeaders(
                  clusterConnectorRequest.getClusterIdentification(),
                  KafkaClustersType.KAFKA_CONNECT);
          HttpEntity<Object> request = new HttpEntity<>(headers);
  
          reqDetails
              .getRight()
              .exchange(
                  reqDetails.getLeft(),
                  HttpMethod.DELETE,
                  request,
                  new ParameterizedTypeReference<>() {});
          return ApiResponse.SUCCESS;
        } catch (HttpServerErrorException | HttpClientErrorException e) {
          log.error("Rest Exception in deleting connector ", e);
          errMsgResponse.add(getErrorMsgeFromRestException(e, CLUSTER_API_ERR_3));
        } catch (RestClientException ex) {
          log.error("Error in deleting connector ", ex);
          errMsgResponse.add(CLUSTER_API_ERR_3);
        }
      }
      return ApiResponse.notOk(StringUtils.join(errMsgResponse, ", "));
    }
  
    public ApiResponse updateConnector(ClusterConnectorRequest clusterConnectorRequest) {
      Set<String> errMsgResponse = new HashSet<>();
      for (String envUrl : getEnvironment(clusterConnectorRequest.getEnv())) {
        try {
          log.info("Into updateConnector {}", clusterConnectorRequest);
          String suffixUrl =
              envUrl + "/connectors/" + clusterConnectorRequest.getConnectorName() + "/config";
          Pair<String, RestTemplate> reqDetails =
              clusterApiUtils.getRequestDetails(suffixUrl, clusterConnectorRequest.getProtocol());
  
          HttpHeaders headers =
              createKafkaConnectHeaders(clusterConnectorRequest.getClusterIdentification());
          HttpEntity<String> request =
              new HttpEntity<>(clusterConnectorRequest.getConnectorConfig(), headers);
  
          reqDetails.getRight().put(reqDetails.getLeft(), request, String.class);
          return ApiResponse.SUCCESS;
        } catch (HttpServerErrorException | HttpClientErrorException e) {
          log.error("Error in updating connector ", e);
          errMsgResponse.add(getErrorMsgeFromRestException(e, CLUSTER_API_ERR_2));
        } catch (Exception ex) {
          log.error("Error in updating connector ", ex);
          errMsgResponse.add(CLUSTER_API_ERR_2);
        }
      }
      return ApiResponse.notOk(StringUtils.join(errMsgResponse, ", "));
    }
  
    private HttpHeaders createKafkaConnectHeaders(String clusterIdentification) {
      HttpHeaders headers =
          clusterApiUtils.createHeaders(clusterIdentification, KafkaClustersType.KAFKA_CONNECT);
      headers.set("Content-Type", "application/json");
      return headers;
    }
  
    private static String getErrorMsgeFromRestException(
        HttpStatusCodeException e, String defaultErrorMsg) {
      RestErrorResponse errorResponse = null;
      try {
        errorResponse = e.getResponseBodyAs(RestErrorResponse.class);
      } catch (Exception ex) {
        log.error("Error caught trying to process the error response. ", ex);
      }
      return errorResponse == null ? defaultErrorMsg : errorResponse.getMessage();
    }
  
    public ApiResponse postNewConnector(ClusterConnectorRequest clusterConnectorRequest) {
      log.info("Into postNewConnector clusterConnectorRequest {} ", clusterConnectorRequest);
      Set<String> errMsgResponse = new HashSet<>();
      ResponseEntity<String> responseNew = null;
      for (String envUrl : getEnvironment(clusterConnectorRequest.getEnv())) {
        try {
          String suffixUrl = envUrl + "/connectors";
          Pair<String, RestTemplate> reqDetails =
              clusterApiUtils.getRequestDetails(suffixUrl, clusterConnectorRequest.getProtocol());
  
          HttpHeaders headers =
              createKafkaConnectHeaders(clusterConnectorRequest.getClusterIdentification());
  
          HttpEntity<String> request =
              new HttpEntity<>(clusterConnectorRequest.getConnectorConfig(), headers);
  
          responseNew =
              reqDetails.getRight().postForEntity(reqDetails.getLeft(), request, String.class);
          if (responseNew.getStatusCode().is2xxSuccessful()) {
            return ApiResponse.SUCCESS;
          }
        } catch (HttpServerErrorException | HttpClientErrorException e) {
          log.error("postNewConnector Rest Exception", e);
          errMsgResponse.add(getErrorMsgeFromRestException(e, CLUSTER_API_ERR_1));
        } catch (Exception ex) {
          log.error("postNewConnector Error", ex);
          errMsgResponse.add(CLUSTER_API_ERR_1);
        }
      }
      return processResponseBody(
          responseNew, ApiResponse.notOk(StringUtils.join(errMsgResponse, ", ")));
    }
  
    public ConnectorsStatus getConnectors(
        String environmentVal,
        KafkaSupportedProtocol protocol,
        String clusterIdentification,
        boolean getConnectorStatuses) {
      ConnectorsStatus connectorsStatus = new ConnectorsStatus();
      List<ConnectorState> connectorStateList = new ArrayList<>();
      connectorsStatus.setConnectorStateList(connectorStateList);
      for (String envUrl : getEnvironment(environmentVal)) {
        try {
          log.info("Into getConnectors {} {}", environmentVal, protocol);
          if (envUrl == null) {
            return null;
          }
  
          String suffixUrl = envUrl + "/connectors";
  
          if (getConnectorStatuses) {
            suffixUrl = suffixUrl + CONNECTOR_URI_EXPAND_STATUS;
          }
  
          Pair<String, RestTemplate> reqDetails =
              clusterApiUtils.getRequestDetails(suffixUrl, protocol);
  
          HttpHeaders headers =
              clusterApiUtils.createHeaders(clusterIdentification, KafkaClustersType.KAFKA_CONNECT);
          HttpEntity<Object> request = new HttpEntity<>(headers);
          Map<String, String> params = new HashMap<>();
  
          if (!getConnectorStatuses) {
            ResponseEntity<List<String>> responseList =
                reqDetails
                    .getRight()
                    .exchange(
                        reqDetails.getLeft(),
                        HttpMethod.GET,
                        request,
                        GET_CONNECTORS_STR_TYPEREF,
                        params);
            log.info("connectors list " + responseList);
            if (responseList.getBody() != null) {
              for (String connectorName : responseList.getBody()) {
                ConnectorState connectorState = new ConnectorState();
                connectorState.setConnectorName(connectorName);
                connectorStateList.add(connectorState);
              }
            }
  
            return connectorsStatus;
          }
  
          ResponseEntity<Map<String, Map<String, Status>>> responseEntity =
              getConnectorStatus(reqDetails, request, params);
          Map<String, Map<String, Status>> responseBody = responseEntity.getBody();
  
          for (String connectorName : Objects.requireNonNull(responseBody).keySet()) {
            Map<String, Status> statusMap = responseBody.get(connectorName);
            Status statusConnector = statusMap.get("status");
            long failedTasksCount =
                statusConnector.getTasks().stream()
                    .filter(task -> task.getState().equals(FAILED_STATUS))
                    .count();
            long runningTasksCount =
                statusConnector.getTasks().stream()
                    .filter(task -> task.getState().equals(RUNNING_STATUS))
                    .count();
  
            ConnectorState connectorState = new ConnectorState();
            connectorState.setConnectorName(connectorName);
            connectorState.setConnectorStatus(statusConnector.getConnector().getState());
            connectorState.setRunningTasks(runningTasksCount);
            connectorState.setFailedTasks(failedTasksCount);
            connectorStateList.add(connectorState);
          }
          connectorsStatus.setConnectorStateList(connectorStateList);
  
          log.info("connectors list " + responseEntity);
          return connectorsStatus;
        } catch (Exception e) {
          log.error("Error in getting connectors " + e);
        }
      }
      return connectorsStatus;
    }
  
    private static ResponseEntity<Map<String, Map<String, Status>>> getConnectorStatus(
        Pair<String, RestTemplate> reqDetails,
        HttpEntity<Object> request,
        Map<String, String> params) {
      return reqDetails
          .getRight()
          .exchange(
              reqDetails.getLeft(),
              HttpMethod.GET,
              request,
              new ParameterizedTypeReference<>() {},
              params);
    }
  
    public Map<String, Object> getConnectorDetails(
        String connector,
        String environmentVal,
        KafkaSupportedProtocol protocol,
        String clusterIdentification) {
      for (String envUrl : getEnvironment(environmentVal)) {
        try {
          log.info("Into getConnectorDetails {} {}", environmentVal, protocol);
          if (envUrl == null) {
            return null;
          }
  
          String suffixUrl = envUrl + "/connectors" + "/" + connector;
          Pair<String, RestTemplate> reqDetails =
              clusterApiUtils.getRequestDetails(suffixUrl, protocol);
  
          HttpHeaders headers =
              clusterApiUtils.createHeaders(clusterIdentification, KafkaClustersType.KAFKA_CONNECT);
          HttpEntity<Object> request = new HttpEntity<>(headers);
          Map<String, String> params = new HashMap<>();
  
          ResponseEntity<Map<String, Object>> responseList =
              reqDetails
                  .getRight()
                  .exchange(
                      reqDetails.getLeft(),
                      HttpMethod.GET,
                      request,
                      GET_CONNECTOR_DETAILS_TYPEREF,
                      params);
          log.info("connectors list " + responseList);
  
          return responseList.getBody();
        } catch (Exception e) {
          log.error("Error in getting connector detail ", e);
        }
      }
      return Collections.emptyMap();
    }
  
    protected ClusterStatus getKafkaConnectStatus(
        String environment, KafkaSupportedProtocol protocol, String clusterIdentification) {
      log.info(
          "env : {} , protocol: {}, clusterIdentification: {}",
          environment,
          protocol,
          clusterIdentification);
      for (String env : getEnvironment(environment)) {
        String suffixUrl = env + "/connectors";
        Pair<String, RestTemplate> reqDetails =
            clusterApiUtils.getRequestDetails(suffixUrl, protocol);
        HttpHeaders headers =
            clusterApiUtils.createHeaders(clusterIdentification, KafkaClustersType.KAFKA_CONNECT);
        HttpEntity<Object> request = new HttpEntity<>(headers);
  
        try {
          reqDetails
              .getRight()
              .exchange(
                  reqDetails.getLeft(),
                  HttpMethod.GET,
                  request,
                  new ParameterizedTypeReference<>() {});
          return ClusterStatus.ONLINE;
        } catch (RestClientException e) {
          log.error("Exception Connectin to {} :", env, e);
        }
      }
      return ClusterStatus.OFFLINE;
    }
  
    public ApiResponse restartConnector(ClusterConnectorRequest clusterConnectorRequest) {
      log.info("Into restartConnector clusterConnectorRequest {} ", clusterConnectorRequest);
      ResponseEntity<String> responseNew = null;
      for (String envUrl : getEnvironment(clusterConnectorRequest.getEnv())) {
        try {
          String suffixUrl =
              envUrl
                  + "/connectors"
                  + "/"
                  + clusterConnectorRequest.getConnectorName()
                  + "/"
                  + "restart";
  
          suffixUrl =
              suffixUrl
                  + "?includeTasks=true&onlyFailed="
                  + clusterConnectorRequest.isIncludeFailedTasksOnly();
  
          Pair<String, RestTemplate> reqDetails =
              clusterApiUtils.getRequestDetails(suffixUrl, clusterConnectorRequest.getProtocol());
  
          HttpHeaders headers =
              createKafkaConnectHeaders(clusterConnectorRequest.getClusterIdentification());
          HttpEntity<String> request = new HttpEntity<>(headers);
  
          responseNew =
              reqDetails.getRight().postForEntity(reqDetails.getLeft(), request, String.class);
        } catch (HttpServerErrorException | HttpClientErrorException e) {
          log.error("restartConnector Rest Exception", e);
        } catch (Exception ex) {
          log.error("restartConnector Error", ex);
        }
      }
      return processResponseBody(responseNew, ApiResponse.notOk("Unable to restart Connector."));
    }
  
    private static ApiResponse processResponseBody(
        ResponseEntity<String> responseNew, ApiResponse customFailureMessage) {
      return responseNew != null && responseNew.getStatusCode().is2xxSuccessful()
          ? ApiResponse.SUCCESS
          : customFailureMessage;
    }
  
    public ApiResponse pauseConnector(ClusterConnectorRequest clusterConnectorRequest) {
      return ApiResponse.notOk("To be implemented");
    }
  
    public ApiResponse resumeConnector(ClusterConnectorRequest clusterConnectorRequest) {
      return ApiResponse.notOk("To be implemented");
    }
  
    public String[] getEnvironment(String environments) {
      return environments.split(",");
    }
  }
  