feat(reindex): Initiate Full Reindex (#642)

* feat(reindex): Initiate Full Reindex

1. Truncate tables: instance, item, holding, ranges (merge and upload), status
2. Request counts of records from mod-inventory-storage (do this for each member in ECS)
3. Construct records' ID ranges and fill Merge Range table (do this for each member in ECS)
4. Call mod-inventory-storage's POST /inventory-reindex-records/publish for constructed records' ranges (do this for each member in ECS)
Do not allow running full reindex from member tenants in ECS

Closes: MSEARCH-794

descriptors/ModuleDescriptor-template.json

@@ -83,6 +83,22 @@
           "permissionsRequired": [
             "search.index.reindex.status.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",
+            "inventory-storage.instances.collection.get",
+            "inventory-storage.items.collection.get",
+            "inventory-storage.holdings.collection.get",
+            "user-tenants.collection.get"
+          ]
         }
       ]
     },
@@ -632,6 +648,11 @@
       "displayName": "Search - updates settings for the index",
       "description": "Updates settings for the index"
     },
+    {
+      "permissionName": "search.index.instance-records.reindex.full.post",
+      "displayName": "Search - starts inventory instance records full reindex operation",
+      "description": "Starts inventory instance records reindex operation"
+    },
     {
       "permissionName": "search.instances.collection.get",
       "displayName": "Search - searches instances by given query",

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/InventoryInstanceClient.java

@@ -0,0 +1,22 @@
+package org.folio.search.client;
+
+import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;
+
+import java.net.URI;
+import org.springframework.cloud.openfeign.FeignClient;
+import org.springframework.web.bind.annotation.GetMapping;
+
+@FeignClient("instance-storage")
+public interface InventoryInstanceClient {
+
+  /**
+   * Retrieves Inventory Records count with given URI.
+   *
+   * @param uri URI to retrieve count of inventory record
+   * @return count represented as {@link InventoryRecordsCountDto}
+   */
+  @GetMapping(produces = APPLICATION_JSON_VALUE)
+  InventoryRecordsCountDto getInventoryRecordsCount(URI uri);
+
+  record InventoryRecordsCountDto(int totalRecords) { }
+}

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

@@ -0,0 +1,17 @@
+package org.folio.search.client;
+
+import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;
+
+import org.springframework.cloud.openfeign.FeignClient;
+import org.springframework.web.bind.annotation.PostMapping;
+
+@FeignClient("inventory-reindex-records")
+public interface InventoryReindexRecordsClient {
+
+  @PostMapping(path = "/publish", consumes = APPLICATION_JSON_VALUE)
+  void publishReindexRecords(ReindexRecords reindexRecords);
+
+  record ReindexRecords(String id, String recordType, ReindexRecordsRange recordIdsRange) {}
+
+  record ReindexRecordsRange(String from, String to) {}
+}

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/AsyncConfig.java

@@ -3,6 +3,7 @@
 import java.util.concurrent.Executor;
 import lombok.RequiredArgsConstructor;
 import org.folio.search.configuration.properties.StreamIdsProperties;
+import org.folio.search.service.FolioExecutor;
 import org.folio.spring.scope.FolioExecutionScopeExecutionContextManager;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -27,5 +28,10 @@ public Executor streamIdsExecutor() {
     executor.initialize();
     return executor;
   }
+
+  @Bean("reindexExecutor")
+  public FolioExecutor remappingExecutor() {
+    return new FolioExecutor(0, 1);
+  }
 }
 

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

@@ -15,7 +15,8 @@
 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.ReindexRangeIndexService;
