package io.aiven.klaw.service;

import static io.aiven.klaw.error.KlawErrorMessages.REQ_ERR_101;
import static io.aiven.klaw.error.KlawErrorMessages.TOPICS_108;
import static io.aiven.klaw.error.KlawErrorMessages.TOPICS_ERR_102;
import static io.aiven.klaw.error.KlawErrorMessages.TOPICS_ERR_103;
import static io.aiven.klaw.error.KlawErrorMessages.TOPICS_ERR_104;
import static io.aiven.klaw.error.KlawErrorMessages.TOPICS_ERR_105;
import static io.aiven.klaw.error.KlawErrorMessages.TOPICS_ERR_106;
import static io.aiven.klaw.error.KlawErrorMessages.TOPICS_ERR_107;
import static io.aiven.klaw.error.KlawErrorMessages.TOPICS_ERR_109;
import static io.aiven.klaw.error.KlawErrorMessages.TOPICS_ERR_110;
import static io.aiven.klaw.error.KlawErrorMessages.TOPICS_ERR_111;
import static io.aiven.klaw.error.KlawErrorMessages.TOPICS_ERR_112;
import static io.aiven.klaw.error.KlawErrorMessages.TOPICS_ERR_113;
import static io.aiven.klaw.error.KlawErrorMessages.TOPICS_ERR_114;
import static io.aiven.klaw.error.KlawErrorMessages.TOPICS_ERR_115;
import static io.aiven.klaw.error.KlawErrorMessages.TOPICS_ERR_116;
import static io.aiven.klaw.error.KlawErrorMessages.TOPICS_ERR_117;
import static io.aiven.klaw.error.KlawErrorMessages.TOPICS_VLD_ERR_121;
import static io.aiven.klaw.error.KlawErrorMessages.TOPICS_VLD_ERR_125;
import static io.aiven.klaw.error.KlawErrorMessages.TOPICS_VLD_ERR_126;
import static io.aiven.klaw.helpers.KwConstants.KLAW_OPTIONAL_PERMISSION_NEW_TOPIC_CREATION_KEY;
import static io.aiven.klaw.helpers.KwConstants.ORDER_OF_TOPIC_ENVS;
import static io.aiven.klaw.helpers.UtilMethods.updateEnvStatus;
import static io.aiven.klaw.service.MailUtils.MailType.TOPIC_CLAIM_REQUESTED;
import static io.aiven.klaw.service.MailUtils.MailType.TOPIC_CREATE_REQUESTED;
import static io.aiven.klaw.service.MailUtils.MailType.TOPIC_DELETE_REQUESTED;
import static io.aiven.klaw.service.MailUtils.MailType.TOPIC_PROMOTION_REQUESTED;
import static io.aiven.klaw.service.MailUtils.MailType.TOPIC_REQUEST_APPROVED;
import static io.aiven.klaw.service.MailUtils.MailType.TOPIC_REQUEST_DENIED;
import static io.aiven.klaw.service.MailUtils.MailType.TOPIC_UPDATE_REQUESTED;
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.MessageSchema;
import io.aiven.klaw.dao.Topic;
import io.aiven.klaw.dao.TopicRequest;
import io.aiven.klaw.dao.UserInfo;
import io.aiven.klaw.error.KlawException;
import io.aiven.klaw.error.KlawNotAuthorizedException;
import io.aiven.klaw.helpers.DisplayHelper;
import io.aiven.klaw.helpers.HandleDbRequests;
import io.aiven.klaw.helpers.HtmlSanitizer;
import io.aiven.klaw.helpers.KlawResourceUtils;
import io.aiven.klaw.helpers.Pager;
import io.aiven.klaw.helpers.UtilMethods;
import io.aiven.klaw.model.ApiResponse;
import io.aiven.klaw.model.ResourceHistory;
import io.aiven.klaw.model.TopicBaseConfig;
import io.aiven.klaw.model.TopicConfigEntry;
import io.aiven.klaw.model.TopicConfiguration;
import io.aiven.klaw.model.TopicConfigurationRequest;
import io.aiven.klaw.model.TopicInfo;
import io.aiven.klaw.model.enums.*;
import io.aiven.klaw.model.requests.TopicRequestModel;
import io.aiven.klaw.model.response.TopicConfig;
import io.aiven.klaw.model.response.TopicDetailsPerEnv;
import io.aiven.klaw.model.response.TopicRequestsResponseModel;
import io.aiven.klaw.model.response.TopicTeamResponse;
import io.aiven.klaw.validation.PermissionAllowed;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.function.Function;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.math.NumberUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;

