package io.aiven.klaw.service;

import static io.aiven.klaw.error.KlawErrorMessages.ENV_CLUSTER_TNT_110;
import static io.aiven.klaw.error.KlawErrorMessages.ENV_CLUSTER_TNT_ERR_101;
import static io.aiven.klaw.error.KlawErrorMessages.ENV_CLUSTER_TNT_ERR_102;
import static io.aiven.klaw.error.KlawErrorMessages.ENV_CLUSTER_TNT_ERR_104;
import static io.aiven.klaw.error.KlawErrorMessages.ENV_CLUSTER_TNT_ERR_105;
import static io.aiven.klaw.error.KlawErrorMessages.ENV_CLUSTER_TNT_ERR_106;
import static io.aiven.klaw.error.KlawErrorMessages.ENV_CLUSTER_TNT_ERR_107;
import static io.aiven.klaw.error.KlawErrorMessages.ENV_CLUSTER_TNT_ERR_108;
import static io.aiven.klaw.helpers.CompareUtils.isEqual;
import static io.aiven.klaw.helpers.CompareUtils.isFalse;
import static io.aiven.klaw.helpers.KwConstants.DATE_TIME_DDMMMYYYY_HHMMSS_FORMATTER;
import static io.aiven.klaw.helpers.KwConstants.DEFAULT_TENANT_ID;
import static io.aiven.klaw.helpers.KwConstants.ORDER_OF_KAFKA_CONNECT_ENVS;
import static io.aiven.klaw.helpers.KwConstants.ORDER_OF_TOPIC_ENVS;
import static io.aiven.klaw.helpers.KwConstants.REQUEST_TOPICS_OF_ENVS;
import static io.aiven.klaw.model.enums.PermissionType.ADD_EDIT_DELETE_CLUSTERS;
import static io.aiven.klaw.model.enums.PermissionType.ADD_EDIT_DELETE_ENVS;
import static io.aiven.klaw.model.enums.PermissionType.ADD_TENANT;
import static io.aiven.klaw.model.enums.PermissionType.UPDATE_DELETE_MY_TENANT;
import static io.aiven.klaw.model.enums.RolesType.SUPERADMIN;
import static io.aiven.klaw.service.UsersTeamsControllerService.MASKED_PWD;
import static java.util.stream.Collectors.toList;
import static org.springframework.beans.BeanUtils.copyProperties;

import io.aiven.klaw.config.ManageDatabase;
import io.aiven.klaw.dao.Env;
import io.aiven.klaw.dao.EnvTag;
import io.aiven.klaw.dao.KwClusters;
import io.aiven.klaw.dao.KwTenants;
import io.aiven.klaw.dao.UserInfo;
import io.aiven.klaw.error.KlawBadRequestException;
import io.aiven.klaw.error.KlawException;
import io.aiven.klaw.error.KlawValidationException;
import io.aiven.klaw.helpers.Pager;
import io.aiven.klaw.helpers.db.rdbms.HandleDbRequestsJdbc;
import io.aiven.klaw.model.ApiResponse;
import io.aiven.klaw.model.KwTenantModel;
import io.aiven.klaw.model.enums.ApiResultStatus;
import io.aiven.klaw.model.enums.ClusterStatus;
import io.aiven.klaw.model.enums.EntityType;
import io.aiven.klaw.model.enums.KafkaClustersType;
import io.aiven.klaw.model.enums.KafkaFlavors;
import io.aiven.klaw.model.enums.KafkaSupportedProtocol;
import io.aiven.klaw.model.enums.MetadataOperationType;
import io.aiven.klaw.model.enums.PermissionType;
import io.aiven.klaw.model.requests.EnvModel;
import io.aiven.klaw.model.requests.KwClustersModel;
import io.aiven.klaw.model.requests.UserInfoModel;
import io.aiven.klaw.model.response.ClusterInfo;
import io.aiven.klaw.model.response.EnvIdInfo;
import io.aiven.klaw.model.response.EnvModelResponse;
import io.aiven.klaw.model.response.EnvParams;
import io.aiven.klaw.model.response.EnvUpdatedStatus;
import io.aiven.klaw.model.response.KwClustersModelResponse;
import io.aiven.klaw.model.response.SupportedProtocolInfo;
import io.aiven.klaw.model.response.TenantInfo;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
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.AtomicBoolean;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.stereotype.Service;

@Service
@Slf4j
public class EnvsClustersTenantsControllerService {

