package io.aiven.klaw.service;

import static io.aiven.klaw.error.KlawErrorMessages.SCH_SYNC_ERR_101;
import static io.aiven.klaw.error.KlawErrorMessages.SCH_SYNC_ERR_102;
import static io.aiven.klaw.error.KlawErrorMessages.SYNC_102;
import static io.aiven.klaw.error.KlawErrorMessages.SYNC_103;

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.MessageSchema;
import io.aiven.klaw.dao.SchemaRequest;
import io.aiven.klaw.dao.Topic;
import io.aiven.klaw.error.KlawException;
import io.aiven.klaw.helpers.Pager;
import io.aiven.klaw.model.ApiResponse;
import io.aiven.klaw.model.SchemaResetCache;
import io.aiven.klaw.model.SyncSchemaUpdates;
import io.aiven.klaw.model.cluster.SchemaInfoOfTopic;
import io.aiven.klaw.model.cluster.SchemasInfoOfClusterResponse;
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.enums.SchemaType;
import io.aiven.klaw.model.response.SchemaDetailsResponse;
import io.aiven.klaw.model.response.SchemaSubjectInfoResponse;
import io.aiven.klaw.model.response.SyncSchemasList;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.function.Function;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;

@Service
@Slf4j
public class SchemaRegistrySyncControllerService {

  public static final String NOT_IN_SYNC = "NOT_IN_SYNC";
  public static final String NOT_ON_CLUSTER = "NOT_ON_CLUSTER";
  public static final String VERSIONS_NOT_IN_SYNC = "VERSIONS_NOT_IN_SYNC";
  public static final String IN_SYNC = "IN_SYNC";
  public static final String SOURCE_METADATA = "metadata";
  public static final String SOURCE_CLUSTER = "cluster";
  private static final String LEGACY_TOPIC_VERSION = "1.0";
  @Autowired ManageDatabase manageDatabase;

  @Autowired ClusterApiService clusterApiService;

  @Autowired private MailUtils mailService;

  @Autowired private CommonUtilsService commonUtilsService;

  public static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

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

  public SchemaRegistrySyncControllerService(
      ClusterApiService clusterApiService, MailUtils mailService) {
    this.clusterApiService = clusterApiService;
    this.mailService = mailService;
  }

  // schema versions only
  public SyncSchemasList getSchemasOfEnvironment(
      String kafkaEnvId,
      String pageNo,
      String currentPage,
      String topicNameSearch,
      boolean showAllTopics,
      String source,
      int teamId)
      throws Exception {

    if (source.equals(SOURCE_METADATA)) {
      return getSchemasOfEnvironmentFromMetadataDb(
          kafkaEnvId, pageNo, currentPage, topicNameSearch, teamId);
    } else if (source.equals(SOURCE_CLUSTER)) {
      return getSchemaOfEnvironmentFromCluster(
          kafkaEnvId, pageNo, currentPage, topicNameSearch, showAllTopics);
    } else {
      return new SyncSchemasList();
    }
  }