@Service
@Slf4j
public class TopicControllerService {

  public static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
  @Autowired private final ClusterApiService clusterApiService;

  @Autowired ManageDatabase manageDatabase;

  @Autowired private final MailUtils mailService;

  @Autowired private CommonUtilsService commonUtilsService;

  @Autowired private RolesPermissionsControllerService rolesPermissionsControllerService;

  private int topicCounter = 0;

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

  public ApiResponse createTopicsCreateRequest(TopicRequestModel topicRequestReq)
      throws KlawException, KlawNotAuthorizedException {
    log.info("createTopicsCreateRequest {}", topicRequestReq);
    return createTopicRequest(topicRequestReq);
  }

  public ApiResponse createTopicsUpdateRequest(TopicRequestModel topicRequestReq)
      throws KlawException, KlawNotAuthorizedException {
    log.info("createTopicsUpdateRequest {}", topicRequestReq);
    return createTopicRequest(topicRequestReq);
  }

  private ApiResponse createTopicRequest(TopicRequestModel topicRequestReq) throws KlawException {
    String userName = getUserName();

    topicRequestReq.setRequestor(userName);
    topicRequestReq.setTeamId(commonUtilsService.getTeamId(userName));

    HandleDbRequests dbHandle = manageDatabase.getHandleDbRequests();
    TopicRequest topicRequestDao = new TopicRequest();

    // Editing a topic request
    if (topicRequestReq.getRequestId() != null) {
      topicRequestDao =
          getTopicRequestFromTopicId(
              topicRequestReq.getRequestId(), commonUtilsService.getTenantId(userName));

      // check if the topic request owned by the logged-in user
      if (!topicRequestDao.getRequestor().equals(userName)) {
        return ApiResponse.notOk(TOPICS_VLD_ERR_121);
      }

      topicRequestDao.setTopicpartitions(topicRequestReq.getTopicpartitions());
      topicRequestDao.setReplicationfactor(topicRequestReq.getReplicationfactor());

      if (topicRequestDao.getRequestOperationType().equals(RequestOperationType.CREATE.value)) {
        topicRequestDao.setTopicname(topicRequestReq.getTopicname());
        topicRequestDao.setEnvironment(topicRequestReq.getEnvironment());
        topicRequestDao.setDescription(topicRequestReq.getDescription());
        topicRequestDao.setRemarks(topicRequestReq.getRemarks());
      }
    } else {
      copyProperties(topicRequestReq, topicRequestDao);
    }
    topicRequestDao.setRequestOperationType(topicRequestReq.getRequestOperationType().value);

    mapAdvancedTopicConfiguration(topicRequestReq, topicRequestDao);
    topicRequestDao.setTenantId(commonUtilsService.getTenantId(userName));

    String result;
    try {
      result = dbHandle.requestForTopic(topicRequestDao).get("result");
      // default to Topic_CREATE which is the old hard coded
    } catch (DataIntegrityViolationException e) {
      log.error("Error in creating a topic ", e);
      if (e.getCause().getMessage().contains("Value too long for column")) {
        return ApiResponse.notOk(TOPICS_VLD_ERR_125);
      }
      throw new KlawException(TOPICS_VLD_ERR_126 + e);
    }

    mailService.sendMail(
        topicRequestReq.getTopicname(),
        null,
        "",
        userName,
        null,
        NumberUtils.toInt(topicRequestReq.getApprovingTeamId(), -1),
        dbHandle,
        getMailType(topicRequestReq),
        commonUtilsService.getLoginUrl());

    return ApiResultStatus.SUCCESS.value.equals(result)
        ? ApiResponse.ok(result)
        : ApiResponse.notOk(result);
  }