  @Autowired private MailUtils mailService;

  @Autowired private CommonUtilsService commonUtilsService;

  @Autowired private ManageDatabase manageDatabase;

  @Autowired private ClusterApiService clusterApiService;

  @Autowired private UsersTeamsControllerService usersTeamsControllerService;

  @Value("${klaw.installation.type:onpremise}")
  private String kwInstallationType;

  @Value("${klaw.max.tenants:1000}")
  private int maxNumberOfTenantsCanBeCreated;

  @Autowired private DefaultDataService defaultDataService;

  @Autowired
  public void setServices(ClusterApiService clusterApiService, MailUtils mailService) {
    this.clusterApiService = clusterApiService;
    this.mailService = mailService;
  }

  private boolean isAuthorizedFor(PermissionType type) {
    return !commonUtilsService.isNotAuthorizedUser(commonUtilsService.getPrincipal(), type);
  }

  public synchronized EnvModelResponse getEnvDetails(String envSelected, String clusterType) {
    String userName = getUserName();
    log.debug("getEnvDetails {}", envSelected);
    int tenantId = commonUtilsService.getTenantId(userName);
    if (!isAuthorizedFor(ADD_EDIT_DELETE_ENVS)
        && !commonUtilsService.getEnvsFromUserId(userName).contains(envSelected)) {
      return null;
    }

    Env env = manageDatabase.getHandleDbRequests().getEnvDetails(envSelected, tenantId);
    if (env == null || isFalse(env.getEnvExists())) {
      return null;
    }
    KwClusters clusters =
        manageDatabase
            .getClusters(KafkaClustersType.of(clusterType), tenantId)
            .get(env.getClusterId());
    String tenantName = manageDatabase.getTenantMap().get(tenantId);

    EnvModelResponse envModel = new EnvModelResponse();
    copyProperties(env, envModel);
    envModel.setClusterName(clusters.getClusterName());
    envModel.setTenantName(tenantName);

    log.debug("Return env model {}", envModel);
    return envModel;
  }

  public UserInfoModel getUserDetails(String userId) {
    UserInfoModel userInfoModel = new UserInfoModel();
    UserInfo userInfo = manageDatabase.getHandleDbRequests().getUsersInfo(userId);
    if (userInfo != null) {
      copyProperties(userInfo, userInfoModel);
      userInfoModel.setUserPassword(MASKED_PWD);
    }
    return userInfo == null ? null : userInfoModel;
  }

  public List<KwClustersModelResponse> getClusters(String typeOfCluster) {
    int tenantId = commonUtilsService.getTenantId(getUserName());
    List<KwClusters> clusters =
        new ArrayList<>(
            manageDatabase.getClusters(KafkaClustersType.of(typeOfCluster), tenantId).values());
    List<KwClustersModelResponse> clustersModels = new ArrayList<>();
    List<Env> allEnvList = manageDatabase.getAllEnvList(tenantId);
    for (KwClusters cluster : clusters) {
      KwClustersModelResponse clusterModel = new KwClustersModelResponse(cluster);
      int clusterId = clusterModel.getClusterId();
      clusterModel.setShowDeleteCluster(true);
      // set only for authorized users
      if (isAuthorizedFor(ADD_EDIT_DELETE_CLUSTERS)) {
        boolean linkedToEnv =
            allEnvList.stream().anyMatch(env -> Objects.equals(env.getClusterId(), clusterId));
        clusterModel.setShowDeleteCluster(!linkedToEnv);
      }
      clustersModels.add(clusterModel);
    }

    return clustersModels;
  }

  public List<KwClustersModelResponse> getClustersPaginated(
      KafkaClustersType typeOfCluster, String clusterId, String pageNo, String searchClusterParam) {

    String clusterTypeValue = typeOfCluster.value;
    List<KwClustersModelResponse> kwClustersModelList = getClusters(clusterTypeValue);

    if (clusterId != null && !clusterId.equals("")) {
      kwClustersModelList =
          kwClustersModelList.stream()
              .filter(cluster -> isEqual(cluster.getClusterId(), clusterId))
              .collect(toList());
    }

    if (searchClusterParam != null && !searchClusterParam.equals("")) {
      kwClustersModelList =
          kwClustersModelList.stream()
              .filter(
                  model ->
                      Stream.of(
                              model.getClusterName(),
                              model.getBootstrapServers(),
                              model.getProtocol().getName())
                          .filter(Objects::nonNull)
                          .map(String::toLowerCase)
                          .anyMatch(s -> s.contains(searchClusterParam.toLowerCase())))
              .sorted(Comparator.comparingInt(KwClustersModelResponse::getClusterId))
              .collect(toList());
    }
    return Pager.getItemsList(
        pageNo, "", 10, kwClustersModelList, (pageContext, mp) -> mp.loadPageContext(pageContext));
  }