  // schema versions only
  public SyncSchemasList getSchemasOfEnvironmentFromMetadataDb(
      String kafkaEnvId,
      String pageNo,
      String currentPage,
      String topicNameSearch,
      Integer teamId) {
    SyncSchemasList syncSchemasList = new SyncSchemasList();
    List<SchemaSubjectInfoResponse> schemaInfoList;
    if (commonUtilsService.isNotAuthorizedUser(
        commonUtilsService.getPrincipal(), PermissionType.SYNC_BACK_SCHEMAS)) {
      return syncSchemasList;
    }

    String userName = getUserName();
    int tenantId = commonUtilsService.getTenantId(userName);
    Env kafkaEnv = manageDatabase.getHandleDbRequests().getEnvDetails(kafkaEnvId, tenantId);
    if (kafkaEnv.getAssociatedEnv() == null) {
      return syncSchemasList;
    }
    String schemaEnvId = kafkaEnv.getAssociatedEnv().getId();
    schemaInfoList = getSchemasFromDb(kafkaEnvId, tenantId, schemaEnvId);
    syncSchemasList.setAllTopicsCount(schemaInfoList.size());

    if (topicNameSearch != null && topicNameSearch.trim().length() > 0) {
      schemaInfoList =
          schemaInfoList.stream()
              .filter(schema -> schema.getTopic().contains(topicNameSearch.trim()))
              .toList();
    }

    if (teamId != 0) {
      schemaInfoList =
          schemaInfoList.stream().filter(schema -> schema.getTeamId() == teamId).toList();
    }

    schemaInfoList =
        Pager.getItemsList(
            pageNo,
            currentPage,
            schemaInfoList,
            (pageContext, mp) -> {
              mp.setTotalNoPages(pageContext.getTotalPages());
              mp.setAllPageNos(pageContext.getAllPageNos());
              mp.setCurrentPage(pageContext.getPageNo());
              mp.setTeamId(mp.getTeamId() == 0 ? getTeamIdFromDb(tenantId, mp) : mp.getTeamId());
              mp.setTeamname(
                  StringUtils.isEmpty(mp.getTeamname())
                      ? manageDatabase.getTeamNameFromTeamId(tenantId, mp.getTeamId())
                      : mp.getTeamname());
              mp.setPossibleTeams(getPossibleTeams(mp));
              return mp;
            });
    syncSchemasList.setSchemaSubjectInfoResponseList(schemaInfoList);
    return syncSchemasList;
  }

  private List<SchemaSubjectInfoResponse> getSchemasFromDb(
      String kafkaEnvId, int tenantId, String schemaEnvId) {
    List<SchemaSubjectInfoResponse> schemaInfoList = new ArrayList<>();
    List<Topic> topicList =
        manageDatabase.getTopicsForTenant(tenantId).stream()
            .filter(topic -> topic.getEnvironment().equals(kafkaEnvId))
            .toList();

    Map<String, Set<String>> topicSchemaVersionsInDb =
        manageDatabase
            .getHandleDbRequests()
            .getTopicAndVersionsForEnvAndTenantId(schemaEnvId, tenantId);
    if (log.isDebugEnabled()) {
      log.debug("Schema Names,Versions retrieved from DB: {}", topicSchemaVersionsInDb);
    }
    for (Topic topic : topicList) {
      if (topicSchemaVersionsInDb.containsKey(topic.getTopicname())) {
        Set<String> schemaVersions = topicSchemaVersionsInDb.get(topic.getTopicname());

        if (schemaVersions.contains(LEGACY_TOPIC_VERSION)) {
          continue;
        }

        Set<Integer> schemaVersionsInt = new TreeSet<>();
        schemaVersions.forEach(ver -> schemaVersionsInt.add(parseIntFromSchemaVersion(ver)));

        SchemaSubjectInfoResponse schemaInfo = new SchemaSubjectInfoResponse();
        schemaInfo.setTopic(topic.getTopicname());
        schemaInfo.setSchemaVersions(schemaVersionsInt);
        schemaInfo.setTeamId(topic.getTeamId());

        schemaInfoList.add(schemaInfo);
        topicSchemaVersionsInDb.remove(topic.getTopicname());
      }
    }
    return schemaInfoList;
  }