  private static MailUtils.MailType getMailType(TopicRequestModel topicRequestReq) {
    // default to Topic_CREATE which is the old hard coded
    return topicRequestReq.getRequestOperationType().equals(RequestOperationType.CREATE)
        ? TOPIC_CREATE_REQUESTED
        : topicRequestReq.getRequestOperationType().equals(RequestOperationType.PROMOTE)
            ? TOPIC_PROMOTION_REQUESTED
            : topicRequestReq.getRequestOperationType().equals(RequestOperationType.UPDATE)
                ? TOPIC_UPDATE_REQUESTED
                : TOPIC_CREATE_REQUESTED;
  }

  private void mapAdvancedTopicConfiguration(
      TopicRequestModel topicRequestModel, TopicRequest topicRequestDao) throws KlawException {
    Map<String, String> topicConfigs = new HashMap<>();
    try {
      List<TopicConfigEntry> advancedTopicConfigEntries =
          topicRequestModel.getAdvancedTopicConfigEntries();
      if (null != advancedTopicConfigEntries) {
        for (TopicConfigEntry advancedTopicConfig : advancedTopicConfigEntries) {
          topicConfigs.put(
              advancedTopicConfig.getConfigKey(), advancedTopicConfig.getConfigValue());
        }
      }
      TopicConfigurationRequest topicConfigurationRequest =
          TopicConfigurationRequest.builder().advancedTopicConfiguration(topicConfigs).build();
      topicRequestDao.setJsonParams(OBJECT_MAPPER.writeValueAsString(topicConfigurationRequest));
    } catch (JsonProcessingException e) {
      log.error("Error in processing topic configs ", e);
      throw new KlawException(TOPICS_ERR_102);
    }
  }

  // create a request to delete topic.
  public ApiResponse createTopicDeleteRequest(
      String topicName, String envId, boolean deleteAssociatedSchema)
      throws KlawException, KlawNotAuthorizedException {
    log.info(
        "createTopicDeleteRequest topicName {} envId {} deleteAssociatedSchema {}",
        topicName,
        envId,
        deleteAssociatedSchema);
    String userName = getUserName();

    int tenantId = commonUtilsService.getTenantId(userName);
    HandleDbRequests dbHandle = manageDatabase.getHandleDbRequests();

    // check if already a delete topic request exists
    if (dbHandle.existsTopicRequests(topicName, envId, RequestStatus.CREATED.value, tenantId)) {
      return ApiResponse.notOk(TOPICS_ERR_103);
    }

    List<Topic> topics = commonUtilsService.getTopicsForTopicName(topicName, tenantId);

    // check if you are part of the same team to delete this request
    Integer userTeamId = commonUtilsService.getTeamId(userName);
    if (topics != null
        && !topics.isEmpty()
        && !Objects.equals(topics.get(0).getTeamId(), userTeamId)) {
      return ApiResponse.notOk(TOPICS_ERR_104);
    }

    TopicRequest topicRequestReq = new TopicRequest();
    topicRequestReq.setRequestor(userName);
    topicRequestReq.setTeamId(userTeamId);
    topicRequestReq.setEnvironment(envId);
    topicRequestReq.setTopicname(topicName);
    topicRequestReq.setRequestOperationType(RequestOperationType.DELETE.value);
    topicRequestReq.setTenantId(tenantId);
    topicRequestReq.setDeleteAssociatedSchema(deleteAssociatedSchema);

    Optional<Topic> topicOb = Optional.empty();
    if (topics != null) {
      topicOb =
          topics.stream()
              .filter(
                  topic -> Objects.equals(topic.getEnvironment(), topicRequestReq.getEnvironment()))
              .findFirst();
    }
    if (topicOb.isPresent()) {
      // Check if any existing subscriptions for this topic
      List<Acl> acls =
          dbHandle.getSyncAcls(
              topicRequestReq.getEnvironment(), topicRequestReq.getTopicname(), tenantId);
      if (!acls.isEmpty()) {
        return ApiResponse.notOk(TOPICS_ERR_105);
      }

      topicRequestReq.setTopicpartitions(topicOb.get().getNoOfPartitions());
      topicRequestReq.setReplicationfactor(topicOb.get().getNoOfReplicas());
      try {
        mailService.sendMail(
            topicRequestReq.getTopicname(),
            null,
            "",
            userName,
            null,
            NumberUtils.toInt(topicRequestReq.getApprovingTeamId(), -1),
            dbHandle,
            TOPIC_DELETE_REQUESTED,
            commonUtilsService.getLoginUrl());

        String result = dbHandle.requestForTopic(topicRequestReq).get("result");

        return ApiResultStatus.SUCCESS.value.equals(result)
            ? ApiResponse.ok(result)
            : ApiResponse.notOk(result);
      } catch (Exception e) {
        log.error("Error ", e);
        throw new KlawException(e.getMessage());
      }
    } else {
      log.error("Topic not found : {}", topicName);
      return ApiResponse.notOk(String.format(TOPICS_ERR_106, topicName));
    }
  }