  public List<EnvIdInfo> getSyncEnvs() {
    log.debug("getSyncEnvs");
    Integer tenantId = getUserDetails(getUserName()).getTenantId();
    String syncCluster;
    try {
      syncCluster = manageDatabase.getTenantConfig().get(tenantId).getBaseSyncEnvironment();
    } catch (Exception e) {
      log.error("Tenant Configuration not found. " + tenantId, e);
      return new ArrayList<>();
    }

    // return empty list, when sync cluster is not configured
    if (syncCluster == null) {
      return new ArrayList<>();
    }

    List<EnvIdInfo> envsOnly = new ArrayList<>();
    List<EnvModelResponse> envList = getKafkaEnvs();
    for (EnvModelResponse env : envList) {
      EnvIdInfo envIdInfo = new EnvIdInfo();
      envIdInfo.setId(env.getId());
      String baseClusterDropDownStr = " (Base Sync cluster)";
      boolean isSyncEnv = Objects.equals(syncCluster, env.getId());
      String envName = isSyncEnv ? env.getName() + baseClusterDropDownStr : env.getName();
      envIdInfo.setName(envName);
      envsOnly.add(envIdInfo);
    }

    return envsOnly;
  }

  public List<EnvModelResponse> getEnvsForRequestTopicsCluster() {
    int tenantId = getUserDetails(getUserName()).getTenantId();

    String requestTopicsEnvs = commonUtilsService.getEnvProperty(tenantId, REQUEST_TOPICS_OF_ENVS);
    if (requestTopicsEnvs == null) {
      return new ArrayList<>();
    }
    String orderOfEnvs = commonUtilsService.getEnvProperty(tenantId, ORDER_OF_TOPIC_ENVS);
    String[] reqTopicsEnvs = requestTopicsEnvs.split(",");
    List<Env> listEnvs = manageDatabase.getKafkaEnvList(tenantId);
    List<EnvModelResponse> envModelList = getEnvModels(listEnvs, KafkaClustersType.KAFKA, tenantId);

    envModelList = filterEnvironmentModelList(reqTopicsEnvs, envModelList);
    if (orderOfEnvs == null) {
      return envModelList;
    }
    envModelList.sort(Comparator.comparingInt(topicEnv -> orderOfEnvs.indexOf(topicEnv.getId())));
    return envModelList;
  }

  public List<EnvModelResponse> getEnvsForRequestTopicsClusterFiltered() {
    return getKafkaEnvs();
  }

  public List<EnvModelResponse> getKafkaEnvs() {
    int tenantId = getUserDetails(getUserName()).getTenantId();
    String orderOfEnvs = commonUtilsService.getEnvProperty(tenantId, ORDER_OF_TOPIC_ENVS);
    List<Env> listEnvs = manageDatabase.getKafkaEnvList(tenantId);
    List<EnvModelResponse> envModelList = getEnvModels(listEnvs, KafkaClustersType.KAFKA, tenantId);

    // set deletable only to authorized, and check for count
    if (isAuthorizedFor(ADD_EDIT_DELETE_ENVS)) {
      envModelList.forEach(
          envModel -> {
            envModel.setShowDeleteEnv(
                !manageDatabase
                    .getHandleDbRequests()
                    .existsKafkaComponentsForEnv(envModel.getId(), tenantId));
          });
    }

    envModelList.sort(Comparator.comparingInt(topicEnv -> orderOfEnvs.indexOf(topicEnv.getId())));
    return envModelList;
  }

  public List<EnvModelResponse> getConnectorEnvs() {
    int tenantId = getUserDetails(getUserName()).getTenantId();
    String orderOfEnvs = commonUtilsService.getEnvProperty(tenantId, ORDER_OF_KAFKA_CONNECT_ENVS);
    List<Env> listEnvs = manageDatabase.getKafkaConnectEnvList(tenantId);
    List<EnvModelResponse> envModelList =
        getEnvModels(listEnvs, KafkaClustersType.KAFKA_CONNECT, tenantId);

    envModelList.sort(Comparator.comparingInt(topicEnv -> orderOfEnvs.indexOf(topicEnv.getId())));
    return envModelList;
  }

