content: package io.aiven.klaw.service;
  
  import static io.aiven.klaw.helpers.KwConstants.ORDER_OF_TOPIC_ENVS;
  
  import io.aiven.klaw.dao.Env;
  import io.aiven.klaw.dao.EnvTag;
  import io.aiven.klaw.dao.KwClusters;
  import io.aiven.klaw.dao.MessageSchema;
  import io.aiven.klaw.dao.Topic;
  import io.aiven.klaw.model.enums.KafkaClustersType;
  import io.aiven.klaw.model.enums.PromotionStatusType;
  import io.aiven.klaw.model.enums.RequestOperationType;
  import io.aiven.klaw.model.enums.RequestStatus;
  import io.aiven.klaw.model.enums.SchemaType;
  import io.aiven.klaw.model.response.PromotionStatus;
  import io.aiven.klaw.model.response.SchemaDetailsPerEnv;
  import io.aiven.klaw.model.response.SchemaOverview;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Set;
  import java.util.SortedMap;
  import java.util.TreeMap;
  import lombok.extern.slf4j.Slf4j;
  import org.springframework.stereotype.Service;
  
  @Slf4j
  @Service
  public class SchemaOverviewService extends BaseOverviewService {
  
    public static final String SCHEMA_ID = "id";
    public static final String SCHEMA_COMPATIBILITY = "compatibility";
    public static final String SCHEMA = "schema";
    private static final String SCHEMA_TYPE = "schemaType";
  
    public SchemaOverviewService(MailUtils mailService) {
      super(mailService);
    }
  
    public SchemaOverview getSchemaOfTopic(
        String topicNameSearch, int schemaVersionSearch, String kafkaEnvId) {
      String userName = getUserName();
      int tenantId = commonUtilsService.getTenantId(userName);
  
      SchemaOverview schemaOverview = new SchemaOverview();
      schemaOverview.setTopicExists(true);
      boolean retrieveSchemas = true;
      updateAvroSchema(
          topicNameSearch,
          schemaVersionSearch,
          kafkaEnvId,
          retrieveSchemas,
          schemaOverview,
          userName,
          tenantId);
      return schemaOverview;
    }
  
    private void updateAvroSchema(
        String topicNameSearch,
        int schemaVersionSearch,
        String kafkaEnvId,
        boolean retrieveSchemas,
        SchemaOverview schemaOverview,
        String userName,
        int tenantId) {
      if (schemaOverview.isTopicExists() && retrieveSchemas) {
        SchemaDetailsPerEnv schemaDetailsPerEnv = new SchemaDetailsPerEnv();
        List<Integer> schemaVersions = new ArrayList<>();
        schemaOverview.setAllSchemaVersions(schemaVersions);
  
        Env schemaEnv;
        // Get first base kafka env
        Env kafkaEnv = manageDatabase.getHandleDbRequests().getEnvDetails(kafkaEnvId, tenantId);
        EnvTag associatedSchemaEnv = kafkaEnv.getAssociatedEnv();
        if (associatedSchemaEnv != null) {
          schemaEnv =
              manageDatabase
                  .getHandleDbRequests()
                  .getEnvDetails(associatedSchemaEnv.getId(), tenantId);
        } else {
          return;
        }
  
        List<Topic> topics =
            manageDatabase
                .getHandleDbRequests()
                .getAllTopicsByTopicNameAndTeamIdAndTenantId(
                    topicNameSearch, commonUtilsService.getTeamId(userName), tenantId);
  
        Object dynamicObj;
        Map<String, Object> hashMapSchemaObj;
        String schemaOfObj;
  
        try {
          log.debug("UpdateAvroSchema - Process env {}", schemaEnv);
  
          KwClusters kwClusters =
              manageDatabase
                  .getClusters(KafkaClustersType.SCHEMA_REGISTRY, tenantId)
                  .get(schemaEnv.getClusterId());
  
          List<MessageSchema> topicSchemaVersionsInDb =
              manageDatabase
                  .getHandleDbRequests()
                  .getSchemaForTenantAndEnvAndTopic(tenantId, schemaEnv.getId(), topicNameSearch);
          boolean schemaUpdated = false;
          if (!topicSchemaVersionsInDb.isEmpty()) {
            schemaUpdated =
                topicSchemaVersionsInDb.stream()
                    .noneMatch(
                        msgSchema ->
                            (msgSchema.getSchemaversion() != null
                                    && msgSchema.getSchemaversion().contains("."))
                                || msgSchema.getCompatibility() == null
                                || msgSchema.getSchemaId() == null);
          }
  
          SortedMap<Integer, Map<String, Object>> schemaObjects = new TreeMap<>();
          schemaObjects =
              getSchemasMap(
                  topicNameSearch,
                  tenantId,
                  kwClusters,
                  topicSchemaVersionsInDb,
                  schemaUpdated,
                  schemaObjects);
          // If the schemaObject is null ie does not exist do not try to manipulate it.
          if (schemaObjects != null && !schemaObjects.isEmpty()) {
            Set<Integer> allVersions = schemaObjects.keySet();
            List<Integer> allVersionsList = new ArrayList<>(allVersions);
            allVersionsList.sort(Collections.reverseOrder());
            Integer latestSchemaVersion = allVersionsList.get(0);
            schemaOverview.setLatestVersion(latestSchemaVersion);
  
            schemaOverview.getAllSchemaVersions().addAll(allVersionsList);
            try {
              if (latestSchemaVersion == schemaVersionSearch) {
                schemaVersionSearch = 0;
              }
            } catch (NumberFormatException ignored) {
            }
  
            // get latest version
            if (schemaVersionSearch == 0) {
              hashMapSchemaObj = schemaObjects.get(latestSchemaVersion);
              schemaOfObj = (String) hashMapSchemaObj.get(SCHEMA);
              schemaDetailsPerEnv.setLatest(true);
              if (!hashMapSchemaObj.containsKey(SCHEMA_TYPE)
                  || Objects.isNull(hashMapSchemaObj.get(SCHEMA_TYPE))) {
                schemaDetailsPerEnv.setSchemaType(SchemaType.AVRO);
              } else {
                schemaDetailsPerEnv.setSchemaType(
                    SchemaType.of(hashMapSchemaObj.get(SCHEMA_TYPE).toString()));
              }
              setSchemaDetailsPerEnvVersionAndCompatibility(
                  tenantId, schemaDetailsPerEnv, hashMapSchemaObj, latestSchemaVersion, schemaEnv);
              if (schemaObjects.size() > 1) {
                schemaDetailsPerEnv.setShowNext(true);
                schemaDetailsPerEnv.setShowPrev(false);
                int indexOfVersion = allVersionsList.indexOf(latestSchemaVersion);
                schemaDetailsPerEnv.setNextVersion(allVersionsList.get(indexOfVersion + 1));
              }
            } else {
              hashMapSchemaObj = schemaObjects.get(schemaVersionSearch);
              schemaOfObj = (String) hashMapSchemaObj.get(SCHEMA);
              schemaDetailsPerEnv.setLatest(false);
              if (!hashMapSchemaObj.containsKey(SCHEMA_TYPE)
                  || Objects.isNull(hashMapSchemaObj.get(SCHEMA_TYPE))) {
                schemaDetailsPerEnv.setSchemaType(SchemaType.AVRO);
              } else {
                schemaDetailsPerEnv.setSchemaType(
                    SchemaType.of(hashMapSchemaObj.get(SCHEMA_TYPE).toString()));
              }
              setSchemaDetailsPerEnvVersionAndCompatibility(
                  tenantId, schemaDetailsPerEnv, hashMapSchemaObj, schemaVersionSearch, schemaEnv);
              if (schemaObjects.size() > 1) {
                int indexOfVersion = allVersionsList.indexOf(schemaVersionSearch);
                if (indexOfVersion + 1 == allVersionsList.size()) {
                  schemaDetailsPerEnv.setShowNext(false);
                  schemaDetailsPerEnv.setShowPrev(true);
                  schemaDetailsPerEnv.setPrevVersion(allVersionsList.get(indexOfVersion - 1));
                } else {
                  schemaDetailsPerEnv.setShowNext(true);
                  schemaDetailsPerEnv.setShowPrev(true);
                  schemaDetailsPerEnv.setPrevVersion(allVersionsList.get(indexOfVersion - 1));
                  schemaDetailsPerEnv.setNextVersion(allVersionsList.get(indexOfVersion + 1));
                }
              }
            }
  
            schemaDetailsPerEnv.setEnv(schemaEnv.getName());
            dynamicObj = OBJECT_MAPPER.readValue(schemaOfObj, Object.class);
            schemaOfObj = WRITER_WITH_DEFAULT_PRETTY_PRINTER.writeValueAsString(dynamicObj);
            schemaDetailsPerEnv.setContent(schemaOfObj);
            schemaOverview.setSchemaExists(true);
  
            // A team owns a topic across all environments so we can assume if the search returned
            // one or more topics it is owned by this users team.
            if (topics.size() > 0) {
  
              // Set Promotion Details
              processSchemaPromotionDetails(
                  topicNameSearch,
                  schemaOverview,
                  tenantId,
                  schemaEnv,
                  topics.stream().map(Topic::getEnvironment).toList());
              log.debug("Getting schema details for: " + topicNameSearch);
            }
          }
        } catch (Exception e) {
          log.error("Error ", e);
        }
  
        if (schemaOverview.isSchemaExists()) {
          log.debug("SchemaDetails {}", schemaDetailsPerEnv);
          schemaOverview.setSchemaDetailsPerEnv(schemaDetailsPerEnv);
        }
        schemaOverview.setCreateSchemaAllowed(
            commonUtilsService.isCreateNewSchemaAllowed(schemaEnv.getId(), tenantId));
      }
    }
  
    private void setSchemaDetailsPerEnvVersionAndCompatibility(
        int tenantId,
        SchemaDetailsPerEnv schemaDetailsPerEnv,
        Map<String, Object> hashMapSchemaObj,
        Integer latestSchemaVersion,
        Env schemaEnv) {
      updateIdAndCompatibility(schemaDetailsPerEnv, hashMapSchemaObj);
      schemaDetailsPerEnv.setVersion(latestSchemaVersion);
    }
  
    private static void updateIdAndCompatibility(
        SchemaDetailsPerEnv schemaDetailsPerEnv, Map<String, Object> hashMapSchemaObj) {
      if (hashMapSchemaObj.containsKey(SCHEMA_ID) && hashMapSchemaObj.get(SCHEMA_ID) != null) {
        schemaDetailsPerEnv.setId((Integer) hashMapSchemaObj.get(SCHEMA_ID));
      }
  
      if (hashMapSchemaObj.containsKey(SCHEMA_COMPATIBILITY)
          && hashMapSchemaObj.get(SCHEMA_COMPATIBILITY) != null) {
        schemaDetailsPerEnv.setCompatibility(hashMapSchemaObj.get(SCHEMA_COMPATIBILITY) + "");
      } else {
        schemaDetailsPerEnv.setCompatibility("Couldn't retrieve");
      }
    }
  
    private SortedMap<Integer, Map<String, Object>> getSchemasMap(
        String topicNameSearch,
        int tenantId,
        KwClusters kwClusters,
        List<MessageSchema> topicSchemaVersionsInDb,
        boolean schemaUpdated,
        SortedMap<Integer, Map<String, Object>> schemaObjects)
        throws Exception {
      if (schemaUpdated) {
        log.debug("GetSchema {} from DB", topicNameSearch);
        for (MessageSchema messageSchema : topicSchemaVersionsInDb) {
          Map<String, Object> schemaObj = new HashMap<>();
          schemaObj.put(SCHEMA, messageSchema.getSchemafull());
          schemaObj.put(SCHEMA_ID, messageSchema.getSchemaId());
          schemaObj.put(SCHEMA_COMPATIBILITY, messageSchema.getCompatibility());
          schemaObj.put(SCHEMA_TYPE, messageSchema.getSchemaType());
          schemaObjects.put(Integer.parseInt(messageSchema.getSchemaversion()), schemaObj);
        }
      } else {
        schemaObjects = getSchemaFromAPI(topicNameSearch, tenantId, kwClusters);
  
        if (schemaObjects != null) {
          Boolean saveChanges = null;
          for (MessageSchema messageSchema : topicSchemaVersionsInDb) {
  
            // The Key for the SchemaObjects map is the version number.
            Map<String, Object> schemaObj =
                schemaObjects.get(Integer.valueOf(messageSchema.getSchemaversion()));
            if (schemaObj != null) {
              if (messageSchema.getSchemaId() == null) {
                saveChanges = true;
                messageSchema.setSchemaId((Integer) schemaObj.get(SCHEMA_ID));
              }
              if (messageSchema.getCompatibility() == null) {
                saveChanges = true;
                messageSchema.setCompatibility((String) schemaObj.get(SCHEMA_COMPATIBILITY));
              }
              if (messageSchema.getSchemaType() == null) {
                saveChanges = true;
                if (schemaObj.containsKey(SCHEMA_TYPE)) {
                  messageSchema.setSchemaType(SchemaType.of(schemaObj.get(SCHEMA_TYPE).toString()));
                } else {
                  messageSchema.setSchemaType(SchemaType.AVRO);
                }
              }
              if (messageSchema.getSchemafull() == null) {
                saveChanges = true;
                messageSchema.setSchemafull((String) schemaObj.get(SCHEMA));
              }
            }
          }
          log.debug("Updated topic {} schemaId, compatibility and schema.", topicNameSearch);
          // Save that back into the DB
          if (saveChanges != null) {
            manageDatabase.getHandleDbRequests().insertIntoMessageSchemaSOT(topicSchemaVersionsInDb);
          }
        }
      }
      return schemaObjects;
    }
  
    private SortedMap<Integer, Map<String, Object>> getSchemaFromAPI(
        String topicNameSearch, int tenantId, KwClusters kwClusters) throws Exception {
      SortedMap<Integer, Map<String, Object>> schemaObjects;
      log.debug("GetSchema {} from API", topicNameSearch);
      schemaObjects =
          clusterApiService.getAvroSchema(
              kwClusters.getBootstrapServers(),
              kwClusters.getProtocol(),
              kwClusters.getClusterName() + kwClusters.getClusterId(),
              topicNameSearch,
              tenantId);
      return schemaObjects;
    }
  
    private void processSchemaPromotionDetails(
        String topicNameSearch,
        SchemaOverview schemaOverview,
        int tenantId,
        Env schemaEnv,
        List<String> kafkaEnvIds) {
      log.debug("SchemaEnv Id {} KafkaEnvIds {}", schemaEnv.getId(), kafkaEnvIds);
      PromotionStatus promotionDetails = new PromotionStatus();
      String orderEnvs = commonUtilsService.getEnvProperty(tenantId, ORDER_OF_TOPIC_ENVS);
      generatePromotionDetails(
          tenantId,
          promotionDetails,
          schemaEnv.getAssociatedEnv() != null
              ? Collections.singletonList(schemaEnv.getAssociatedEnv().getId())
              : null,
          orderEnvs);
      if (schemaOverview.getSchemaPromotionDetails() == null) {
        PromotionStatus searchOverviewPromotionDetails = new PromotionStatus();
        schemaOverview.setSchemaPromotionDetails(searchOverviewPromotionDetails);
      }
  
      // verify if topic exists in target env
      if (!verifyIfTopicExistsInTargetSchemaEnv(kafkaEnvIds, promotionDetails, tenantId)) {
        promotionDetails.setStatus(PromotionStatusType.NO_PROMOTION);
      } else if (isSchemaPromoteRequestOpen(
          topicNameSearch, promotionDetails.getTargetEnvId(), tenantId)) {
        promotionDetails.setStatus(PromotionStatusType.REQUEST_OPEN);
      }
      schemaOverview.setSchemaPromotionDetails(promotionDetails);
    }
  
    private boolean verifyIfTopicExistsInTargetSchemaEnv(
        List<String> kafkaEnvIds, PromotionStatus promotionDetails, int tenantId) {
      if (promotionDetails.getTargetEnvId() == null) {
        return false;
      }
      // Get kafka env and ensure that it has an associated env with it.
      Env promotedEnv =
          manageDatabase
              .getHandleDbRequests()
              .getEnvDetails(promotionDetails.getTargetEnvId(), tenantId);
  
      String kafkaEnvId = promotedEnv.getId();
      return kafkaEnvIds.contains(kafkaEnvId) && promotedEnv.getAssociatedEnv() != null;
    }
  
    private boolean isSchemaPromoteRequestOpen(String topicName, String envId, int tenantId) {
      return manageDatabase
          .getAssociatedSchemaEnvIdFromTopicId(envId, tenantId)
          .filter(
              s ->
                  manageDatabase
                      .getHandleDbRequests()
                      .existsSchemaRequest(
                          topicName,
                          RequestStatus.CREATED.value,
                          RequestOperationType.PROMOTE.value,
                          s,
                          tenantId))
          .isPresent();
    }
  }
  