  private List<SchemaSubjectInfoResponse> annotateAnomalousSchemasNotInDB(
      String schemaEnvId,
      List<SchemaSubjectInfoResponse> schemaInfoList,
      Map<String, Set<String>> topicSchemaVersionsInDb,
      Map<String, SchemaInfoOfTopic> schemaTopicAndVersions) {
    // Annotate anomalous Schemas
    Set<Map.Entry<String, Set<String>>> schemaTopicNameToVersionsDB =
        topicSchemaVersionsInDb.entrySet();
    for (Map.Entry<String, Set<String>> schemaDetail : schemaTopicNameToVersionsDB) {

      Set<Integer> schemaVersionsInt = new TreeSet<>();
      schemaDetail.getValue().forEach(ver -> schemaVersionsInt.add(parseIntFromSchemaVersion(ver)));
      SchemaSubjectInfoResponse schemaInfo = new SchemaSubjectInfoResponse();

      schemaInfo.setTopic(schemaDetail.getKey());
      schemaInfo.setSchemaVersions(schemaVersionsInt);
      schemaInfo.setEnvId(schemaEnvId);
      schemaInfo.setRemarks(annotateSchemasNotInCluster(schemaInfo, schemaTopicAndVersions));

      schemaInfoList.add(schemaInfo);
    }
    return schemaInfoList;
  }

  private static int parseIntFromSchemaVersion(String ver) {
    double dub = Double.parseDouble(ver);
    return (int) dub;
  }

  // schema versions only
  private SyncSchemasList getSchemaOfEnvironmentFromCluster(
      String kafkaEnvId,
      String pageNo,
      String currentPage,
      String topicNameSearch,
      boolean showAllTopics)
      throws KlawException {
    SyncSchemasList syncSchemasList = new SyncSchemasList();
    if (commonUtilsService.isNotAuthorizedUser(
        commonUtilsService.getPrincipal(), PermissionType.SYNC_SCHEMAS)) {
      return syncSchemasList;
    }

    List<SchemaSubjectInfoResponse> schemaSubjectInfoResponseList = new ArrayList<>();

    String userDetails = getUserName();
    int tenantId = commonUtilsService.getTenantId(userDetails);
    Env kafkaEnv = manageDatabase.getHandleDbRequests().getEnvDetails(kafkaEnvId, tenantId);
    if (kafkaEnv.getAssociatedEnv() == null) {
      return syncSchemasList;
    }
    Env schemaEnvSelected =
        manageDatabase
            .getHandleDbRequests()
            .getEnvDetails(kafkaEnv.getAssociatedEnv().getId(), tenantId);
    KwClusters kwClusters =
        manageDatabase
            .getClusters(KafkaClustersType.SCHEMA_REGISTRY, tenantId)
            .get(schemaEnvSelected.getClusterId());
    try {
      SchemasInfoOfClusterResponse schemasInfoOfClusterResponse =
          clusterApiService.getSchemasFromCluster(
              kwClusters.getBootstrapServers(),
              kwClusters.getProtocol(),
              kwClusters.getClusterName() + kwClusters.getClusterId(),
              tenantId);

      List<Topic> topicsFromSOT =
          manageDatabase.getHandleDbRequests().getSyncTopics(kafkaEnvId, null, tenantId);

      List<SchemaInfoOfTopic> schemaInfoOfTopicList =
          schemasInfoOfClusterResponse.getSchemaInfoOfTopicList();

      Map<String, SchemaInfoOfTopic> schemaTopicAndVersions =
          schemasInfoOfClusterResponse.getSchemaInfoOfTopicList().stream()
              .collect(Collectors.toMap(SchemaInfoOfTopic::getTopic, Function.identity()));

      for (SchemaInfoOfTopic schemaInfoOfTopic : schemaInfoOfTopicList) {
        SchemaSubjectInfoResponse schemaSubjectInfoResponse = new SchemaSubjectInfoResponse();
        schemaSubjectInfoResponse.setSchemaVersions(schemaInfoOfTopic.getSchemaVersions());
        schemaSubjectInfoResponse.setTopic(schemaInfoOfTopic.getTopic());
        schemaSubjectInfoResponseList.add(schemaSubjectInfoResponse);
      }

      schemaSubjectInfoResponseList =
          annotateAllSchemaWithStatus(
              schemaSubjectInfoResponseList,
              topicsFromSOT,
              schemaTopicAndVersions,
              schemaEnvSelected.getId(),
              tenantId);

      if (!showAllTopics) {
        schemaSubjectInfoResponseList =
            schemaSubjectInfoResponseList.stream()
                .filter(schema -> schema.getRemarks().equals(NOT_IN_SYNC))
                .toList();
      }

      if (topicNameSearch != null && topicNameSearch.trim().length() > 0) {
        schemaSubjectInfoResponseList =
            schemaSubjectInfoResponseList.stream()
                .filter(schema -> schema.getTopic().contains(topicNameSearch.trim()))
                .toList();
      }

      syncSchemasList.setSchemaSubjectInfoResponseList(
          Pager.getItemsList(
              pageNo,
              currentPage,
              schemaSubjectInfoResponseList,
              (pageContext, mp) -> {
                mp.setTotalNoPages(pageContext.getTotalPages());
                mp.setAllPageNos(pageContext.getAllPageNos());
                mp.setCurrentPage(pageContext.getPageNo());
                mp.setTeamId(mp.getTeamId() == 0 ? getTeamIdFromDb(tenantId, mp) : mp.getTeamId());
                mp.setTeamname(
                    StringUtils.isEmpty(mp.getTeamname())
                        ? manageDatabase.getTeamNameFromTeamId(tenantId, mp.getTeamId())
                        : mp.getTeamname());
                mp.setPossibleTeams(getPossibleTeams(mp));
                return mp;
              }));
      return syncSchemasList;
    } catch (Exception e) {
      log.error("Exception:", e);
      throw new KlawException(e.getMessage());
    }
  }