  public List<EnvModelResponse> getEnvsPaginated(
      KafkaClustersType type, String envId, String pageNo, String searchEnvParam) {
    List<EnvModelResponse> envListMap;
    envListMap =
        switch (type) {
          case KAFKA:
            yield getKafkaEnvs();
          case SCHEMA_REGISTRY:
            yield getSchemaRegEnvs();
          case KAFKA_CONNECT:
            yield getKafkaConnectEnvs();
          case ALL:
            yield Stream.of(getKafkaEnvs(), getSchemaRegEnvs(), getKafkaConnectEnvs())
                .flatMap(Collection::stream)
                .toList();
        };

    if (envId != null && !envId.equals("")) {
      envListMap =
          envListMap.stream().filter(env -> Objects.equals(env.getId(), envId)).collect(toList());
    }

    if (searchEnvParam != null && !searchEnvParam.equals("")) {
      envListMap =
          envListMap.stream()
              .filter(
                  env ->
                      Stream.of(
                              env.getName(),
                              env.getClusterName(),
                              env.getOtherParams(),
                              env.getTenantName())
                          .filter(Objects::nonNull)
                          .map(String::toLowerCase)
                          .anyMatch(s -> s.contains(searchEnvParam.toLowerCase())))
              .collect(toList());
    }

    return Pager.getItemsList(
        pageNo, "", 10, envListMap, (pageContext, mp) -> mp.loadPageContext(pageContext));
  }

  private List<EnvModelResponse> getEnvModels(
      List<Env> listEnvs, KafkaClustersType clusterType, int tenantId) {
    List<EnvModelResponse> envModelList = new ArrayList<>();
    String tenantName = manageDatabase.getTenantMap().get(tenantId);
    for (Env listEnv : listEnvs) {
      log.debug("Params {} for env {}", listEnv.getParams(), listEnv.getName());
      KwClusters kwCluster =
          manageDatabase.getClusters(clusterType, tenantId).get(listEnv.getClusterId());
      if (kwCluster != null) {
        EnvModelResponse envModel = new EnvModelResponse();
        copyProperties(listEnv, envModel);
        envModel.setClusterName(kwCluster.getClusterName());
        envModel.setTenantName(tenantName);
        envModelList.add(envModel);
      } else {
        log.error("Error : Environment/cluster not loaded :{}", listEnv);
      }
    }
    return envModelList;
  }

  public EnvParams getEnvParams(Integer targetEnv) {
    return manageDatabase.getEnvParams(commonUtilsService.getTenantId(getUserName()), targetEnv);
  }

  public List<EnvModelResponse> getSchemaRegEnvs() {
    int tenantId = getUserDetails(getUserName()).getTenantId();
    List<Env> listEnvs = manageDatabase.getSchemaRegEnvList(tenantId);

    List<EnvModelResponse> envModelList =
        getEnvModels(listEnvs, KafkaClustersType.SCHEMA_REGISTRY, tenantId);

    // set deletable only to authorized, and check for count
    if (isAuthorizedFor(ADD_EDIT_DELETE_ENVS)) {
      envModelList.forEach(
          envModel -> {
            envModel.setShowDeleteEnv(
                !manageDatabase
                    .getHandleDbRequests()
                    .existsSchemaComponentsForEnv(envModel.getId(), tenantId));
          });
    }

    return envModelList;
  }

  private List<EnvModelResponse> filterEnvironmentModelList(
      String[] reqEnvs, List<EnvModelResponse> envModelList) {
    envModelList =
        envModelList.stream()
            .filter(env -> Arrays.asList(reqEnvs).contains(env.getId()))
            .collect(toList());
    return envModelList;
  }

  public List<EnvModelResponse> getKafkaConnectEnvs() {
    String userName = getUserName();
    int tenantId = getUserDetails(userName).getTenantId();
    List<Env> listEnvs = manageDatabase.getKafkaConnectEnvList(tenantId);

    if (commonUtilsService.isNotAuthorizedUser(
        commonUtilsService.getPrincipal(), ADD_EDIT_DELETE_ENVS)) {
      final Set<String> allowedEnvIdSet = commonUtilsService.getEnvsFromUserId(userName);
      listEnvs =
          listEnvs.stream().filter(env -> allowedEnvIdSet.contains(env.getId())).collect(toList());
    }

    List<EnvModelResponse> envModelList =
        getEnvModels(listEnvs, KafkaClustersType.KAFKA_CONNECT, tenantId);

    // set deletable only to authorized, and check for count
    if (isAuthorizedFor(ADD_EDIT_DELETE_ENVS)) {
      envModelList.forEach(
          envModel -> {
            envModel.setShowDeleteEnv(
                !manageDatabase
                    .getHandleDbRequests()
                    .existsConnectorComponentsForEnv(envModel.getId(), tenantId));
          });
    }

    return envModelList;
  }

