[MODORDERS-1209]. Implement API to Send claims for multiple pieces

ramls/claim.raml

@@ -0,0 +1,39 @@
+#%RAML 1.0
+title: Claim
+baseUri: https://github.com/folio-org/mod-orders
+version: v1
+protocols: [ HTTP, HTTPS ]
+
+documentation:
+  - title: Orders Business Logic API
+    content: <b>API for claiming pieces</b>
+
+types:
+  claiming-collection: !include acq-models/mod-orders/schemas/claimingCollection.json
+  claiming-results: !include acq-models/mod-orders/schemas/claimingResults.json
+  errors: !include raml-util/schemas/errors.schema
+  UUID:
+    type: string
+    pattern: ^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$
+
+traits:
+  validate: !include raml-util/traits/validation.raml
+
+resourceTypes:
+  post-with-200: !include rtypes/post-json-200.raml
+
+/orders/claim:
+  displayName: Claim pieces
+  description: |
+    Claim pieces. The endpoint is used to:
+    - Claims pieces grouped by organizations
+    - Triggers jobs in mod-data-export per each organization that contains an integration detail
+  type:
+    post-with-200:
+      requestSchema: claiming-collection
+      responseSchema: claiming-results
+      requestExample: !include acq-models/mod-orders/examples/claimingCollection.sample
+      responseExample: !include acq-models/mod-orders/examples/claimingResults.sample
+  is: [validate]
+  post:
+    description: Claim pieces

src/main/java/org/folio/models/ClaimingHolder.java

@@ -0,0 +1,22 @@
+package org.folio.models;
+
+import org.folio.rest.jaxrs.model.Piece;
+
+import java.util.List;
+
+public class ClaimingHolder {
+
+  private List<Piece> pieces;
+
+  public ClaimingHolder() {
+  }
+
+  public ClaimingHolder withPieces(List<Piece> pieces) {
+    this.pieces = pieces;
+    return this;
+  }
+
+  public List<Piece> getPieces() {
+    return pieces;
+  }
+}

src/main/java/org/folio/orders/utils/HelperUtils.java