+import org.folio.search.service.reindex.ReindexService;
+import org.folio.search.service.reindex.ReindexStatusService;
 import org.springframework.http.ResponseEntity;
 import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -32,8 +33,9 @@
 public class IndexManagementController implements IndexManagementApi {
 
   private final IndexService indexService;
-  private final ReindexRangeIndexService reindexRangeService;
   private final ResourceService resourceService;
+  private final ReindexService reindexService;
+  private final ReindexStatusService reindexStatusService;
 
   @Override
   public ResponseEntity<FolioCreateIndexResponse> createIndices(String tenantId, CreateIndexRequest request) {
@@ -45,6 +47,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);
@@ -65,6 +74,6 @@ public ResponseEntity<FolioIndexOperationResponse> updateMappings(String tenantI
 
   @Override
   public ResponseEntity<List<ReindexStatusItem>> getReindexStatus(String tenantId) {
-    return ResponseEntity.ok(reindexRangeService.getReindexStatuses(tenantId));
+    return ResponseEntity.ok(reindexStatusService.getReindexStatuses(tenantId));
   }
 }

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,71 @@
+package org.folio.search.integration;
+
+import lombok.extern.log4j.Log4j2;
+import org.apache.commons.lang3.ObjectUtils;
+import org.folio.search.client.InventoryInstanceClient;
+import org.folio.search.client.InventoryReindexRecordsClient;
+import org.folio.search.exception.FolioIntegrationException;
+import org.folio.search.model.reindex.MergeRangeEntity;
+import org.folio.search.model.types.InventoryRecordType;
+import org.springframework.stereotype.Service;
+import org.springframework.web.util.UriComponentsBuilder;
+
+@Service
+@Log4j2
+public class InventoryService {
+
+  private final InventoryInstanceClient inventoryInstanceClient;
+  private final InventoryReindexRecordsClient reindexRecordsClient;
+
+  public InventoryService(InventoryInstanceClient inventoryInstanceClient,
+                          InventoryReindexRecordsClient reindexRecordsClient) {
+    this.inventoryInstanceClient = inventoryInstanceClient;
+    this.reindexRecordsClient = reindexRecordsClient;
+  }
+
+  public int fetchInventoryRecordsCount(InventoryRecordType recordType) {
+    if (recordType == null) {
+      log.warn("No Inventory Record Type was provided to fetch Inventory Count");
+      return 0;
+    }
+
+    try {
+      var countPath = "%s?limit=0&totalRecords=exact".formatted(recordType.getPath());
+      var uri = UriComponentsBuilder.fromUriString(countPath).build().toUri();
+      var result = inventoryInstanceClient.getInventoryRecordsCount(uri);
+
+      if (result == null) {
+        log.warn("Failed to retrieve Inventory Records count");
+        return 0;
+      }
+
+      return result.totalRecords();
+    } catch (Exception e) {
+      log.warn("Failed to fetch Inventory record counts for {}", recordType);
+      throw new FolioIntegrationException(
+        "Failed to fetch inventory record counts for %s : %s".formatted(recordType.name(), e.getMessage()));
+    }
+  }
+
+  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 InventoryReindexRecordsClient.ReindexRecords(
+      rangeEntity.getId().toString(),
+      rangeEntity.getEntityType().getType(),
+      new InventoryReindexRecordsClient.ReindexRecordsRange(from, to));
+
+    try {
+      reindexRecordsClient.publishReindexRecords(recordsRange);
+    } catch (Exception e) {
+      log.warn("Failed to publish reindex records range {} : {}", recordsRange, e.getMessage());
+      throw new FolioIntegrationException("Failed to publish reindex records range", e);
+    }
+  }
+}

src/main/java/org/folio/search/model/client/CqlQuery.java

@@ -21,6 +21,14 @@ public static CqlQuery exactMatchAny(CqlQueryParam param, Iterable<String> value
     return fromTemplate("%s==(%s)", param.getCqlParam(), valuesConcatenated);
   }
 