  public ApiResponse addNewEnv(EnvModel newEnv) throws KlawException, KlawValidationException {
    log.info("addNewEnv {}", newEnv);
    int tenantId = getUserDetails(getUserName()).getTenantId();
    if (!isAuthorizedFor(ADD_EDIT_DELETE_ENVS)) {
      return ApiResponse.NOT_AUTHORIZED;
    }

    newEnv.setTenantId(tenantId);

    newEnv.setName(newEnv.getName().toUpperCase());
    if (KafkaClustersType.KAFKA.value.equalsIgnoreCase(newEnv.getType())) {
      newEnv
          .getParams()
          .setPartitionsList(
              buildListWithDefault(
                  newEnv.getParams().getDefaultPartitions(),
                  newEnv.getParams().getMaxPartitions()));
      newEnv
          .getParams()
          .setReplicationFactorList(
              buildListWithDefault(
                  newEnv.getParams().getDefaultRepFactor(), newEnv.getParams().getMaxRepFactor()));
    }

    List<Env> envActualList = manageDatabase.getHandleDbRequests().getAllEnvs(tenantId);
    List<Env> kafkaEnvs = manageDatabase.getKafkaEnvList(tenantId);
    List<Env> schemaEnvs = manageDatabase.getSchemaRegEnvList(tenantId);
    List<Integer> kafkaClusterIds = kafkaEnvs.stream().map(Env::getClusterId).toList();
    List<Integer> schemaClusterIds = schemaEnvs.stream().map(Env::getClusterId).toList();

    if (newEnv.getId() == null || newEnv.getId().length() == 0) {
      if (validateConnectedClusters(newEnv, kafkaClusterIds, schemaClusterIds)) {
        return ApiResponse.notOk(ENV_CLUSTER_TNT_110);
      }
      Integer id =
          manageDatabase
              .getHandleDbRequests()
              .getNextSeqIdAndUpdate(EntityType.ENVIRONMENT.name(), tenantId);
      if (id != null) {
        newEnv.setId(String.valueOf(id));
      }
    } else {
      // modify env
      envActualList =
          envActualList.stream()
              .filter(env -> !env.getId().equals(newEnv.getId()))
              .collect(toList());
    }

    // Same name per type (kafka, kafkaconnect) in tenant not posssible.
    boolean envNameAlreadyPresent =
        envActualList.stream()
            .anyMatch(
                en ->
                    Objects.equals(en.getName().toLowerCase(), newEnv.getName().toLowerCase())
                        && Objects.equals(en.getType(), newEnv.getType())
                        && Objects.equals(en.getTenantId(), newEnv.getTenantId())
                        && Objects.equals(en.getEnvExists(), "true"));
    if (envNameAlreadyPresent) {
      return ApiResponse.notOk(ENV_CLUSTER_TNT_ERR_101);
    }

    Env env = new Env();
    copyProperties(newEnv, env);
    env.setEnvExists("true");

    try {
      EnvTag envTag =
          addEnvironmentMapping(
              env.getAssociatedEnv(), env.getId(), env.getName(), env.getTenantId(), env.getType());
      env.setAssociatedEnv(envTag);
      String result = manageDatabase.getHandleDbRequests().addNewEnv(env);
      if (result.equals(ApiResultStatus.SUCCESS.value)) {
        manageDatabase.addEnvToCache(tenantId, env, false);
        return ApiResponse.ok(result);
      } else {
        return ApiResponse.notOk(result);
      }
    } catch (KlawValidationException ex) {
      log.error("KlawValidationException:", ex);
      throw ex;
    } catch (Exception e) {
      log.error("Exception:", e);
      throw new KlawException(e.getMessage());
    }
  }