  private String annotateSchemasNotInCluster(
      SchemaSubjectInfoResponse schemaInfo, Map<String, SchemaInfoOfTopic> schemas) {

    try {

      SchemaInfoOfTopic clusterSchema = schemas.get(schemaInfo.getTopic());
      // Missing From Cluster
      if (clusterSchema == null) {
        return NOT_ON_CLUSTER;
      } else {
        // Schema version differences
        if (!(schemaInfo.getSchemaVersions().size() == clusterSchema.getSchemaVersions().size()
            && clusterSchema.getSchemaVersions().containsAll(schemaInfo.getSchemaVersions()))) {
          return VERSIONS_NOT_IN_SYNC;
        }
      }
    } catch (Exception ex) {
      log.error("Error Caught while finding differences between cluster and metadata.");
    }
    return null;
  }

  /**
   * Annotate All Schemas With State cycles through all schemas from the Cluster Api and the DB and
   * finds all differences between them. these differences are annotated in the remarks section of
   * the SchemaSubjectInfoResponse Object
   */
  private List<SchemaSubjectInfoResponse> annotateAllSchemaWithStatus(
      List<SchemaSubjectInfoResponse> schemaSubjectInfoResponseList,
      List<Topic> topicsFromSOT,
      Map<String, SchemaInfoOfTopic> schemaTopicAndVersions,
      String schemaEnvId,
      int tenantId) {

    List<SchemaSubjectInfoResponse> updatedList = new ArrayList<>();
    Map<String, Set<String>> topicSchemaVersionsInDb =
        manageDatabase
            .getHandleDbRequests()
            .getTopicAndVersionsForEnvAndTenantId(schemaEnvId, tenantId);
    for (SchemaSubjectInfoResponse schemaSubjectInfoResponse : schemaSubjectInfoResponseList) {
      Optional<Topic> optionalTopic =
          topicsFromSOT.stream()
              .filter(t -> t.getTopicname().equals(schemaSubjectInfoResponse.getTopic()))
              .findAny();
      if (optionalTopic.isPresent()) {
        schemaSubjectInfoResponse.setTeamId(optionalTopic.get().getTeamId());
        validateSchemas(
            schemaSubjectInfoResponse,
            topicSchemaVersionsInDb.get(schemaSubjectInfoResponse.getTopic()));
        updatedList.add(schemaSubjectInfoResponse);
      } else {
        // Here The Topic does not exist in the Klaw DB we check if the Schema exists in the DB.
        // If it does we mark it as an Orphan.
        List<MessageSchema> sch =
            manageDatabase
                .getHandleDbRequests()
                .getSchemaForTenantAndEnvAndTopic(
                    tenantId, schemaEnvId, schemaSubjectInfoResponse.getTopic());
        if (!sch.isEmpty()) {
          schemaSubjectInfoResponse.setTeamId(sch.get(0).getTeamId());
          schemaSubjectInfoResponse.setTeamname(
              manageDatabase.getTeamNameFromTeamId(tenantId, sch.get(0).getTeamId()));
          schemaSubjectInfoResponse.setRemarks(SYNC_103);
          updatedList.add(schemaSubjectInfoResponse);
        }
        // Do not add anything here as there is no topic or schema associated with the cluster
        // schema in Klaw.
      }
      // We remove each schema which has been processed from the cluster from our list of schemas we
      // have from the db.
      topicSchemaVersionsInDb.remove(schemaSubjectInfoResponse.getTopic());
    }
    if (log.isDebugEnabled()) {
      log.debug("Remaining schemas not removed {}", topicSchemaVersionsInDb);
    }
    // Any Schema not in the DB will be annotated below with not on cluster, versions not in sync or
    // orphaned.
    return annotateAnomalousSchemasNotInDB(
        schemaEnvId, updatedList, topicSchemaVersionsInDb, schemaTopicAndVersions);
  }

