content: package io.aiven.klaw.controller;
  
  import io.aiven.klaw.error.KlawException;
  import io.aiven.klaw.model.ApiResponse;
  import io.aiven.klaw.model.SchemaResetCache;
  import io.aiven.klaw.model.SyncSchemaUpdates;
  import io.aiven.klaw.model.enums.PermissionType;
  import io.aiven.klaw.model.response.SchemaDetailsResponse;
  import io.aiven.klaw.model.response.SyncSchemasList;
  import io.aiven.klaw.service.SchemaRegistrySyncControllerService;
  import io.aiven.klaw.validation.PermissionAllowed;
  import lombok.extern.slf4j.Slf4j;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.http.HttpStatus;
  import org.springframework.http.MediaType;
  import org.springframework.http.ResponseEntity;
  import org.springframework.web.bind.annotation.PathVariable;
  import org.springframework.web.bind.annotation.PostMapping;
  import org.springframework.web.bind.annotation.RequestBody;
  import org.springframework.web.bind.annotation.RequestMapping;
  import org.springframework.web.bind.annotation.RequestMethod;
  import org.springframework.web.bind.annotation.RequestParam;
  import org.springframework.web.bind.annotation.RestController;
  
  @RestController
  @RequestMapping("/")
  @Slf4j
  public class SchemaRegistrySyncController {
  
    @Autowired SchemaRegistrySyncControllerService schemaRegistrySyncControllerService;
  
    /**
     * Retrieve schema subjects and its versions in a given kafka env (associated schema env) from
     * cluster or metadata
     *
     * @param kafkaEnvId kafka env
     * @return
     * @throws Exception
     */
    @PermissionAllowed(
        permissionAllowed = {PermissionType.SYNC_SCHEMAS, PermissionType.SYNC_BACK_SCHEMAS})
    @RequestMapping(
        value = "/schemas",
        method = RequestMethod.GET,
        produces = {MediaType.APPLICATION_JSON_VALUE})
    public ResponseEntity<SyncSchemasList> getSchemasOfEnvironment(
        @RequestParam(value = "envId") String kafkaEnvId,
        @RequestParam(value = "pageNo") String pageNo,
        @RequestParam(value = "showAllTopics", defaultValue = "false") boolean showAllTopics,
        @RequestParam(value = "currentPage", defaultValue = "") String currentPage,
        @RequestParam(value = "topicnamesearch", required = false) String topicNameSearch,
        @RequestParam(value = "source", defaultValue = "cluster") String source,
        @RequestParam(value = "teamId", required = false, defaultValue = "0") Integer teamId)
        throws Exception {
      return new ResponseEntity<>(
          schemaRegistrySyncControllerService.getSchemasOfEnvironment(
              kafkaEnvId, pageNo, currentPage, topicNameSearch, showAllTopics, source, teamId),
          HttpStatus.OK);
    }
  
    /**
     * Store schemas of all versions in metadata db or cluster
     *
     * @param syncSchemaUpdates topic list and environment
     * @return
     * @throws Exception
     */
    @PermissionAllowed(
        permissionAllowed = {PermissionType.SYNC_SCHEMAS, PermissionType.SYNC_BACK_SCHEMAS})
    @PostMapping(
        value = "/schemas",
        produces = {MediaType.APPLICATION_JSON_VALUE})
    public ResponseEntity<ApiResponse> updateSyncSchemas(
        @RequestBody SyncSchemaUpdates syncSchemaUpdates) throws Exception {
      return new ResponseEntity<>(
          schemaRegistrySyncControllerService.updateSyncSchemas(syncSchemaUpdates), HttpStatus.OK);
    }
  
    /**
     * Retrieve schema of a topic and specific version from cluster or metadata
     *
     * @param source cluster or metadata
     * @param topicName
     * @param schemaVersion
     * @param kafkaEnvId kafka environment id
     * @return
     * @throws Exception
     */
    @PermissionAllowed(
        permissionAllowed = {PermissionType.SYNC_SCHEMAS, PermissionType.SYNC_BACK_SCHEMAS})
    @RequestMapping(
        value =
            "/schemas/source/{source}/kafkaEnv/{kafkaEnvId}/topic/{topicName}/schemaVersion/{schemaVersion}",
        method = RequestMethod.GET,
        produces = {MediaType.APPLICATION_JSON_VALUE})
    public ResponseEntity<SchemaDetailsResponse> getSchemaOfTopicFromSource(
        @PathVariable(value = "source") String source,
        @PathVariable(value = "topicName") String topicName,
        @PathVariable(value = "schemaVersion") int schemaVersion,
        @PathVariable(value = "kafkaEnvId") String kafkaEnvId)
        throws Exception {
      return new ResponseEntity<>(
          schemaRegistrySyncControllerService.getSchemaOfTopicFromSource(
              source, topicName, schemaVersion, kafkaEnvId),
          HttpStatus.OK);
    }
  
    @PermissionAllowed(
        permissionAllowed = {PermissionType.SYNC_SCHEMAS, PermissionType.SYNC_BACK_SCHEMAS})
    @RequestMapping(
        value = "/schemas/resetCache",
        method = RequestMethod.POST,
        produces = {MediaType.APPLICATION_JSON_VALUE})
    public ResponseEntity<ApiResponse> resetCacheClusterApi(
        @RequestBody SchemaResetCache schemaResetCache) throws KlawException {
      return new ResponseEntity<>(
          schemaRegistrySyncControllerService.resetCacheClusterApi(schemaResetCache), HttpStatus.OK);
    }
  }
  