  private List<String> buildListWithDefault(String defaultNumber, String maxNumber) {
    int defaultNum = Integer.parseInt(defaultNumber);
    int maxNum = Integer.parseInt(maxNumber);
    List<String> parameterList = new ArrayList();
    for (int i = 1; i <= maxNum; i++) {
      String value = i == defaultNum ? i + " (default)" : String.valueOf(i);
      parameterList.add(value);
    }
    return parameterList;
  }

  private boolean validateConnectedClusters(
      EnvModel newEnv, List<Integer> kafkaClusterIds, List<Integer> schemaClusterIds) {
    KafkaClustersType type = KafkaClustersType.of(newEnv.getType());
    switch (type) {
      case KAFKA -> {
        // don't allow same cluster id be assigned to another kafka env, if regex is not defined
        return kafkaClusterIds.contains(newEnv.getClusterId())
            && newEnv.getParams().getTopicPrefix() == null
            && newEnv.getParams().getTopicSuffix() == null;
      }
      case SCHEMA_REGISTRY -> {
        return schemaClusterIds.contains(newEnv.getClusterId());
      }
      default -> {
        return false;
      }
    }
  }

  public ApiResponse addNewCluster(KwClustersModel kwClustersModel) {
    log.info("addNewCluster {}", kwClustersModel);

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

    if (!isAuthorizedFor(ADD_EDIT_DELETE_CLUSTERS)) {
      return ApiResponse.NOT_AUTHORIZED;
    }

    AtomicBoolean clusterNameAlreadyExists = new AtomicBoolean(false);
    if (kwClustersModel.getClusterId() == null) {
      manageDatabase
          .getClusters(KafkaClustersType.ALL, tenantId)
          .forEach(
              (k, v) -> {
                if (Objects.equals(v.getClusterName(), kwClustersModel.getClusterName())
                    && Objects.equals(v.getClusterType(), kwClustersModel.getClusterType().value)) {
                  clusterNameAlreadyExists.set(true);
                }
              });

      if (clusterNameAlreadyExists.get()) {
        return ApiResponse.notOk(ENV_CLUSTER_TNT_ERR_102);
      }
    }
    KwClusters kwCluster = new KwClusters();
    copyProperties(kwClustersModel, kwCluster);
    kwCluster.setTenantId(tenantId);
    kwCluster.setClusterName(kwClustersModel.getClusterName().toUpperCase());
    kwCluster.setKafkaFlavor(kwClustersModel.getKafkaFlavor().value);
    kwCluster.setClusterType(kwClustersModel.getClusterType().value);

    // only for new cluster requests
    String result = manageDatabase.getHandleDbRequests().addNewCluster(kwCluster);

    if (result.equals(ApiResultStatus.SUCCESS.value)) {
      commonUtilsService.updateMetadata(
          tenantId, EntityType.CLUSTER, MetadataOperationType.CREATE, null);
      return ApiResponse.SUCCESS;
    } else {
      return ApiResponse.FAILURE;
    }
  }

  public ApiResponse deleteCluster(String clusterId) throws KlawException {
    log.info("deleteCluster {}", clusterId);
    if (!isAuthorizedFor(ADD_EDIT_DELETE_CLUSTERS)) {
      return ApiResponse.NOT_AUTHORIZED;
    }
    int tenantId = commonUtilsService.getTenantId(getUserName());
    List<Env> allEnvList = manageDatabase.getAllEnvList(tenantId);
    if (allEnvList.stream().anyMatch(env -> isEqual(env.getClusterId(), clusterId))) {
      return ApiResponse.notOk(ENV_CLUSTER_TNT_ERR_104);
    }

    try {
      String result =
          manageDatabase.getHandleDbRequests().deleteCluster(Integer.parseInt(clusterId), tenantId);
      if (result.equals(ApiResultStatus.SUCCESS.value)) {
        commonUtilsService.updateMetadata(
            tenantId, EntityType.CLUSTER, MetadataOperationType.DELETE, null);
        return ApiResponse.ok(result);
      } else {
        return ApiResponse.notOk(result);
      }
    } catch (Exception e) {
      log.error("Exception:", e);
      throw new KlawException(e.getMessage());
    }
  }