+  public static CqlQuery greaterThan(CqlQueryParam param, String value) {
+    return fromTemplate("%s>(%s)", param.getCqlParam(), value);
+  }
+
+  public static CqlQuery sortBy(CqlQuery cqlQuery, CqlQueryParam param) {
+    return fromTemplate("%s sortBy %s", cqlQuery.query, param.getCqlParam());
+  }
+
   private static CqlQuery fromTemplate(String format, Object... args) {
     return new CqlQuery(String.format(format, args));
   }

src/main/java/org/folio/search/model/reindex/MergeRangeEntity.java

@@ -0,0 +1,27 @@
+package org.folio.search.model.reindex;
+
+import java.sql.Timestamp;
+import java.util.UUID;
+import lombok.Data;
+import org.folio.search.model.types.ReindexEntityType;
+
+@Data
+public class MergeRangeEntity {
+
+  public static final String ID_COLUMN = "id";
+  public static final String ENTITY_TYPE_COLUMN = "entity_type";
+  public static final String TENANT_ID_COLUMN = "tenant_id";
+  public static final String RANGE_LOWER_COLUMN = "lower";
+  public static final String RANGE_UPPER_COLUMN = "upper";
+  public static final String CREATED_AT_COLUMN = "created_at";
+  public static final String FINISHED_AT_COLUMN = "finished_at";
+
+  private final UUID id;
+  private final ReindexEntityType entityType;
+  private final String tenantId;
+  private final UUID lowerId;
+  private final UUID upperId;
+  private final Timestamp createdAt;
+  private Timestamp finishedAt;
+
+}

src/main/java/org/folio/search/model/reindex/ReindexStatusEntity.java

@@ -8,7 +8,6 @@
 @Data
 public class ReindexStatusEntity {
 
-  public static final String REINDEX_STATUS_TABLE = "reindex_status";
   public static final String ENTITY_TYPE_COLUMN = "entity_type";
   public static final String STATUS_COLUMN = "status";
   public static final String TOTAL_MERGE_RANGES_COLUMN = "total_merge_ranges";
@@ -30,5 +29,4 @@ public class ReindexStatusEntity {
   private Timestamp endTimeMerge;
   private Timestamp startTimeUpload;
   private Timestamp endTimeUpload;
-
 }

src/main/java/org/folio/search/model/reindex/UploadRangeEntity.java

@@ -8,7 +8,6 @@
 @Data
 public class UploadRangeEntity {
 
-  public static final String UPLOAD_RANGE_TABLE = "upload_range";
   public static final String ID_COLUMN = "id";
   public static final String ENTITY_TYPE_COLUMN = "entity_type";
   public static final String RANGE_LIMIT_COLUMN = "range_limit";

src/main/java/org/folio/search/model/types/ErrorCode.java

@@ -13,7 +13,8 @@ public enum ErrorCode {
   ELASTICSEARCH_ERROR("elasticsearch_error"),
   VALIDATION_ERROR("validation_error"),
   NOT_FOUND_ERROR("not_found_error"),
-  CONSTRAINT_VIOLATION("constraint_violation_error");
+  CONSTRAINT_VIOLATION("constraint_violation_error"),
+  INTEGRATION_ERROR("integration-error");
 
   @JsonValue
   private final String value;

src/main/java/org/folio/search/model/types/InventoryRecordType.java

@@ -0,0 +1,32 @@
+package org.folio.search.model.types;
+
+import java.util.Locale;
+import lombok.Getter;
+
+@Getter
+public enum InventoryRecordType {
+  INSTANCE("instance", "http://instance-storage/instances"),
+  ITEM("item", "http://item-storage/items"),
+  HOLDING("holding", "http://holdings-storage/holdings");
+
+
+  /**
+   * record name.
+   */
+  private final String recordName;
+
+  /**
+   * Request path for the record.
+   */
+  private final String path;
+
+  InventoryRecordType(String recordName, String path) {
+    this.recordName = recordName;
+    this.path = path;
+  }
+
+  @Override
+  public String toString() {
+    return name().toLowerCase(Locale.ROOT);
+  }
+}