package io.aiven.klaw.helpers.db.rdbms;

import static io.aiven.klaw.helpers.KwConstants.DATE_DDMMMYYYY_FORMATTER;
import static io.aiven.klaw.helpers.KwConstants.DATE_TIME_DDMMMYYYY_HHMMSS_FORMATTER;
import static io.aiven.klaw.helpers.KwConstants.REQUESTOR_SUBSCRIPTIONS;

import com.google.common.collect.Lists;
import io.aiven.klaw.dao.*;
import io.aiven.klaw.model.enums.AclPatternType;
import io.aiven.klaw.model.enums.AclType;
import io.aiven.klaw.model.enums.KafkaClustersType;
import io.aiven.klaw.model.enums.NewUserStatus;
import io.aiven.klaw.model.enums.OperationalRequestType;
import io.aiven.klaw.model.enums.OrderBy;
import io.aiven.klaw.model.enums.RequestMode;
import io.aiven.klaw.model.enums.RequestOperationType;
import io.aiven.klaw.model.enums.RequestStatus;
import io.aiven.klaw.model.enums.RolesType;
import io.aiven.klaw.model.response.DashboardStats;
import io.aiven.klaw.repository.*;
import io.aiven.klaw.service.CommonUtilsService;
import java.sql.Timestamp;
import java.time.Instant;
import java.time.LocalDate;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.TreeSet;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Example;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class SelectDataJdbc {

  @Autowired(required = false)
  private UserInfoRepo userInfoRepo;

  @Autowired(required = false)
  private KwPropertiesRepo kwPropertiesRepo;

  @Autowired(required = false)
  private RegisterInfoRepo registerInfoRepo;

  @Autowired(required = false)
  private TeamRepo teamRepo;

  @Autowired(required = false)
  private EnvRepo envRepo;

  @Autowired(required = false)
  private ActivityLogRepo activityLogRepo;

  @Autowired(required = false)
  private AclRequestsRepo aclRequestsRepo;

  @Autowired(required = false)
  private OperationalRequestsRepo operationalRequestsRepo;

  @Autowired(required = false)
  private TopicRepo topicRepo;

  @Autowired(required = false)
  private KwKafkaConnectorRepo kafkaConnectorRepo;

  @Autowired(required = false)
  private KwKafkaConnectorRequestsRepo kafkaConnectorRequestsRepo;

  @Autowired(required = false)
  private AclRepo aclRepo;

  @Autowired(required = false)
  private TopicRequestsRepo topicRequestsRepo;

  @Autowired(required = false)
  private SchemaRequestRepo schemaRequestRepo;

  @Autowired(required = false)
  private MessageSchemaRepo messageSchemaRepo;

  @Autowired(required = false)
  private TenantRepo tenantRepo;

  @Autowired(required = false)
  private KwRolesPermsRepo kwRolesPermsRepo;

  @Autowired(required = false)
  private KwClusterRepo kwClusterRepo;

  @Autowired(required = false)
  private KwMetricsRepo kwMetricsRepo;

  @Autowired(required = false)
  private ProductDetailsRepo productDetailsRepo;

  @Autowired(required = false)
  private KwEntitySequenceRepo kwEntitySequenceRepo;

  public boolean existsAclRequest(
      String topicName, String requestStatus, String env, int tenantId) {
    return aclRequestsRepo.existsByTenantIdAndEnvironmentAndRequestStatusAndTopicname(
        tenantId, env, requestStatus, topicName);
  }

  public boolean existsAclSslInTeam(int teamId, int tenantId, String aclSsl) {
    return aclRepo.existsByTeamIdAndTenantIdAndAclssl(teamId, tenantId, aclSsl);
  }

  public boolean existsSpecificAclRequest(
      String topicName, String requestStatus, String env, int tenantId, int associatedAclId) {
    log.debug("associatedAclId = {}", associatedAclId);

    return aclRequestsRepo
        .existsByTenantIdAndEnvironmentAndRequestStatusAndTopicnameAndAssociatedAclId(
            tenantId, env, requestStatus, topicName, associatedAclId);
  }

  public List<AclRequests> selectFilteredAclRequests(
      boolean isApproval,
      String requestor,
      String role,
      String status,
      RequestOperationType requestOperationType,
      boolean showRequestsOfAllTeams,
      String topic,
      String environment,
      String wildcardSearch,
      AclType aclType,
      boolean isMyRequest,
      int tenantId) {
    log.debug("selectAclRequests {}", requestor);
    List<AclRequests> aclList = new ArrayList<>();
    List<AclRequests> aclListSub;
    Integer teamSelected = selectUserInfo(requestor).getTeamId();
    Optional<Team> team = teamRepo.findById(new TeamID(teamSelected, tenantId));
    aclListSub =
        Lists.newArrayList(
            findAclRequestsByExample(
                topic,
                environment,
                aclType,
                status,
                isMyRequest ? requestor : null,
                tenantId,
                requestOperationType));
    if (isApproval) {
      // Only filter when returning to approvers view.
      // in the acl request the username is mapped to the requestor column in the database.

      aclListSub =
          aclListSub.stream()
              .filter(
                  req ->
                      !req.getRequestor().equals(requestor)
                          && !req.getRequestOperationType()
                              .equals(RequestOperationType.CLAIM.value))
              .collect(Collectors.toList());
      getAclClaimRequestsForApproval(requestor, team, aclListSub, status);
    }

    if (wildcardSearch != null && !wildcardSearch.isEmpty()) {
      aclListSub =
          aclListSub.stream()
              .filter(
                  req -> req.getTopicname().toLowerCase().contains(wildcardSearch.toLowerCase()))
              .collect(Collectors.toList());
    }

    for (AclRequests row : aclListSub) {

      Integer teamId;
      String rowRequestOperationType = row.getRequestOperationType();
      if (isApproval) {
        teamSelected = showRequestsOfAllTeams ? null : teamSelected;
        if (REQUESTOR_SUBSCRIPTIONS.equals(role)) {
          teamId = row.getRequestingteam();
        } else {
          teamId = row.getTeamId();
        }

        if (RequestOperationType.DELETE.value.equals(rowRequestOperationType)) {
          teamId = row.getRequestingteam();
        } else if (RequestOperationType.CLAIM.value.equals(rowRequestOperationType)
            && row.getApprovals().stream()
                .anyMatch(approval -> requestor.equals(approval.getApproverName()))) {
          // Multi Approval filtering remove this row as it is an approval that has already been
          // approved by this user.
          continue;
        }

      } else {
        teamId = row.getRequestingteam();
      }

      if (showRequestsOfAllTeams) { // show all requests of all teams
        aclList.add(row);
      } else if ((teamSelected != null && teamSelected.equals(teamId))
          || row.getRequestOperationType().equals(RequestOperationType.CLAIM.value)) {
        aclList.add(row);
      }

      try {
        row.setRequesttimestring(
            DATE_TIME_DDMMMYYYY_HHMMSS_FORMATTER.format(row.getRequesttime().toLocalDateTime()));
      } catch (Exception ignored) {
      }
    }

    return aclList;
  }

  private void getAclClaimRequestsForApproval(
      String requestor, Optional<Team> team, List<AclRequests> aclListSub, String requestStatus) {
    if (team.isPresent()) {

      if (requestStatus == null || requestStatus.equals(RequestStatus.ALL.value)) {
        aclListSub.addAll(
            aclRequestsRepo.getAllByApprovalsRequiredApprover(team.get().getTeamname()));
      } else if (requestStatus.equals(RequestStatus.CREATED.value)) {
        // GET CLAIMS that have not been approved yet
        aclListSub.addAll(
            aclRequestsRepo
                .getAllByRequestStatusAndRequestorNotAndApprovalsRequiredApproverAndApprovalsApproverName(
                    requestStatus, requestor, team.get().getTeamname(), null));
      } else {
        aclListSub.addAll(
            aclRequestsRepo.getAllByRequestStatusAndRequestorNotAndApprovalsRequiredApprover(
                requestStatus, requestor, team.get().getTeamname()));
      }
    }
  }

  /**
   * Query the AclRequestsRepo by supplying optional search parameters any given search parameters
   * will be utilised in the search.
   *
   * @param topic The topic Name
   * @param environment the environment
   * @param aclType Producer or consumer
   * @param requestStatus created/declined/approved
   * @param requestor the name of the user who created the request in klaw
   * @param tenantId The tenantId
   * @return
   */
  public Iterable<AclRequests> findAclRequestsByExample(
      String topic,
      String environment,
      AclType aclType,
      String requestStatus,
      String requestor,
      int tenantId,
      RequestOperationType requestOperationType) {

    AclRequests request = new AclRequests();

    request.setTenantId(tenantId);
    if (aclType != null) {
      request.setAclType(aclType.value);
    }
    if (environment != null) {
      request.setEnvironment(environment);
    }
    if (topic != null && !topic.isEmpty()) {
      request.setTopicname(topic);
    }
    if (requestStatus != null && !requestStatus.equalsIgnoreCase("all")) {
      request.setRequestStatus(requestStatus);
    }

    if (requestor != null && !requestor.isEmpty()) {
      request.setRequestor(requestor);
    }
    if (requestOperationType != null && requestOperationType != RequestOperationType.ALL) {
      request.setRequestOperationType(requestOperationType.value);
    }
    // check if debug is enabled so the logger doesnt waste resources converting object request to a
    // string
    if (log.isDebugEnabled()) {
      log.debug("find By topic etc example {}", request);
    }
    return aclRequestsRepo.findAll(Example.of(request));
  }

  public List<SchemaRequest> selectFilteredSchemaRequests(
      boolean isApproval,
      String requestor,
      int tenantId,
      RequestOperationType requestOperationType,
      String topic,
      String env,
      String status,
      String wildcardSearch,
      boolean showRequestsOfAllTeams,
      boolean isMyRequest) {
    if (log.isDebugEnabled()) {
      log.debug(
          "selectSchemaRequests isApproval {} Requestor: {} , tenantIf:{} , topic: {}, env: {}, status: {}, wildcardSearch: {}, showRequestsOfAllTeams {}, isMyRequest: {}",
          isApproval,
          requestor,
          tenantId,
          topic,
          env,
          status,
          wildcardSearch,
          showRequestsOfAllTeams,
          isMyRequest);
    }
    List<SchemaRequest> schemaList = new ArrayList<>();
    List<SchemaRequest> schemaListSub;
    Integer teamSelected = selectUserInfo(requestor).getTeamId();
    if (isApproval) {
      schemaListSub =
          Lists.newArrayList(
              findSchemaRequestsByExample(
                  topic,
                  env,
                  status != null ? status : RequestStatus.CREATED.value,
                  showRequestsOfAllTeams ? null : teamSelected,
                  tenantId,
                  isMyRequest ? requestor : null,
                  requestOperationType));

      // Placed here as it should only apply for approvers.
      schemaListSub =
          schemaListSub.stream()
              .filter(request -> !request.getRequestor().equals(requestor))
              .collect(Collectors.toList());
    } else {
      // Previously Status was being filtered by the calling method but it is more efficient to have
      // the database do this.
      // Similarily Team Selected now included in the database call as previously they were all
      // being filtered out after the data was returned.
      schemaListSub =
          Lists.newArrayList(
              findSchemaRequestsByExample(
                  topic,
                  env,
                  status,
                  teamSelected,
                  tenantId,
                  isMyRequest ? requestor : null,
                  requestOperationType));
    }

    boolean wildcardFilter = (wildcardSearch != null && !wildcardSearch.isEmpty());
    for (SchemaRequest row : schemaListSub) {

      try {
        row.setRequesttimestring(
            DATE_TIME_DDMMMYYYY_HHMMSS_FORMATTER.format(row.getRequesttime().toLocalDateTime()));
      } catch (Exception ignored) {
      }
      if (!wildcardFilter
          || row.getTopicname().toLowerCase().contains(wildcardSearch.toLowerCase())) {
        schemaList.add(row);
      }
    }

    return schemaList;
  }

  public Iterable<SchemaRequest> findSchemaRequestsByExample(
      String topic,
      String environment,
      String status,
      Integer teamId,
      int tenantId,
      String userName,
      RequestOperationType requestOperationType) {

    SchemaRequest request = new SchemaRequest();

    request.setTenantId(tenantId);

    if (environment != null) {
      request.setEnvironment(environment);
    }
    if (topic != null && !topic.isEmpty()) {
      request.setTopicname(topic);
    }
    if (status != null && !status.equalsIgnoreCase("all")) {
      request.setRequestStatus(status);
    }
    if (teamId != null) {
      request.setTeamId(teamId);
    }
    if (userName != null && !userName.isEmpty()) {
      request.setRequestor(userName);
    }
    if (requestOperationType != null && requestOperationType != RequestOperationType.ALL) {
      request.setRequestOperationType(requestOperationType.value);
    }

    request.setForceRegister(null);
    // check if debug is enabled so the logger doesnt waste resources converting object request to a
    // string
    if (log.isDebugEnabled()) {
      log.debug("find By topic etc example {}", request);
    }
    return schemaRequestRepo.findAll(Example.of(request));
  }

  public SchemaRequest selectSchemaRequest(int avroSchemaId, int tenantId) {
    log.debug("selectSchemaRequest {}", avroSchemaId);
    SchemaRequestID schemaRequestID = new SchemaRequestID(avroSchemaId, tenantId);
    Optional<SchemaRequest> sR = schemaRequestRepo.findById(schemaRequestID);
    return sR.orElse(null);
  }

  public List<Topic> selectTopicDetails(String topic, int tenantId) {
    List<Topic> topicOpt = topicRepo.findAllByTopicnameAndTenantId(topic, tenantId);
    return topicOpt.isEmpty() ? Collections.emptyList() : topicOpt;
  }

  public List<KwKafkaConnector> selectConnectorDetails(String connectorName, int tenantId) {
    List<KwKafkaConnector> topicOpt =
        kafkaConnectorRepo.findAllByConnectorNameAndTenantId(connectorName, tenantId);
    return topicOpt.isEmpty() ? Collections.emptyList() : topicOpt;
  }

  // "All teams"
  // teamId 1 is All teams
  public List<Topic> selectSyncTopics(String env, Integer teamId, int tenantId) {
    log.debug("selectSyncTopics {}", teamId);
    if (teamId == null || teamId.equals(1)) {
      if (env == null || env.equals("ALL")) {
        return topicRepo.findAllByTenantId(tenantId);
      } else {
        List<String> topics = topicRepo.findAllTopicNamesForEnv(env, tenantId);
        return topicRepo.findAllByTenantIdAndTopicnameIn(tenantId, topics);
      }
    } else {
      if (env == null || "ALL".equals(env)) {
        return topicRepo.findAllByTeamIdAndTenantId(teamId, tenantId);
      } else {
        List<String> topics = topicRepo.findAllTopicNamesForEnvAndTeam(env, teamId, tenantId);
        return topicRepo.findAllByTenantIdAndTopicnameIn(tenantId, topics);
      }
    }
  }

  public DashboardStats getDashboardStats(Integer teamId, int tenantId) {
    DashboardStats dashboardStats = new DashboardStats();
    dashboardStats.setProducerCount(
        aclRepo.countDistinctTopicNameByAclTypeAndTeamIdAndTenantId(
            AclType.PRODUCER.value, teamId, tenantId));

    dashboardStats.setConsumerCount(
        aclRepo.countDistinctTopicNameByAclTypeAndTeamIdAndTenantId(
            AclType.CONSUMER.value, teamId, tenantId));

    dashboardStats.setTeamMembersCount(userInfoRepo.countByTeamIdAndTenantId(teamId, tenantId));

    return dashboardStats;
  }

  public List<Topic> selectAllTopicsByTopictypeAndTeamname(
      String isProducerConsumer, Integer teamId, int tenantId, String env) {
    log.debug("selectAllByTopictypeAndTeamname {} {}", isProducerConsumer, teamId);
    List<Topic> topics = new ArrayList<>();
    String topicType, aclPatternType;
    if (isProducerConsumer != null && isProducerConsumer.equals(AclType.PRODUCER.value)) {
      topicType = AclType.PRODUCER.value;
    } else {
      topicType = AclType.CONSUMER.value;
    }
    List<Acl> acls;

    if (Objects.isNull(env) || Objects.equals(env, "ALL")) {
      acls = aclRepo.findAllByAclTypeAndTeamIdAndTenantId(topicType, teamId, tenantId);
    } else {
      acls =
          aclRepo.findAllByAclTypeAndTeamIdAndTenantIdAndEnvironment(
              topicType, teamId, tenantId, env);
    }

    Topic t;
    Map<String, Set<String>> topicEnvMap = new HashMap<>();
    String tmpTopicName;
    Set<String> envSet;

    for (Acl acl : acls) {
      t = new Topic();
      tmpTopicName = acl.getTopicname();
      aclPatternType = acl.getAclPatternType();

      if (aclPatternType != null && aclPatternType.equals(AclPatternType.PREFIXED.value))
        tmpTopicName = tmpTopicName + "--" + AclPatternType.PREFIXED.value + "--";

      t.setEnvironment(acl.getEnvironment());
      t.setTopicname(tmpTopicName);

      if (topicEnvMap.containsKey(tmpTopicName)) {
        envSet = topicEnvMap.get(tmpTopicName);
        if (!envSet.contains(acl.getEnvironment())) {
          envSet.add(acl.getEnvironment());
          topicEnvMap.put(tmpTopicName, envSet);
        }
      } else {
        envSet = new HashSet<>();
        envSet.add(acl.getEnvironment());
        topicEnvMap.put(tmpTopicName, envSet);
      }

      topics.add(t);
    }

    for (Topic topic : topics) {
      topic.setEnvironmentsSet(topicEnvMap.get(topic.getTopicname()));
    }

    topics =
        topics.stream()
            .collect(
                Collectors.collectingAndThen(
                    Collectors.toCollection(
                        () -> new TreeSet<>(Comparator.comparing(Topic::getTopicname))),
                    ArrayList::new));

    return topics;
  }

  public List<Acl> selectSyncAcls(String env, int tenantId) {
    return aclRepo.findAllByEnvironmentAndTenantId(env, tenantId);
  }

  public List<Acl> getPrefixedAclsSOT(String env, int tenantId) {
    return aclRepo.findAllByEnvironmentAndAclPatternTypeAndTenantId(
        env, AclPatternType.PREFIXED.value, tenantId);
  }

  /**
   * @param isApproval boolean should all requests be returned (true if requesting for an approvers
   *     view)
   * @param requestor The UserName of the person who is requesting the list of topics
   * @param status Filter the response by the status of the requests e.g.
   *     all,created,declined,deleted,approved
   * @param showRequestsOfAllTeams boolean true if a superadmin with the permission to view
   *     requests. otherwise false
   * @param tenantId The identifier of the tenancy with which the requestor is from.
   * @param teamId filter the results by the name of the team who created the request
   * @param env filter the results by the environment identifier e.g. 1,2,3,4. Applied only when
   *     allReqs is true
   * @param wildcardSearch filter the results by a wildcard search of the topicnames. Applied only
   *     when allReqs is true
   * @return A list of TopicRequests that meet the filtered criteria. Applied only when allReqs is
   *     true
   */
  public List<TopicRequest> selectFilteredTopicRequests(
      boolean isApproval,
      String requestor,
      String status,
      boolean showRequestsOfAllTeams,
      int tenantId,
      Integer teamId,
      RequestOperationType requestOperationType,
      String env,
      String wildcardSearch,
      boolean isMyRequest) {
    if (log.isDebugEnabled()) {
      log.debug(
          "selectTopicRequests {} {} {} {} {} {}",
          showRequestsOfAllTeams,
          requestor,
          status,
          teamId,
          env,
          wildcardSearch);
    }
    Integer teamSelected = selectUserInfo(requestor).getTeamId();
    List<TopicRequest> topicRequests = new ArrayList<>();
    List<TopicRequest> topicRequestListSub;
    if (isApproval) { // approvers
      // On Approvals this should always be filtered by the users current team
      topicRequestListSub =
          Lists.newArrayList(
              findTopicRequestsByExample(
                  requestOperationType != null ? requestOperationType.value : null,
                  showRequestsOfAllTeams ? null : teamSelected,
                  env,
                  status,
                  tenantId,
                  null,
                  isMyRequest ? requestor : null));

      // Only execute just before adding the separate claim list as this will make sure only the
      // claim topics this team is able to approve will be returned.
      topicRequestListSub =
          topicRequestListSub.stream()
              .filter(
                  topicRequest ->
                      !RequestOperationType.CLAIM.value.equals(
                          topicRequest.getRequestOperationType()))
              .collect(Collectors.toList());

      if (requestOperationType == null
          || requestOperationType.equals(RequestOperationType.CLAIM)
          || requestOperationType.equals(RequestOperationType.ALL)) {
        // Team Name is null here as it is the team who created the requested
        // Approving Team includes the teamId as that is how Claim topics know who is the owning
        // team.
        List<TopicRequest> claimTopicReqs =
            Lists.newArrayList(
                findTopicRequestsByExample(
                    RequestOperationType.CLAIM.value,
                    null,
                    env,
                    status,
                    tenantId,
                    showRequestsOfAllTeams ? null : String.valueOf(teamSelected),
                    isMyRequest ? requestor : null));
        topicRequestListSub.addAll(claimTopicReqs);
      }

      // remove users own requests to approve/show in the list
      // Placed here as it should only apply for approvers.
      topicRequestListSub =
          topicRequestListSub.stream()
              .filter(topicRequest -> !topicRequest.getRequestor().equals(requestor))
              .toList();
    } else {
      if (showRequestsOfAllTeams) {
        topicRequestListSub =
            Lists.newArrayList(
                findTopicRequestsByExample(
                    requestOperationType != null ? requestOperationType.value : null,
                    null,
                    env,
                    status,
                    tenantId,
                    null,
                    isMyRequest ? requestor : null));
      } else {

        topicRequestListSub =
            Lists.newArrayList(
                findTopicRequestsByExample(
                    requestOperationType != null ? requestOperationType.value : null,
                    teamSelected,
                    env,
                    status,
                    tenantId,
                    null,
                    isMyRequest ? requestor : null));
      }
    }

    if (teamId != null) {
      topicRequestListSub =
          topicRequestListSub.stream()
              .filter(topicRequest -> topicRequest.getTeamId().equals(teamId))
              .toList();
    }

    boolean wildcardFilter = (wildcardSearch != null && !wildcardSearch.isEmpty());

    for (TopicRequest row : topicRequestListSub) {

      try {
        row.setRequesttimestring(
            DATE_TIME_DDMMMYYYY_HHMMSS_FORMATTER.format(row.getRequesttime().toLocalDateTime()));
      } catch (Exception ignored) {
      }
      filterAndAddTopicRequest(topicRequests, row, wildcardSearch, wildcardFilter);
    }

    return topicRequests;
  }

  private void filterAndAddTopicRequest(
      List<TopicRequest> topicRequestList,
      TopicRequest row,
      String wildcardSearch,
      boolean applyWildcardFilter) {
    if (!applyWildcardFilter
        || row.getTopicname().toLowerCase().contains(wildcardSearch.toLowerCase())) {
      topicRequestList.add(row);
    }
  }

  private void filterAndAddOperationalRequest(
      List<OperationalRequest> operationalRequestList,
      OperationalRequest row,
      String wildcardSearch,
      boolean applyWildcardFilter) {
    if (!applyWildcardFilter
        || row.getTopicname()
            .toLowerCase(Locale.ROOT)
            .contains(wildcardSearch.toLowerCase(Locale.ROOT))) {
      operationalRequestList.add(row);
    }
  }

  /**
   * Query the TopicRequestsRepo by supplying optional search parameters any given search parameters
   * will be utilised in the search.
   *
   * @param requestOperationType The type of topic request Create/claim etc
   * @param teamId The identifier of the team
   * @param environment the environment
   * @param status created/declined/approved
   * @param tenantId The tenantId @Param approvingTeam The team who is able to approve this request
   *     will be filled in the case of claim requests. @Param userName of the person who created the
   *     request
   * @return An Iterable of all TopicRequests that match the parameters that have been supplied
   */
  private Iterable<TopicRequest> findTopicRequestsByExample(
      String requestOperationType,
      Integer teamId,
      String environment,
      String status,
      int tenantId,
      String approvingTeam,
      String userName) {

    TopicRequest request = new TopicRequest();
    request.setTenantId(tenantId);

    if (requestOperationType != null
        && !requestOperationType.equals(RequestOperationType.ALL.value)) {
      request.setRequestOperationType(requestOperationType);
    }
    if (environment != null) {
      request.setEnvironment(environment);
    }
    if (teamId != null) {
      request.setTeamId(teamId);
    }

    if (approvingTeam != null && !approvingTeam.isEmpty()) {
      request.setApprovingTeamId(approvingTeam);
    }

    if (status != null && !status.equalsIgnoreCase("all")) {
      request.setRequestStatus(status);
    }
    // userName is transient and so not available in the database to be queried.
    if (userName != null && !userName.isEmpty()) {
      request.setRequestor(userName);
    }

    // check if debug is enabled so the logger doesn't waste resources converting object request to
    // a
    // string
    if (log.isDebugEnabled()) {
      log.debug("find By topic etc example {}", request);
    }

    return topicRequestsRepo.findAll(Example.of(request));
  }

  public List<KafkaConnectorRequest> selectFilteredKafkaConnectorRequests(
      boolean isApproval,
      String requestor,
      String status,
      RequestOperationType requestOperationType,
      boolean showRequestsOfAllTeams,
      int tenantId,
      String env,
      String search,
      boolean isMyRequest) {
    log.debug("selectConnectorRequestsByStatus {} {}", requestor, status);
    List<KafkaConnectorRequest> topicRequestList = new ArrayList<>();

    List<KafkaConnectorRequest> topicRequestListSub;
    Integer teamSelected = selectUserInfo(requestor).getTeamId();

    if (isApproval) { // approvers

      // Team Name is null here as it is the team who created the requested
      // Approving Team includes the teamId as that is how Claim topics know who is the owning team.
      List<KafkaConnectorRequest> claimTopicReqs =
          Lists.newArrayList(
              findKafkaConnectorRequestsByExample(
                  RequestOperationType.CLAIM.value,
                  null,
                  env,
                  status,
                  tenantId,
                  showRequestsOfAllTeams ? null : String.valueOf(teamSelected),
                  isMyRequest ? requestor : null));

      topicRequestListSub =
          Lists.newArrayList(
              findKafkaConnectorRequestsByExample(
                  requestOperationType != null ? requestOperationType.value : null,
                  showRequestsOfAllTeams ? null : teamSelected,
                  env,
                  status,
                  tenantId,
                  null,
                  isMyRequest ? requestor : null));

      // Only execute just before adding the separate claim list as this will make sure only the
      // claim topics this team is able to approve will be returned.
      topicRequestListSub =
          topicRequestListSub.stream()
              .filter(
                  request ->
                      !RequestOperationType.CLAIM.value.equals(request.getRequestOperationType()))
              .collect(Collectors.toList());

      topicRequestListSub.addAll(claimTopicReqs);
      // remove users own requests to approve/show in the list
      // Placed here as it should only apply for approvers.
      topicRequestListSub =
          topicRequestListSub.stream()
              .filter(topicRequest -> !topicRequest.getRequestor().equals(requestor))
              .collect(Collectors.toList());
    } else {
      // show my teams requests
      topicRequestListSub =
          Lists.newArrayList(
              findKafkaConnectorRequestsByExample(
                  requestOperationType != null ? requestOperationType.value : null,
                  showRequestsOfAllTeams ? null : teamSelected,
                  env,
                  status,
                  tenantId,
                  null,
                  isMyRequest ? requestor : null));
    }

    boolean wildcardSearch = search != null && !search.isEmpty();

    for (KafkaConnectorRequest row : topicRequestListSub) {
      try {
        row.setRequesttimestring(
            DATE_TIME_DDMMMYYYY_HHMMSS_FORMATTER.format(row.getRequesttime().toLocalDateTime()));
      } catch (Exception ignored) {
      }
      if (!wildcardSearch || row.getConnectorName().toLowerCase().contains(search.toLowerCase()))
        topicRequestList.add(row); // no team filter
    }

    return topicRequestList;
  }

  public List<Topic> getTopicsByTopicNameAndTeamId(String topicName, int teamId, int tenantId) {

    return topicRepo.findAllByTopicnameAndTeamIdAndTenantId(topicName, teamId, tenantId);
  }

  /**
   * Query the KafkaConnectorRequestsRepo by supplying optional search parameters any given search
   * parameters will be utilised in the search.
   *
   * @param requestOperationType The type of topic request Create/claim etc
   * @param teamId The identifier of the team
   * @param environment the environment
   * @param status created/declined/approved
   * @param tenantId The tenantId @Param approvingTeam The team who is able to approve this request
   *     will be filled in the case of claim requests.
   * @return An Iterable of all TopicRequests that match the parameters that have been supplied
   */
  private Iterable<KafkaConnectorRequest> findKafkaConnectorRequestsByExample(
      String requestOperationType,
      Integer teamId,
      String environment,
      String status,
      int tenantId,
      String approvingTeam,
      String requestor) {

    KafkaConnectorRequest request = new KafkaConnectorRequest();
    request.setTenantId(tenantId);

    if (requestOperationType != null
        && !requestOperationType.equals(RequestOperationType.ALL.value)) {
      request.setRequestOperationType(requestOperationType);
    }
    if (environment != null) {
      request.setEnvironment(environment);
    }
    if (teamId != null) {
      request.setTeamId(teamId);
    }

    if (requestor != null) {
      request.setRequestor(requestor);
    }

    if (approvingTeam != null && !approvingTeam.isEmpty()) {
      request.setApprovingTeamId(approvingTeam);
    }

    if (status != null && !status.equalsIgnoreCase("all")) {
      request.setRequestStatus(status);
    }

    // check if debug is enabled so the logger doesn't waste resources converting object request to
    // a
    // string
    if (log.isDebugEnabled()) {
      log.debug("find By topic etc example {}", request);
    }

    return kafkaConnectorRequestsRepo.findAll(Example.of(request));
  }

  public TopicRequest selectTopicRequestsForTopic(int topicId, int tenantId) {
    log.debug("selectTopicRequestsForTopic {}", topicId);