  private void validateSchemas(
      SchemaSubjectInfoResponse schemaSubjectInfoResponse, Set<String> schemaVersionsOnDb) {

    if (schemaVersionsOnDb == null) {
      schemaSubjectInfoResponse.setRemarks(NOT_IN_SYNC);
      return;
    }

    Set<Integer> schemaVersionsOnCluster = schemaSubjectInfoResponse.getSchemaVersions();
    if (schemaVersionsOnCluster.size() != schemaVersionsOnDb.size()) {
      schemaSubjectInfoResponse.setRemarks(NOT_IN_SYNC);
      return;
    } else {
      schemaSubjectInfoResponse.setRemarks(IN_SYNC);
    }

    schemaVersionsOnCluster.forEach(
        ver -> {
          if (!schemaVersionsOnDb.contains(ver + "")) {
            schemaSubjectInfoResponse.setRemarks(NOT_IN_SYNC);
          }
        });
  }

  public ApiResponse updateSyncSchemas(SyncSchemaUpdates syncSchemaUpdates) throws Exception {
    log.debug("syncSchemaUpdates {}", syncSchemaUpdates);

    if (syncSchemaUpdates.getTypeOfSync().equals(PermissionType.SYNC_SCHEMAS.name())) {
      return updateSyncSchemasToMetadata(syncSchemaUpdates);
    } else if (syncSchemaUpdates.getTypeOfSync().equals(PermissionType.SYNC_BACK_SCHEMAS.name())) {
      return updateSyncSchemasToCluster(syncSchemaUpdates);
    } else {
      return ApiResponse.builder().success(false).build();
    }
  }

