package io.aiven.klaw.service;

import static io.aiven.klaw.error.KlawErrorMessages.*;
import static io.aiven.klaw.helpers.KwConstants.ORDER_OF_KAFKA_CONNECT_ENVS;
import static io.aiven.klaw.service.MailUtils.MailType.CONNECTOR_CLAIM_REQUESTED;
import static io.aiven.klaw.service.MailUtils.MailType.CONNECTOR_CREATE_REQUESTED;
import static io.aiven.klaw.service.MailUtils.MailType.CONNECTOR_DELETE_REQUESTED;
import static io.aiven.klaw.service.MailUtils.MailType.CONNECTOR_REQUEST_APPROVED;
import static io.aiven.klaw.service.MailUtils.MailType.CONNECTOR_REQUEST_DENIED;
import static org.springframework.beans.BeanUtils.copyProperties;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.fasterxml.jackson.databind.node.ObjectNode;
import io.aiven.klaw.config.ManageDatabase;
import io.aiven.klaw.dao.Env;
import io.aiven.klaw.dao.KafkaConnectorRequest;
import io.aiven.klaw.dao.KwClusters;
import io.aiven.klaw.dao.KwKafkaConnector;
import io.aiven.klaw.dao.UserInfo;
import io.aiven.klaw.error.KlawBadRequestException;
import io.aiven.klaw.error.KlawException;
import io.aiven.klaw.error.KlawRestException;
import io.aiven.klaw.error.RestErrorResponse;
import io.aiven.klaw.helpers.DisplayHelper;
import io.aiven.klaw.helpers.HandleDbRequests;
import io.aiven.klaw.helpers.HtmlSanitizer;
import io.aiven.klaw.helpers.KlawResourceUtils;
import io.aiven.klaw.helpers.Pager;
import io.aiven.klaw.model.ApiResponse;
import io.aiven.klaw.model.ConnectorConfig;
import io.aiven.klaw.model.ResourceHistory;
import io.aiven.klaw.model.enums.ApiResultStatus;
import io.aiven.klaw.model.enums.KafkaClustersType;
import io.aiven.klaw.model.enums.KafkaSupportedProtocol;
import io.aiven.klaw.model.enums.Order;
import io.aiven.klaw.model.enums.PermissionType;
import io.aiven.klaw.model.enums.PromotionStatusType;
import io.aiven.klaw.model.enums.RequestEntityType;
import io.aiven.klaw.model.enums.RequestOperationType;
import io.aiven.klaw.model.enums.RequestStatus;
import io.aiven.klaw.model.requests.KafkaConnectorModel;
import io.aiven.klaw.model.requests.KafkaConnectorRequestModel;
import io.aiven.klaw.model.requests.KafkaConnectorRestartModel;
import io.aiven.klaw.model.response.*;
import io.aiven.klaw.model.response.ConnectorOverview;
import io.aiven.klaw.model.response.ConnectorOverviewPerEnv;
import io.aiven.klaw.model.response.EnvIdInfo;
import io.aiven.klaw.model.response.KafkaConnectorModelResponse;
import io.aiven.klaw.model.response.KafkaConnectorRequestsResponseModel;
import io.aiven.klaw.validation.PermissionAllowed;
import java.time.Instant;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.jasypt.util.text.BasicTextEncryptor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.web.client.HttpClientErrorException;
import org.springframework.web.client.HttpServerErrorException;
import org.springframework.web.client.HttpStatusCodeException;

@Service
@Slf4j
public class KafkaConnectControllerService {

  public static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
  public static final ObjectWriter WRITER_WITH_DEFAULT_PRETTY_PRINTER =
      OBJECT_MAPPER.writerWithDefaultPrettyPrinter();

  public static final TypeReference<ArrayList<ResourceHistory>> VALUE_TYPE_REF =
      new TypeReference<>() {};

  public static final DateTimeFormatter DATE_TIME_FORMATTER =
      DateTimeFormatter.ofPattern("yyyy-MMM-dd HH:mm:ss");

  @Value("${klaw.connect.sensitive.fields:password}")
  private String kafkaConnectorSensitiveFields;

  @Autowired private CommonUtilsService commonUtilsService;

  @Autowired ClusterApiService clusterApiService;

  @Autowired private MailUtils mailService;