  public ApiResponse deleteEnvironment(String envId, String envType) throws KlawException {
    log.info("deleteEnvironment {}", envId);
    int tenantId = commonUtilsService.getTenantId(getUserName());
    if (!isAuthorizedFor(ADD_EDIT_DELETE_ENVS)) {
      return ApiResponse.NOT_AUTHORIZED;
    }

    HandleDbRequestsJdbc jdbc = manageDatabase.getHandleDbRequests();
    KafkaClustersType type = KafkaClustersType.of(envType);
    switch (type) {
      case KAFKA:
        if (jdbc.existsKafkaComponentsForEnv(envId, tenantId)) {
          return ApiResponse.notOk(ENV_CLUSTER_TNT_ERR_105);
        }
        break;
      case KAFKA_CONNECT:
        if (jdbc.existsConnectorComponentsForEnv(envId, tenantId)) {
          return ApiResponse.notOk(ENV_CLUSTER_TNT_ERR_106);
        }
        break;
      case SCHEMA_REGISTRY:
        if (jdbc.existsSchemaComponentsForEnv(envId, tenantId)) {
          return ApiResponse.notOk(ENV_CLUSTER_TNT_ERR_107);
        }
        break;
    }

    try {
      removeAssociatedKafkaOrSchemaEnvironment(envId, tenantId, envType);
      String result = jdbc.deleteEnvironmentRequest(envId, tenantId);
      if (result.equals(ApiResultStatus.SUCCESS.value)) {
        manageDatabase.removeEnvFromCache(tenantId, Integer.valueOf(envId), false);
        return ApiResponse.ok(result);
      } else {
        return ApiResponse.notOk(result);
      }
    } catch (Exception e) {
      log.error("Exception:", e);
      throw new KlawException(e.getMessage());
    }
  }

  private void removeAssociatedKafkaOrSchemaEnvironment(
      String envId, int tenantId, String envType) {

    if (KafkaClustersType.KAFKA.value.equals(envType)
        || KafkaClustersType.SCHEMA_REGISTRY.value.equals(envType)) {
      Env env = manageDatabase.getHandleDbRequests().getEnvDetails(envId, tenantId);
      if (env != null && env.getAssociatedEnv() != null) {
        Env linkedEnv =
            manageDatabase
                .getHandleDbRequests()
                .getEnvDetails(env.getAssociatedEnv().getId(), tenantId);
        linkedEnv.setAssociatedEnv(null);
        manageDatabase.getHandleDbRequests().addNewEnv(linkedEnv);
      }
    }
  }

  private EnvTag addEnvironmentMapping(
      EnvTag envTag, String envId, String envName, int tenantId, String envType)
      throws KlawValidationException {
    KafkaClustersType type = KafkaClustersType.of(envType);
    switch (type) {
      case SCHEMA_REGISTRY -> {
        // only assignable on a schema registry
        log.debug("Env Tag supplied = {}", envTag);
        if (envTag != null && !envTag.getId().isEmpty()) {
          associateWithKafkaEnv(envTag, envId, envName, tenantId);
          // remove existing association if it exists
          removeAssociationWithKafkaEnv(envTag, envId, tenantId);
        } else {
          // envTag is always null here
          removeAssociationWithKafkaEnv(null, envId, tenantId);
        }
      }
      case KAFKA -> envTag = getKafkaAssociation(envTag, envId, tenantId);
    }

    return envTag;
  }

  private EnvTag getKafkaAssociation(EnvTag kafkaEnvTag, String kafkaEnvId, int tenantId) {
    if (kafkaEnvTag == null) {
      Env existing = manageDatabase.getHandleDbRequests().getEnvDetails(kafkaEnvId, tenantId);
      kafkaEnvTag = existing != null ? existing.getAssociatedEnv() : null;
    }
    return kafkaEnvTag;
  }

  private void removeAssociationWithKafkaEnv(EnvTag envTag, String envId, int tenantId) {
    Env existingEnv = manageDatabase.getHandleDbRequests().getEnvDetails(envId, tenantId);
    // envTag is equal to null we don't need to check if the associated env is the same as the
    // passed env tag because this is actualy an operation to remove an association.
    if (existingEnv != null
        && existingEnv.getAssociatedEnv() != null
        && (envTag == null || !existingEnv.getAssociatedEnv().getId().equals(envTag.getId()))) {
      Env linkedEnv =
          manageDatabase
              .getHandleDbRequests()
              .getEnvDetails(existingEnv.getAssociatedEnv().getId(), tenantId);
      linkedEnv.setAssociatedEnv(null);
      manageDatabase.getHandleDbRequests().addNewEnv(linkedEnv);
      // add to cache
      manageDatabase.addEnvToCache(tenantId, linkedEnv, false);
    }
  }