  private ApiResponse updateSyncSchemasToCluster(SyncSchemaUpdates syncSchemaUpdates)
      throws KlawException {

    if (commonUtilsService.isNotAuthorizedUser(
        commonUtilsService.getPrincipal(), PermissionType.SYNC_BACK_SCHEMAS)) {
      return ApiResponse.NOT_AUTHORIZED;
    }
    List<String> logArray = new ArrayList<>();
    logArray.add("Topics/Schemas result");

    String userDetails = getUserName();
    int tenantId = commonUtilsService.getTenantId(userDetails);
    Env kafkaEnv =
        manageDatabase
            .getHandleDbRequests()
            .getEnvDetails(syncSchemaUpdates.getSourceKafkaEnvSelected(), tenantId);

    if (kafkaEnv.getAssociatedEnv() == null) {
      return ApiResponse.notOk(SCH_SYNC_ERR_101);
    }

    if (syncSchemaUpdates.getTopicsSelectionType().equals("ALL_TOPICS")) {
      List<SchemaSubjectInfoResponse> schemaInfoList =
          getSchemasFromDb(kafkaEnv.getId(), tenantId, kafkaEnv.getAssociatedEnv().getId());
      List<String> topicList = new ArrayList<>();
      schemaInfoList.forEach(schemaInfo -> topicList.add(schemaInfo.getTopic()));
      syncSchemaUpdates.setTopicList(topicList);
    }

    for (String topicName : syncSchemaUpdates.getTopicList()) {
      // delete all schemas
      ResponseEntity<ApiResponse> apiResponseEntity =
          clusterApiService.deleteSchema(
              topicName, syncSchemaUpdates.getTargetKafkaEnvSelected(), tenantId);

      // check for success or schema may not exist
      if (apiResponseEntity.getBody() != null
          && (apiResponseEntity.getBody().isSuccess()
              || (!apiResponseEntity.getBody().isSuccess()
                  && apiResponseEntity.getBody().getMessage().contains(SCH_SYNC_ERR_102)))) {
        logArray.add("Schemas deleted for " + topicName);
        // create new schemas
        List<MessageSchema> schemaList =
            manageDatabase
                .getHandleDbRequests()
                .getSchemaForTenantAndEnvAndTopic(
                    tenantId, kafkaEnv.getAssociatedEnv().getId(), topicName);
        schemaList =
            schemaList.stream()
                .sorted(Comparator.comparing(a -> parseIntFromSchemaVersion(a.getSchemaversion())))
                .toList();

        List<MessageSchema> schemaListUpdated = new ArrayList<>();
        for (MessageSchema messageSchema : schemaList) {
          SchemaRequest schemaRequest = new SchemaRequest();
          schemaRequest.setForceRegister(syncSchemaUpdates.isForceRegisterSchema());
          schemaRequest.setSchemafull(messageSchema.getSchemafull());
          if (messageSchema.getSchemaType() == null) {
            schemaRequest.setSchemaType(SchemaType.AVRO);
          } else {
            schemaRequest.setSchemaType(messageSchema.getSchemaType());
          }

          ResponseEntity<ApiResponse> apiResponseCreateEntity =
              clusterApiService.postSchema(
                  schemaRequest, kafkaEnv.getAssociatedEnv().getId(), topicName, tenantId);
          ApiResponse apiResponse = apiResponseCreateEntity.getBody();
          Map<String, Object> registerSchemaCustomResponse = null;
          boolean schemaRegistered = false;
          if (apiResponse != null
              && apiResponse.getData() != null
              && apiResponse.getData() instanceof Map<?, ?>) {
            registerSchemaCustomResponse = (Map) apiResponse.getData();
            schemaRegistered = (Boolean) registerSchemaCustomResponse.get("schemaRegistered");
          }
          if (registerSchemaCustomResponse != null
              && (registerSchemaCustomResponse.containsKey("id")
                  && schemaRegistered
                  && registerSchemaCustomResponse.containsKey("compatibility"))) {

            Integer schemaVersion = (Integer) registerSchemaCustomResponse.get("version");
            messageSchema.setSchemaversion(schemaVersion + "");
            schemaRequest.setSchemaId((Integer) registerSchemaCustomResponse.get("id"));
            schemaRequest.setCompatibility(
                (String) registerSchemaCustomResponse.get("compatibility"));
            schemaListUpdated.add(messageSchema);

            logArray.add(
                "Schemas registered on cluster for " + topicName + " Version " + schemaVersion);
          } else {
            logArray.add("Schema NOT updated :" + topicName);
          }
        }

        manageDatabase.getHandleDbRequests().updateDbWithUpdatedVersions(schemaListUpdated);
      } else {
        logArray.add("Schema NOT updated :" + topicName);
      }
    }

    return ApiResponse.builder()
        .success(true)
        .message(ApiResultStatus.SUCCESS.value)
        .data(logArray)
        .build();
  }