  @Autowired ManageDatabase manageDatabase;

  @Autowired private RolesPermissionsControllerService rolesPermissionsControllerService;

  public ApiResponse createConnectorRequest(KafkaConnectorRequestModel connectorRequestModel)
      throws KlawException {
    log.info("createConnectorRequest {}", connectorRequestModel);
    String userName = getUserName();

    if (commonUtilsService.isNotAuthorizedUser(
        commonUtilsService.getPrincipal(), PermissionType.REQUEST_CREATE_CONNECTORS)) {
      return ApiResponse.NOT_AUTHORIZED;
    }

    try {
      if (connectorRequestModel.getConnectorConfig() != null
          && connectorRequestModel.getConnectorConfig().trim().length() > 0) {
        JsonNode jsonNode =
            OBJECT_MAPPER.readTree(connectorRequestModel.getConnectorConfig().trim());
        updateJsonNode("encrypt", jsonNode);

        if (!jsonNode.has("tasks.max")) {
          return ApiResponse.notOk(KAFKA_CONNECT_ERR_101);
        } else if (!jsonNode.has("connector.class")) {
          return ApiResponse.notOk(KAFKA_CONNECT_ERR_102);
        } else if (!jsonNode.has("topics") && !jsonNode.has("topics.regex")) {
          return ApiResponse.notOk(KAFKA_CONNECT_ERR_103);
        }

        if (jsonNode.has("topics") && jsonNode.has("topics.regex")) {
          return ApiResponse.notOk(KAFKA_CONNECT_ERR_104);
        }

        Map<String, Object> resultMap =
            OBJECT_MAPPER.convertValue(jsonNode, new TypeReference<>() {});
        connectorRequestModel.setConnectorConfig(
            OBJECT_MAPPER.writerWithDefaultPrettyPrinter().writeValueAsString(resultMap));
      }
    } catch (JsonProcessingException e) {
      log.error("Exception:", e);
      throw new KlawException(e.getMessage());
    }

    connectorRequestModel.setRequestor(userName);
    Integer userTeamId = commonUtilsService.getTeamId(userName);
    connectorRequestModel.setTeamId(userTeamId);
    String envSelected = connectorRequestModel.getEnvironment();

    // tenant filtering
    if (!commonUtilsService.getEnvsFromUserId(userName).contains(envSelected)) {
      return ApiResponse.notOk(KAFKA_CONNECT_ERR_105);
    }

    HandleDbRequests dbHandle = manageDatabase.getHandleDbRequests();
    // tenant filtering
    int tenantId = commonUtilsService.getTenantId(getUserName());
    String syncCluster;
    try {
      syncCluster = manageDatabase.getTenantConfig().get(tenantId).getBaseSyncKafkaConnectCluster();
    } catch (Exception e) {
      log.error("Tenant Configuration not found. " + tenantId, e);
      throw new KlawException(e.getMessage());
    }
    String orderOfEnvs = commonUtilsService.getEnvProperty(tenantId, "ORDER_OF_KAFKA_CONNECT_ENVS");

    List<KwKafkaConnector> kafkaConnectorList =
        getConnectorsFromName(connectorRequestModel.getConnectorName(), tenantId);

    if (!kafkaConnectorList.isEmpty()
        && !Objects.equals(
            kafkaConnectorList.get(0).getTeamId(), connectorRequestModel.getTeamId())) {
      return ApiResponse.notOk(KAFKA_CONNECT_ERR_106);
    }

    boolean promotionOrderCheck =
        checkInPromotionOrder(connectorRequestModel.getEnvironment(), orderOfEnvs);

    if (!kafkaConnectorList.isEmpty()) {
      if (promotionOrderCheck) {
        int devTopicFound =
            (int)
                kafkaConnectorList.stream()
                    .filter(topic -> Objects.equals(topic.getEnvironment(), syncCluster))
                    .count();
        if (devTopicFound != 1) {
          if (getKafkaConnectEnvDetails(syncCluster) == null) {
            return ApiResponse.notOk(KAFKA_CONNECT_ERR_107);
          } else {
            return ApiResponse.notOk(
                String.format(
                    KAFKA_CONNECT_ERR_108, getKafkaConnectEnvDetails(syncCluster).getName()));
          }
        }
      }
    } else if (!Objects.equals(connectorRequestModel.getEnvironment(), syncCluster)) {
      if (promotionOrderCheck) {
        return ApiResponse.notOk(
            String.format(KAFKA_CONNECT_ERR_109, getKafkaConnectEnvDetails(syncCluster).getName()));
      }
    }

    if (manageDatabase
        .getHandleDbRequests()
        .existsConnectorRequest(
            connectorRequestModel.getConnectorName(),
            connectorRequestModel.getEnvironment(),
            RequestStatus.CREATED.value,
            tenantId)) {
      return ApiResponse.notOk(KAFKA_CONNECT_ERR_110);
    }

    // Ignore connector exists check if Update request
    if (RequestOperationType.UPDATE != connectorRequestModel.getRequestOperationType()) {
      boolean topicExists = false;
      if (!kafkaConnectorList.isEmpty()) {
        topicExists =
            kafkaConnectorList.stream()
                .anyMatch(
                    topicEx ->
                        Objects.equals(
                            topicEx.getEnvironment(), connectorRequestModel.getEnvironment()));
      }
      if (topicExists) {
        return ApiResponse.notOk(KAFKA_CONNECT_ERR_111);
      }
    }

    KafkaConnectorRequest topicRequestDao = new KafkaConnectorRequest();
    copyProperties(connectorRequestModel, topicRequestDao);
    topicRequestDao.setRequestOperationType(connectorRequestModel.getRequestOperationType().value);
    topicRequestDao.setTenantId(tenantId);

    String result = dbHandle.requestForConnector(topicRequestDao).get("result");
    mailService.sendMail(
        connectorRequestModel.getConnectorName(),
        null,
        "",
        userName,
        null,
        NumberUtils.toInt(topicRequestDao.getApprovingTeamId(), -1),
        dbHandle,
        CONNECTOR_CREATE_REQUESTED,
        commonUtilsService.getLoginUrl());
    if (result.equals(ApiResultStatus.SUCCESS.value)) {
      return ApiResponse.ok(result);
    } else {
      return ApiResponse.notOk(result);
    }
  }

