content: package io.aiven.klaw.service;
  
  import static io.aiven.klaw.error.KlawErrorMessages.KAFKA_CONNECT_SYNC_102;
  import static io.aiven.klaw.error.KlawErrorMessages.KAFKA_CONNECT_SYNC_ERR_101;
  import static io.aiven.klaw.error.KlawErrorMessages.KAFKA_CONNECT_SYNC_ERR_102;
  import static io.aiven.klaw.error.KlawErrorMessages.KAFKA_CONNECT_SYNC_ERR_103;
  import static io.aiven.klaw.error.KlawErrorMessages.SYNC_102;
  import static io.aiven.klaw.error.KlawErrorMessages.SYNC_ERR_101;
  
  import com.fasterxml.jackson.core.JsonProcessingException;
  import com.fasterxml.jackson.databind.ObjectMapper;
  import com.fasterxml.jackson.databind.ObjectWriter;
  import io.aiven.klaw.config.ManageDatabase;
  import io.aiven.klaw.dao.Env;
  import io.aiven.klaw.dao.KwClusters;
  import io.aiven.klaw.dao.KwKafkaConnector;
  import io.aiven.klaw.dao.Team;
  import io.aiven.klaw.error.KlawException;
  import io.aiven.klaw.helpers.Pager;
  import io.aiven.klaw.model.ApiResponse;
  import io.aiven.klaw.model.SyncConnectorUpdates;
  import io.aiven.klaw.model.cluster.ConnectorState;
  import io.aiven.klaw.model.cluster.ConnectorsStatus;
  import io.aiven.klaw.model.enums.ApiResultStatus;
  import io.aiven.klaw.model.enums.KafkaClustersType;
  import io.aiven.klaw.model.enums.PermissionType;
  import io.aiven.klaw.model.response.KafkaConnectorModelResponse;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Optional;
  import java.util.stream.Collectors;
  import lombok.extern.slf4j.Slf4j;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.stereotype.Service;
  
  @Service
  @Slf4j
  public class KafkaConnectSyncControllerService {
  
    public static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
    public static final ObjectWriter WRITER_WITH_DEFAULT_PRETTY_PRINTER =
        OBJECT_MAPPER.writerWithDefaultPrettyPrinter();
    @Autowired private CommonUtilsService commonUtilsService;
  
    @Autowired ClusterApiService clusterApiService;
  
    @Autowired private MailUtils mailService;
  
    @Autowired ManageDatabase manageDatabase;
  
    public ApiResponse getConnectorDetails(String connectorName, String envId) throws KlawException {
      int tenantId = commonUtilsService.getTenantId(getUserName());
      KwClusters kwClusters =
          manageDatabase
              .getClusters(KafkaClustersType.KAFKA_CONNECT, tenantId)
              .get(getKafkaConnectorEnvDetails(envId).getClusterId());
  
      Map<String, Object> res =
          clusterApiService.getConnectorDetails(
              connectorName,
              kwClusters.getBootstrapServers(),
              kwClusters.getProtocol(),
              kwClusters.getClusterName() + kwClusters.getClusterId(),
              tenantId);
  
      try {
        String schemaOfObj = OBJECT_MAPPER.writerWithDefaultPrettyPrinter().writeValueAsString(res);
        return ApiResponse.ok(schemaOfObj);
      } catch (JsonProcessingException e) {
        log.error("Exception:", e);
        return ApiResponse.notOk(e.getMessage());
      }
    }
  
    public ApiResponse updateSyncConnectors(List<SyncConnectorUpdates> updatedSyncTopics)
        throws KlawException {
      log.info("updateSyncConnectors {}", updatedSyncTopics);
      String userName = getUserName();
  
      if (commonUtilsService.isNotAuthorizedUser(
          commonUtilsService.getPrincipal(), PermissionType.SYNC_CONNECTORS)) {
        return ApiResponse.NOT_AUTHORIZED;
      }
  
      // tenant filtering
      int tenantId = commonUtilsService.getTenantId(userName);
      String syncCluster =
          manageDatabase.getTenantConfig().get(tenantId).getBaseSyncKafkaConnectCluster();
      String orderOfEnvs = commonUtilsService.getEnvProperty(tenantId, "ORDER_OF_KAFKA_CONNECT_ENVS");
  
      List<KwKafkaConnector> existingTopics;
      List<KwKafkaConnector> kafkaConnectorList = new ArrayList<>();
      KwKafkaConnector t;
  
      StringBuilder erroredTopics = new StringBuilder();
      boolean topicsWithDiffTeams = false;
  
      StringBuilder erroredTopicsExist = new StringBuilder();
      boolean topicsDontExistInMainCluster = false;
      int topicId =
          manageDatabase.getHandleDbRequests().getNextConnectorRequestId("CONNECTOR_ID", tenantId);
  
      List<Integer> updatedSyncTopicsDelete = new ArrayList<>();
      updatedSyncTopics = handleConnectorDeletes(updatedSyncTopics, updatedSyncTopicsDelete);
  
      if (updatedSyncTopics.size() > 0) {
        for (SyncConnectorUpdates topicUpdate : updatedSyncTopics) {
          // tenant filtering
          if (!commonUtilsService
              .getEnvsFromUserId(userName)
              .contains(topicUpdate.getEnvSelected())) {
            return ApiResponse.NOT_AUTHORIZED;
          }
          existingTopics = getConnectorsFromName(topicUpdate.getConnectorName(), tenantId);
  
          if (existingTopics != null) {
            for (KwKafkaConnector existingTopic : existingTopics) {
              if (Objects.equals(existingTopic.getEnvironment(), syncCluster)) {
                if (!Objects.equals(
                        manageDatabase.getTeamNameFromTeamId(tenantId, existingTopic.getTeamId()),
                        topicUpdate.getTeamSelected())
                    && !Objects.equals(topicUpdate.getEnvSelected(), syncCluster)) {
                  erroredTopics.append(topicUpdate.getConnectorName()).append(" ");
                  topicsWithDiffTeams = true;
                }
                break;
              }
            }
          } else if (!Objects.equals(topicUpdate.getEnvSelected(), syncCluster)) {
            erroredTopicsExist.append(topicUpdate.getConnectorName()).append(" ");
            if (checkInPromotionOrder(topicUpdate.getEnvSelected(), orderOfEnvs))
              topicsDontExistInMainCluster = true;
          }
  
          String connectorConfig;
          try {
            connectorConfig =
                getConnectorConfiguration(
                    topicUpdate.getConnectorName(), topicUpdate.getEnvSelected(), tenantId);
          } catch (KlawException | JsonProcessingException e) {
            log.error("Exception:", e);
            return ApiResponse.notOk(
                String.format(KAFKA_CONNECT_SYNC_ERR_101, topicUpdate.getConnectorName()));
          }
  
          boolean topicAdded = false;
          if (existingTopics == null) {
            t = new KwKafkaConnector();
  
            topicId = topicId + 1;
            t.setConnectorId(topicId);
            t.setConnectorName(topicUpdate.getConnectorName());
            t.setConnectorConfig(connectorConfig);
            t.setEnvironment(topicUpdate.getEnvSelected());
            t.setTeamId(
                manageDatabase.getTeamIdFromTeamName(tenantId, topicUpdate.getTeamSelected()));
            t.setDescription(KAFKA_CONNECT_SYNC_102);
            t.setExistingConnector(false);
            t.setTenantId(tenantId);
  
            kafkaConnectorList.add(t);
          } else {
            for (KwKafkaConnector existingTopic : existingTopics) {
              if (Objects.equals(existingTopic.getEnvironment(), topicUpdate.getEnvSelected())) {
                t = existingTopic;
                t.setTeamId(
                    manageDatabase.getTeamIdFromTeamName(tenantId, topicUpdate.getTeamSelected()));
                t.setConnectorName(topicUpdate.getConnectorName());
                t.setConnectorConfig(connectorConfig);
                t.setEnvironment(existingTopic.getEnvironment());
                t.setExistingConnector(true);
                t.setTenantId(tenantId);
                kafkaConnectorList.add(t);
                topicAdded = true;
              } else if (!Objects.equals(
                  existingTopic.getTeamId(),
                  manageDatabase.getTeamIdFromTeamName(tenantId, topicUpdate.getTeamSelected()))) {
                t = existingTopic;
                t.setTeamId(
                    manageDatabase.getTeamIdFromTeamName(tenantId, topicUpdate.getTeamSelected()));
                t.setConnectorName(topicUpdate.getConnectorName());
                t.setConnectorConfig(connectorConfig);
                t.setEnvironment(existingTopic.getEnvironment());
                t.setExistingConnector(true);
                t.setTenantId(tenantId);
                kafkaConnectorList.add(t);
                topicAdded = true;
              }
            }
          }
  
          boolean envFound = false;
          if (existingTopics != null) {
            for (KwKafkaConnector existingTopic : existingTopics) {
              if (Objects.equals(existingTopic.getEnvironment(), topicUpdate.getEnvSelected())) {
                envFound = true;
                break;
              }
            }
            if (!envFound && !topicAdded) {
              t = new KwKafkaConnector();
              topicId = topicId + 1;
              t.setConnectorId(topicId);
              t.setConnectorName(topicUpdate.getConnectorName());
              t.setConnectorConfig(connectorConfig);
              t.setEnvironment(topicUpdate.getEnvSelected());
              t.setTeamId(
                  manageDatabase.getTeamIdFromTeamName(tenantId, topicUpdate.getTeamSelected()));
              t.setDescription(KAFKA_CONNECT_SYNC_102);
              t.setExistingConnector(false);
              t.setTenantId(tenantId);
  
              kafkaConnectorList.add(t);
            }
          }
        }
      }
  
      if (updatedSyncTopics.size() == 0 && updatedSyncTopicsDelete.size() > 0) {
        return ApiResponse.SUCCESS;
      }
  
      if (topicsDontExistInMainCluster) {
        return ApiResponse.notOk(
            KAFKA_CONNECT_SYNC_ERR_102 + " :" + syncCluster + ". \n Topics : " + erroredTopicsExist);
      }
  
      if (topicsWithDiffTeams) {
        return ApiResponse.notOk(
            KAFKA_CONNECT_SYNC_ERR_103 + " :" + syncCluster + ". \n Topics : " + erroredTopics);
      }
  
      if (kafkaConnectorList.size() > 0) {
        try {
          String result =
              manageDatabase.getHandleDbRequests().addToSyncConnectors(kafkaConnectorList);
          return ApiResultStatus.SUCCESS.value.equals(result)
              ? ApiResponse.ok(result)
              : ApiResponse.notOk(result);
        } catch (Exception e) {
          throw new KlawException(e.getMessage());
        }
      } else {
        return ApiResponse.notOk(SYNC_ERR_101);
      }
    }
  
    private String getConnectorConfiguration(String connectorName, String environmentId, int tenantId)
        throws KlawException, JsonProcessingException {
      KwClusters kwClusters =
          manageDatabase
              .getClusters(KafkaClustersType.KAFKA_CONNECT, tenantId)
              .get(getKafkaConnectorEnvDetails(environmentId).getClusterId());
  
      Object configMap =
          clusterApiService
              .getConnectorDetails(
                  connectorName,
                  kwClusters.getBootstrapServers(),
                  kwClusters.getProtocol(),
                  kwClusters.getClusterName() + kwClusters.getClusterId(),
                  tenantId)
              .get("config");
      return WRITER_WITH_DEFAULT_PRETTY_PRINTER.writeValueAsString(configMap);
    }
  
    private List<SyncConnectorUpdates> handleConnectorDeletes(
        List<SyncConnectorUpdates> updatedSyncTopics, List<Integer> updatedSyncConnectorsDelete) {
      List<SyncConnectorUpdates> updatedSyncTopicsUpdated = new ArrayList<>();
      for (SyncConnectorUpdates updatedSyncConn : updatedSyncTopics) {
        if (SYNC_102.equals(updatedSyncConn.getTeamSelected())) {
          updatedSyncConnectorsDelete.add(Integer.parseInt(updatedSyncConn.getConnectorId()));
        } else {
          updatedSyncTopicsUpdated.add(updatedSyncConn);
        }
      }
  
      // delete topic
      for (Integer connectorId : updatedSyncConnectorsDelete) {
        manageDatabase
            .getHandleDbRequests()
            .deleteConnector(connectorId, commonUtilsService.getTenantId(getUserName()));
      }
  
      return updatedSyncTopicsUpdated;
    }
  
    private boolean checkInPromotionOrder(String envId, String orderOfEnvs) {
      List<String> orderedEnv = Arrays.asList(orderOfEnvs.split(","));
      return orderedEnv.contains(envId);
    }
  
    public List<KwKafkaConnector> getConnectorsFromName(String connectorName, int tenantId) {
      return manageDatabase.getHandleDbRequests().getConnectorsFromName(connectorName, tenantId);
    }
  
    public List<KafkaConnectorModelResponse> getSyncConnectors(
        String envId,
        String pageNo,
        String currentPage,
        String connectorNameSearch,
        boolean getConnectorsStatuses) {
      Env envSelected = getKafkaConnectorEnvDetails(envId);
      List<String> teamList = new ArrayList<>();
      teamList = tenantFilterTeams(teamList);
      int tenantId = commonUtilsService.getTenantId(getUserName());
  
      // get from metastore
      List<KafkaConnectorModelResponse> kafkaConnectorModelSourceList =
          getSyncConnectorsList(envId, teamList, tenantId);
      if (connectorNameSearch != null && connectorNameSearch.length() > 0) {
        final String topicSearchFilter = connectorNameSearch;
        kafkaConnectorModelSourceList =
            kafkaConnectorModelSourceList.stream()
                .filter(topic -> topic.getConnectorName().contains(topicSearchFilter))
                .collect(Collectors.toList());
      }
      List<String> allSyncConnectors = new ArrayList<>();
      for (KafkaConnectorModelResponse kafkaConnectorModel : kafkaConnectorModelSourceList) {
        allSyncConnectors.add(kafkaConnectorModel.getConnectorName());
      }
  
      // get from cluster
      List<KafkaConnectorModelResponse> kafkaConnectorModelClusterList = new ArrayList<>();
      KwClusters kwClusters =
          manageDatabase
              .getClusters(KafkaClustersType.KAFKA_CONNECT, tenantId)
              .get(envSelected.getClusterId());
      String bootstrapHost = kwClusters.getBootstrapServers();
      try {
        ConnectorsStatus allConnectors =
            clusterApiService.getAllKafkaConnectors(
                bootstrapHost,
                kwClusters.getProtocol().getName(),
                kwClusters.getClusterName() + kwClusters.getClusterId(),
                tenantId,
                getConnectorsStatuses);
        List<ConnectorState> connectorsList = allConnectors.getConnectorStateList();
        if (connectorNameSearch != null && connectorNameSearch.length() > 0) {
          final String topicSearchFilter = connectorNameSearch;
          connectorsList =
              connectorsList.stream()
                  .filter(
                      connectorState -> connectorState.getConnectorName().contains(topicSearchFilter))
                  .toList();
        }
  
        for (ConnectorState connectorState : connectorsList) {
          KafkaConnectorModelResponse kafkaConnectorModel = new KafkaConnectorModelResponse();
          kafkaConnectorModel.setConnectorName(connectorState.getConnectorName());
          kafkaConnectorModel.setRunningTasks(connectorState.getRunningTasks());
          kafkaConnectorModel.setConnectorStatus(connectorState.getConnectorStatus());
          kafkaConnectorModel.setFailedTasks(connectorState.getFailedTasks());
          kafkaConnectorModel.setEnvironmentId(envId);
          kafkaConnectorModel.setEnvironmentName(getKafkaConnectorEnvDetails(envId).getName());
          kafkaConnectorModel.setPossibleTeams(teamList);
  
          kafkaConnectorModelClusterList.add(kafkaConnectorModel);
        }
  
        // remove if any which already exist in metastore
        kafkaConnectorModelClusterList.removeIf(
            p -> !allSyncConnectors.isEmpty() && allSyncConnectors.contains(p.getConnectorName()));
  
        for (KafkaConnectorModelResponse kafkaConnectorModelResponse :
            kafkaConnectorModelSourceList) {
          kafkaConnectorModelResponse.setRemarks("IN_SYNC"); // default
          Optional<ConnectorState> optionalConnectorState =
              connectorsList.stream()
                  .filter(
                      connectorState ->
                          connectorState
                              .getConnectorName()
                              .equals(kafkaConnectorModelResponse.getConnectorName()))
                  .findFirst();
          if (optionalConnectorState.isPresent()) {
            ConnectorState connectorState = optionalConnectorState.get();
            kafkaConnectorModelResponse.setConnectorStatus(connectorState.getConnectorStatus());
            kafkaConnectorModelResponse.setFailedTasks(connectorState.getFailedTasks());
            kafkaConnectorModelResponse.setRunningTasks(connectorState.getRunningTasks());
          } else {
            kafkaConnectorModelResponse.setConnectorStatus("NOT_KNOWN");
          }
        }
  
        kafkaConnectorModelClusterList.addAll(kafkaConnectorModelSourceList);
  
        for (KafkaConnectorModelResponse kafkaConnectorModel : kafkaConnectorModelSourceList) {
          if (connectorsList.stream()
              .noneMatch(
                  connectorState ->
                      connectorState
                          .getConnectorName()
                          .equals(kafkaConnectorModel.getConnectorName()))) {
            for (KafkaConnectorModelResponse kafkaConnectorModelCluster :
                kafkaConnectorModelClusterList) {
              if (Objects.equals(
                  kafkaConnectorModelCluster.getConnectorName(),
                  kafkaConnectorModel.getConnectorName())) {
                kafkaConnectorModelCluster.setRemarks("DELETED");
                // Remove the other team options added and replace with existing team and option to
                // remove it.
                List<String> possibleTeams = new ArrayList<>();
                possibleTeams.add(kafkaConnectorModelCluster.getTeamName());
                possibleTeams.add(SYNC_102);
                kafkaConnectorModelCluster.setPossibleTeams(possibleTeams);
              }
            }
          }
        }
  
        // set sequence
        int i = 0;
        for (KafkaConnectorModelResponse kafkaConnectorModel : kafkaConnectorModelClusterList) {
          kafkaConnectorModel.setSequence(i);
          i++;
  
          if (kafkaConnectorModel.getTeamName() == null
              || kafkaConnectorModel.getTeamName().equals("")) {
            kafkaConnectorModel.setRemarks("ADDED");
            kafkaConnectorModel.setTeamName("");
          }
        }
  
        // don't show connectors which are deleted on cluster
        if (getConnectorsStatuses) {
          kafkaConnectorModelClusterList =
              kafkaConnectorModelClusterList.stream()
                  .filter(connectorList -> connectorList.getRemarks().equals("IN_SYNC"))
                  .toList();
        }
  
        // pagination
        return Pager.getItemsList(
            pageNo,
            currentPage,
            kafkaConnectorModelClusterList,
            (pageContext, kafkaConnectorModelResponse) -> {
              kafkaConnectorModelResponse.setAllPageNos(pageContext.getAllPageNos());
              kafkaConnectorModelResponse.setTotalNoPages(pageContext.getTotalPages());
              kafkaConnectorModelResponse.setCurrentPage(currentPage);
              return kafkaConnectorModelResponse;
            });
      } catch (KlawException e) {
        log.error("Exception:", e);
      }
  
      return new ArrayList<>();
    }
  
    private List<KafkaConnectorModelResponse> getSyncConnectorsList(
        String envId, List<String> teamList, int tenantId) {
      //         Get Sync connectors
      List<KwKafkaConnector> connectorsFromSOT =
          manageDatabase.getHandleDbRequests().getSyncConnectors(envId, null, tenantId);
  
      List<KafkaConnectorModelResponse> kafkaConnectorModelSourceList = new ArrayList<>();
  
      for (KwKafkaConnector kwKafkaConnector : connectorsFromSOT) {
        KafkaConnectorModelResponse kafkaConnectorModel = new KafkaConnectorModelResponse();
        kafkaConnectorModel.setEnvironmentName(
            getKafkaConnectorEnvDetails(kwKafkaConnector.getEnvironment()).getName());
        kafkaConnectorModel.setEnvironmentId(kwKafkaConnector.getEnvironment());
        kafkaConnectorModel.setConnectorName(kwKafkaConnector.getConnectorName());
        kafkaConnectorModel.setTeamName(
            manageDatabase.getTeamNameFromTeamId(tenantId, kwKafkaConnector.getTeamId()));
        kafkaConnectorModel.setConnectorId(kwKafkaConnector.getConnectorId());
        kafkaConnectorModel.setPossibleTeams(teamList);
  
        kafkaConnectorModelSourceList.add(kafkaConnectorModel);
      }
  
      return kafkaConnectorModelSourceList;
    }
  
    private List<String> tenantFilterTeams(List<String> teamList) {
      if (!commonUtilsService.isNotAuthorizedUser(
          commonUtilsService.getPrincipal(), PermissionType.SYNC_CONNECTORS)) {
        // tenant filtering
        int tenantId = commonUtilsService.getTenantId(getUserName());
        List<Team> teams = manageDatabase.getHandleDbRequests().getAllTeams(tenantId);
  
        List<String> teamListUpdated = new ArrayList<>();
        for (Team teamsItem : teams) {
          teamListUpdated.add(teamsItem.getTeamname());
        }
        teamList = teamListUpdated;
      }
      return teamList;
    }
  
    private String getUserName() {
      return mailService.getUserName(commonUtilsService.getPrincipal());
    }
  
    public Env getKafkaConnectorEnvDetails(String envId) {
      Optional<Env> envFound =
          manageDatabase
              .getKafkaConnectEnvList(commonUtilsService.getTenantId(getUserName()))
              .stream()
              .filter(env -> Objects.equals(env.getId(), envId))
              .findFirst();
      return envFound.orElse(null);
    }
  
    public List<KafkaConnectorModelResponse> getConnectorsToManage(
        String envId, String pageNo, String currentPage, String connectorNameSearch) {
      return null;
    }
  }
  