  public ApiResponse createClaimTopicRequest(String topicName, String envId) throws KlawException {
    log.info("createClaimTopicRequest {}", topicName);
    String userName = getUserName();

    HandleDbRequests dbHandle = manageDatabase.getHandleDbRequests();
    TopicRequest topicRequestReq = new TopicRequest();
    int tenantId = commonUtilsService.getTenantId(userName);

    if (dbHandle.existsTopicRequests(topicName, envId, RequestStatus.CREATED.value, tenantId)) {
      return ApiResponse.notOk(TOPICS_ERR_107);
    }

    List<Topic> topics = commonUtilsService.getTopicsForTopicName(topicName, tenantId);
    Integer topicOwnerTeamId = topics.get(0).getTeamId();
    Optional<UserInfo> topicOwnerContact =
        dbHandle.getAllUsersInfo(tenantId).stream()
            .filter(user -> Objects.equals(user.getTeamId(), topicOwnerTeamId))
            .findFirst();

    Integer userTeamId = commonUtilsService.getTeamId(userName);

    topicRequestReq.setRequestor(userName);
    topicRequestReq.setTeamId(userTeamId);
    topicRequestReq.setTenantId(tenantId);
    topicRequestReq.setEnvironment(envId);
    topicRequestReq.setTopicname(topicName);
    topicRequestReq.setRequestOperationType(RequestOperationType.CLAIM.value);
    topicRequestReq.setApprovingTeamId(topicOwnerTeamId + "");
    topicRequestReq.setRemarks(TOPICS_108);

    String approverName = null;
    Integer approverTeamId = null;
    if (topicOwnerContact.isPresent()) {
      approverName = topicOwnerContact.get().getUsername();
      approverTeamId = topicOwnerContact.get().getTeamId();
    }

    mailService.sendMail(
        topicRequestReq.getTopicname(),
        null,
        "",
        userName,
        approverName,
        approverTeamId,
        dbHandle,
        TOPIC_CLAIM_REQUESTED,
        commonUtilsService.getLoginUrl());

    try {
      String result = dbHandle.requestForTopic(topicRequestReq).get("result");
      return ApiResultStatus.SUCCESS.value.equals(result)
          ? ApiResponse.ok(result)
          : ApiResponse.notOk(result);
    } catch (Exception e) {
      log.error("Error ", e);
      throw new KlawException(e.getMessage());
    }
  }