  public List<List<KafkaConnectorModelResponse>> getConnectors(
      String env, String pageNo, String currentPage, String connectorNameSearch, Integer teamId)
      throws Exception {
    log.debug("getConnectors {}", connectorNameSearch);
    List<KafkaConnectorModelResponse> topicListUpdated =
        getConnectorsPaginated(env, pageNo, currentPage, connectorNameSearch, teamId);

    if (topicListUpdated != null && topicListUpdated.size() > 0) {
      DisplayHelper.updateTeamNamesForDisplay(
          topicListUpdated,
          KafkaConnectorModelResponse::getTeamName,
          KafkaConnectorModelResponse::setTeamName);
      return getPagedList(topicListUpdated);
    }

    return null;
  }

  private List<List<KafkaConnectorModelResponse>> getPagedList(
      List<KafkaConnectorModelResponse> topicsList) {

    List<List<KafkaConnectorModelResponse>> newList = new ArrayList<>();
    List<KafkaConnectorModelResponse> innerList = new ArrayList<>();
    int modulusFactor = 3;
    int i = 0;
    for (KafkaConnectorModelResponse topicInfo : topicsList) {

      innerList.add(topicInfo);

      if (i % modulusFactor == (modulusFactor - 1)) {
        newList.add(innerList);
        innerList = new ArrayList<>();
      }
      i++;
    }

    if (innerList.size() > 0) {
      newList.add(innerList);
    }

    return newList;
  }