  private ApiResponse updateSyncSchemasToMetadata(SyncSchemaUpdates syncSchemaUpdates)
      throws Exception {

    if (commonUtilsService.isNotAuthorizedUser(
        commonUtilsService.getPrincipal(), PermissionType.SYNC_SCHEMAS)) {
      return ApiResponse.NOT_AUTHORIZED;
    }
    String userDetails = getUserName();
    int tenantId = commonUtilsService.getTenantId(userDetails);
    Env kafkaEnv =
        manageDatabase
            .getHandleDbRequests()
            .getEnvDetails(syncSchemaUpdates.getSourceKafkaEnvSelected(), tenantId);

    if (kafkaEnv.getAssociatedEnv() == null) {
      return ApiResponse.notOk(SCH_SYNC_ERR_101);
    }

    Env schemaEnvSelected =
        manageDatabase
            .getHandleDbRequests()
            .getEnvDetails(kafkaEnv.getAssociatedEnv().getId(), tenantId);
    KwClusters kwClusters =
        manageDatabase
            .getClusters(KafkaClustersType.SCHEMA_REGISTRY, tenantId)
            .get(schemaEnvSelected.getClusterId());

    for (String topicName : syncSchemaUpdates.getTopicList()) {
      TreeMap<Integer, Map<String, Object>> schemaObject =
          clusterApiService.getAvroSchema(
              kwClusters.getBootstrapServers(),
              kwClusters.getProtocol(),
              kwClusters.getClusterName() + kwClusters.getClusterId(),
              topicName,
              tenantId);

      int teamId = commonUtilsService.getTopicsForTopicName(topicName, tenantId).get(0).getTeamId();

      // delete all versions of topic
      Topic topic = new Topic();
      topic.setEnvironment(schemaEnvSelected.getId());
      topic.setTenantId(tenantId);
      topic.setTopicname(topicName);
      manageDatabase.getHandleDbRequests().deleteSchemas(topic);

      // insert all versions of topic
      List<MessageSchema> schemaList = new ArrayList<>();
      for (Integer schemaVersion : schemaObject.keySet()) {
        MessageSchema messageSchema = new MessageSchema();
        messageSchema.setEnvironment(schemaEnvSelected.getId());
        messageSchema.setTopicname(topicName);
        messageSchema.setTenantId(tenantId);
        messageSchema.setSchemaversion(schemaVersion + "");
        messageSchema.setSchemafull((String) schemaObject.get(schemaVersion).get("schema"));
        messageSchema.setTeamId(teamId);
        messageSchema.setSchemaId((Integer) schemaObject.get(schemaVersion).get("id"));
        messageSchema.setCompatibility(
            (String) schemaObject.get(schemaVersion).get("compatibility"));
        schemaList.add(messageSchema);
      }

      manageDatabase.getHandleDbRequests().insertIntoMessageSchemaSOT(schemaList);
    }

    if (syncSchemaUpdates.getTypeOfSync().equalsIgnoreCase("SYNC_SCHEMAS")
        && syncSchemaUpdates.getTopicListForRemoval() != null) {
      // This is a list of the Schemas refrenced by the topic that owned them originally.
      for (String schemaByTopicName : syncSchemaUpdates.getTopicListForRemoval()) {
        manageDatabase
            .getHandleDbRequests()
            .deleteSchema(tenantId, schemaByTopicName, kafkaEnv.getAssociatedEnv().getId());
      }
    }

    return ApiResponse.ok(
        "Topics/Schemas "
            + CollectionUtils.emptyIfNull(syncSchemaUpdates.getTopicList())
            + "\nSchemas removed "
            + CollectionUtils.emptyIfNull(syncSchemaUpdates.getTopicListForRemoval()));
  }