  public List<TopicRequestsResponseModel> getTopicRequests(
      String pageNo,
      String currentPage,
      RequestOperationType requestOperationType,
      String requestsType,
      String env,
      String wildcardSearch,
      Order order,
      boolean isMyRequest) {
    log.debug("getTopicRequests page {} requestsType {}", pageNo, requestsType);
    String userName = getUserName();
    List<TopicRequest> topicReqs =
        manageDatabase
            .getHandleDbRequests()
            .getAllTopicRequests(
                userName,
                requestsType,
                requestOperationType,
                env,
                wildcardSearch,
                isMyRequest,
                commonUtilsService.getTenantId(userName));

    // tenant filtering
    topicReqs = filterByTenantAndSort(order, userName, topicReqs);

    topicReqs =
        Pager.getItemsList(
            pageNo,
            currentPage,
            10,
            topicReqs,
            (pageContext, activityLog) -> {
              activityLog.setAllPageNos(pageContext.getAllPageNos());
              activityLog.setTotalNoPages(pageContext.getTotalPages());
              activityLog.setCurrentPage(pageContext.getPageNo());
              activityLog.setEnvironmentName(getEnvDetails(activityLog.getEnvironment()).getName());
              return activityLog;
            });
    return getTopicRequestModels(topicReqs);
  }

  private List<TopicRequest> filterByTenantAndSort(
      Order order, String userName, List<TopicRequest> topicReqs) {
    try {
      final Set<String> allowedEnvIdSet = commonUtilsService.getEnvsFromUserId(userName);
      topicReqs =
          topicReqs.stream()
              .filter(topicRequest -> allowedEnvIdSet.contains(topicRequest.getEnvironment()))
              .sorted(getPreferredOrdering(order))
              .collect(Collectors.toList());
    } catch (Exception e) {
      log.error("No environments/clusters found.", e);
      return new ArrayList<>();
    }
    return topicReqs;
  }

  private Comparator<TopicRequest> getPreferredOrdering(Order order) {
    return switch (order) {
      case ASC_REQUESTED_TIME -> compareByTime();
      case DESC_REQUESTED_TIME -> Collections.reverseOrder(compareByTime());
    };
  }

  private static Comparator<TopicRequest> compareByTime() {
    return Comparator.comparing(TopicRequest::getRequesttime);
  }

  private TopicRequestsResponseModel setRequestorPermissions(
      TopicRequestsResponseModel req, String userName) {
    log.debug(
        " My Topic Status {} and userName {} and userName {}",
        req.getRequestStatus(),
        userName,
        req.getRequestor());
    if (RequestStatus.CREATED == req.getRequestStatus()
        && userName != null
        && userName.equals(req.getRequestor())) {
      req.setDeletable(true);
      req.setEditable(true);
    }

    return req;
  }

  public TopicTeamResponse getTopicTeamOnly(String topicName, AclPatternType patternType) {
    log.debug("getTopicTeamOnly {} {}", topicName, patternType);
    TopicTeamResponse topicTeamResponse = new TopicTeamResponse();
    topicTeamResponse.setStatus(false);
    String userName = getUserName();
    int tenantId = commonUtilsService.getTenantId(userName);
    List<Topic> topics;

    if (AclPatternType.PREFIXED == patternType) {
      topics = manageDatabase.getHandleDbRequests().getAllTopics(tenantId);

      // tenant filtering
      Set<String> allowedEnvIdSet = new HashSet<>(commonUtilsService.getEnvsFromUserId(userName));
      List<Topic> allTopicsStartingWithPattern = new ArrayList<>();

      topics.stream()
          .filter(topicRequest -> allowedEnvIdSet.contains(topicRequest.getEnvironment()))
          .distinct()
          .forEach(
              topic -> {
                if (topic.getTopicname().startsWith(topicName)) {
                  allTopicsStartingWithPattern.add(topic);
                }
              });

      if (allTopicsStartingWithPattern.isEmpty()) {
        topicTeamResponse.setError(TOPICS_ERR_109);
        return topicTeamResponse;
      }

      Set<Integer> stringTeamsFound = new HashSet<>();
      allTopicsStartingWithPattern.forEach(top -> stringTeamsFound.add(top.getTeamId()));

      if (stringTeamsFound.size() > 1) {
        topicTeamResponse.setError(TOPICS_ERR_110);
      } else {
        int teamId = stringTeamsFound.iterator().next();
        topicTeamResponse.setTeam(manageDatabase.getTeamNameFromTeamId(tenantId, teamId));
        topicTeamResponse.setTeamId(teamId);
      }
    } else {
      topics = commonUtilsService.getTopicsForTopicName(topicName, tenantId);

      if (!topics.isEmpty()) {
        topicTeamResponse.setTeam(
            manageDatabase.getTeamNameFromTeamId(tenantId, topics.get(0).getTeamId()));
        topicTeamResponse.setTeamId(topics.get(0).getTeamId());
        topicTeamResponse.setStatus(true);
      } else {
        topicTeamResponse.setError(TOPICS_ERR_111);
      }
    }
    return topicTeamResponse;
  }