  private List<KafkaConnectorModelResponse> getConnectorsPaginated(
      String env, String pageNo, String currentPage, String connectorNameSearch, Integer teamId) {
    if (connectorNameSearch != null) {
      connectorNameSearch = connectorNameSearch.trim();
    }

    int tenantId = commonUtilsService.getTenantId(getUserName());
    HandleDbRequests handleDbRequests = manageDatabase.getHandleDbRequests();

    // Get Sync topics

    List<KwKafkaConnector> connectorsFromSOT =
        handleDbRequests.getSyncConnectors(env, teamId, tenantId);
    connectorsFromSOT = getFilteredConnectorsForTenant(connectorsFromSOT);
    List<Env> listAllEnvs = manageDatabase.getKafkaConnectEnvList(tenantId);
    // tenant filtering

    String orderOfEnvs = commonUtilsService.getEnvProperty(tenantId, "ORDER_OF_KAFKA_CONNECT_ENVS");
    connectorsFromSOT = groupConnectorsByEnv(connectorsFromSOT);

    List<KwKafkaConnector> connectorsFilteredList = connectorsFromSOT;
    // Filter topics on topic name for search
    if (connectorNameSearch != null && connectorNameSearch.length() > 0) {
      final String topicSearchFilter = connectorNameSearch;
      connectorsFilteredList =
          connectorsFromSOT.stream()
              .filter(
                  topic ->
                      topic
                          .getConnectorName()
                          .toLowerCase()
                          .contains(topicSearchFilter.toLowerCase()))
              .collect(Collectors.toList());

      // searching documentation
      List<KwKafkaConnector> searchDocList =
          connectorsFromSOT.stream()
              .filter(
                  topic ->
                      (topic.getDocumentation() != null
                          && topic
                              .getDocumentation()
                              .toLowerCase()
                              .contains(topicSearchFilter.toLowerCase())))
              .toList();

      connectorsFilteredList.addAll(searchDocList);
      connectorsFilteredList =
          new ArrayList<>(
              connectorsFilteredList.stream()
                  .collect(
                      Collectors.toConcurrentMap(
                          KwKafkaConnector::getConnectorName, Function.identity(), (p, q) -> p))
                  .values());
    }

    connectorsFromSOT =
        connectorsFilteredList.stream()
            .sorted(new TopicNameComparator())
            .collect(Collectors.toList());

    final UserInfo user = manageDatabase.getHandleDbRequests().getUsersInfo(getUserName());
    return connectorsFromSOT.isEmpty()
        ? null
        : Pager.getItemsList(
            pageNo,
            currentPage,
            21,
            connectorsFromSOT,
            (pageContext, connectorSOT) -> {
              int counterInc = counterIncrement();
              KafkaConnectorModelResponse kafkaConnectorModelResponse =
                  new KafkaConnectorModelResponse();
              kafkaConnectorModelResponse.setSequence(counterInc);

              Set<String> envSet = new HashSet<>(connectorSOT.getEnvironmentsList());

              kafkaConnectorModelResponse.setConnectorId(connectorSOT.getConnectorId());
              kafkaConnectorModelResponse.setEnvironmentId(connectorSOT.getEnvironment());
              kafkaConnectorModelResponse.setEnvironmentsList(
                  KlawResourceUtils.getConvertedEnvs(listAllEnvs, envSet));
              kafkaConnectorModelResponse.setConnectorName(connectorSOT.getConnectorName());
              kafkaConnectorModelResponse.setTeamName(
                  manageDatabase.getTeamNameFromTeamId(tenantId, connectorSOT.getTeamId()));
              kafkaConnectorModelResponse.setTeamId(connectorSOT.getTeamId());

              kafkaConnectorModelResponse.setDescription(connectorSOT.getDescription());

              kafkaConnectorModelResponse.setTotalNoPages(pageContext.getTotalPages());
              kafkaConnectorModelResponse.setCurrentPage(pageContext.getPageNo());

              kafkaConnectorModelResponse.setAllPageNos(pageContext.getAllPageNos());

              setKafkaConnectorBooleans(tenantId, user, kafkaConnectorModelResponse, connectorSOT);
              return kafkaConnectorModelResponse;
            });
  }

  private void setKafkaConnectorBooleans(
      int tenantId,
      UserInfo user,
      KafkaConnectorModelResponse connectorInfo,
      KwKafkaConnector connectorSOT) {
    connectorInfo.setConnectorOwner(Objects.equals(connectorSOT.getTeamId(), user.getTeamId()));

    connectorInfo.setHighestEnv(
        checkIsHighestEnv(connectorInfo.getEnvironmentId(), connectorInfo.getEnvironmentsList()));
    connectorInfo.setHasOpenClaimRequest(
        isClaimRequestOpen(tenantId, connectorSOT.getConnectorName()));
    connectorInfo.setHasOpenRequest(
        connectorInfo.isHasOpenClaimRequest()
            || isConnectorRequestOpen(
                tenantId, connectorSOT.getConnectorName(), connectorSOT.getEnvironment()));
    connectorInfo.setHasOpenRequestOnAnyEnv(
        isConnectorRequestOpen(tenantId, connectorSOT.getConnectorName()));
  }

