feat(bulk/#2802): merge integration tests for bulk parties and quotes

.circleci/config.yml

@@ -518,7 +518,7 @@ jobs:
           at: /tmp
       - run:
           name: Load the pre-built docker image from workspace
-          command: | 
+          command: |
             docker load -i /tmp/docker-image.tar
       - run:
           name: Login to Docker Hub
@@ -576,7 +576,7 @@ jobs:
           at: /tmp
       - run:
           name: Load the pre-built docker image from workspace
-          command: | 
+          command: |
             docker load -i /tmp/docker-image.tar
       - run:
           name: Login to Docker Hub

docker-compose.yml

@@ -150,3 +150,23 @@ services:
       - kafka
     profiles:
       - debug
+
+  init-kafka:
+    networks:
+      - mojaloop-net
+    image: docker.io/bitnami/kafka:3.2
+    depends_on:
+      - kafka
+    entrypoint: [ '/bin/sh', '-c' ]
+    command: |
+      "
+      # blocks until kafka is reachable
+      kafka-topics.sh --bootstrap-server kafka:9093 --list
+
+      echo -e 'Creating kafka topics'
+      kafka-topics.sh --bootstrap-server kafka:9093 --create --if-not-exists --topic topic-sdk-outbound-command-events --replication-factor 1 --partitions 1
+      kafka-topics.sh --bootstrap-server kafka:9093 --create --if-not-exists --topic topic-sdk-outbound-domain-events --replication-factor 1 --partitions 1
+
+      echo -e 'Successfully created the following topics:'
+      kafka-topics.sh --bootstrap-server kafka:9093 --list
+      "

docs/design-bulk-transfers/outbound-sequence.md

@@ -47,13 +47,14 @@ sequenceDiagram
         SDKOutboundCommandEventHandler->>SDKOutboundCommandEventHandler: Update the party response
         SDKOutboundCommandEventHandler->>SDKOutboundDomainEventHandler: PartyInfoCallbackProcessed
         Note right of SDKOutboundDomainEventHandler: topic-sdk-outbound-domain-events
-        SDKOutboundDomainEventHandler->>SDKOutboundDomainEventHandler: Check the status of the remaining items in the bulk
+        SDKOutboundCommandEventHandler->>SDKOutboundCommandEventHandler: Check the status of the remaining items in the bulk
     end
-    SDKOutboundDomainEventHandler->>SDKOutboundCommandEventHandler: ProcessSDKOutboundBulkPartyInfoRequestComplete
-    Note left of SDKOutboundCommandEventHandler: topic-sdk-outbound-command-events
-
     SDKOutboundCommandEventHandler->>SDKOutboundCommandEventHandler: Update global state "DISCOVERY_COMPLETED"
-    SDKOutboundCommandEventHandler->>SDKOutboundCommandEventHandler: check optiions.autoAcceptParty in redis
+
+    SDKOutboundCommandEventHandler->>SDKOutboundDomainEventHandler: SDKOutboundBulkPartyInfoRequestProcessed
+    Note right of SDKOutboundDomainEventHandler: topic-sdk-outbound-domain-events
+
+    SDKOutboundCommandEventHandler->>SDKOutboundCommandEventHandler: check options.autoAcceptParty in redis
 
     alt autoAcceptParty == false
         SDKOutboundCommandEventHandler->>SDKOutboundAPI: SDKOutboundBulkAcceptPartyInfoRequested

modules/api-svc/src/FSPIOPEventHandler/handlers/party-info-requested.js

@@ -72,7 +72,7 @@ module.exports.handlePartyInfoRequestedDmEvt = async (
         await options.producer.sendDomainEvent(partyInfoCallbackReceivedDmEvt);
     } catch (err) {
         logger.push({ err }).log('Error in handlePartyInfoRequestedDmEvt');
-        const { code, message } = Errors.MojaloopApiErrorCodes.SERVER_TIMED_OUT
+        const { code, message } = Errors.MojaloopApiErrorCodes.SERVER_TIMED_OUT;
         const partyInfoCallbackReceivedDmEvt = new PartyInfoCallbackReceivedDmEvt({
             bulkId: event.getKey(),
             content: {

modules/api-svc/src/config.js

@@ -93,7 +93,7 @@ module.exports = {
         enabled: env.get('ENABLE_BACKEND_EVENT_HANDLER').default('true').asBool(),
         domainEventConsumer: {
             brokerList: env.get('BACKEND_EVENT_CONSUMER_BROKER_LIST').default('localhost:9092').asString(),
-            groupId: env.get('BACKEND_EVENT_CONSUMER_GROUP_ID').default('domain_events_consumer_group').asString(),
+            groupId: env.get('BACKEND_EVENT_CONSUMER_GROUP_ID').default('domain_events_consumer_api_svc_backend_group').asString(),
             clientId: env.get('BACKEND_EVENT_CONSUMER_CLIENT_ID').default('backend_consumer_client_id').asString(),
             topics: env.get('BACKEND_EVENT_CONSUMER_TOPICS').default('topic-sdk-outbound-domain-events').asArray(),
         },
@@ -107,7 +107,7 @@ module.exports = {
         enabled: env.get('ENABLE_FSPIOP_EVENT_HANDLER').default('true').asBool(),
         domainEventConsumer: {
             brokerList: env.get('FSPIOP_EVENT_CONSUMER_BROKER_LIST').default('localhost:9092').asString(),
-            groupId: env.get('FSPIOP_EVENT_CONSUMER_GROUP_ID').default('domain_events_consumer_group').asString(),
+            groupId: env.get('FSPIOP_EVENT_CONSUMER_GROUP_ID').default('domain_events_consumer_api_svc_fspiop_group').asString(),
             clientId: env.get('FSPIOP_EVENT_CONSUMER_CLIENT_ID').default('fspiop_consumer_client_id').asString(),
             topics: env.get('FSPIOP_EVENT_CONSUMER_TOPICS').default('topic-sdk-outbound-domain-events').asArray(),
         },

modules/outbound-command-event-handler/src/application/handler.ts

@@ -38,7 +38,6 @@ import {
     ProcessSDKOutboundBulkRequestCmdEvt,
     ProcessSDKOutboundBulkPartyInfoRequestCmdEvt,
     ProcessPartyInfoCallbackCmdEvt,
-    ProcessSDKOutboundBulkPartyInfoRequestCompleteCmdEvt,
     ProcessSDKOutboundBulkAcceptPartyInfoCmdEvt,
     ProcessSDKOutboundBulkQuotesRequestCmdEvt,
     ProcessBulkQuotesCallbackCmdEvt,
@@ -126,14 +125,6 @@ export class OutboundEventHandler implements IRunHandler {
                 );
                 break;
             }
-            case ProcessSDKOutboundBulkPartyInfoRequestCompleteCmdEvt.name: {
-                BulkTransactionAgg.ProcessCommandEvent(
-                    ProcessSDKOutboundBulkPartyInfoRequestCompleteCmdEvt.CreateFromCommandEvent(message),
-                    this._commandEventHandlerOptions,
-                    this._logger,
-                );
-                break;
-            }
             case ProcessSDKOutboundBulkAcceptPartyInfoCmdEvt.name: {
                 BulkTransactionAgg.ProcessCommandEvent(
                     ProcessSDKOutboundBulkAcceptPartyInfoCmdEvt.CreateFromCommandEvent(message),

modules/outbound-command-event-handler/src/domain/bulk_transaction_agg/handlers/index.ts

@@ -24,10 +24,9 @@
 
 import * as ProcessSDKOutboundBulkRequestHandler from './process_sdk_outbound_bulk_request';
 import * as ProcessSDKOutboundBulkPartyInfoRequestHandler from './process_sdk_outbound_bulk_party_info_request';
-import * as ProcessSDKOutboundBulkPartyInfoRequestCompleteHandler from './process_sdk_outbound_bulk_party_info_request_complete';
 import * as ProcessPartyInfoCallbackHandler from './process_party_info_callback';
 import * as ProcessSDKOutboundBulkAcceptPartyInfoHandler from './process_sdk_outbound_bulk_accept_party_info';
-import * as ProcessBulkQuotesCallbackHandler from './process-bulk-quotes-callback';
+import * as ProcessBulkQuotesCallbackHandler from './process_bulk_quotes_callback';
 import * as ProcessSDKOutboundBulkQuotesRequestHandler from './process_sdk_outbound_bulk_quotes_request';
 
 import { CommandEvent } from '@mojaloop/sdk-scheme-adapter-private-shared-lib';
@@ -37,7 +36,6 @@ import { ILogger } from '@mojaloop/logging-bc-public-types-lib';
 export default  {
     ...ProcessSDKOutboundBulkRequestHandler,
     ...ProcessSDKOutboundBulkPartyInfoRequestHandler,
-    ...ProcessSDKOutboundBulkPartyInfoRequestCompleteHandler,
     ...ProcessPartyInfoCallbackHandler,
     ...ProcessSDKOutboundBulkAcceptPartyInfoHandler,
     ...ProcessBulkQuotesCallbackHandler,

modules/outbound-command-event-handler/src/domain/bulk_transaction_agg/handlers/process-bulk-quotes-callback.ts → modules/outbound-command-event-handler/src/domain/bulk_transaction_agg/handlers/process_bulk_quotes_callback.ts

@@ -38,6 +38,8 @@ export async function handleProcessBulkQuotesCallbackCmdEvt(
     const processBulkQuotesCallbackMessage = message as ProcessBulkQuotesCallbackCmdEvt;
     try {
         logger.info(`Got ProcessBulkQuotesCallbackCmdEvt: id=${processBulkQuotesCallbackMessage.getKey()}`);
+        let successCountAfterIncrement;
+        let failedCountAfterIncrement;
 
         // Create aggregate
         const bulkTransactionAgg = await BulkTransactionAgg.CreateFromRepo(
@@ -50,30 +52,45 @@ export async function handleProcessBulkQuotesCallbackCmdEvt(
             processBulkQuotesCallbackMessage.batchId,
         );
         const bulkQuotesResult = processBulkQuotesCallbackMessage.bulkQuotesResult;
-        // TODO: There is no currentState in the bulkQuotesResult specification, but its there in the response. Need to discuss on this.
-        // if(bulkQuotesResult.currentState && bulkQuotesResult.currentState === 'COMPLETED') {
-        if(bulkQuotesResult.individualQuoteResults.length > 0) {
-            bulkBatch.setState(BulkBatchInternalState.AGREEMENT_SUCCESS);
-            bulkTransactionAgg.incrementBulkQuotesSuccessCount();
+
+        // If individual quote result contains `lastError` the individual transfer state should be AGREEMENT_FAILED.
+        // bulkQuotesResult.currentState === 'ERROR_OCCURRED' necessitates erroring out all individual transfers in that bulk batch.
+        if(bulkQuotesResult.currentState &&
+           bulkQuotesResult.currentState === 'COMPLETED') {
+            bulkBatch.setState(BulkBatchInternalState.AGREEMENT_COMPLETED);
+            successCountAfterIncrement = await bulkTransactionAgg.incrementBulkQuotesSuccessCount();
+
+            // Iterate through items in batch and update the individual states
+            for await (const quoteResult of bulkQuotesResult.individualQuoteResults) {
+                if(quoteResult.quoteId && !quoteResult.lastError) {
+                    const individualTransferId = bulkBatch.getReferenceIdForQuoteId(quoteResult.quoteId);
+                    const individualTransfer = await bulkTransactionAgg.getIndividualTransferById(individualTransferId);
+                    individualTransfer.setTransferState(IndividualTransferInternalState.AGREEMENT_SUCCESS);
+                    individualTransfer.setQuoteResponse(quoteResult);
+                    await bulkTransactionAgg.setIndividualTransferById(individualTransfer.id, individualTransfer);
+                } else {
+                    const individualTransferId = bulkBatch.getReferenceIdForQuoteId(quoteResult.quoteId);
+                    const individualTransfer = await bulkTransactionAgg.getIndividualTransferById(individualTransferId);
+                    individualTransfer.setTransferState(IndividualTransferInternalState.AGREEMENT_FAILED);
+                    individualTransfer.setQuoteResponse(quoteResult);
+                    await bulkTransactionAgg.setIndividualTransferById(individualTransfer.id, individualTransfer);
+                }
+            }
+        // If the bulk quote is in any other state, update the bulk batch and all individual transfers
+        // to AGREEMENT_FAILED.
         } else {
             bulkBatch.setState(BulkBatchInternalState.AGREEMENT_FAILED);
-            bulkTransactionAgg.incrementBulkQuotesFailedCount();
-        }
-        bulkBatch.setBulkQuotesResponse(bulkQuotesResult);
-        await bulkTransactionAgg.setBulkBatchById(bulkBatch.id, bulkBatch);
-
+            failedCountAfterIncrement = await bulkTransactionAgg.incrementBulkQuotesFailedCount();
 
-        // Iterate through items in batch and update the individual states
-        for await (const quoteResult of bulkQuotesResult.individualQuoteResults) {
-            // TODO: quoteId should be required field in the quoteResponse. But it is optional now, so if it is empty we are ignoring the result for now
-            if(quoteResult.quoteId) {
-                const individualTransferId = bulkBatch.getReferenceIdForQuoteId(quoteResult.quoteId);
+            const individualTransferIds = Object.values(bulkBatch.quoteIdReferenceIdMap);
+            for await (const individualTransferId of individualTransferIds) {
                 const individualTransfer = await bulkTransactionAgg.getIndividualTransferById(individualTransferId);
-                individualTransfer.setTransferState(IndividualTransferInternalState.AGREEMENT_SUCCESS);
-                individualTransfer.setQuoteResponse(quoteResult);
+                individualTransfer.setTransferState(IndividualTransferInternalState.AGREEMENT_FAILED);
                 await bulkTransactionAgg.setIndividualTransferById(individualTransfer.id, individualTransfer);
             }
         }
+        bulkBatch.setBulkQuotesResponse(bulkQuotesResult);
+        await bulkTransactionAgg.setBulkBatchById(bulkBatch.id, bulkBatch);
 
         const bulkQuotesCallbackProcessedDmEvt = new BulkQuotesCallbackProcessedDmEvt({
             bulkId: bulkTransactionAgg.bulkId,
@@ -88,12 +105,12 @@ export async function handleProcessBulkQuotesCallbackCmdEvt(
         // Progressing to the next step
         // Check the status of the remaining items in the bulk
         const bulkQuotesTotalCount = await bulkTransactionAgg.getBulkQuotesTotalCount();
-        const bulkQuotesSuccessCount = await bulkTransactionAgg.getBulkQuotesSuccessCount();
-        const bulkQuotesFailedCount = await bulkTransactionAgg.getBulkQuotesFailedCount();
+        const bulkQuotesSuccessCount = successCountAfterIncrement || await bulkTransactionAgg.getBulkQuotesSuccessCount();
+        const bulkQuotesFailedCount = failedCountAfterIncrement || await bulkTransactionAgg.getBulkQuotesFailedCount();
         if(bulkQuotesTotalCount === (bulkQuotesSuccessCount + bulkQuotesFailedCount)) {
             // Update global state "AGREEMENT_COMPLETED"
             await bulkTransactionAgg.setGlobalState(BulkTransactionInternalState.AGREEMENT_COMPLETED);
-    
+
             // Send the domain message SDKOutboundBulkQuotesRequestProcessed
             const sdkOutboundBulkQuotesRequestProcessedDmEvt = new SDKOutboundBulkQuotesRequestProcessedDmEvt({
                 bulkId: bulkTransactionAgg.bulkId,

modules/outbound-command-event-handler/src/domain/bulk_transaction_agg/handlers/process_party_info_callback.ts

@@ -30,8 +30,11 @@ import {
     IndividualTransferInternalState,
     ProcessPartyInfoCallbackCmdEvt,
     PartyInfoCallbackProcessedDmEvt,
-    IPartyResult,
     SDKOutboundTransferState,
+    BulkTransactionInternalState,
+    SDKOutboundBulkPartyInfoRequestProcessedDmEvt,
+    SDKOutboundBulkAutoAcceptPartyInfoRequestedDmEvt,
+    SDKOutboundBulkAcceptPartyInfoRequestedDmEvt,
 } from '@mojaloop/sdk-scheme-adapter-private-shared-lib';
 import { BulkTransactionAgg } from '..';
 import { ICommandEventHandlerOptions } from '@module-types';
@@ -43,6 +46,8 @@ export async function handleProcessPartyInfoCallbackCmdEvt(
 ): Promise<void> {
     const processPartyInfoCallback = message as ProcessPartyInfoCallbackCmdEvt;
     try {
+        let successCountAfterIncrement;
+        let failedCountAfterIncrement;
         logger.info(`Got ProcessPartyInfoCallbackCmdEvt: id=${processPartyInfoCallback.getKey()}`);
 
         // Create aggregate
@@ -58,13 +63,12 @@ export async function handleProcessPartyInfoCallbackCmdEvt(
         const partyResult = processPartyInfoCallback.getPartyResult();
         if(partyResult.currentState && partyResult.currentState === SDKOutboundTransferState.COMPLETED && !partyResult.errorInformation) {
             individualTransfer.setTransferState(IndividualTransferInternalState.DISCOVERY_SUCCESS);
-            await bulkTransactionAgg.incrementPartyLookupSuccessCount();
+            successCountAfterIncrement = await bulkTransactionAgg.incrementPartyLookupSuccessCount();
         } else {
             individualTransfer.setTransferState(IndividualTransferInternalState.DISCOVERY_FAILED);
-            await bulkTransactionAgg.incrementPartyLookupFailedCount();
+            failedCountAfterIncrement = await bulkTransactionAgg.incrementPartyLookupFailedCount();
         }
         individualTransfer.setPartyResponse(partyResult);
-
         await bulkTransactionAgg.setIndividualTransferById(individualTransfer.id, individualTransfer);
 
         const msg = new PartyInfoCallbackProcessedDmEvt({
@@ -76,6 +80,69 @@ export async function handleProcessPartyInfoCallbackCmdEvt(
             headers: [],
         });
         await options.domainProducer.sendDomainEvent(msg);
+
+        // Progressing to the next step
+        // Check the status of the remaining party lookups
+        const partyLookupTotalCount = await bulkTransactionAgg.getPartyLookupTotalCount();
+        const partyLookupSuccessCount = successCountAfterIncrement || await bulkTransactionAgg.getPartyLookupSuccessCount();
+        const partyLookupFailedCount = failedCountAfterIncrement || await bulkTransactionAgg.getPartyLookupFailedCount();
+        if(partyLookupTotalCount === (partyLookupSuccessCount + partyLookupFailedCount)) {
+            // Update global state "DISCOVERY_COMPLETED"
+            await bulkTransactionAgg.setGlobalState(BulkTransactionInternalState.DISCOVERY_COMPLETED);
+
+            // Send the domain message SDKOutboundBulkPartyInfoRequestProcessedDmEvt
+            const sdkOutboundBulkPartyInfoRequestProcessedDmEvt = new SDKOutboundBulkPartyInfoRequestProcessedDmEvt({
+                bulkId: bulkTransactionAgg.bulkId,
+                timestamp: Date.now(),
+                headers: [],
+            });
+            await options.domainProducer.sendDomainEvent(sdkOutboundBulkPartyInfoRequestProcessedDmEvt);
+            logger.info(`Sent domain event message ${SDKOutboundBulkPartyInfoRequestProcessedDmEvt.name}`);
+
+            // Progressing to the next step
+            // Check configuration parameter isAutoAcceptPartyEnabled
+            const bulkTx = bulkTransactionAgg.getBulkTransaction();
+            if(bulkTx.isAutoAcceptPartyEnabled()) {
+                const autoAcceptPartyMsg = new SDKOutboundBulkAutoAcceptPartyInfoRequestedDmEvt({
+                    bulkId: bulkTx.id,
+                    timestamp: Date.now(),
+                    headers: [],
+                });
+                await options.domainProducer.sendDomainEvent(autoAcceptPartyMsg);
+            } else {
+                const individualTransferResults = [];
+                const allIndividualTransferIds = await bulkTransactionAgg.getAllIndividualTransferIds();
+                for await (const individualTransferId of allIndividualTransferIds) {
+                    const individualTransferData = await bulkTransactionAgg
+                        .getIndividualTransferById(individualTransferId);
+                    if(individualTransferData.partyResponse) {
+                        individualTransferResults.push({
+                            homeTransactionId: individualTransferData.request.homeTransactionId,
+                            transactionId: individualTransferData.id,
+                            to: individualTransferData.partyResponse?.party,
+                            lastError: individualTransferData.partyResponse?.errorInformation && {
+                                mojaloopError: individualTransferData.partyResponse?.errorInformation,
+                            },
+                        });
+                    }
+                }
+                const sdkOutboundBulkAcceptPartyInfoRequestedDmEvt = new SDKOutboundBulkAcceptPartyInfoRequestedDmEvt({
+                    bulkId: bulkTransactionAgg.bulkId,
+                    request: {
+                        bulkHomeTransactionID: bulkTx.bulkHomeTransactionID,
+                        bulkTransactionId: bulkTransactionAgg.bulkId,
+                        individualTransferResults,
+                    },
+                    timestamp: Date.now(),
+                    headers: [],
+                });
+                await options.domainProducer.sendDomainEvent(sdkOutboundBulkAcceptPartyInfoRequestedDmEvt);
+                logger.info(`Sent domain event message ${SDKOutboundBulkAcceptPartyInfoRequestedDmEvt.name}`);
+
+                bulkTx.setTxState(BulkTransactionInternalState.DISCOVERY_ACCEPTANCE_PENDING);
+            }
+            await bulkTransactionAgg.setTransaction(bulkTx);
+        }
     } catch (err) {
         logger.error(`Failed to create BulkTransactionAggregate. ${(err as Error).message}`);
     }