  public List<TopicRequestsResponseModel> getTopicRequestsForApprover(
      String pageNo,
      String currentPage,
      String requestsType,
      Integer teamId,
      String env,
      RequestOperationType requestOperationType,
      String wildcardSearch,
      Order order) {
    if (log.isDebugEnabled()) {
      log.debug(
          "getCreatedTopicRequests pageNo {} requestsType {} teamId {} env {} requestOperationType {} wildcardSearch {}",
          pageNo,
          requestsType,
          teamId,
          env,
          requestOperationType,
          wildcardSearch);
    }

    String userName = getUserName();
    List<TopicRequest> createdTopicReqList;
    int tenantId = commonUtilsService.getTenantId(userName);
    // get requests relevant to your teams or all teams
    if (commonUtilsService.isNotAuthorizedUser(
        commonUtilsService.getPrincipal(), PermissionType.APPROVE_ALL_REQUESTS_TEAMS)) {
      createdTopicReqList =
          manageDatabase
              .getHandleDbRequests()
              .getCreatedTopicRequests(
                  userName,
                  requestsType,
                  false,
                  tenantId,
                  teamId,
                  env,
                  requestOperationType,
                  wildcardSearch);
    } else {
      createdTopicReqList =
          manageDatabase
              .getHandleDbRequests()
              .getCreatedTopicRequests(
                  userName,
                  requestsType,
                  true,
                  tenantId,
                  teamId,
                  env,
                  requestOperationType,
                  wildcardSearch);
    }

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

    return updateCreateTopicReqsList(createdTopicReqList, tenantId);
  }

  private List<TopicRequestsResponseModel> updateCreateTopicReqsList(
      List<TopicRequest> topicsList, int tenantId) {
    List<TopicRequestsResponseModel> topicRequestModelList = getTopicRequestModels(topicsList);

    for (TopicRequestsResponseModel topicInfo : topicRequestModelList) {
      topicInfo.setTeamname(manageDatabase.getTeamNameFromTeamId(tenantId, topicInfo.getTeamId()));
      topicInfo.setEnvironmentName(getEnvDetails(topicInfo.getEnvironment()).getName());
    }

    return topicRequestModelList;
  }