  private boolean isConnectorRequestOpen(int tenantId, String connectorName, String environment) {

    return manageDatabase
        .getHandleDbRequests()
        .existsConnectorRequest(connectorName, RequestStatus.CREATED.value, environment, tenantId);
  }

  private boolean isClaimRequestOpen(int tenantId, String connectorName) {

    return manageDatabase
        .getHandleDbRequests()
        .existsClaimConnectorRequest(connectorName, RequestStatus.CREATED.value, tenantId);
  }

  /**
   * Checks if any connector request is open
   *
   * @param tenantId the id of the tenant
   * @param connectorName the name of the connector
   * @return true/false if there is any request open on any environment for a connector
   */
  private boolean isConnectorRequestOpen(int tenantId, String connectorName) {

    return manageDatabase
        .getHandleDbRequests()
        .existsConnectorRequest(connectorName, RequestStatus.CREATED.value, tenantId);
  }

  private boolean isPromoteRequestOpen(int tenantId, String connectorName, String environment) {

    return manageDatabase
        .getHandleDbRequests()
        .existsConnectorRequest(
            connectorName,
            RequestStatus.CREATED.value,
            RequestOperationType.PROMOTE.value,
            environment,
            tenantId);
  }

  private int topicCounter = 0;

  private int counterIncrement() {
    topicCounter++;
    return topicCounter;
  }

  static class TopicNameComparator implements Comparator<KwKafkaConnector> {
    @Override
    public int compare(KwKafkaConnector topic1, KwKafkaConnector topic2) {
      return topic1.getConnectorName().compareTo(topic2.getConnectorName());
    }
  }

  private List<KwKafkaConnector> groupConnectorsByEnv(List<KwKafkaConnector> topicsFromSOT) {
    List<KwKafkaConnector> tmpTopicList = new ArrayList<>();

    Map<String, List<KwKafkaConnector>> groupedList =
        topicsFromSOT.stream().collect(Collectors.groupingBy(KwKafkaConnector::getConnectorName));
    groupedList.forEach(
        (k, v) -> {
          KwKafkaConnector t = v.get(0);
          List<String> tmpEnvList = new ArrayList<>();
          for (KwKafkaConnector topic : v) {
            tmpEnvList.add(topic.getEnvironment());
          }
          t.setEnvironmentsList(tmpEnvList);
          tmpTopicList.add(t);
        });
    return tmpTopicList;
  }

  @PermissionAllowed(permissionAllowed = {PermissionType.REQUEST_CREATE_CONNECTORS})
  public ApiResponse deleteConnectorRequests(String topicId) throws KlawException {
    log.info("deleteConnectorRequests {}", topicId);

    try {
      String deleteTopicReqStatus =
          manageDatabase
              .getHandleDbRequests()
              .deleteConnectorRequest(
                  Integer.parseInt(topicId), commonUtilsService.getTenantId(getUserName()));

      return ApiResponse.ok(deleteTopicReqStatus);
    } catch (Exception e) {
      throw new KlawException(e.getMessage());
    }
  }

  public ApiResponse restartConnector(KafkaConnectorRestartModel kafkaConnectorRestartModel) {
    int tenantId = commonUtilsService.getTenantId(getUserName());
    try {
      if (commonUtilsService.isNotAuthorizedUser(
          commonUtilsService.getPrincipal(), PermissionType.MANAGE_CONNECTORS)) {
        return ApiResponse.NOT_AUTHORIZED;
      }
      return clusterApiService.restartConnector(kafkaConnectorRestartModel, tenantId);
    } catch (KlawException e) {
      return ApiResponse.notOk("Connector could not be restarted");
    }
  }

