feat(mojaloop/#2883): add PartyInfoCallbackProcessed domain handler

docker-compose.yml

@@ -40,8 +40,8 @@ services:
 
   redis:
     networks:
-        - mojaloop-net
-    image: "redis:5.0.4-alpine"
+      - mojaloop-net
+    image: "redis:6.2.4-alpine"
     container_name: redis
     ports:
       - "6379:6379"
@@ -103,9 +103,9 @@ services:
     environment:
       ALLOW_PLAINTEXT_LISTENER: "yes"
       # KAFKA_ADVERTISED_HOST_NAME: kafka
-      KAFKA_LISTENERS: LISTENER_INTERN://kafka:9093,LISTENER_EXTERN://0.0.0.0:9092
-      KAFKA_ADVERTISED_LISTENERS: LISTENER_INTERN://kafka:9093,LISTENER_EXTERN://127.0.0.1:9092
-      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_INTERN:PLAINTEXT,LISTENER_EXTERN:PLAINTEXT
+      KAFKA_LISTENERS: LISTENER_DOCKER://kafka:29092,LISTENER_INTERN://kafka:9093,LISTENER_EXTERN://0.0.0.0:9092
+      KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER://kafka:29092,LISTENER_INTERN://kafka:9093,LISTENER_EXTERN://127.0.0.1:9092
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER:PLAINTEXT,LISTENER_INTERN:PLAINTEXT,LISTENER_EXTERN:PLAINTEXT
       KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_INTERN
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
       KAFKA_CFG_ZOOKEEPER_CONNECT: zookeeper:2181
@@ -117,3 +117,28 @@ services:
       retries: 10
       start_period: 40s
       interval: 30s
+
+  insights:
+    image: redislabs/redisinsight
+    ports:
+      - "9001:8001"
+    volumes: []
+    restart: on-failure
+
+  kowl:
+    image: quay.io/cloudhut/kowl:v1.4.0
+    container_name: cl_kowl
+    deploy:
+      replicas: 1
+    restart: on-failure
+    hostname: kowl
+    ports:
+    - "9080:8080"
+    networks:
+      - mojaloop-net
+    environment:
+      - KAFKA_BROKERS=kafka:29092
+    depends_on:
+      - kafka
+    profiles:
+      - debug

modules/outbound-command-event-handler/package.json

@@ -40,8 +40,8 @@
   },
   "dependencies": {
     "@mojaloop/api-snippets": "^14.2.4",
-    "@mojaloop/logging-bc-client-lib": "^0.1.12",
-    "@mojaloop/logging-bc-public-types-lib": "^0.1.9",
+    "@mojaloop/logging-bc-client-lib": "^0.1.13",
+    "@mojaloop/logging-bc-public-types-lib": "^0.1.10",
     "@mojaloop/sdk-scheme-adapter-private-shared-lib": "workspace:^",
     "ajv": "^8.11.0",
     "convict": "^6.2.3",

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

@@ -30,6 +30,7 @@ import {
     IndividualTransferInternalState,
     ProcessPartyInfoCallbackCmdEvt,
     PartyInfoCallbackProcessedDmEvt,
+    IPartyResult,
     SDKOutboundTransferState,
 } from '@mojaloop/sdk-scheme-adapter-private-shared-lib';
 import { BulkTransactionAgg } from '..';
@@ -54,11 +55,13 @@ export async function handleProcessPartyInfoCallbackCmdEvt(
         const individualTransfer = await bulkTransactionAgg.getIndividualTransferById(
             processPartyInfoCallback.getTransferId(),
         );
-        const partyResult = processPartyInfoCallback.getPartyResult();
+        const partyResult = processPartyInfoCallback.getPartyResult() as IPartyResult;
         if(partyResult.currentState && partyResult.currentState === SDKOutboundTransferState.COMPLETED) {
             individualTransfer.setTransferState(IndividualTransferInternalState.DISCOVERY_SUCCESS);
+            await bulkTransactionAgg.incrementPartyLookupSuccessCount();
         } else {
             individualTransfer.setTransferState(IndividualTransferInternalState.DISCOVERY_FAILED);
+            await bulkTransactionAgg.incrementPartyLookupFailedCount();
         }
         individualTransfer.setPartyResponse(partyResult);
 

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

@@ -56,16 +56,19 @@ export async function handleProcessSDKOutboundBulkPartyInfoRequestCmdEvt(
 
         bulkTx.setTxState(BulkTransactionInternalState.DISCOVERY_PROCESSING);
         await bulkTransactionAgg.setTransaction(bulkTx);
-
         const allIndividualTransferIds = await bulkTransactionAgg.getAllIndividualTransferIds();
+        await bulkTransactionAgg.setPartyLookupTotalCount(allIndividualTransferIds.length);
+
         for await (const individualTransferId of allIndividualTransferIds) {
             const individualTransfer = await bulkTransactionAgg.getIndividualTransferById(individualTransferId);
 
             if(bulkTx.isSkipPartyLookupEnabled()) {
                 if(individualTransfer.isPartyInfoExists) {
                     individualTransfer.setTransferState(IndividualTransferInternalState.DISCOVERY_SUCCESS);
+                    await bulkTransactionAgg.incrementPartyLookupSuccessCount();
                 } else {
                     individualTransfer.setTransferState(IndividualTransferInternalState.DISCOVERY_FAILED);
+                    await bulkTransactionAgg.incrementPartyLookupFailedCount();
                 }
                 await bulkTransactionAgg.setIndividualTransferById(individualTransferId, individualTransfer);
                 continue;

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

@@ -47,7 +47,6 @@ import { randomUUID } from 'crypto';
 
 
 export class BulkTransactionAgg extends BaseAggregate<BulkTransactionEntity, BulkTransactionState> {
-    // TODO: These counts can be part of bulk transaction entity?
     // private _partyLookupTotalCount?: number;
     // private _partyLookupSuccessCount?: number;
     // private _partyLookupFailedCount?: number;
@@ -100,6 +99,9 @@ export class BulkTransactionAgg extends BaseAggregate<BulkTransactionEntity, Bul
         await agg.setBulkQuotesTotalCount(0);
         await agg.setBulkQuotesSuccessCount(0);
         await agg.setBulkQuotesFailedCount(0);
+        await agg.setPartyLookupTotalCount(0);
+        await agg.setPartyLookupSuccessCount(0);
+        await agg.setPartyLookupFailedCount(0);
 
         // Return the aggregate
         return agg;
@@ -314,6 +316,46 @@ export class BulkTransactionAgg extends BaseAggregate<BulkTransactionEntity, Bul
 
     }
 
+    async setPartyLookupTotalCount(count: number): Promise<void> {
+        await (<IBulkTransactionEntityRepo> this._entity_state_repo)
+            .setPartyLookupTotalCount(this._rootEntity.id, count);
+    }
+
+    async setPartyLookupSuccessCount(count: number): Promise<void> {
+        await (<IBulkTransactionEntityRepo> this._entity_state_repo)
+            .setPartyLookupSuccessCount(this._rootEntity.id, count);
+    }
+
+    async setPartyLookupFailedCount(count: number): Promise<void> {
+        await (<IBulkTransactionEntityRepo> this._entity_state_repo)
+            .setPartyLookupFailedCount(this._rootEntity.id, count);
+    }
+
+    async getPartyLookupTotalCount(): Promise<any> {
+        await (<IBulkTransactionEntityRepo> this._entity_state_repo)
+            .getPartyLookupTotalCount(this._rootEntity.id);
+    }
+
+    async getPartyLookupSuccessCount(): Promise<any> {
+        await (<IBulkTransactionEntityRepo> this._entity_state_repo)
+            .getPartyLookupSuccessCount(this._rootEntity.id);
+    }
+
+    async getPartyLookupFailedCount(): Promise<any> {
+        await (<IBulkTransactionEntityRepo> this._entity_state_repo)
+            .getPartyLookupFailedCount(this._rootEntity.id);
+    }
+
+    async incrementPartyLookupSuccessCount(increment = 1): Promise<void> {
+        await (<IBulkTransactionEntityRepo> this._entity_state_repo)
+            .incrementPartyLookupSuccessCount(this._rootEntity.id, increment);
+    }
+
+    async incrementPartyLookupFailedCount(increment = 1): Promise<void> {
+        await (<IBulkTransactionEntityRepo> this._entity_state_repo)
+            .incrementPartyLookupFailedCount(this._rootEntity.id, increment);
+    }
+
     async destroy() : Promise<void> {
         if(this._rootEntity) {
             // Cleanup repo

modules/outbound-command-event-handler/test/integration/application/outbound_command_event_handler.test.ts

@@ -28,8 +28,11 @@ import { DefaultLogger } from "@mojaloop/logging-bc-client-lib";
 import { ILogger } from "@mojaloop/logging-bc-public-types-lib";
 import { SDKSchemeAdapter } from '@mojaloop/api-snippets';
 
-import { CommandEvent, ICommandEventData, DomainEvent,
-         KafkaCommandEventProducer, IKafkaEventProducerOptions, KafkaDomainEventConsumer, IKafkaEventConsumerOptions,
+import { DomainEvent,
+         KafkaCommandEventProducer,
+         IKafkaEventProducerOptions,
+         KafkaDomainEventConsumer,
+         IKafkaEventConsumerOptions,
          ProcessSDKOutboundBulkRequestCmdEvt,
          ProcessSDKOutboundBulkPartyInfoRequestCompleteCmdEvt,
          ProcessSDKOutboundBulkPartyInfoRequestCmdEvt,
@@ -44,10 +47,10 @@ import { CommandEvent, ICommandEventData, DomainEvent,
 import { randomUUID } from "crypto";
 
 // Tests can timeout in a CI pipeline so giving it leeway
-jest.setTimeout(30000)
+jest.setTimeout(10000)
 
 const logger: ILogger = new DefaultLogger('bc', 'appName', 'appVersion'); //TODO: parameterize the names here
-const messageTimeout = 5000;
+const messageTimeout = 2000;
 
 // Setup for Kafka Producer
 const commandEventProducerOptions: IKafkaEventProducerOptions = {
@@ -254,7 +257,7 @@ describe("Tests for Outbound Command Event Handler", () => {
     const bulkPartyInfoRequestCommandEventObj = new ProcessSDKOutboundBulkPartyInfoRequestCmdEvt(bulkPartyInfoRequestCommandEventData);
     await producer.sendCommandEvent(bulkPartyInfoRequestCommandEventObj);
 
-    await new Promise(resolve => setTimeout(resolve, 3000));
+    await new Promise(resolve => setTimeout(resolve, messageTimeout));
     // Check the state in Redis
     console.log('bulk id: ', bulkTransactionId);
     const bulkState = await bulkTransactionEntityRepo.load(bulkTransactionId);
@@ -266,9 +269,17 @@ describe("Tests for Outbound Command Event Handler", () => {
     expect((await bulkTransactionEntityRepo.getIndividualTransfer(bulkTransactionId, individualTransfers[0])).state).toBe('DISCOVERY_PROCESSING');
     expect((await bulkTransactionEntityRepo.getIndividualTransfer(bulkTransactionId, individualTransfers[1])).state).toBe('DISCOVERY_PROCESSING');
 
+    // Check number of transfers to be looked up have been saved in Redis
+    expect(await bulkTransactionEntityRepo.getPartyLookupTotalCount(bulkTransactionId)).toEqual(individualTransfers.length)
+
+    // Check counts have been initialized
+    expect(await bulkTransactionEntityRepo.getPartyLookupSuccessCount(bulkTransactionId)).toEqual(0)
+    expect(await bulkTransactionEntityRepo.getPartyLookupFailedCount(bulkTransactionId)).toEqual(0)
+
     // Check domain events published to kafka
     const filteredEvents = domainEvents.filter(domainEvent => domainEvent.getName() === 'PartyInfoRequestedDmEvt');
     expect(filteredEvents.length).toBe(2);
+
     // Check the data contents for domain event
     expect(filteredEvents[0].getName()).toBe('PartyInfoRequestedDmEvt');
     expect(JSON.parse(JSON.stringify(filteredEvents[0].getContent()))).toBeDefined();
@@ -351,6 +362,13 @@ describe("Tests for Outbound Command Event Handler", () => {
     expect(individualTransfers.length).toBe(1);
     expect((await bulkTransactionEntityRepo.getIndividualTransfer(bulkTransactionId, individualTransfers[0])).state).toBe('DISCOVERY_SUCCESS');
 
+    // Check number of transfers to be looked up have been saved in Redis
+    expect(await bulkTransactionEntityRepo.getPartyLookupTotalCount(bulkTransactionId)).toEqual(individualTransfers.length)
+
+    // Check counts have been initialized
+    expect(await bulkTransactionEntityRepo.getPartyLookupSuccessCount(bulkTransactionId)).toEqual(0)
+    expect(await bulkTransactionEntityRepo.getPartyLookupFailedCount(bulkTransactionId)).toEqual(0)
+
     const filteredEvents = domainEvents.filter(domainEvent => domainEvent.getName() === 'PartyInfoRequestedDmEvt');
     // Check domain events published to kafka
     expect(filteredEvents.length).toBe(0)
@@ -448,6 +466,7 @@ describe("Tests for Outbound Command Event Handler", () => {
       headers: []
     }
     const processPartyInfoCallbackMessageObj = new ProcessPartyInfoCallbackCmdEvt(processPartyInfoCallbackMessageData);
+    const previousPartyLookupSuccessCount = await bulkTransactionEntityRepo.getPartyLookupSuccessCount(bulkTransactionId)
     await producer.sendCommandEvent(processPartyInfoCallbackMessageObj);
     await new Promise(resolve => setTimeout(resolve, messageTimeout));
 
@@ -458,6 +477,16 @@ describe("Tests for Outbound Command Event Handler", () => {
     expect(individualTransferData.state).toBe('DISCOVERY_SUCCESS');
     expect(individualTransferData.partyResponse?.party.partyIdInfo.fspId).toBe('receiverfsp');
 
+    // Check number of transfers to be looked up have been saved in Redis
+    expect(await bulkTransactionEntityRepo.getPartyLookupTotalCount(bulkTransactionId)).toEqual(individualTransfers.length)
+
+    // Check counts have been updated
+    expect(await bulkTransactionEntityRepo.getPartyLookupSuccessCount(bulkTransactionId)).toEqual(1)
+    expect(await bulkTransactionEntityRepo.getPartyLookupFailedCount(bulkTransactionId)).toEqual(0)
+
+    // Check that the party lookup success count has been incremented
+    const followingPartyLookupSuccessCount = await bulkTransactionEntityRepo.getPartyLookupSuccessCount(bulkTransactionId);
+    expect(followingPartyLookupSuccessCount).toBe(previousPartyLookupSuccessCount! + 1);
 
     // // Check domain events published to kafka
     expect(domainEvents[2].getName()).toBe('PartyInfoCallbackProcessedDmEvt');
@@ -556,6 +585,8 @@ describe("Tests for Outbound Command Event Handler", () => {
       headers: []
     }
     const processPartyInfoCallbackMessageObj = new ProcessPartyInfoCallbackCmdEvt(processPartyInfoCallbackMessageData);
+    const previousPartyLookupFailedCount = await bulkTransactionEntityRepo.getPartyLookupFailedCount(bulkTransactionId)
+
     await producer.sendCommandEvent(processPartyInfoCallbackMessageObj);
     await new Promise(resolve => setTimeout(resolve, messageTimeout));
 
@@ -567,6 +598,17 @@ describe("Tests for Outbound Command Event Handler", () => {
     expect(individualTransferData.partyResponse?.errorInformation?.errorCode).toBe('12345');
     expect(individualTransferData.partyResponse?.errorInformation?.errorDescription).toBe('ID Not Found');
 
+    // Check number of transfers to be looked up have been saved in Redis
+    expect(await bulkTransactionEntityRepo.getPartyLookupTotalCount(bulkTransactionId)).toEqual(individualTransfers.length)
+
+    // Check counts have been updated
+    expect(await bulkTransactionEntityRepo.getPartyLookupSuccessCount(bulkTransactionId)).toEqual(0)
+    expect(await bulkTransactionEntityRepo.getPartyLookupFailedCount(bulkTransactionId)).toEqual(1)
+
+    // Check that the party lookup failed count has been incremented
+    const followingPartyLookupFailedCount = await bulkTransactionEntityRepo.getPartyLookupFailedCount(bulkTransactionId)
+    expect(followingPartyLookupFailedCount).toBe(previousPartyLookupFailedCount + 1);
+
     // // Check domain events published to kafka
     expect(domainEvents[2].getName()).toBe('PartyInfoCallbackProcessedDmEvt')
   });

modules/outbound-domain-event-handler/package.json

@@ -29,7 +29,7 @@
         "lint": "eslint --ext .js,.ts src --color",
         "lint:fix": "eslint --ext .js,.ts src --color --fix",
         "test:unit": "jest --passWithNoTests --testMatch '**/test/unit/**/*.test.ts'",
-        "test:integration": "jest --passWithNoTests --testMatch '**/test/integration/**/*.test.ts'",
+        "test:integration": "jest --runInBand --passWithNoTests --testMatch '**/test/integration/**/*.test.ts'",
         "test:coverage": "jest --passWithNoTests --coverage --testMatch '**/test/unit/**/*.test.ts'",
         "test:coverage-check": "jest --coverage --testMatch '**/test/unit/**/*.test.ts'",
         "dep:check": "ncu -e 2",
@@ -38,8 +38,8 @@
         "snapshot": "standard-version --no-verify --skip.changelog --prerelease snapshot --releaseCommitMessageFormat 'chore(snapshot): {{currentTag}}'"
     },
     "dependencies": {
-        "@mojaloop/logging-bc-client-lib": "^0.1.12",
-        "@mojaloop/logging-bc-public-types-lib": "^0.1.9",
+        "@mojaloop/logging-bc-client-lib": "^0.1.13",
+        "@mojaloop/logging-bc-public-types-lib": "^0.1.10",
         "@mojaloop/sdk-scheme-adapter-private-shared-lib": "workspace:^",
         "convict": "^6.2.3"
     },

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

@@ -37,6 +37,8 @@ import {
     SDKOutboundBulkRequestReceivedDmEvt,
     SDKOutboundBulkPartyInfoRequestedDmEvt,
     PartyInfoCallbackReceivedDmEvt,
+    IBulkTransactionEntityReadOnlyRepo,
+    PartyInfoCallbackProcessedDmEvt,
     SDKOutboundBulkAcceptPartyInfoReceivedDmEvt,
     SDKOutboundBulkAcceptPartyInfoProcessedDmEvt,
     BulkQuotesCallbackReceivedDmEvt,
@@ -49,17 +51,29 @@ import {
     handleSDKOutboundBulkAcceptPartyInfoReceived,
     handleSDKOutboundBulkAcceptPartyInfoProcessed,
     handleBulkQuotesCallbackReceived,
+    handlePartyInfoCallbackProcessed,
 } from './handlers';
 
+export interface IOutboundEventHandlerOptions {
+    bulkTransactionEntityRepo: IBulkTransactionEntityReadOnlyRepo;
+}
+
 export class OutboundEventHandler implements IRunHandler {
     private _logger: ILogger;
 
     private _consumer: IEventConsumer;
 
     private _commandProducer: ICommandEventProducer;
 
+    private _bulkTransactionEntityStateRepo: IBulkTransactionEntityReadOnlyRepo;
+
     private _domainEventHandlerOptions: IDomainEventHandlerOptions;
 
+    /* eslint-disable-next-line @typescript-eslint/no-useless-constructor */
+    constructor(options: IOutboundEventHandlerOptions) {
+        this._bulkTransactionEntityStateRepo = options.bulkTransactionEntityRepo;
+    }
+
     /* eslint-disable-next-line @typescript-eslint/no-explicit-any */
     async start(appConfig: any, logger: ILogger): Promise<void> {
         this._logger = logger;
@@ -81,6 +95,7 @@ export class OutboundEventHandler implements IRunHandler {
         // Create options for handlers
         this._domainEventHandlerOptions = {
             commandProducer: this._commandProducer,
+            bulkTransactionEntityRepo: this._bulkTransactionEntityStateRepo,
         };
     }
 
@@ -119,6 +134,10 @@ export class OutboundEventHandler implements IRunHandler {
                 await handleBulkQuotesCallbackReceived(message, this._domainEventHandlerOptions, this._logger);
                 break;
             }
+            case PartyInfoCallbackProcessedDmEvt.name: {
+                await handlePartyInfoCallbackProcessed(message, this._domainEventHandlerOptions, this._logger);
+                break;
+            }
             default: {
                 this._logger.debug(`${message?.getName()}:${message?.getKey()} - Skipping unknown domain event`);
                 return;

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

@@ -28,3 +28,4 @@ export * from './party-info-callback-received';
 export * from './sdk_outbound_bulk_accept_party_info_received';
 export * from './sdk_outbound_bulk_accept_party_info_processed';
 export * from './bulk-quotes-callback-received';
+export * from './party-info-callback-processed';