feat(reindex): Initiate Full Reindex

Closes: MSEARCH-794

descriptors/ModuleDescriptor-template.json

@@ -4,7 +4,7 @@
   "provides": [
     {
       "id": "indices",
-      "version": "0.7",
+      "version": "0.8",
       "handlers": [
         {
           "methods": [
@@ -74,6 +74,19 @@
           "modulePermissions": [
             "user-tenants.collection.get"
           ]
+        },
+        {
+          "methods": [
+            "POST"
+          ],
+          "pathPattern": "/search/index/instance-records/reindex/full",
+          "permissionsRequired": [
+            "search.index.instance-records.reindex.full.post"
+          ],
+          "modulePermissions": [
+            "inventory-storage.reindex-records.publish.post",
+            "user-tenants.collection.get"
+          ]
         }
       ]
     },

src/main/java/org/folio/search/client/ConsortiumTenantsClient.java

@@ -0,0 +1,26 @@
+package org.folio.search.client;
+
+import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;
+
+import java.util.List;
+import org.springframework.cloud.openfeign.FeignClient;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestParam;
+
+@FeignClient("consortia")
+public interface ConsortiumTenantsClient {
+
+  /**
+   * Get tenants by consortium id.
+   *
+   * @return consortium tenants if executed under consortium central 'tenantId' context
+   * */
+  @GetMapping(value = "/{consortiumId}/tenants", produces = APPLICATION_JSON_VALUE)
+  ConsortiumTenants getConsortiumTenants(@PathVariable("consortiumId") String consortiumId,
+                                         @RequestParam("limit") int limit);
+
+  record ConsortiumTenants(List<ConsortiumTenant> tenants) { }
+
+  record ConsortiumTenant(String id, boolean isCentral) { }
+}

src/main/java/org/folio/search/client/InventoryClient.java

@@ -0,0 +1,64 @@
+package org.folio.search.client;
+
+import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;
+import static org.springframework.http.MediaType.APPLICATION_OCTET_STREAM_VALUE;
+
+import org.folio.search.model.client.CqlQuery;
+import org.folio.search.model.service.ResultList;
+import org.springframework.cloud.openfeign.FeignClient;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+
+@FeignClient
+public interface InventoryClient {
+
+  @GetMapping(
+    path = "/instance-storage/instances",
+    consumes = APPLICATION_OCTET_STREAM_VALUE,
+    produces = APPLICATION_JSON_VALUE)
+  ResultList<InventoryInstanceDto> getInstances(@RequestParam("query") CqlQuery cql,
+                                               @RequestParam("offset") int offset,
+                                               @RequestParam("limit") int limit);
+
+  @GetMapping(path = "/instance-storage/instances", produces = APPLICATION_JSON_VALUE)
+  ResultList<InventoryInstanceDto> getInstances(@RequestParam("limit") int limit,
+                                                @RequestParam("totalRecords") TotalRecordsType totalRecordsType);
+
+  @GetMapping(
+    path = "/item-storage/items",
+    consumes = APPLICATION_OCTET_STREAM_VALUE,
+    produces = APPLICATION_JSON_VALUE)
+  ResultList<InventoryItemDto> getItems(@RequestParam("query") CqlQuery cql,
+                                        @RequestParam("offset") int offset,
+                                        @RequestParam("limit") int limit);
+
+  @GetMapping(path = "/item-storage/items", produces = APPLICATION_JSON_VALUE)
+  ResultList<InventoryItemDto> getItems(@RequestParam("limit") int limit,
+                                        @RequestParam("totalRecords") TotalRecordsType totalRecordsType);
+
+  @GetMapping(
+    path = "/holdings-storage/holdings",
+    consumes = APPLICATION_OCTET_STREAM_VALUE,
+    produces = APPLICATION_JSON_VALUE)
+  ResultList<InventoryHoldingDto> getHoldings(@RequestParam("query") CqlQuery cql,
+                                              @RequestParam("offset") int offset,
+                                              @RequestParam("limit") int limit);
+
+  @GetMapping(path = "/holdings-storage/holdings", produces = APPLICATION_JSON_VALUE)
+  ResultList<InventoryHoldingDto> getHoldings(@RequestParam("limit") int limit,
+                                              @RequestParam("totalRecords") TotalRecordsType totalRecordsType);
+
+  @PostMapping(path = "/inventory-reindex-records/publish", consumes = APPLICATION_JSON_VALUE)
+  void publishReindexRecords(ReindexRecords reindexRecords);
+
+  record InventoryInstanceDto(String id) {}
+
+  record InventoryItemDto(String id) {}
+
+  record InventoryHoldingDto(String id) {}
+
+  record ReindexRecords(String id, String recordType, ReindexRecordsRange recordIdsRange) {}
+
+  record ReindexRecordsRange(String from, String to) {}
+}

src/main/java/org/folio/search/client/TotalRecordsType.java

@@ -0,0 +1,16 @@
+package org.folio.search.client;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+@Getter
+@AllArgsConstructor
+public enum TotalRecordsType {
+
+  EXACT("exact"),
+  ESTIMATED("estimated"),
+  NONE("none"),
+  AUTO("auto");
+
+  private final String value;
+}

src/main/java/org/folio/search/client/UserTenantsClient.java

@@ -15,5 +15,5 @@ public interface UserTenantsClient {
 
   record UserTenants(List<UserTenant> userTenants) { }
 
-  record UserTenant(String centralTenantId) { }
+  record UserTenant(String centralTenantId, String consortiumId) { }
 }

src/main/java/org/folio/search/configuration/SearchCacheNames.java

@@ -13,4 +13,5 @@ public class SearchCacheNames {
   public static final String TENANT_FEATURES_CACHE = "tenant-features";
   public static final String SEARCH_PREFERENCE_CACHE = "search-preference";
   public static final String USER_TENANTS_CACHE = "user-tenants";
+  public static final String CONSORTIUM_TENANTS_CACHE = "consortium-tenants-cache";
 }

src/main/java/org/folio/search/configuration/properties/ReindexConfigurationProperties.java

@@ -19,4 +19,6 @@ public class ReindexConfigurationProperties {
   private Integer locationBatchSize = 1_000;
 
   private Integer uploadRangeSize = 1_000;
+
+  private Integer mergeRangeSize = 1_000;
 }

src/main/java/org/folio/search/controller/IndexManagementController.java

@@ -14,6 +14,7 @@
 import org.folio.search.rest.resource.IndexManagementApi;
 import org.folio.search.service.IndexService;
 import org.folio.search.service.ResourceService;
+import org.folio.search.service.reindex.ReindexService;
 import org.springframework.http.ResponseEntity;
 import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -31,6 +32,7 @@ public class IndexManagementController implements IndexManagementApi {
 
   private final IndexService indexService;
   private final ResourceService resourceService;
+  private final ReindexService reindexService;
 
   @Override
   public ResponseEntity<FolioCreateIndexResponse> createIndices(String tenantId, CreateIndexRequest request) {
@@ -42,6 +44,13 @@ public ResponseEntity<FolioIndexOperationResponse> indexRecords(List<ResourceEve
     return ResponseEntity.ok(resourceService.indexResources(events));
   }
 
+  @Override
+  public ResponseEntity<Void> reindexInstanceRecords(String tenantId) {
+    log.info("Attempting to run full-reindex for instance records [tenant: {}]", tenantId);
+    reindexService.initFullReindex(tenantId);
+    return ResponseEntity.ok().build();
+  }
+
   @Override
   public ResponseEntity<ReindexJob> reindexInventoryRecords(String tenantId, ReindexRequest request) {
     log.info("Attempting to start reindex for inventory [tenant: {}]", tenantId);

src/main/java/org/folio/search/controller/SearchConsortiumController.java

@@ -29,7 +29,7 @@
 import org.folio.search.service.consortium.ConsortiumInstitutionService;
 import org.folio.search.service.consortium.ConsortiumLibraryService;
 import org.folio.search.service.consortium.ConsortiumLocationService;
-import org.folio.search.service.consortium.ConsortiumTenantService;
+import org.folio.search.service.consortium.UserTenantsService;
 import org.folio.spring.integration.XOkapiHeaders;
 import org.springframework.http.ResponseEntity;
 import org.springframework.validation.annotation.Validated;
@@ -46,7 +46,7 @@ public class SearchConsortiumController implements SearchConsortiumApi {
   static final String REQUEST_NOT_ALLOWED_MSG =
     "The request allowed only for central tenant of consortium environment";
 
-  private final ConsortiumTenantService consortiumTenantService;
+  private final UserTenantsService userTenantsService;
   private final ConsortiumInstanceService instanceService;
   private final ConsortiumLocationService locationService;
   private final ConsortiumInstanceSearchService searchService;
@@ -201,7 +201,7 @@ public ResponseEntity<ConsortiumItemCollection> fetchConsortiumBatchItems(String
   }
 
   private String verifyAndGetTenant(String tenantHeader) {
-    var centralTenant = consortiumTenantService.getCentralTenant(tenantHeader);
+    var centralTenant = userTenantsService.getCentralTenant(tenantHeader);
     if (centralTenant.isEmpty() || !centralTenant.get().equals(tenantHeader)) {
       throw new RequestValidationException(REQUEST_NOT_ALLOWED_MSG, XOkapiHeaders.TENANT, tenantHeader);
     }

src/main/java/org/folio/search/exception/FolioIntegrationException.java

@@ -0,0 +1,28 @@
+package org.folio.search.exception;
+
+import static org.folio.search.model.types.ErrorCode.INTEGRATION_ERROR;
+
+/**
+ * Handles exceptional cases of module integration with other Folio modules.
+ */
+public class FolioIntegrationException extends BaseSearchException {
+
+  /**
+   * Initialize exception with provided message and error code.
+   *
+   * @param message exception message
+   */
+  public FolioIntegrationException(String message) {
+    super(message, INTEGRATION_ERROR);
+  }
+
+  /**
+   * Initialize exception with provided message and error code.
+   *
+   * @param message exception message
+   * @param cause   cause Exception
+   */
+  public FolioIntegrationException(String message, Throwable cause) {
+    super(message, cause, INTEGRATION_ERROR);
+  }
+}

src/main/java/org/folio/search/integration/InventoryService.java

@@ -0,0 +1,133 @@
+package org.folio.search.integration;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import lombok.extern.log4j.Log4j2;
+import org.apache.commons.lang3.ObjectUtils;
+import org.folio.search.client.InventoryClient;
+import org.folio.search.client.TotalRecordsType;
+import org.folio.search.exception.FolioIntegrationException;
+import org.folio.search.model.client.CqlQuery;
+import org.folio.search.model.reindex.MergeRangeEntity;
+import org.folio.search.model.types.InventoryRecordType;
+import org.springframework.stereotype.Service;
+
+@Service
+@Log4j2
+public class InventoryService {
+
+  private final InventoryClient inventoryClient;
+
+  public InventoryService(InventoryClient inventoryClient) {
+    this.inventoryClient = inventoryClient;
+  }
+
+  public List<UUID> fetchInventoryRecordIds(InventoryRecordType recordType, CqlQuery cqlQuery, int offset, int limit) {
+    if (recordType == null) {
+      log.warn("No Inventory Record Type was provided to fetch Inventory Record");
+      return Collections.emptyList();
+    }
+
+    try {
+      return switch (recordType) {
+        case INSTANCE -> fetchInstances(cqlQuery, offset, limit);
+        case ITEM -> fetchItems(cqlQuery, offset, limit);
+        case HOLDING -> fetchHoldings(cqlQuery, offset, limit);
+      };
+    } catch (Exception e) {
+      log.warn("Failed to fetch Inventory records for {}", recordType);
+      throw new FolioIntegrationException("Failed to fetch inventory records for %s".formatted(recordType.name()), e);
+    }
+  }
+
+  public int fetchInventoryRecordCount(InventoryRecordType recordType) {
+    if (recordType == null) {
+      log.warn("No Inventory Record Type was provided to fetch Inventory Count");
+      return 0;
+    }
+
+    try {
+      var result = switch (recordType) {
+        case INSTANCE -> inventoryClient.getInstances(0, TotalRecordsType.EXACT);
+        case ITEM -> inventoryClient.getItems(0, TotalRecordsType.EXACT);
+        case HOLDING -> inventoryClient.getHoldings(0, TotalRecordsType.EXACT);
+      };
+
+      if (result == null) {
+        log.warn("Failed to retrieve Inventory Instances count");
+        return 0;
+      }
+
+      return result.getTotalRecords();
+    } catch (Exception e) {
+      log.warn("Failed to fetch Inventory record counts for {}", recordType);
+      throw new FolioIntegrationException(
+        "Failed to fetch inventory record counts for %s".formatted(recordType.name()), e);
+    }
+  }
+
+  public void publishReindexRecordsRange(MergeRangeEntity rangeEntity) {
+    if (rangeEntity == null
+      || ObjectUtils.anyNull(rangeEntity.getId(), rangeEntity.getLowerId(), rangeEntity.getUpperId())) {
+      log.warn("invalid Range Entity: [rangeEntity: {}]", rangeEntity);
+      return;
+    }
+
+    var from = rangeEntity.getLowerId().toString();
+    var to = rangeEntity.getUpperId().toString();
+    var recordsRange = new InventoryClient.ReindexRecords(
+      rangeEntity.getId().toString(),
+      rangeEntity.getEntityType().name(),
+      new InventoryClient.ReindexRecordsRange(from, to));
+
+    try {
+      inventoryClient.publishReindexRecords(recordsRange);
+    } catch (Exception e) {
+      log.warn("Failed to publish reindex records range {}", recordsRange);
+      throw new FolioIntegrationException("Failed to publish reindex records range", e);
+    }
+  }
+
+  private List<UUID> fetchInstances(CqlQuery cqlQuery, int offset, int limit) {
+    var result = inventoryClient.getInstances(cqlQuery, offset, limit);
+
+    if (result == null) {
+      log.warn("Failed to retrieve Inventory Instances, [query: {}, offset: {}, limit: {}]", cqlQuery, offset, limit);
+      return Collections.emptyList();
+    }
+
+    return result.getResult().stream()
+      .map(InventoryClient.InventoryInstanceDto::id)
+      .map(UUID::fromString)
+      .toList();
+  }
+
+  private List<UUID> fetchItems(CqlQuery cqlQuery, int offset, int limit) {
+    var result = inventoryClient.getItems(cqlQuery, offset, limit);
+
+    if (result == null) {
+      log.warn("Failed to retrieve Inventory Items, [query: {}, offset: {}, limit: {}]", cqlQuery, offset, limit);
+      return Collections.emptyList();
+    }
+
+    return result.getResult().stream()
+      .map(InventoryClient.InventoryItemDto::id)
+      .map(UUID::fromString)
+      .toList();
+  }
+
+  private List<UUID> fetchHoldings(CqlQuery cqlQuery, int offset, int limit) {
+    var result = inventoryClient.getHoldings(cqlQuery, offset, limit);
+
+    if (result == null) {
+      log.warn("Failed to retrieve Inventory Holdings, [query: {}, offset: {}, limit: {}]", cqlQuery, offset, limit);
+      return Collections.emptyList();
+    }
+
+    return result.getResult().stream()
+      .map(InventoryClient.InventoryHoldingDto::id)
+      .map(UUID::fromString)
+      .toList();
+  }
+}