  public List<KafkaConnectorRequestsResponseModel> getCreatedConnectorRequests(
      String pageNo,
      String currentPage,
      String requestsType,
      String env,
      Order order,
      RequestOperationType requestOperationType,
      String search) {
    log.debug("getCreatedTopicRequests {} {}", pageNo, requestsType);
    String userDetails = getUserName();
    List<KafkaConnectorRequest> createdTopicReqList;
    int tenantId = commonUtilsService.getTenantId(getUserName());

    // get requests relevant to your teams or all teams
    if (commonUtilsService.isNotAuthorizedUser(
        commonUtilsService.getPrincipal(), PermissionType.APPROVE_ALL_REQUESTS_TEAMS))
      createdTopicReqList =
          manageDatabase
              .getHandleDbRequests()
              .getCreatedConnectorRequests(
                  userDetails, requestsType, false, tenantId, env, requestOperationType, search);
    else
      createdTopicReqList =
          manageDatabase
              .getHandleDbRequests()
              .getCreatedConnectorRequests(
                  userDetails, requestsType, true, tenantId, env, requestOperationType, search);

    createdTopicReqList = filterByTenantAndOrder(userDetails, createdTopicReqList, order);
    createdTopicReqList =
        Pager.getItemsList(
            pageNo,
            currentPage,
            10,
            createdTopicReqList,
            (pageContext, activityLog) -> {
              activityLog.setAllPageNos(pageContext.getAllPageNos());
              activityLog.setTotalNoPages(pageContext.getTotalPages());
              activityLog.setCurrentPage(pageContext.getPageNo());
              activityLog.setEnvironmentName(
                  getKafkaConnectEnvDetails(activityLog.getEnvironment()).getName());
              return activityLog;
            });

    return updateCreateConnectorReqsList(createdTopicReqList, tenantId);
  }

  private List<KafkaConnectorRequestsResponseModel> updateCreateConnectorReqsList(
      List<KafkaConnectorRequest> kafkaConnectorRequests, int tenantId) {
    List<KafkaConnectorRequestsResponseModel> connectorRequestModels =
        getConnectorRequestModels(kafkaConnectorRequests);

    for (KafkaConnectorRequestsResponseModel kafkaConnectorRequestModel : connectorRequestModels) {
      kafkaConnectorRequestModel.setTeamname(
          manageDatabase.getTeamNameFromTeamId(tenantId, kafkaConnectorRequestModel.getTeamId()));
      kafkaConnectorRequestModel.setEnvironmentName(
          getKafkaConnectEnvDetails(kafkaConnectorRequestModel.getEnvironment()).getName());
    }

    return connectorRequestModels;
  }

  private String createConnectorConfig(
      KafkaConnectorRequest connectorRequest, String typeOfRequest) {
    ConnectorConfig connectorConfig = new ConnectorConfig();
    connectorConfig.setName(connectorRequest.getConnectorName());

    try {
      JsonNode jsonNode = OBJECT_MAPPER.readTree(connectorRequest.getConnectorConfig());
      updateJsonNode("decrypt", jsonNode);

      ObjectNode objectNode = (ObjectNode) jsonNode;
      connectorConfig.setConfig(objectNode);

      if (typeOfRequest.equals(RequestOperationType.UPDATE.value)) {
        return WRITER_WITH_DEFAULT_PRETTY_PRINTER.writeValueAsString(objectNode);
      } else {
        return WRITER_WITH_DEFAULT_PRETTY_PRINTER.writeValueAsString(connectorConfig);
      }
    } catch (JsonProcessingException e) {
      log.error("Exception:", e);
      return e.toString();
    }
  }

  void updateJsonNode(String encType, JsonNode jsonNode) {
    List<String> sensitiveKeys = new ArrayList<>();
    String[] defaultSensitiveFieldsList = kafkaConnectorSensitiveFields.split(",");
    jsonNode
        .fieldNames()
        .forEachRemaining(
            key -> {
              for (String sensitiveField : defaultSensitiveFieldsList) {
                if (key.toLowerCase().contains(sensitiveField.toLowerCase())) {
                  sensitiveKeys.add(key);
                }
              }
            });
    try {
      BasicTextEncryptor jasyptEncryptor = commonUtilsService.getJasyptEncryptor();
      for (String sensitiveKey : sensitiveKeys) {
        if (encType.equals("decrypt")) {
          ((ObjectNode) jsonNode)
              .put(sensitiveKey, jasyptEncryptor.decrypt(jsonNode.get(sensitiveKey).asText()));
        } else if (encType.equals("encrypt")) {
          {
            ((ObjectNode) jsonNode)
                .put(sensitiveKey, jasyptEncryptor.encrypt(jsonNode.get(sensitiveKey).asText()));
          }
        }
      }
    } catch (Exception e) {
      // existing requests where passwords are not encrypted
      log.error("Ignore errors during decryption");
    }
  }