  private void associateWithKafkaEnv(EnvTag envTag, String envId, String envName, int tenantId)
      throws KlawValidationException {
    Env linkedEnv = manageDatabase.getHandleDbRequests().getEnvDetails(envTag.getId(), tenantId);

    if (linkedEnv.getAssociatedEnv() != null
        && !linkedEnv.getAssociatedEnv().getId().equals(envId)) {
      throw new KlawValidationException(
          "Target Environment "
              + linkedEnv.getName()
              + " is already assigned to env "
              + linkedEnv.getAssociatedEnv().getName());
    }
    linkedEnv.setAssociatedEnv(new EnvTag(envId, envName));
    manageDatabase.getHandleDbRequests().addNewEnv(linkedEnv);
    // add update to cache
    manageDatabase.addEnvToCache(tenantId, linkedEnv, false);
  }

  private String getUserName() {
    return mailService.getUserName(commonUtilsService.getPrincipal());
  }

  private Boolean isUserSuperAdmin() {
    return SUPERADMIN
        .name()
        .equals(manageDatabase.getHandleDbRequests().getUsersInfo(getUserName()).getRole());
  }

  public List<KwTenantModel> getAllTenants() {
    boolean allowToRetrieve =
        isUserSuperAdmin() && commonUtilsService.getTenantId(getUserName()) == DEFAULT_TENANT_ID;

    if (!allowToRetrieve) {
      return new ArrayList<>();
    }

    List<KwTenants> tenants = manageDatabase.getHandleDbRequests().getTenants();

    Map<Integer, UserInfo> superAdminMap = manageDatabase.getUserInfoMap(SUPERADMIN);

    List<KwTenantModel> tenantModels =
        tenants.stream()
            .map(
                tenant -> {
                  KwTenantModel kwTenantModel = new KwTenantModel(tenant);
                  UserInfo superAdmin = superAdminMap.get(tenant.getTenantId());
                  kwTenantModel.setEmailId(superAdmin == null ? null : superAdmin.getMailid());
                  return kwTenantModel;
                })
            .collect(Collectors.toList());

    return tenantModels;
  }

  public KwClustersModelResponse getClusterDetails(String clusterId) {
    try {
      int tenantId = commonUtilsService.getTenantId(getUserName());
      KwClusters kwClusters =
          manageDatabase
              .getHandleDbRequests()
              .getClusterDetails(Integer.parseInt(clusterId), tenantId);

      if (kwClusters == null) {
        return null;
      }

      KwClustersModelResponse kwClustersModel = new KwClustersModelResponse(kwClusters);
      return kwClustersModel;
    } catch (Exception e) {
      log.error("Exception:", e);
      return null;
    }
  }

  public List<String> getStandardEnvNames() {
    int tenantId = commonUtilsService.getTenantId(getUserName());
    String standardNames = manageDatabase.getKwPropertyValue("klaw.envs.standardnames", tenantId);
    List<String> envList = Arrays.asList(standardNames.split(","));
    Collections.sort(envList);
    return envList;
  }

  public ApiResponse addTenantId(KwTenantModel kwTenantModel, boolean isExternal)
      throws KlawException {
    if (manageDatabase.getHandleDbRequests().getTenants().size()
        >= maxNumberOfTenantsCanBeCreated) {
      return ApiResponse.notOk(ENV_CLUSTER_TNT_ERR_108);
    }

    if (isExternal && !isAuthorizedFor(ADD_TENANT)) {
      return ApiResponse.NOT_AUTHORIZED;
    }

    KwTenants kwTenants = new KwTenants();
    copyProperties(kwTenantModel, kwTenants);
    Boolean active = isExternal || kwTenantModel.isActiveTenant();
    kwTenants.setIsActive(active.toString());

    try {
      String addNewTenantStatus = manageDatabase.getHandleDbRequests().addNewTenant(kwTenants);
      int tenantId =
          manageDatabase.getHandleDbRequests().getTenants().stream()
              .filter(
                  kwTenant ->
                      Objects.equals(kwTenant.getTenantName(), kwTenantModel.getTenantName()))
              .findFirst()
              .get()
              .getTenantId();

      commonUtilsService.updateMetadata(
          tenantId, EntityType.TENANT, MetadataOperationType.CREATE, null);
      if (isExternal) {
        manageDatabase
            .getHandleDbRequests()
            .insertDefaultKwProperties(defaultDataService.createDefaultProperties(tenantId, ""));
        manageDatabase
            .getHandleDbRequests()
            .insertDefaultRolesPermissions(
