content: package io.aiven.klaw.service;
  
  import static io.aiven.klaw.error.KlawErrorMessages.SCHEMA_ERR_101;
  import static io.aiven.klaw.error.KlawErrorMessages.SCHEMA_ERR_102;
  import static io.aiven.klaw.error.KlawErrorMessages.SCHEMA_ERR_103;
  import static io.aiven.klaw.error.KlawErrorMessages.SCHEMA_ERR_104;
  import static io.aiven.klaw.error.KlawErrorMessages.SCHEMA_ERR_105;
  import static io.aiven.klaw.error.KlawErrorMessages.SCHEMA_ERR_106;
  import static io.aiven.klaw.error.KlawErrorMessages.SCHEMA_ERR_107;
  import static io.aiven.klaw.error.KlawErrorMessages.SCHEMA_ERR_108;
  import static io.aiven.klaw.error.KlawErrorMessages.SCHEMA_ERR_109;
  import static io.aiven.klaw.error.KlawErrorMessages.SCHEMA_ERR_110;
  import static io.aiven.klaw.error.KlawErrorMessages.SCHEMA_ERR_111;
  import static io.aiven.klaw.helpers.UtilMethods.updateEnvStatus;
  import static io.aiven.klaw.service.MailUtils.MailType.SCHEMA_APPROVED_NOTIFY_SUBSCRIBERS;
  import static io.aiven.klaw.service.MailUtils.MailType.SCHEMA_REQUESTED;
  import static io.aiven.klaw.service.MailUtils.MailType.SCHEMA_REQUEST_APPROVED;
  import static io.aiven.klaw.service.MailUtils.MailType.SCHEMA_REQUEST_DENIED;
  import static org.springframework.beans.BeanUtils.copyProperties;
  
  import com.fasterxml.jackson.core.JsonProcessingException;
  import com.fasterxml.jackson.databind.ObjectMapper;
  import io.aiven.klaw.config.ManageDatabase;
  import io.aiven.klaw.dao.Acl;
  import io.aiven.klaw.dao.Env;
  import io.aiven.klaw.dao.KwClusters;
  import io.aiven.klaw.dao.SchemaRequest;
  import io.aiven.klaw.dao.Team;
  import io.aiven.klaw.dao.Topic;
  import io.aiven.klaw.dao.UserInfo;
  import io.aiven.klaw.error.KlawException;
  import io.aiven.klaw.helpers.HandleDbRequests;
  import io.aiven.klaw.helpers.Pager;
  import io.aiven.klaw.model.ApiResponse;
  import io.aiven.klaw.model.enums.*;
  import io.aiven.klaw.model.requests.SchemaPromotion;
  import io.aiven.klaw.model.requests.SchemaRequestModel;
  import io.aiven.klaw.model.response.BaseRequestsResponseModel;
  import io.aiven.klaw.model.response.SchemaRequestsResponseModel;
  import io.aiven.klaw.validation.PermissionAllowed;
  import java.io.IOException;
  import java.util.*;
  import java.util.stream.Collectors;
  import lombok.extern.slf4j.Slf4j;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.beans.factory.annotation.Value;
  import org.springframework.http.ResponseEntity;
  import org.springframework.stereotype.Service;
  
  @Service
  @Slf4j
  public class SchemaRegistryControllerService {
    @Value("${klaw.schema.validate.compatibility.onSave:true}")
    private Boolean validateCompatiblityOnSave;
  
    @Autowired ManageDatabase manageDatabase;
  
    @Autowired ClusterApiService clusterApiService;
  
    @Autowired private MailUtils mailService;
  
    @Autowired private CommonUtilsService commonUtilsService;
  
    @Autowired private RolesPermissionsControllerService rolesPermissionsControllerService;
  
    private static final String SCHEMA_TYPE = "schemaType";
  
    public SchemaRegistryControllerService(
        ClusterApiService clusterApiService, MailUtils mailService) {
      this.clusterApiService = clusterApiService;
      this.mailService = mailService;
    }
  
    public List<SchemaRequestsResponseModel> getSchemaRequests(
        String pageNo,
        String currentPage,
        String requestStatus,
        RequestOperationType requestOperationType,
        boolean isApproval,
        String topic,
        String env,
        String search,
        Order order,
        boolean isMyRequest) {
      log.debug("getSchemaRequests page {} requestsType {}", pageNo, requestStatus);
  
      String userName = getUserName();
      int tenantId = commonUtilsService.getTenantId(userName);
  
      List<SchemaRequest> schemaReqs =
          manageDatabase
              .getHandleDbRequests()
              .getAllSchemaRequests(
                  isApproval,
                  userName,
                  tenantId,
                  requestOperationType,
                  topic,
                  env,
                  requestStatus,
                  search,
                  isApproval
                      && !commonUtilsService.isNotAuthorizedUser(
                          commonUtilsService.getPrincipal(),
                          PermissionType.APPROVE_ALL_REQUESTS_TEAMS),
                  isMyRequest);
  
      // tenant filtering
      final Set<String> allowedEnvIdSet = commonUtilsService.getEnvsFromUserId(userName);
      if (schemaReqs != null) {
        schemaReqs =
            schemaReqs.stream()
                .filter(request -> allowedEnvIdSet.contains(request.getEnvironment()))
                .collect(Collectors.toList());
      }
  
      Integer userTeamId = commonUtilsService.getTeamId(userName);
      List<UserInfo> userList = manageDatabase.getUsersPerTeamAndTenant(userTeamId, tenantId);
  
      Set<String> approverRoles =
          rolesPermissionsControllerService.getApproverRoles("CONNECTORS", tenantId);
      List<SchemaRequestsResponseModel> schemaRequestModels = new ArrayList<>();
  
      SchemaRequestsResponseModel schemaRequestModel;
      if (schemaReqs != null) {
        for (SchemaRequest schemaReq : schemaReqs) {
          schemaRequestModel = new SchemaRequestsResponseModel();
          schemaReq.setEnvironmentName(
              manageDatabase
                  .getHandleDbRequests()
                  .getEnvDetails(schemaReq.getEnvironment(), tenantId)
                  .getName());
          copyProperties(schemaReq, schemaRequestModel);
          schemaRequestModel.setRequestStatus(RequestStatus.of(schemaReq.getRequestStatus()));
          schemaRequestModel.setRequestOperationType(
              RequestOperationType.of(schemaReq.getRequestOperationType()));
  
          // show approving info only before approvals
          if (RequestStatus.APPROVED != schemaRequestModel.getRequestStatus()) {
            schemaRequestModel.setApprovingTeamDetails(
                updateApproverInfo(
                    userList,
                    manageDatabase.getTeamNameFromTeamId(tenantId, userTeamId),
                    approverRoles,
                    schemaRequestModel.getRequestor()));
          }
          schemaRequestModels.add(setRequestorPermissions(schemaRequestModel, userName));
        }
      }
  
      schemaRequestModels = schemaRequestModels.stream().sorted(getPreferredOrder(order)).toList();
  
      return Pager.getItemsList(
          pageNo,
          currentPage,
          10,
          schemaRequestModels,
          (pageContext, schemaRequestModel1) -> {
            schemaRequestModel1.setAllPageNos(pageContext.getAllPageNos());
            schemaRequestModel1.setTotalNoPages(pageContext.getTotalPages());
            schemaRequestModel1.setCurrentPage(pageContext.getPageNo());
            schemaRequestModel1.setTeamname(
                manageDatabase.getTeamNameFromTeamId(tenantId, schemaRequestModel1.getTeamId()));
            return schemaRequestModel1;
          });
    }
  
    private Comparator<SchemaRequestsResponseModel> getPreferredOrder(Order order) {
      return switch (order) {
        case ASC_REQUESTED_TIME -> Comparator.comparing(BaseRequestsResponseModel::getRequesttime);
        case DESC_REQUESTED_TIME -> Collections.reverseOrder(
            Comparator.comparing(BaseRequestsResponseModel::getRequesttime));
      };
    }
  
    private String updateApproverInfo(
        List<UserInfo> userList, String teamName, Set<String> approverRoles, String requestor) {
      StringBuilder approvingInfo = new StringBuilder("Team : " + teamName + ", Users : ");
  
      for (UserInfo userInfo : userList) {
        if (approverRoles.contains(userInfo.getRole())
            && !Objects.equals(requestor, userInfo.getUsername())) {
          approvingInfo.append(userInfo.getUsername()).append(",");
        }
      }
  
      return String.valueOf(approvingInfo);
    }
  
    private SchemaRequestsResponseModel setRequestorPermissions(
        SchemaRequestsResponseModel req, String userName) {
  
      if (RequestStatus.CREATED == req.getRequestStatus()
          && userName != null
          && userName.equals(req.getRequestor())) {
        req.setDeletable(true);
        req.setEditable(true);
      }
  
      return req;
    }
  
    @PermissionAllowed(permissionAllowed = {PermissionType.REQUEST_CREATE_SCHEMAS})
    public ApiResponse deleteSchemaRequests(String avroSchemaId) throws KlawException {
      log.info("deleteSchemaRequests {}", avroSchemaId);
  
      if (commonUtilsService.isNotAuthorizedUser(
          commonUtilsService.getPrincipal(), PermissionType.REQUEST_DELETE_SCHEMAS)) {
        return ApiResponse.NOT_AUTHORIZED;
      }
      String userName = getUserName();
      try {
        String result =
            manageDatabase
                .getHandleDbRequests()
                .deleteSchemaRequest(
                    Integer.parseInt(avroSchemaId),
                    userName,
                    commonUtilsService.getTenantId(getUserName()));
        return ApiResultStatus.SUCCESS.value.equals(result)
            ? ApiResponse.ok(result)
            : ApiResponse.notOk(result);
      } catch (Exception e) {
        log.error("Exception:", e);
        throw new KlawException(e.getMessage());
      }
    }
  
    @PermissionAllowed(permissionAllowed = {PermissionType.APPROVE_SCHEMAS})
    public ApiResponse execSchemaRequests(String avroSchemaId) throws KlawException {
      log.info("execSchemaRequests {}", avroSchemaId);
      String userDetails = getUserName();
      int tenantId = commonUtilsService.getTenantId(getUserName());
      if (commonUtilsService.isNotAuthorizedUser(
          commonUtilsService.getPrincipal(), PermissionType.APPROVE_SCHEMAS)) {
        return ApiResponse.NOT_AUTHORIZED;
      }
  
      SchemaRequest schemaRequest =
          manageDatabase
              .getHandleDbRequests()
              .getSchemaRequest(Integer.parseInt(avroSchemaId), tenantId);
  
      if (Objects.equals(schemaRequest.getRequestor(), userDetails))
        return ApiResponse.notOk(SCHEMA_ERR_101);
  
      final Set<String> allowedEnvIdSet = commonUtilsService.getEnvsFromUserId(getUserName());
  
      if (!allowedEnvIdSet.contains(schemaRequest.getEnvironment())) {
        return ApiResponse.NOT_AUTHORIZED;
      }
  
      ResponseEntity<ApiResponse> response =
          clusterApiService.postSchema(
              schemaRequest, schemaRequest.getEnvironment(), schemaRequest.getTopicname(), tenantId);
      updateEnvStatus(response, manageDatabase, tenantId, schemaRequest.getEnvironment());
      HandleDbRequests dbHandle = manageDatabase.getHandleDbRequests();
      ApiResponse apiResponse = response.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"))
          && (registerSchemaCustomResponse.containsKey("compatibility"))) {
        try {
          if (schemaRegistered) {
            schemaRequest.setSchemaversion(registerSchemaCustomResponse.get("version") + "");
            schemaRequest.setSchemaId((Integer) registerSchemaCustomResponse.get("id"));
            schemaRequest.setCompatibility(
                (String) registerSchemaCustomResponse.get("compatibility"));
          }
          String responseDb = dbHandle.updateSchemaRequest(schemaRequest, userDetails);
          if (responseDb.equals(ApiResultStatus.SUCCESS.value)) {
            updateAuditAndHistory(
                userDetails, tenantId, schemaRequest, dbHandle, registerSchemaCustomResponse);
          }
  
          // send mail to producers and consumers
          notifySubscribers(schemaRequest, tenantId);
  
          mailService.sendMail(
              schemaRequest.getTopicname(),
              null,
              "",
              schemaRequest.getRequestor(),
              schemaRequest.getApprover(),
              schemaRequest.getTeamId(),
              dbHandle,
              SCHEMA_REQUEST_APPROVED,
              commonUtilsService.getLoginUrl());
          return ApiResultStatus.SUCCESS.value.equals(responseDb)
              ? ApiResponse.ok(responseDb)
              : ApiResponse.notOk(responseDb);
        } catch (Exception e) {
          log.error("Exception:", e);
          throw new KlawException(e.getMessage());
        }
  
      } else {
        String errStr = response.getBody().getMessage();
        if (errStr.length() > 100) {
          errStr = errStr.substring(0, 98) + "...";
        }
        return ApiResponse.notOk(String.format(SCHEMA_ERR_102, errStr));
      }
    }
  
    private void updateAuditAndHistory(
        String userDetails,
        int tenantId,
        SchemaRequest schemaRequest,
        HandleDbRequests dbHandle,
        Map<String, Object> registerSchemaCustomResponse) {
      saveToTopicHistory(userDetails, tenantId, schemaRequest);
      dbHandle.insertIntoActivityLog(
          RequestEntityType.SCHEMA.value,
          tenantId,
          schemaRequest.getRequestOperationType(),
          schemaRequest.getTeamId(),
          "Topic : "
              + schemaRequest.getTopicname()
              + " version : "
              + registerSchemaCustomResponse.get("version")
              + " id : "
              + registerSchemaCustomResponse.get("id"),
          schemaRequest.getEnvironment(),
          schemaRequest.getRequestor());
    }
  
    public void notifySubscribers(SchemaRequest schemaRequest, int tenantId) {
      String topic = schemaRequest.getTopicname();
      String schemaRequestEnvironment = schemaRequest.getEnvironment();
  
      Optional<Env> optSchemaEnv =
          manageDatabase.getEnv(tenantId, Integer.valueOf(schemaRequestEnvironment));
      Optional<Env> optionalKafkaEnv = Optional.empty();
      // add null pointer checks.
      if (optSchemaEnv.isPresent()) {
        optionalKafkaEnv =
            manageDatabase.getEnv(
                tenantId, Integer.valueOf(optSchemaEnv.get().getAssociatedEnv().getId()));
      }
  
      // get all producer and consumer acls for topic, schemaRequestEnvironment
      List<Acl> acls = new ArrayList<>();
      if (optionalKafkaEnv.isPresent()) {
        acls =
            manageDatabase
                .getHandleDbRequests()
                .getSyncAcls(optionalKafkaEnv.get().getId(), topic, tenantId);
      }
  
      // get all teams to be notified based on above acls
      List<Integer> teamIdsToBeNotified = acls.stream().map(Acl::getTeamId).toList();
      List<String> teamsToBeNotified =
          manageDatabase.getTeamObjForTenant(tenantId).stream()
              .filter(team -> teamIdsToBeNotified.contains(team.getTeamId()))
              .map(Team::getTeammail)
              .toList();
  
      Optional<Team> optionalOwnerTeam =
          manageDatabase.getTeamObjForTenant(tenantId).stream()
              .filter(team -> Objects.equals(team.getTeamId(), schemaRequest.getTeamId()))
              .findFirst();
  
      // send notifications
      if (optionalOwnerTeam.isPresent() && optionalKafkaEnv.isPresent()) {
        mailService.notifySubscribersOnSchemaChange(
            SCHEMA_APPROVED_NOTIFY_SUBSCRIBERS,
            topic,
            optionalKafkaEnv.get().getName(),
            optionalOwnerTeam.get().getTeamname(),
            teamsToBeNotified,
            optionalOwnerTeam.get().getTeammail(),
            tenantId,
            commonUtilsService.getLoginUrl());
      }
    }
  
    private void saveToTopicHistory(String userDetails, int tenantId, SchemaRequest schemaRequest) {
      manageDatabase.getKafkaEnvList(tenantId).stream()
          .filter(
              kafkaEnv -> {
                if (kafkaEnv.getAssociatedEnv() != null) {
                  return kafkaEnv.getAssociatedEnv().getId().equals(schemaRequest.getEnvironment());
                }
                return false;
              })
          .findFirst()
          .ifPresent(
              env ->
                  commonUtilsService.saveTopicHistory(
                      schemaRequest.getRequestOperationType(),
                      schemaRequest.getTopicname(),
                      env.getId(),
                      schemaRequest.getRequestor(),
                      schemaRequest.getRequesttime(),
                      schemaRequest.getTeamId(),
                      userDetails,
                      tenantId,
                      RequestEntityType.SCHEMA.name(),
                      RequestEntityType.SCHEMA.name()
                          + " "
                          + schemaRequest.getRequestOperationType()
                          + " Version : "
                          + schemaRequest.getSchemaversion()));
    }
  
    @PermissionAllowed(permissionAllowed = {PermissionType.APPROVE_SCHEMAS})
    public ApiResponse execSchemaRequestsDecline(String avroSchemaId, String reasonForDecline)
        throws KlawException {
      log.info("execSchemaRequestsDecline {}", avroSchemaId);
      String userDetails = getUserName();
      if (commonUtilsService.isNotAuthorizedUser(
          commonUtilsService.getPrincipal(), PermissionType.APPROVE_SCHEMAS)) {
        return ApiResponse.NOT_AUTHORIZED;
      }
      int tenantId = commonUtilsService.getTenantId(getUserName());
      HandleDbRequests dbHandle = manageDatabase.getHandleDbRequests();
      SchemaRequest schemaRequest =
          dbHandle.getSchemaRequest(Integer.parseInt(avroSchemaId), tenantId);
      final Set<String> allowedEnvIdSet = commonUtilsService.getEnvsFromUserId(getUserName());
  
      if (!allowedEnvIdSet.contains(schemaRequest.getEnvironment())) {
        return ApiResponse.NOT_AUTHORIZED;
      }
  
      try {
        String responseDb = dbHandle.updateSchemaRequestDecline(schemaRequest, userDetails);
        mailService.sendMail(
            schemaRequest.getTopicname(),
            null,
            reasonForDecline,
            schemaRequest.getRequestor(),
            schemaRequest.getApprover(),
            schemaRequest.getTeamId(),
            dbHandle,
            SCHEMA_REQUEST_DENIED,
            commonUtilsService.getLoginUrl());
        return ApiResultStatus.SUCCESS.value.equals(responseDb)
            ? ApiResponse.ok(responseDb)
            : ApiResponse.notOk(responseDb);
      } catch (Exception e) {
        throw new KlawException(e.getMessage());
      }
    }
  
    public ApiResponse promoteSchema(SchemaPromotion schemaPromotion) throws Exception {
      String userDetails = getUserName();
  
      if (commonUtilsService.isNotAuthorizedUser(
          commonUtilsService.getPrincipal(), PermissionType.REQUEST_CREATE_SCHEMAS)) {
        return ApiResponse.NOT_AUTHORIZED;
      }
  
      Integer userTeamId = commonUtilsService.getTeamId(userDetails);
      int tenantId = commonUtilsService.getTenantId(getUserName());
  
      if (!userAndTopicOwnerAreOnTheSameTeam(schemaPromotion.getTopicName(), userTeamId, tenantId)) {
        return ApiResponse.notOk(SCHEMA_ERR_103);
      }
  
      SchemaRequestModel schemaRequest = buildSchemaRequestFromPromotionRequest(schemaPromotion);
      Optional<Env> optionalEnv = getSchemaEnvFromKafkaEnvId(schemaPromotion.getSourceEnvironment());
  
      if (optionalEnv.isEmpty()) {
        return ApiResponse.notOk(SCHEMA_ERR_104);
      }
      Env schemaSourceEnv = optionalEnv.get();
      SortedMap<Integer, Map<String, Object>> schemaObjects =
          getSchemasFromTopicName(schemaPromotion.getTopicName(), tenantId, schemaSourceEnv);
      log.debug(
          "getSchemaVersion {}, schemaObjects keySet {}",
          schemaPromotion.getSchemaVersion(),
          schemaObjects.keySet());
      Map<String, Object> schemaObject =
          schemaObjects.get(Integer.valueOf(schemaPromotion.getSchemaVersion()));
      // Pretty Print the Json String so that it can be seen clearly in the UI.
      schemaRequest.setSchemafull(prettyPrintUglyJsonString((String) schemaObject.get("schema")));
      if (schemaObject.containsKey(SCHEMA_TYPE)) {
        schemaRequest.setSchemaType(SchemaType.of(schemaObject.get(SCHEMA_TYPE).toString()));
      } else {
        schemaRequest.setSchemaType(SchemaType.AVRO);
      }
      // sending request operation type along with function call
      return uploadSchema(schemaRequest, RequestOperationType.PROMOTE);
    }
  
    private boolean userAndTopicOwnerAreOnTheSameTeam(
        String topicName, Integer userTeamId, Integer tenantId) {
      List<Topic> topicsSearchList = commonUtilsService.getTopicsForTopicName(topicName, tenantId);
  
      // tenant filtering
      Integer topicOwnerTeam = topicsSearchList.get(0).getTeamId();
      return Objects.equals(userTeamId, topicOwnerTeam);
    }
  
    private SortedMap<Integer, Map<String, Object>> getSchemasFromTopicName(
        String topicName, int tenantId, Env schemaEnv) throws Exception {
      KwClusters kwClusters =
          manageDatabase
              .getClusters(KafkaClustersType.SCHEMA_REGISTRY, tenantId)
              .get(schemaEnv.getClusterId());
  
      return clusterApiService.getAvroSchema(
          kwClusters.getBootstrapServers(),
          kwClusters.getProtocol(),
          kwClusters.getClusterName() + kwClusters.getClusterId(),
          topicName,
          tenantId);
    }
  
    private Optional<Env> getSchemaEnvFromKafkaEnvId(String envId) {
      return manageDatabase
          .getSchemaRegEnvList(commonUtilsService.getTenantId(getUserName()))
          .stream()
          .filter(
              env -> env.getAssociatedEnv() != null && env.getAssociatedEnv().getId().equals(envId))
          .findFirst();
    }
  
    private SchemaRequestModel buildSchemaRequestFromPromotionRequest(
        SchemaPromotion schemaPromotion) {
      SchemaRequestModel schemaRequest = new SchemaRequestModel();
      // setup schema Request
      schemaRequest.setRemarks(schemaPromotion.getRemarks());
  
      schemaRequest.setSchemaversion(schemaPromotion.getSchemaVersion());
      schemaRequest.setTopicname(schemaPromotion.getTopicName());
      schemaRequest.setForceRegister(schemaPromotion.isForceRegister());
      schemaRequest.setEnvironment(schemaPromotion.getTargetEnvironment());
      return schemaRequest;
    }
  
    public ApiResponse uploadSchema(
        SchemaRequestModel schemaRequest, RequestOperationType requestOperationType)
        throws KlawException {
      log.debug("uploadSchema {}, requestOperationType {}", schemaRequest, requestOperationType);
      String userName = getUserName();
  
      if (commonUtilsService.isNotAuthorizedUser(
          commonUtilsService.getPrincipal(), PermissionType.REQUEST_CREATE_SCHEMAS)) {
        return ApiResponse.NOT_AUTHORIZED;
      }
      schemaRequest.setRequestor(userName);
  
      int tenantId = commonUtilsService.getTenantId(getUserName());
      Optional<Env> schemaEnv = getSchemaEnvFromKafkaEnvId(schemaRequest.getEnvironment());
      if (schemaEnv.isPresent()) {
        schemaRequest.setEnvironment(schemaEnv.get().getId());
        // Check to see if the schema Registry exists
        Optional<Env> schemaReg =
            manageDatabase.getEnv(
                schemaEnv.get().getTenantId(), Integer.valueOf(schemaEnv.get().getId()));
        if (schemaReg.isEmpty()) {
          return ApiResponse.notOk(SCHEMA_ERR_109);
        }
  
      } else {
        return ApiResponse.notOk(SCHEMA_ERR_108);
      }
  
      // If force register is not set validate the schema
      if (validateCompatiblityOnSave
          && (schemaRequest.getForceRegister() == null || !schemaRequest.getForceRegister())) {
        // check if Schema is valid
        ApiResponse isValid = validateSchema(schemaRequest);
        if (!isValid.isSuccess()) {
          // Return on Failure response
          return isValid;
        }
      }
  
      try {
        new ObjectMapper().readValue(schemaRequest.getSchemafull(), Object.class);
      } catch (IOException e) {
        log.error("Exception:", e);
        return ApiResponse.notOk(SCHEMA_ERR_105);
      }
  
      Integer userTeamId = commonUtilsService.getTeamId(userName);
      schemaRequest.setTeamId(userTeamId);
  
      if (!userAndTopicOwnerAreOnTheSameTeam(schemaRequest.getTopicname(), userTeamId, tenantId)) {
        return ApiResponse.notOk(SCHEMA_ERR_106);
      }
  
      List<SchemaRequest> schemaReqs =
          manageDatabase
              .getHandleDbRequests()
              .getAllSchemaRequests(
                  false, userName, tenantId, null, null, null, null, null, false, false);
  
      // tenant filtering
      final Set<String> allowedEnvIdSet = commonUtilsService.getEnvsFromUserId(getUserName());
      if (schemaReqs != null)
        schemaReqs =
            schemaReqs.stream()
                .filter(request -> allowedEnvIdSet.contains(request.getEnvironment()))
                .collect(Collectors.toList());
  
      // request status filtering
      if (schemaReqs != null && schemaRequest.getRequestId() == null) {
        schemaReqs =
            schemaReqs.stream()
                .filter(
                    schemaRequest1 ->
                        RequestStatus.CREATED.value.equals(schemaRequest1.getRequestStatus())
                            && Objects.equals(
                                schemaRequest1.getTopicname(), schemaRequest.getTopicname()))
                .collect(Collectors.toList());
        if (schemaReqs.size() > 0) {
          return ApiResponse.notOk(SCHEMA_ERR_107);
        }
      } else if (schemaReqs != null && schemaRequest.getRequestId() != null) {
        // edit schema request
        Optional<SchemaRequest> optionalSchemaRequest =
            schemaReqs.stream()
                .filter(schemaReq -> schemaReq.getReq_no().equals(schemaRequest.getRequestId()))
                .findFirst();
  
        if (optionalSchemaRequest.isPresent()) {
          // verify if request is in CREATED state
          if (!optionalSchemaRequest.get().getRequestStatus().equals(RequestStatus.CREATED.value)) {
            return ApiResponse.notOk(SCHEMA_ERR_110);
          }
  
          // verify if the edit request is being submitted by request owner
          if (!optionalSchemaRequest.get().getRequestor().equals(schemaRequest.getRequestor())) {
            return ApiResponse.notOk(SCHEMA_ERR_111);
          }
        }
      }
  
      SchemaRequest schemaRequestDao = new SchemaRequest();
      copyProperties(schemaRequest, schemaRequestDao);
      schemaRequestDao.setReq_no(schemaRequest.getRequestId());
      schemaRequestDao.setRequestOperationType(requestOperationType.value);
      HandleDbRequests dbHandle = manageDatabase.getHandleDbRequests();
      schemaRequestDao.setTenantId(tenantId);
      try {
        String responseDb = dbHandle.requestForSchema(schemaRequestDao);
        mailService.sendMail(
            schemaRequest.getTopicname(),
            null,
            "",
            schemaRequest.getRequestor(),
            null,
            schemaRequest.getTeamId(),
            dbHandle,
            SCHEMA_REQUESTED,
            commonUtilsService.getLoginUrl());
  
        return ApiResultStatus.SUCCESS.value.equals(responseDb)
            ? ApiResponse.ok(responseDb)
            : ApiResponse.notOk(responseDb);
      } catch (Exception e) {
        log.error("Exception:", e);
        throw new KlawException(e.getMessage());
      }
    }
  
    public ApiResponse validateSchema(SchemaRequestModel schemaRequest) throws KlawException {
      log.info("validateSchema {}", schemaRequest);
      String userDetails = getUserName();
      int tenantId = commonUtilsService.getTenantId(userDetails);
      try {
        return clusterApiService
            .validateSchema(
                schemaRequest.getSchemafull(),
                schemaRequest.getEnvironment(),
                schemaRequest.getTopicname(),
                tenantId,
                schemaRequest.getSchemaType())
            .getBody();
      } catch (Exception e) {
        log.error("Exception:", e);
        throw new KlawException(e.getMessage());
      }
    }
  
    public SchemaRequestsResponseModel getSchemaRequest(Integer schemaReqId) {
      String userName = getUserName();
      int tenantId = commonUtilsService.getTenantId(userName);
      SchemaRequest schemaRequest =
          manageDatabase.getHandleDbRequests().getSchemaRequest(schemaReqId, tenantId);
  
      if (schemaRequest == null) {
        return null;
      } else {
        SchemaRequestsResponseModel schemaRequestsResponseModel = new SchemaRequestsResponseModel();
        copyProperties(schemaRequest, schemaRequestsResponseModel);
        schemaRequestsResponseModel.setRequestStatus(
            RequestStatus.of(schemaRequest.getRequestStatus()));
        schemaRequestsResponseModel.setRequestOperationType(
            RequestOperationType.of(schemaRequest.getRequestOperationType()));
        // Set kafka env name, id
        manageDatabase.getKafkaEnvList(tenantId).stream()
            .filter(
                kafkaEnv -> {
                  if (kafkaEnv.getAssociatedEnv() != null) {
                    return kafkaEnv.getAssociatedEnv().getId().equals(schemaRequest.getEnvironment());
                  }
                  return false;
                })
            .findFirst()
            .ifPresent(
                env -> {
                  schemaRequestsResponseModel.setEnvironmentName(env.getName());
                  schemaRequestsResponseModel.setEnvironment(env.getId());
                });
  
        schemaRequestsResponseModel.setTeamname(
            manageDatabase.getTeamNameFromTeamId(tenantId, schemaRequest.getTeamId()));
  
        return schemaRequestsResponseModel;
      }
    }
  
    private String prettyPrintUglyJsonString(String json) {
      ObjectMapper mapper = new ObjectMapper();
  
      try {
        return mapper.writerWithDefaultPrettyPrinter().writeValueAsString(mapper.readTree(json));
      } catch (JsonProcessingException e) {
        log.error("Unable to pretty print json : ", e);
      }
      return json;
    }
  
    private String getUserName() {
      return mailService.getUserName(commonUtilsService.getPrincipal());
    }
  }
  