  @PermissionAllowed(permissionAllowed = {PermissionType.APPROVE_CONNECTORS})
  public ApiResponse approveConnectorRequests(String connectorId)
      throws KlawException, KlawRestException {
    log.info("approveConnectorRequests {}", connectorId);
    String userDetails = getUserName();
    int tenantId = commonUtilsService.getTenantId(getUserName());

    if (commonUtilsService.isNotAuthorizedUser(
        commonUtilsService.getPrincipal(), PermissionType.APPROVE_CONNECTORS)) {
      return ApiResponse.NOT_AUTHORIZED;
    }

    KafkaConnectorRequest connectorRequest =
        manageDatabase
            .getHandleDbRequests()
            .getConnectorRequestsForConnector(Integer.parseInt(connectorId), tenantId);

    String jsonConnectorConfig;
    try {
      jsonConnectorConfig =
          createConnectorConfig(connectorRequest, connectorRequest.getRequestOperationType());
    } catch (Exception e) {
      log.error("Exception:", e);
      throw new KlawException(KAFKA_CONNECT_ERR_112);
    }

    if (connectorRequest.getRequestor().equals(userDetails)) {
      return ApiResponse.notOk(KAFKA_CONNECT_ERR_113);
    }

    if (!RequestStatus.CREATED.value.equals(connectorRequest.getRequestStatus())) {
      return ApiResponse.notOk(REQ_ERR_101);
    }

    // tenant filtering
    final Set<String> allowedEnvIdSet = commonUtilsService.getEnvsFromUserId(getUserName());
    if (!allowedEnvIdSet.contains(connectorRequest.getEnvironment())) {
      return ApiResponse.NOT_AUTHORIZED;
    }

    HandleDbRequests dbHandle = manageDatabase.getHandleDbRequests();
    String updateConnectorReqStatus;
    if (RequestOperationType.CLAIM.value.equals(connectorRequest.getRequestOperationType())) {
      List<KwKafkaConnector> allConnectors =
          getConnectorsFromName(connectorRequest.getConnectorName(), tenantId);
      for (KwKafkaConnector connector : allConnectors) {
        connector.setTeamId(
            connectorRequest.getTeamId()); // for claim reqs, team stored in approving Team
        connector.setExistingConnector(true);
      }

      updateConnectorReqStatus = dbHandle.addToSyncConnectors(allConnectors);
      if (ApiResultStatus.SUCCESS.value.equals(updateConnectorReqStatus)) {
        updateConnectorReqStatus =
            dbHandle.updateConnectorRequestStatus(connectorRequest, userDetails);
      }
    } else {
      Env envSelected =
          manageDatabase
              .getHandleDbRequests()
              .getEnvDetails(connectorRequest.getEnvironment(), tenantId);
      KwClusters kwClusters =
          manageDatabase
              .getClusters(KafkaClustersType.KAFKA_CONNECT, tenantId)
              .get(envSelected.getClusterId());
      KafkaSupportedProtocol protocol = kwClusters.getProtocol();
      String kafkaConnectHost = kwClusters.getBootstrapServers();

      updateConnectorReqStatus =
          clusterApiService.approveConnectorRequests(
              connectorRequest.getConnectorName(),
              protocol,
              connectorRequest.getRequestOperationType(),
              jsonConnectorConfig,
              kafkaConnectHost,
              kwClusters.getClusterName() + kwClusters.getClusterId(),
              connectorRequest.getEnvironment(),
              tenantId);

      if (Objects.equals(updateConnectorReqStatus, ApiResultStatus.SUCCESS.value)) {
        updateConnectorReqStatus = dbHandle.updateConnectorRequest(connectorRequest, userDetails);
        updateAuditAndHistory(userDetails, tenantId, connectorRequest, dbHandle);
        mailService.sendMail(
            connectorRequest.getConnectorName(),
            null,
            "",
            connectorRequest.getRequestor(),
            connectorRequest.getApprover(),
            NumberUtils.toInt(connectorRequest.getApprovingTeamId(), -1),
            dbHandle,
            CONNECTOR_REQUEST_APPROVED,
            commonUtilsService.getLoginUrl());
      }
    }
    return ApiResultStatus.SUCCESS.value.equalsIgnoreCase(updateConnectorReqStatus)
        ? ApiResponse.ok(updateConnectorReqStatus)
        : ApiResponse.notOk(updateConnectorReqStatus);
  }