  private List<TopicRequestsResponseModel> getTopicRequestModels(List<TopicRequest> topicsList) {
    List<TopicRequestsResponseModel> topicRequestModelList = new ArrayList<>();
    TopicRequestsResponseModel topicRequestModel;
    String userName = getUserName();
    Integer userTeamId = commonUtilsService.getTeamId(userName);

    int tenantId = commonUtilsService.getTenantId(userName);
    Set<String> approverRoles =
        rolesPermissionsControllerService.getApproverRoles("TOPICS", tenantId);
    List<UserInfo> userList = manageDatabase.getUsersPerTeamAndTenant(userTeamId, tenantId);

    for (TopicRequest topicReq : topicsList) {
      topicRequestModel = new TopicRequestsResponseModel();
      copyProperties(topicReq, topicRequestModel);
      topicRequestModel.setRequestStatus(RequestStatus.of(topicReq.getRequestStatus()));
      topicRequestModel.setRequestOperationType(
          RequestOperationType.of(topicReq.getRequestOperationType()));

      validateAndCopyTopicConfigs(topicReq, topicRequestModel);

      topicRequestModel.setTeamname(
          manageDatabase.getTeamNameFromTeamId(tenantId, topicReq.getTeamId()));

      // show approving info only before approvals
      if (RequestStatus.APPROVED != topicRequestModel.getRequestStatus()) {
        if (topicRequestModel.getRequestOperationType() != null
            && RequestOperationType.CLAIM == topicRequestModel.getRequestOperationType()) {
          List<Topic> topics =
              commonUtilsService.getTopicsForTopicName(topicRequestModel.getTopicname(), tenantId);
          if (!topics.isEmpty()) {
            topicRequestModel.setApprovingTeamDetails(
                updateApproverInfo(
                    manageDatabase.getUsersPerTeamAndTenant(topics.get(0).getTeamId(), tenantId),
                    manageDatabase.getTeamNameFromTeamId(tenantId, topics.get(0).getTeamId()),
                    approverRoles,
                    topicRequestModel.getRequestor()));
          } else {
            log.warn(
                "Request Exists for Topic {} in env {} and Topic does not exist.",
                topicReq.getTopicname(),
                topicReq.getEnvironmentName());
            topicRequestModel.setRemarks(
                "This topic is not found in Klaw. Please contact your Administrator.");
          }
        } else {
          topicRequestModel.setApprovingTeamDetails(
              updateApproverInfo(
                  userList,
                  manageDatabase.getTeamNameFromTeamId(tenantId, userTeamId),
                  approverRoles,
                  topicRequestModel.getRequestor()));
        }
      }

      topicRequestModelList.add(setRequestorPermissions(topicRequestModel, userName));
    }
    return topicRequestModelList;
  }