@@ -65,6 +65,7 @@ public class HelperUtils {
 
   public static final String SYSTEM_CONFIG_MODULE_NAME = "ORG";
   public static final String ORDER_CONFIG_MODULE_NAME = "ORDERS";
+  public static final String DATA_EXPORT_SPRING_CONFIG_MODULE_NAME = "mod-data-export-spring";
 
   public static final String DEFAULT_POLINE_LIMIT = "1";
   public static final String REASON_COMPLETE = "Complete";

src/main/java/org/folio/orders/utils/ResourcePathResolver.java

@@ -14,6 +14,7 @@ private ResourcePathResolver() {
   public static final String PO_LINES_BATCH_STORAGE = "poLines.batch";
   public static final String PO_LINES_BUSINESS = "poLinesBusinessEndpoint";
   public static final String ORDERS_BUSINESS = "ordersBusinessEndpoint";
+  public static final String CLAIMING_BUSINESS = "claimingBusinessEndpoint";
   public static final String PO_NUMBER = "poNumber";
   public static final String VENDOR_ID = "vendor";
   public static final String PO_LINE_NUMBER = "poLineNumber";
@@ -72,6 +73,7 @@ private ResourcePathResolver() {
     apis.put(PO_LINES_BATCH_STORAGE, "/orders-storage/po-lines-batch");
     apis.put(PO_LINES_BUSINESS, "/orders/order-lines");
     apis.put(ORDERS_BUSINESS, "/orders/composite-orders");
+    apis.put(CLAIMING_BUSINESS, "/orders/claim");
     apis.put(PO_NUMBER, "/orders-storage/po-number");
     apis.put(PURCHASE_ORDER_STORAGE, "/orders-storage/purchase-orders");
     apis.put(PIECES_STORAGE, "/orders-storage/pieces");

src/main/java/org/folio/rest/impl/ClaimingApi.java

@@ -0,0 +1,46 @@
+package org.folio.rest.impl;
+
+import io.vertx.core.AsyncResult;
+import io.vertx.core.Context;
+import io.vertx.core.Future;
+import io.vertx.core.Handler;
+import io.vertx.core.Vertx;
+import org.folio.rest.annotations.Validate;
+import org.folio.rest.core.models.RequestContext;
+import org.folio.rest.jaxrs.model.ClaimingCollection;
+import org.folio.rest.jaxrs.resource.OrdersClaim;
+import org.folio.service.claims.ClaimingService;
+import org.folio.spring.SpringContextUtil;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import javax.ws.rs.core.Response;
+import java.util.Map;
+
+import static org.folio.orders.utils.ResourcePathResolver.CLAIMING_BUSINESS;
+import static org.folio.orders.utils.ResourcePathResolver.resourceByIdPath;
+import static org.folio.rest.RestConstants.OKAPI_URL;
+
+public class ClaimingApi extends BaseApi implements OrdersClaim {
+
+  @Autowired
+  private ClaimingService claimingService;
+
+  public ClaimingApi() {
+    SpringContextUtil.autowireDependencies(this, Vertx.currentContext());
+  }
+
+  @Override
+  @Validate
+  public void postOrdersClaim(ClaimingCollection claimingCollection, Map<String, String> okapiHeaders,
+                              Handler<AsyncResult<Response>> asyncResultHandler, Context vertxContext) {
+    var requestContext = new RequestContext(vertxContext, okapiHeaders);
+    claimingService.sendClaims(claimingCollection, requestContext)
+      .onSuccess(claimingResults -> {
+        var okapiUrl = okapiHeaders.get(OKAPI_URL);
+        var url = resourceByIdPath(CLAIMING_BUSINESS);
+        var response = buildResponseWithLocation(okapiUrl, url, claimingResults);
+        asyncResultHandler.handle(Future.succeededFuture(response));
+      })
+      .onFailure(t -> handleErrorResponse(asyncResultHandler, t));
+  }
+}

src/main/java/org/folio/service/claims/ClaimingService.java

@@ -0,0 +1,180 @@
+package org.folio.service.claims;
+
+import io.vertx.core.Future;
+import io.vertx.core.json.JsonObject;
+import lombok.extern.log4j.Log4j2;
+import one.util.streamex.StreamEx;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.folio.models.ClaimingHolder;
+import org.folio.rest.core.RestClient;
+import org.folio.rest.core.models.RequestContext;
+import org.folio.rest.jaxrs.model.ClaimingCollection;
+import org.folio.rest.jaxrs.model.ClaimingResult;
+import org.folio.rest.jaxrs.model.ClaimingResults;
+import org.folio.rest.jaxrs.model.Piece;
+import org.folio.service.caches.ConfigurationEntriesCache;
+import org.folio.service.orders.PurchaseOrderLineService;
+import org.folio.service.orders.PurchaseOrderStorageService;
+import org.folio.service.organization.OrganizationService;
+import org.folio.service.pieces.PieceStorageService;
+import org.springframework.stereotype.Service;
+import org.springframework.util.CollectionUtils;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static java.util.stream.Collectors.collectingAndThen;
+import static java.util.stream.Collectors.mapping;
+import static java.util.stream.Collectors.toList;
+import static org.folio.orders.utils.HelperUtils.DATA_EXPORT_SPRING_CONFIG_MODULE_NAME;
+import static org.folio.orders.utils.HelperUtils.collectResultsOnSuccess;
+
+@Log4j2
+@Service
+public class ClaimingService {
+
+  private static final Logger logger = LogManager.getLogger(ClaimingService.class);
+  private static final String CREATE_JOB = "/data-export-spring/jobs";
+  private static final String EXECUTE_JOB = "/data-export-spring/jobs/send";
+  private static final String JOB_STATUS = "status";
+
+  private final ConfigurationEntriesCache configurationEntriesCache;
+  private final PieceStorageService pieceStorageService;
+  private final PurchaseOrderLineService purchaseOrderLineService;
+  private final PurchaseOrderStorageService purchaseOrderStorageService;
+  private final OrganizationService organizationService;
+  private final RestClient restClient;
+
+  public ClaimingService(ConfigurationEntriesCache configurationEntriesCache, PieceStorageService pieceStorageService,
+                         PurchaseOrderLineService purchaseOrderLineService, PurchaseOrderStorageService purchaseOrderStorageService,
+                         OrganizationService organizationService, RestClient restClient) {
+    this.configurationEntriesCache = configurationEntriesCache;
+    this.pieceStorageService = pieceStorageService;
+    this.purchaseOrderLineService = purchaseOrderLineService;
+    this.purchaseOrderStorageService = purchaseOrderStorageService;
+    this.organizationService = organizationService;
+    this.restClient = restClient;
+  }
+
+  /**
+   * Sends claims by receiving pieces to be claimed, groups them by vendor,
+   * updates piece statuses and finally creates jobs per vendor and associated integration details
+   * @param claimingCollection An array of pieces ids
+   * @param requestContext Headers to make HTTP or Kafka requests
+   * @return Future of an array of claimingResults
+   */
+  public Future<ClaimingResults> sendClaims(ClaimingCollection claimingCollection, RequestContext requestContext) {
+    var claimingHolder = new ClaimingHolder();
+    return configurationEntriesCache.loadConfiguration(DATA_EXPORT_SPRING_CONFIG_MODULE_NAME, requestContext)
+      .compose(config -> {
+        var pieceIds = claimingCollection.getClaiming().stream().toList();
+        logger.info("sendClaims:: Received pieces to claim, pieceIds: {}", pieceIds);
+        return groupPieceIdsByVendorId(claimingHolder, pieceIds, requestContext)
+          .compose(pieceIdsByVendorIds -> createJobsByVendor(claimingHolder, config, pieceIdsByVendorIds, requestContext));
+      })
+      .onFailure(t -> logger.error("sendClaims :: Failed send claims: {}",
+        JsonObject.mapFrom(claimingCollection).encodePrettily(), t));
+  }
+
+  private Future<Map<String, List<String>>> groupPieceIdsByVendorId(ClaimingHolder claimingHolder, List<String> pieceIds, RequestContext requestContext) {
+    if (CollectionUtils.isEmpty(pieceIds)) {
+      logger.info("groupPieceIdsByVendorId:: No pieces are grouped by vendor, pieceIds is empty");
+      return Future.succeededFuture();
+    }
+    return pieceStorageService.getPiecesByIds(pieceIds, requestContext)
+      .compose(pieces -> {
+        claimingHolder.withPieces(pieces);
+        var uniquePiecePoLinePairs = pieces.stream()
+          .map(piece -> Pair.of(piece.getPoLineId(), piece.getId())).distinct()
+          .toList();
+        var pieceIdByVendorIdFutures = new ArrayList<Future<Pair<String, String>>>();
+        uniquePiecePoLinePairs.forEach(piecePoLinePairs -> {
+           var pieceIdByVendorIdFuture = pieceStorageService.getPieceById(piecePoLinePairs.getRight(), requestContext)
+            .compose(piece -> createVendorPiecePair(piecePoLinePairs, piece, requestContext));
+            if (Objects.nonNull(pieceIdByVendorIdFuture)) {
+              pieceIdByVendorIdFutures.add(pieceIdByVendorIdFuture);
+            }
+        });
+        return collectResultsOnSuccess(pieceIdByVendorIdFutures)
+          .map(ClaimingService::transformAndGroupPieceIdsByVendorId);
+      });
+  }
+
+  private Future<Pair<String, String>> createVendorPiecePair(Pair<String, String> piecePoLinePairs, Piece piece, RequestContext requestContext) {
+    if (Objects.nonNull(piece) && !piece.getReceivingStatus().equals(Piece.ReceivingStatus.EXPECTED)) {
+      logger.info("createVendorPiecePair:: Ignoring processing of a piece not in expected state, piece id: {}", piece.getId());
+      return Future.succeededFuture();
+    }
+    return purchaseOrderLineService.getOrderLineById(piecePoLinePairs.getLeft(), requestContext)
+      .compose(poLine -> purchaseOrderStorageService.getPurchaseOrderById(poLine.getPurchaseOrderId(), requestContext)
+        .compose(purchaseOrder -> organizationService.getVendorById(purchaseOrder.getVendor(), requestContext)))
+      .map(vendor -> {
+        if (Objects.nonNull(vendor) && Boolean.TRUE.equals(vendor.getIsVendor())) {
+          return Pair.of(vendor.getId(), piecePoLinePairs.getRight());
+        }
+        return null;
+      });
+  }
+
+  private static Map<String, List<String>> transformAndGroupPieceIdsByVendorId(List<Pair<String, String>> piecesByVendorList) {
+    return StreamEx.of(piecesByVendorList).distinct().filter(Objects::nonNull)
+      .groupingBy(Pair::getKey, mapping(Pair::getValue, collectingAndThen(toList(), lists -> StreamEx.of(lists).toList())));
+  }
+
+  private Future<ClaimingResults> createJobsByVendor(ClaimingHolder claimingHolder, JsonObject config, Map<String,
+                                                     List<String>> pieceIdsByVendorId, RequestContext requestContext) {
+    if (CollectionUtils.isEmpty(pieceIdsByVendorId)) {
+      logger.info("createJobsByVendor:: No jobs are created, pieceIdsByVendorId is empty");
+      return Future.succeededFuture();
+    }
+    var updatePiecesAndJobFutures = new ArrayList<Future<List<String>>>();
+    pieceIdsByVendorId.forEach((vendorId, pieceIds) -> {
+      logger.info("createJobsByVendor:: Preparing job integration detail for vendor, vendor id: {}, pieceIds: {}", vendorId, pieceIds);
+      config.stream()
+        .filter(entry -> entry.getKey().contains(vendorId))
+        .forEach(entry -> {
+          var updatePiecesAndJobFuture = updatePieces(claimingHolder, pieceIds, requestContext)
+            .compose(updatePieceIds -> createJob(entry.getKey(), entry.getValue(), requestContext).map(updatePieceIds));
+          updatePiecesAndJobFutures.add(updatePiecesAndJobFuture);
+        });
+    });
+    return collectResultsOnSuccess(updatePiecesAndJobFutures)
+      .map(updatedLists -> {
+        var processedPieces = updatedLists.stream().flatMap(Collection::stream).distinct()
+          .map(pieceId -> new ClaimingResult().withPieceId(pieceId).withType(ClaimingResult.Type.SUCCESS))
+          .toList();
+        logger.info("createJobsByVendor:: Processed pieces for claiming, count: {}", processedPieces.size());
+        return new ClaimingResults().withClaimingResults(processedPieces)
+          .withTotalRecords(processedPieces.size());
+      });
+  }
+
+  private Future<List<String>> updatePieces(ClaimingHolder claimingHolder, List<String> pieceIds, RequestContext requestContext) {
+    var piecesByVendorFutures = new ArrayList<Future<String>>();
+    pieceIds.forEach(pieceId -> {
+      var piece = claimingHolder.getPieces().stream()
+        .filter(pieceFromStorage -> pieceFromStorage.getId().equals(pieceId))
+        .findFirst().orElseThrow()
+        .withReceivingStatus(Piece.ReceivingStatus.CLAIM_SENT);
+      piecesByVendorFutures.add(pieceStorageService.updatePiece(piece, requestContext).map(pieceId));
+    });
+    return collectResultsOnSuccess(piecesByVendorFutures);
+  }
+
+  private Future<Void> createJob(String configKey, Object configValue, RequestContext requestContext) {
+    var integrationDetail = new JsonObject(String.valueOf(configValue));
+    return restClient.post(CREATE_JOB, integrationDetail, Object.class, requestContext)
+      .map(response -> {
+        var createdJob = new JsonObject(String.valueOf(response));
+        logger.info("createJob:: Created job, config key: {}, job status: {}", configKey, createdJob.getString(JOB_STATUS));
+        return restClient.postEmptyResponse(EXECUTE_JOB, createdJob, requestContext)
+          .onSuccess(v -> logger.info("createJob:: Executed job, config key: {}, job status: {}", configKey, createdJob.getString(JOB_STATUS)));
+      })
+      .mapEmpty();
+  }
+}

src/test/java/org/folio/service/claims/ClaimingServiceTest.java

@@ -0,0 +1,10 @@
+package org.folio.service.claims;
+
+import org.junit.jupiter.api.Test;
+
+class ClaimingServiceTest {
+
+  @Test
+  void sendClaims() {
+  }
+}