[MODORDERS-1210] Implement API to change piece statuses in batch

descriptors/ModuleDescriptor-template.json

@@ -627,6 +627,31 @@
             "invoice.invoice-lines.item.put"
           ]
         },
+        {
+          "methods": ["PUT"],
+          "pathPattern": "/orders/pieces-batch/status",
+          "permissionsRequired": ["orders.pieces.collection.put"],
+          "modulePermissions": [
+            "orders-storage.pieces.item.put",
+            "orders-storage.pieces.collection.get",
+            "orders-storage.po-lines.item.get",
+            "orders-storage.po-lines.item.put",
+            "orders-storage.po-lines.collection.get",
+            "orders-storage.purchase-orders.item.get",
+            "orders-storage.purchase-orders.item.put",
+            "orders-storage.titles.collection.get",
+            "finance.funds.item.get",
+            "finance.ledgers.current-fiscal-year.item.get",
+            "finance.transactions.collection.get",
+            "finance.transactions.batch.execute",
+            "inventory.items.item.get",
+            "inventory.items.item.put",
+            "inventory.items.collection.get",
+            "inventory-storage.holdings.collection.get",
+            "acquisitions-units-storage.units.collection.get",
+            "acquisitions-units-storage.memberships.collection.get"
+          ]
+        },
         {
           "methods": ["GET"],
           "pathPattern": "/orders/pieces-requests",
@@ -1519,6 +1544,11 @@
       "displayName": "orders - delete an existing piece record",
       "description": "Delete an existing piece"
     },