  private void validateAndCopyTopicConfigs(
      TopicRequest topicReq, TopicRequestsResponseModel topicRequestModel) {

    if (topicReq.getJsonParams() != null) {
      List<TopicConfigEntry> topicConfigEntryList = new ArrayList<>();
      TopicConfigurationRequest topicConfigurationRequest =
          UtilMethods.createTopicConfigurationRequestFromJson(
              topicReq.getJsonParams(), OBJECT_MAPPER);
      for (Map.Entry<String, String> entry :
          topicConfigurationRequest.getAdvancedTopicConfiguration().entrySet()) {
        topicConfigEntryList.add(
            TopicConfigEntry.builder()
                .configKey(entry.getKey())
                .configValue(entry.getValue())
                .build());
      }
      topicRequestModel.setAdvancedTopicConfigEntries(topicConfigEntryList);
    }
  }

  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);
  }

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

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

      return ApiResultStatus.SUCCESS.value.equals(deleteTopicReqStatus)
          ? ApiResponse.ok(deleteTopicReqStatus)
          : ApiResponse.notOk(deleteTopicReqStatus);
    } catch (Exception e) {
      log.error(e.getMessage());
      throw new KlawException(e.getMessage());
    }
  }

  /*
  - On approval, create,delete,update the topics on kafka cluster with api call
  - For claim topics, there are no kafka cluster operations
   */
  @PermissionAllowed(
      permissionAllowed = {PermissionType.APPROVE_TOPICS, PermissionType.APPROVE_TOPICS_CREATE})
  public ApiResponse approveTopicRequests(String topicId) throws KlawException {
    log.info("approveTopicRequests {}", topicId);
    String userName = getUserName();
    int tenantId = commonUtilsService.getTenantId(userName);
    TopicRequest topicRequest = getTopicRequestFromTopicId(Integer.parseInt(topicId), tenantId);

    String isOptionalExtraPermissionForPromote =
        manageDatabase.getKwPropertyValue(
            KLAW_OPTIONAL_PERMISSION_NEW_TOPIC_CREATION_KEY, tenantId);
    if (topicRequest.getRequestOperationType().equals(RequestOperationType.CREATE.value)
        && Boolean.parseBoolean(isOptionalExtraPermissionForPromote)
        && commonUtilsService.isNotAuthorizedUser(
            commonUtilsService.getPrincipal(), PermissionType.APPROVE_TOPICS_CREATE)) {
      return ApiResponse.notOk(ApiResultStatus.NOT_AUTHORIZED.value + ". " + TOPICS_ERR_116);
    } else if (commonUtilsService.isNotAuthorizedUser(
        commonUtilsService.getPrincipal(), PermissionType.APPROVE_TOPICS)) {
      return ApiResponse.NOT_AUTHORIZED;
    }

    ApiResponse validationResponse = validateTopicRequest(topicRequest, userName);
    if (!validationResponse.isSuccess()) {
      return validationResponse;
    }

    HandleDbRequests dbHandle = manageDatabase.getHandleDbRequests();
    String updateTopicReqStatus;
    // Starts as success as their may be no schema related to this topic.
    String schemaUpdateStatus = ApiResultStatus.SUCCESS.value;
    String updateSchemaMsg = "";
    if (RequestOperationType.CLAIM.value.equals(topicRequest.getRequestOperationType())) {
      List<Topic> allTopics =
          commonUtilsService.getTopicsForTopicName(topicRequest.getTopicname(), tenantId);
      for (Topic allTopic : allTopics) {
        allTopic.setTeamId(
            topicRequest.getTeamId()); // for claim reqs, team stored in approving team field
        allTopic.setExistingTopic(true);
        // check if there is any associated schema.

      }
      List<MessageSchema> updatedSchemas = new ArrayList<>();
      for (Env schemaEnv : manageDatabase.getSchemaRegEnvList(tenantId)) {
        // check every schema Env for these schemas
        List<MessageSchema> schemas =
            manageDatabase
                .getHandleDbRequests()
                .getSchemaForTenantAndEnvAndTopic(
                    tenantId, schemaEnv.getId(), topicRequest.getTopicname());
        for (MessageSchema schema : schemas) {
          schema.setTeamId(topicRequest.getTeamId());
          updatedSchemas.add(schema);
        }
      }

      if (!updatedSchemas.isEmpty()) {
        schemaUpdateStatus =
            manageDatabase.getHandleDbRequests().insertIntoMessageSchemaSOT(updatedSchemas);
        updateSchemaMsg = ", TopicSchemaStatus: " + schemaUpdateStatus;
      }

      updateTopicReqStatus = dbHandle.addToSynctopics(allTopics).getResultStatus();
      if (ApiResultStatus.SUCCESS.value.equals(updateTopicReqStatus)
          && ApiResultStatus.SUCCESS.value.equalsIgnoreCase(schemaUpdateStatus)) {
        updateTopicReqStatus = dbHandle.updateTopicRequestStatus(topicRequest, userName);
      }
    } else {
      Map<String, String> topicConfig =
          UtilMethods.createAdvancedConfigFromJson(topicRequest.getJsonParams(), OBJECT_MAPPER);
      updateTopicReqStatus =
          invokeClusterApiForTopicRequest(userName, tenantId, topicRequest, dbHandle, topicConfig);
    }

    if (updateTopicReqStatus.equals(ApiResultStatus.SUCCESS.value)) {
      commonUtilsService.updateMetadata(
          tenantId, EntityType.TOPICS, MetadataOperationType.CREATE, null);
    }

    String message = "Topic Status: " + updateTopicReqStatus + updateSchemaMsg;
    return ApiResultStatus.SUCCESS.value.equals(updateTopicReqStatus)
            && ApiResultStatus.SUCCESS.value.equalsIgnoreCase(schemaUpdateStatus)
        ? ApiResponse.ok(message)
        : ApiResponse.notOk(message);
  }

  private String invokeClusterApiForTopicRequest(
      String userName,
      int tenantId,
      TopicRequest topicRequest,
      HandleDbRequests dbHandle,
      Map<String, String> topicConfig)
      throws KlawException {
    String updateTopicReqStatus;
    ResponseEntity<ApiResponse> response =
        clusterApiService.approveTopicRequests(
            topicRequest.getTopicname(),
            topicRequest.getRequestOperationType(),
            topicRequest.getTopicpartitions(),
            topicRequest.getReplicationfactor(),
            topicRequest.getEnvironment(),
            topicConfig,
            tenantId,
            topicRequest.getDeleteAssociatedSchema());