  private void updateAuditAndHistory(
      String userDetails,
      int tenantId,
      KafkaConnectorRequest connectorRequest,
      HandleDbRequests dbHandle) {
    setConnectorHistory(connectorRequest, userDetails, tenantId);
    dbHandle.insertIntoActivityLog(
        RequestEntityType.CONNECTOR.value,
        tenantId,
        connectorRequest.getRequestOperationType(),
        connectorRequest.getTeamId(),
        "Connector : " + connectorRequest.getConnectorName(),
        connectorRequest.getEnvironment(),
        connectorRequest.getRequestor());
  }

  private void setConnectorHistory(
      KafkaConnectorRequest connectorRequest, String userName, int tenantId) {
    try {
      AtomicReference<String> existingHistory = new AtomicReference<>("");
      List<ResourceHistory> existingConnectorHistory;
      List<ResourceHistory> connectorHistoryList = new ArrayList<>();

      if (RequestOperationType.UPDATE.value.equals(connectorRequest.getRequestOperationType())) {
        List<KwKafkaConnector> existingConnectorList =
            getConnectorsFromName(connectorRequest.getConnectorName(), tenantId);
        existingConnectorList.stream()
            .filter(
                connector ->
                    Objects.equals(connector.getEnvironment(), connectorRequest.getEnvironment()))
            .findFirst()
            .ifPresent(a -> existingHistory.set(a.getHistory()));
        existingConnectorHistory = OBJECT_MAPPER.readValue(existingHistory.get(), VALUE_TYPE_REF);
        connectorHistoryList.addAll(existingConnectorHistory);
      }

      ResourceHistory connectorHistory = new ResourceHistory();
      connectorHistory.setTeamName(
          manageDatabase.getTeamNameFromTeamId(tenantId, connectorRequest.getTeamId()));
      connectorHistory.setEnvironmentName(
          getKafkaConnectEnvDetails(connectorRequest.getEnvironment()).getName());
      connectorHistory.setRequestedBy(connectorRequest.getRequestor());
      connectorHistory.setRequestedTime(
          DATE_TIME_FORMATTER.format(connectorRequest.getRequesttime().toInstant()));
      connectorHistory.setApprovedBy(userName);
      connectorHistory.setApprovedTime(DATE_TIME_FORMATTER.format(Instant.now()));
      connectorHistory.setRemarks("Connector " + connectorRequest.getRequestOperationType());
      connectorHistoryList.add(connectorHistory);

      connectorRequest.setHistory(OBJECT_MAPPER.writer().writeValueAsString(connectorHistoryList));
    } catch (Exception e) {
      log.error("setConnectorHistory Docs ", e);
    }
  }

  @PermissionAllowed(permissionAllowed = {PermissionType.APPROVE_CONNECTORS})
  public ApiResponse declineConnectorRequests(String connectorId, String reasonForDecline)
      throws KlawException {
    log.info("declineConnectorRequests {} {}", connectorId, reasonForDecline);
    String userDetails = getUserName();
    if (commonUtilsService.isNotAuthorizedUser(
        commonUtilsService.getPrincipal(), PermissionType.APPROVE_CONNECTORS)) {
      return ApiResponse.NOT_AUTHORIZED;
    }

    int tenantId = commonUtilsService.getTenantId(getUserName());

    HandleDbRequests dbHandle = manageDatabase.getHandleDbRequests();
    KafkaConnectorRequest connectorRequest =
        dbHandle.getConnectorRequestsForConnector(Integer.parseInt(connectorId), tenantId);

    if (!RequestStatus.CREATED.value.equals(connectorRequest.getRequestStatus())) {
      return ApiResponse.notOk(REQ_ERR_101);
    }

    // tenant filtering
    final Set<String> allowedEnvIdSet = commonUtilsService.getEnvsFromUserId(getUserName());