  // schema content either from metadata or cluster
  public SchemaDetailsResponse getSchemaOfTopicFromSource(
      String source, String topicName, int schemaVersion, String kafkaEnvId) throws Exception {
    String userName = getUserName();
    SchemaDetailsResponse schemaDetailsResponse = new SchemaDetailsResponse();
    int tenantId = commonUtilsService.getTenantId(userName);

    Env kafkaEnv = manageDatabase.getHandleDbRequests().getEnvDetails(kafkaEnvId, tenantId);
    if (kafkaEnv.getAssociatedEnv() == null) {
      return schemaDetailsResponse;
    }

    TreeMap<Integer, Map<String, Object>> schemaObject = null;
    Env schemaEnvSelected =
        manageDatabase
            .getHandleDbRequests()
            .getEnvDetails(kafkaEnv.getAssociatedEnv().getId(), tenantId);
    Object dynamicObj = null;

    if (source.equals(SOURCE_CLUSTER)) {
      KwClusters kwClusters =
          manageDatabase
              .getClusters(KafkaClustersType.SCHEMA_REGISTRY, tenantId)
              .get(schemaEnvSelected.getClusterId());

      schemaObject =
          clusterApiService.getAvroSchema(
              kwClusters.getBootstrapServers(),
              kwClusters.getProtocol(),
              kwClusters.getClusterName() + kwClusters.getClusterId(),
              topicName,
              tenantId);
      if (schemaObject != null && schemaObject.containsKey(schemaVersion)) {
        dynamicObj =
            OBJECT_MAPPER.readValue(
                (String) schemaObject.get(schemaVersion).get("schema"), Object.class);
      }
    } else if (source.equals(SOURCE_METADATA)) {
      Optional<MessageSchema> messageSchema =
          manageDatabase
              .getHandleDbRequests()
              .getFirstSchemaForTenantAndEnvAndTopicAndVersion(
                  tenantId, kafkaEnv.getAssociatedEnv().getId(), topicName, schemaVersion + "");
      if (messageSchema.isPresent()) {
        String schemaString = messageSchema.get().getSchemafull();
        dynamicObj = OBJECT_MAPPER.readValue(schemaString, Object.class);
      }
    }

    if (dynamicObj != null) {
      String jsonSchema = WRITER_WITH_DEFAULT_PRETTY_PRINTER.writeValueAsString(dynamicObj);
      schemaDetailsResponse.setSchemaContent(jsonSchema);
      schemaDetailsResponse.setSchemaVersion(schemaVersion + "");
      schemaDetailsResponse.setTopicName(topicName);
      schemaDetailsResponse.setEnvName(kafkaEnv.getName());
    }

    return schemaDetailsResponse;
  }

  private static List<String> getPossibleTeams(SchemaSubjectInfoResponse mp) {
    if (!StringUtils.isEmpty(mp.getRemarks())
        && (mp.getRemarks().equals(SYNC_103) || mp.getRemarks().equals(NOT_ON_CLUSTER))) {
      return List.of(mp.getTeamname(), SYNC_102);
    }
    return List.of(mp.getTeamname());
  }

  private Integer getTeamIdFromDb(int tenantId, SchemaSubjectInfoResponse mp) {
    MessageSchema schema =
        manageDatabase
            .getHandleDbRequests()
            .getTeamIdFromSchemaTopicNameAndEnvAndTenantId(
                mp.getTopic(), String.valueOf(mp.getEnvId()), tenantId);
    if (schema != null) {
      return schema.getTeamId();
    } else {
      return mp.getTeamId();
    }
  }

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

  public ApiResponse resetCacheClusterApi(SchemaResetCache schemaResetCache) throws KlawException {
    String userName = getUserName();
    int tenantId = commonUtilsService.getTenantId(userName);
    ResponseEntity<ApiResponse> apiResponseResponseEntity =
        clusterApiService.resetSchemaInfoCache(schemaResetCache.getKafkaEnvId(), tenantId);
    return apiResponseResponseEntity.getBody();
  }
}