+    {
+      "permissionName": "orders.pieces.collection.put",
+      "displayName": "orders - batch update piece statuses",
+      "description": "Batch update piece statuses"
+    },
     {
       "permissionName": "orders.piece-requests.collection.get",
       "displayName": "Orders - Get piece requests",
@@ -1530,6 +1560,7 @@
       "description" : "All permissions for the pieces",
       "subPermissions" : [
         "orders.pieces.collection.get",
+        "orders.pieces.collection.put",
         "orders.pieces.item.post",
         "orders.pieces.item.get",
         "orders.pieces.item.put",

ramls/pieces-batch.raml

@@ -0,0 +1,49 @@
+#%RAML 1.0
+title: Pieces Batch Operations
+baseUri: https://github.com/folio-org/mod-orders
+version: v1
+protocols: [ HTTP, HTTPS ]
+
+documentation:
+  - title: Endpoint for batch operations on Pieces
+    content: <b>Endpoint for batch operations on Pieces</b>
+
+types:
+  piece-batch-status-collection: !include acq-models/mod-orders/schemas/pieceStatusBatchCollection.json
+
+/orders/pieces-batch:
+  /status:
+    put:
+      description: Batch status update for pieces
+      body:
+        application/json:
+          type: piece-batch-status-collection
+          example:
+            strict: false
+            value: !include acq-models/mod-orders/examples/pieceStatusBatchCollection.sample
+      responses:
+        204:
+          description: "Piece records successfully updated"
+        400:
+          description: "Bad request"
+          body:
+            application/json:
+              example:
+                strict: false
+                value: !include examples/errors_400.sample
+            text/plain:
+              example: "Unable to update pieces - Bad request"
+        404:
+          description: "Pieces do not exist"
+          body:
+            text/plain:
+              example: "Following pieces are not found: [id1, id2]"
+        500:
+          description: "Internal server error, e.g. due to misconfiguration"
+          body:
+            application/json:
+              example:
+                strict: false
+                value: !include examples/errors_500.sample
+            text/plain:
+              example: "Unable to update pieces - Internal server error, e.g. due to misconfiguration"

src/main/java/org/folio/config/ApplicationConfig.java

@@ -641,13 +641,12 @@ PieceUpdateFlowPoLineService pieceUpdateFlowPoLineService(PurchaseOrderStorageSe
   }
 
   @Bean
-  PieceUpdateFlowManager pieceUpdateFlowManager(PieceStorageService pieceStorageService, PieceService pieceService, ProtectionService protectionService,
-            PieceUpdateFlowPoLineService pieceUpdateFlowPoLineService, PieceUpdateFlowInventoryManager pieceUpdateFlowInventoryManager,
-            BasePieceFlowHolderBuilder basePieceFlowHolderBuilder, DefaultPieceFlowsValidator defaultPieceFlowsValidator,
-            PurchaseOrderLineService purchaseOrderLineService) {
-    return new PieceUpdateFlowManager(pieceStorageService, pieceService, protectionService, pieceUpdateFlowPoLineService,
-                            pieceUpdateFlowInventoryManager, basePieceFlowHolderBuilder, defaultPieceFlowsValidator,
-                            purchaseOrderLineService);
+  PieceUpdateFlowManager pieceUpdateFlowManager(PieceStorageService pieceStorageService, PieceService pieceService, TitlesService titlesService,
+                                                ProtectionService protectionService, PieceUpdateFlowPoLineService pieceUpdateFlowPoLineService,
+                                                PieceUpdateFlowInventoryManager pieceUpdateFlowInventoryManager, BasePieceFlowHolderBuilder basePieceFlowHolderBuilder,
+                                                DefaultPieceFlowsValidator defaultPieceFlowsValidator, PurchaseOrderLineService purchaseOrderLineService) {
+    return new PieceUpdateFlowManager(pieceStorageService, pieceService, titlesService, protectionService, pieceUpdateFlowPoLineService,
+      pieceUpdateFlowInventoryManager, basePieceFlowHolderBuilder, defaultPieceFlowsValidator, purchaseOrderLineService);
   }
 
   @Bean

src/main/java/org/folio/models/pieces/PieceBatchStatusUpdateHolder.java

@@ -0,0 +1,28 @@
+package org.folio.models.pieces;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import org.folio.rest.jaxrs.model.Piece;
+
+import java.util.List;
+
+@AllArgsConstructor
+public class PieceBatchStatusUpdateHolder extends BasePieceFlowHolder {
+
+  @Getter
+  private Piece.ReceivingStatus receivingStatus;
+  @Getter
+  private List<Piece> pieces;
+  private String poLineId;
+
+  @Override
+  public String getOrderLineId() {
+    return poLineId;
+  }
+
+  @Override
+  public String getTitleId() {
+    return null;
+  }
+
+}

src/main/java/org/folio/orders/events/handlers/ReceiptStatusConsistency.java

@@ -1,162 +1,88 @@
 package org.folio.orders.events.handlers;
 
-import static org.folio.helper.CheckinReceivePiecesHelper.EXPECTED_STATUSES;
-import static org.folio.helper.CheckinReceivePiecesHelper.RECEIVED_STATUSES;
-import static org.folio.orders.utils.ResourcePathResolver.PIECES_STORAGE;
-import static org.folio.orders.utils.ResourcePathResolver.resourcesPath;
-import static org.folio.rest.jaxrs.model.PoLine.ReceiptStatus.AWAITING_RECEIPT;
-import static org.folio.rest.jaxrs.model.PoLine.ReceiptStatus.FULLY_RECEIVED;
-import static org.folio.rest.jaxrs.model.PoLine.ReceiptStatus.PARTIALLY_RECEIVED;
+import static org.folio.orders.events.utils.EventUtils.getPoLineId;
+import static org.folio.orders.utils.HelperUtils.getOkapiHeaders;
+import static org.folio.service.orders.utils.StatusUtils.calculatePoLineReceiptStatus;
 
-import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 
-import org.apache.commons.collections4.CollectionUtils;
 import org.folio.helper.BaseHelper;
 import org.folio.orders.utils.HelperUtils;
 import org.folio.orders.utils.PoLineCommonUtil;
-import org.folio.rest.core.RestClient;
 import org.folio.rest.core.models.RequestContext;
 import org.folio.rest.jaxrs.model.Piece;
-import org.folio.rest.jaxrs.model.Piece.ReceivingStatus;
-import org.folio.rest.jaxrs.model.PieceCollection;
 import org.folio.rest.jaxrs.model.PoLine;
-import org.folio.rest.jaxrs.model.PoLine.ReceiptStatus;
 import org.folio.service.orders.PurchaseOrderLineService;
+import org.folio.service.pieces.PieceStorageService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
-import io.vertx.core.Promise;
 import io.vertx.core.Vertx;
 import io.vertx.core.eventbus.Message;
 import io.vertx.core.json.JsonArray;
 import io.vertx.core.json.JsonObject;
-import one.util.streamex.StreamEx;
 
 @Component("receiptStatusHandler")
 public class ReceiptStatusConsistency extends BaseHelper implements Handler<Message<JsonObject>> {
 
-  private static final int LIMIT = Integer.MAX_VALUE;
-  private static final String PIECES_ENDPOINT = resourcesPath(PIECES_STORAGE) + "?query=poLineId==%s&limit=%s";
-
+  private final PieceStorageService pieceStorageService;
   private final PurchaseOrderLineService purchaseOrderLineService;
 
 
   @Autowired
-  public ReceiptStatusConsistency(Vertx vertx, PurchaseOrderLineService purchaseOrderLineService) {
+  public ReceiptStatusConsistency(Vertx vertx, PieceStorageService pieceStorageService, PurchaseOrderLineService purchaseOrderLineService) {
     super(vertx.getOrCreateContext());
+    this.pieceStorageService = pieceStorageService;
     this.purchaseOrderLineService = purchaseOrderLineService;
   }
 
   @Override
   public void handle(Message<JsonObject> message) {
-    JsonObject messageFromEventBus = message.body();
-
+    var messageFromEventBus = message.body();
     logger.info("Received message body: {}", messageFromEventBus);
 
-    Map<String, String> okapiHeaders = org.folio.orders.utils.HelperUtils.getOkapiHeaders(message);
+    var okapiHeaders = getOkapiHeaders(message);
     var requestContext = new RequestContext(ctx, okapiHeaders);
-    List<Future<Void>> futures = new ArrayList<>();
-    Promise<Void> promise = Promise.promise();
-    futures.add(promise.future());
-
-    String poLineIdUpdate = messageFromEventBus.getString("poLineIdUpdate");
-    String query = String.format(PIECES_ENDPOINT, poLineIdUpdate, LIMIT);
-
-    // 1. Get all pieces for poLineId
-    getPieces(query, requestContext)
-      .compose(listOfPieces ->
-        // 2. Get PoLine for the poLineId which will be used to calculate PoLineReceiptStatus
-        purchaseOrderLineService.getOrderLineById(poLineIdUpdate, requestContext)
-          .map(poLine -> {
-            if (PoLineCommonUtil.isCancelledOrOngoingStatus(poLine)) {
-              promise.complete();
-              return null;
-            }
-            ReceiptStatus receivingStatus = calculatePoLineReceiptStatus(poLine, listOfPieces);
-            boolean statusUpdated = purchaseOrderLineService.updatePoLineReceiptStatusWithoutSave(poLine, receivingStatus);
-            if (statusUpdated) {
-              purchaseOrderLineService.saveOrderLine(poLine, requestContext)
-                .map(aVoid -> {
-                  // send event to update order status
-                  updateOrderStatus(poLine, okapiHeaders, requestContext);
-                  promise.complete();
-                  return null;
-                })
-                .onFailure(e -> {
-                  logger.error("The error updating poLine by id {}", poLineIdUpdate, e);
-                  promise.fail(e);
-                });
-            } else {
-              promise.complete();
-            }
-            return null;
-          })
-          .onFailure(e -> {
-            logger.error("The error getting poLine by id {}", poLineIdUpdate, e);
-            promise.fail(e);
-          }))
-      .onFailure(e -> {
-        logger.error("The error happened getting all pieces by poLine {}", poLineIdUpdate, e);
-        promise.fail(e);
-      });
 
-    // Now wait for all operations to be completed and send reply
-    completeAllFutures(futures, message);
-  }
+    var poLineId = getPoLineId(messageFromEventBus);
+    var future = pieceStorageService.getPiecesByLineId(poLineId, requestContext)
+      .compose(pieces -> purchaseOrderLineService.getOrderLineById(poLineId, requestContext)
+        .compose(poLine -> updatePoLineAndOrderStatuses(pieces, poLine, requestContext))
+        .onFailure(e -> logger.error("Exception occurred while fetching PoLine by id: '{}'", poLineId, e)))
+      .onFailure(e -> logger.error("Exception occurred while fetching pieces by PoLine id: '{}'", poLineId, e));
 
-  private void updateOrderStatus(PoLine poLine, Map<String, String> okapiHeaders, RequestContext requestContext) {
-    List<JsonObject> poIds = StreamEx
-      .of(poLine)
-      .map(PoLine::getPurchaseOrderId)
-      .distinct()
-      .map(orderId -> new JsonObject().put(ORDER_ID, orderId))
-      .toList();
-    JsonObject messageContent = new JsonObject();
-    messageContent.put(OKAPI_HEADERS, okapiHeaders);
-    // Collect order ids which should be processed
-    messageContent.put(EVENT_PAYLOAD, new JsonArray(poIds));
-    HelperUtils.sendEvent(MessageAddress.RECEIVE_ORDER_STATUS_UPDATE, messageContent, requestContext);
+    completeAllFutures(List.of(future), message);
   }
 
-  private ReceiptStatus calculatePoLineReceiptStatus(PoLine poLine, List<Piece> pieces) {
-
-    if (CollectionUtils.isEmpty(pieces)) {
-      logger.info("No pieces processed - receipt status unchanged for PO Line '{}'", poLine.getId());
-      return poLine.getReceiptStatus();
+  private Future<Void> updatePoLineAndOrderStatuses(List<Piece> pieces, PoLine poLine, RequestContext requestContext) {
+    if (PoLineCommonUtil.isCancelledOrOngoingStatus(poLine)) {
+      logger.info("updatePoLineAndOrderStatuses:: PoLine with id: '{}' has status: '{}', skipping...", poLine.getId(), poLine.getReceiptStatus());
+      return Future.succeededFuture();
     }
-
-    long expectedQty = getPiecesQuantityByPoLineAndStatus(EXPECTED_STATUSES, pieces);
-    return calculatePoLineReceiptStatus(expectedQty, pieces);
-  }
-
-  private ReceiptStatus calculatePoLineReceiptStatus(long expectedPiecesQuantity, List<Piece> pieces) {
-    if (expectedPiecesQuantity == 0) {
-      logger.info("calculatePoLineReceiptStatus:: Fully received");
-      return FULLY_RECEIVED;
-    }
-
-    if (StreamEx.of(pieces).anyMatch(piece -> RECEIVED_STATUSES.contains(piece.getReceivingStatus()))) {
-      logger.info("calculatePoLineReceiptStatus:: Partially Received - In case there is at least one successfully received piece");
-      return PARTIALLY_RECEIVED;
+    var newStatus = pieces.isEmpty()
+      ? poLine.getReceiptStatus()
+      : calculatePoLineReceiptStatus(poLine.getId(), pieces);
+    boolean statusUpdated = purchaseOrderLineService.updatePoLineReceiptStatusWithoutSave(poLine, newStatus);
+    if (!statusUpdated) {
+      logger.info("updatePoLineAndOrderStatuses:: PoLine receipt status is not updated, skipping...");
+      return Future.succeededFuture();
     }
-
-    logger.info("calculatePoLineReceiptStatus::Pieces were rolled-back to Expected, checking if there is any Received piece in the storage");
-    long receivedQty = getPiecesQuantityByPoLineAndStatus(RECEIVED_STATUSES, pieces);
-    return receivedQty == 0 ? AWAITING_RECEIPT : PARTIALLY_RECEIVED;
+    return purchaseOrderLineService.saveOrderLine(poLine, requestContext)
+      .compose(v -> updateOrderStatus(poLine, okapiHeaders, requestContext))
+      .onSuccess(v -> logger.info("updatePoLineAndOrderStatuses:: Order '{}' and PoLine '{}' updated successfully", poLine.getId(), poLine.getPurchaseOrderId()))
+      .onFailure(e -> logger.error("Exception occurred while updating Order '{}' and PoLine '{}'", poLine.getId(), poLine.getPurchaseOrderId(), e));
   }
 
-  private long getPiecesQuantityByPoLineAndStatus(List<ReceivingStatus> receivingStatuses, List<Piece> pieces) {
-    return pieces.stream()
-      .filter(piece -> receivingStatuses.contains(piece.getReceivingStatus()))
-      .count();
+  private Future<Void> updateOrderStatus(PoLine poLine, Map<String, String> okapiHeaders, RequestContext requestContext) {
+    var messageContent = JsonObject.of(
+      OKAPI_HEADERS, okapiHeaders,
+      EVENT_PAYLOAD, JsonArray.of(JsonObject.of(ORDER_ID, poLine.getPurchaseOrderId()))
+    );
+    HelperUtils.sendEvent(MessageAddress.RECEIVE_ORDER_STATUS_UPDATE, messageContent, requestContext);
+    return Future.succeededFuture();
   }
 
-  Future<List<Piece>> getPieces(String endpoint, RequestContext requestContext) {
-    return new RestClient().get(endpoint, PieceCollection.class, requestContext)
-      .map(PieceCollection::getPieces);
-  }
 }

src/main/java/org/folio/orders/events/utils/EventUtils.java

@@ -0,0 +1,19 @@
+package org.folio.orders.events.utils;
+
+import io.vertx.core.json.JsonObject;
+
+public class EventUtils {
+
+  public static final String POL_UPDATE_FIELD = "poLineIdUpdate";
+
+  public static JsonObject createPoLineUpdateEvent(String poLineId) {
+    return JsonObject.of(POL_UPDATE_FIELD, poLineId);
+  }
+
+  public static String getPoLineId(JsonObject eventPayload) {
+    return eventPayload.getString(POL_UPDATE_FIELD);
+  }
+
+  private EventUtils() {}
+
+}