feat(mojaloop/#2803): add event handlers (#380)

* feat(mojaloop/#2803): add BulkTransfersRequestedDmEvt implementation and handling

* add BulkTransfersCallbackReceived domain handler

* chore: key

* Update modules/outbound-domain-event-handler/src/application/handlers/bulk_transfers_callback_received.ts

Co-authored-by: Miguel de Barros <miguel@debarros.me>

Co-authored-by: Miguel de Barros <miguel@debarros.me>

modules/api-svc/package.json

@@ -104,7 +104,7 @@
     "jest": "^29.0.3",
     "jest-junit": "^14.0.1",
     "nock": "^13.2.9",
-    "npm-check-updates": "^16.1.3",
+    "npm-check-updates": "^16.2.1",
     "openapi-response-validator": "^12.0.2",
     "openapi-typescript": "^5.4.1",
     "redis-mock": "^0.56.3",

modules/api-svc/src/FSPIOPEventHandler/handlers/bulk-transfers-requested.js

@@ -0,0 +1,63 @@
+/*****
+ License
+ --------------
+ Copyright © 2017 Bill & Melinda Gates Foundation
+ The Mojaloop files are made available by the Bill & Melinda Gates Foundation under the Apache License, Version 2.0 (the "License") and you may not use these files except in compliance with the License. You may obtain a copy of the License at
+ http://www.apache.org/licenses/LICENSE-2.0
+ Unless required by applicable law or agreed to in writing, the Mojaloop files are distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.
+ Contributors
+ --------------
+ This is the official list (alphabetical ordering) of the Mojaloop project contributors for this file.
+ Names of the original copyright holders (individuals or organizations)
+ should be listed with a '*' in the first column. People who have
+ contributed from an organization can be listed under the organization
+ that actually holds the copyright for their contributions (see the
+ Gates Foundation organization for an example). Those individuals should have
+ their names indented and be marked with a '-'. Email address can be added
+ optionally within square brackets <email>.
+ * Gates Foundation
+ - Name Surname <name.surname@gatesfoundation.com>
+ * Modusbox
+ - Kevin Leyow <kevin.leyow@modusbox.com>
+ --------------
+ ******/
+
+const { BulkTransfersRequestedDmEvt } = require('@mojaloop/sdk-scheme-adapter-private-shared-lib');
+const { OutboundBulkTransfersModel } = require('../../lib/model');
+const { BulkTransfersCallbackReceivedDmEvt } = require('@mojaloop/sdk-scheme-adapter-private-shared-lib');
+
+module.exports.handleBulkTransfersRequestedDmEvt = async (
+    message,
+    options,
+    logger,
+) => {
+    const event = BulkTransfersRequestedDmEvt.CreateFromDomainEvent(message);
+
+    try {
+        // use the bulk transfers model to execute asynchronous request with the switch
+        const model = new OutboundBulkTransfersModel({
+            ...options.config,
+            cache: options.cache,
+            logger: logger,
+            wso2: options.wso2Auth,
+        });
+
+        await model.initialize(event.request);
+        const response = await model.run();
+
+        const bulkTransfersCallbackReceivedDmEvt = new BulkTransfersCallbackReceivedDmEvt({
+            bulkId: event.getKey(),
+            content: {
+                batchId: event.batchId,
+                bulkTransferId: response.bulkTransferId,
+                bulkTransfersResult: response,
+            },
+            timestamp: Date.now(),
+            headers: [],
+        });
+
+        await options.producer.sendDomainEvent(bulkTransfersCallbackReceivedDmEvt);
+    } catch (err) {
+        logger.push({ err }).log('Error in handleBulkTransfersRequestedDmEvt');
+    }
+};

modules/api-svc/src/FSPIOPEventHandler/handlers/index.js

@@ -25,4 +25,5 @@
 module.exports = {
     ...require('./bulk-quotes-requested'),
     ...require('./party-info-requested'),
+    ...require('./bulk-transfers-requested'),
 };

modules/api-svc/src/FSPIOPEventHandler/index.js

@@ -29,15 +29,15 @@ const {
     BC_CONFIG,
     KafkaDomainEventConsumer,
     KafkaDomainEventProducer,
+    BulkTransfersRequestedDmEvt,
+    PartyInfoRequestedDmEvt,
+    BulkQuotesRequestedDmEvt,
 } = require('@mojaloop/sdk-scheme-adapter-private-shared-lib');
 const {
     handlePartyInfoRequestedDmEvt,
     handleBulkQuotesRequestedDmEvt,
+    handleBulkTransfersRequestedDmEvt,
 } = require('./handlers');
-const {
-    PartyInfoRequestedDmEvt,
-    BulkQuotesRequestedDmEvt,
-} = require('@mojaloop/sdk-scheme-adapter-private-shared-lib');
 
 class FSPIOPEventHandler {
     constructor({ config, logger, cache, wso2Auth }) {
@@ -94,6 +94,10 @@ class FSPIOPEventHandler {
                 await handleBulkQuotesRequestedDmEvt(message, this._domainEventHandlerOptions, this._logger);
                 break;
             }
+            case BulkTransfersRequestedDmEvt.name: {
+                await handleBulkTransfersRequestedDmEvt(message, this._domainEventHandlerOptions, this._logger);
+                break;
+            }
             default: {
                 this._logger.debug(`${message?.getName()}:${message?.getKey()} - Skipping unknown domain event`);
                 return;

modules/api-svc/test/unit/FSPIOPEventHandler.test.js

@@ -5,27 +5,35 @@ const config = require('./data/defaultConfig.json');
 const {
     PartyInfoRequestedDmEvt,
     BulkQuotesRequestedDmEvt,
+    BulkTransfersRequestedDmEvt,
+    KafkaDomainEventConsumer,
+    KafkaDomainEventProducer,
 } = require('@mojaloop/sdk-scheme-adapter-private-shared-lib');
 
 const {
     OutboundBulkQuotesModel,
     PartiesModel,
+    OutboundBulkTransfersModel,
 } = require('~/lib/model');
 
-const {
-    KafkaDomainEventConsumer,
-    KafkaDomainEventProducer,
-} = require('@mojaloop/sdk-scheme-adapter-private-shared-lib');
-
 const logger = new Logger.Logger({ context: { app: 'FSPIOPEventHandler' }, stringify: () => '' });
 
 describe('FSPIOPEventHandler', () => {
-    test('should handle PartyInfoRequestedDmEvt event', async () => {
-        const fspiopEventHandler = new FSPIOPEventHandler({
+    let fspiopEventHandler;
+
+    beforeEach(async () => {
+        fspiopEventHandler = new FSPIOPEventHandler({
             config,
             logger,
         });
         await fspiopEventHandler.start();
+    });
+
+    afterAll(async () => {
+        await fspiopEventHandler.stop();
+    });
+
+    test('should handle PartyInfoRequestedDmEvt event', async () => {
         const request = {
             partyIdType: 'PERSONAL_ID',
             partyIdentifier: '16135551212',
@@ -99,16 +107,9 @@ describe('FSPIOPEventHandler', () => {
                 currentState: 'COMPLETED',
             }
         });
-
-        await fspiopEventHandler.stop();
     });
 
     test('should return error information for PartyInfoRequestedDmEvt event', async () => {
-        const fspiopEventHandler = new FSPIOPEventHandler({
-            config,
-            logger,
-        });
-        await fspiopEventHandler.start();
         const request = {
             partyIdType: 'PERSONAL_ID',
             partyIdentifier: '16135551212',
@@ -180,16 +181,9 @@ describe('FSPIOPEventHandler', () => {
                 },
             },
         });
-
-        await fspiopEventHandler.stop();
     });
 
     test('should handle BulkQuotesRequestedDmEvt event', async () => {
-        const fspiopEventHandler = new FSPIOPEventHandler({
-            config,
-            logger,
-        });
-        await fspiopEventHandler.start();
         const bulkId = 'bulk-tx-test';
         const event = new BulkQuotesRequestedDmEvt({
             bulkId,
@@ -251,8 +245,81 @@ describe('FSPIOPEventHandler', () => {
             bulkQuoteId: bulkQuoteResponse.bulkQuoteId,
             bulkQuotesResult: bulkQuoteResponse,
         });
+    });
 
-        await fspiopEventHandler.stop();
+    test('should handle BulkTransfersRequestedDmEvt event', async () => {
+        const bulkTransfersRequest = {
+            homeTransactionId: 'home-transaction-id',
+            from: {
+                idType: 'MSISDN',
+                idValue: '123456'
+            },
+            individualTransfers: [
+                {
+                    homeTransactionId: 'home-individual-transfer-id',
+                    to: {
+                        partyIdInfo: {
+                            partyIdType: 'MSISDN',
+                            partyIdentifier: '1'
+                        },
+                    },
+                    amountType: 'SEND',
+                    currency: 'USD',
+                    amount: '1'
+                },
+            ]
+        };
+        const bulkTransfersRequestedDmEvt = new BulkTransfersRequestedDmEvt({
+            bulkId: 'bulk-tx-test',
+            headers: [],
+            timestamp: Date.now(),
+            content: {
+                batchId: '61c35bae-77d0-4f7d-b894-be375b838ff6',
+                bulkTransfersRequest,
+            },
+        });
+
+        const bulkTransfersResult = {
+            bulkTransferId: '81c35bae-77d0-4f7d-b894-be375b838ff6',
+            currentState: 'COMPLETED',
+            individualTransferResults: [
+                {
+                    transferId: 'individual-transfer-id',
+                    to: {
+                        partyIdInfo: {
+                            partyIdType: 'MSISDN',
+                            partyIdentifier: '1'
+                        },
+                    },
+                    amountType: 'SEND',
+                    currency: 'USD',
+                    amount: '1'
+                },
+            ]
+        };
+
+        const initializeSpy = jest.spyOn(OutboundBulkTransfersModel.prototype, 'initialize')
+            .mockImplementationOnce(async () => bulkTransfersResult);
+
+        jest.spyOn(OutboundBulkTransfersModel.prototype, 'run')
+            .mockImplementationOnce(async () => bulkTransfersResult);
+
+
+        const handler = KafkaDomainEventConsumer.mock.ctor.mock.calls[0][0];
+        await handler(bulkTransfersRequestedDmEvt);
+
+        // run workflow
+        expect(initializeSpy).toBeCalledWith(bulkTransfersRequestedDmEvt.request);
+
+        await new Promise((resolve) => setTimeout(resolve, 1500));
+
+        const sent = KafkaDomainEventProducer.mock.sendDomainEvent.mock.calls[0][0];
+        expect(sent._data.name).toEqual('BulkTransfersCallbackReceivedDmEvt');
+        expect(sent._data.content).toEqual({
+            batchId: '61c35bae-77d0-4f7d-b894-be375b838ff6',
+            bulkTransferId: '81c35bae-77d0-4f7d-b894-be375b838ff6',
+            bulkTransfersResult
+        });
     });
 });
 

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

@@ -40,8 +40,8 @@
   },
   "dependencies": {
     "@mojaloop/api-snippets": "^15.0.1",
-    "@mojaloop/logging-bc-client-lib": "^0.1.13",
-    "@mojaloop/logging-bc-public-types-lib": "^0.1.10",
+    "@mojaloop/logging-bc-client-lib": "^0.1.14",
+    "@mojaloop/logging-bc-public-types-lib": "^0.1.11",
     "@mojaloop/sdk-scheme-adapter-private-shared-lib": "workspace:^",
     "ajv": "^8.11.0",
     "convict": "^6.2.3",
@@ -60,13 +60,13 @@
     "@types/supertest": "^2.0.12",
     "@types/swagger-ui-express": "^4.1.3",
     "@types/yamljs": "^0.2.31",
-    "@typescript-eslint/eslint-plugin": "^5.37.0",
-    "@typescript-eslint/parser": "^5.37.0",
+    "@typescript-eslint/eslint-plugin": "^5.38.0",
+    "@typescript-eslint/parser": "^5.38.0",
     "copyfiles": "^2.4.1",
     "eslint": "^8.23.1",
     "jest": "^29.0.3",
     "nodemon": "^2.0.20",
-    "npm-check-updates": "^16.1.3",
+    "npm-check-updates": "^16.2.1",
     "replace": "^1.2.1",
     "standard-version": "^9.5.0",
     "ts-jest": "^29.0.1",

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

@@ -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.13",
-        "@mojaloop/logging-bc-public-types-lib": "^0.1.10",
+        "@mojaloop/logging-bc-client-lib": "^0.1.14",
+        "@mojaloop/logging-bc-public-types-lib": "^0.1.11",
         "@mojaloop/sdk-scheme-adapter-private-shared-lib": "workspace:^",
         "convict": "^6.2.3"
     },
@@ -48,12 +48,12 @@
         "@types/jest": "^29.0.3",
         "@types/node": "^18.7.18",
         "@types/node-cache": "^4.2.5",
-        "@typescript-eslint/eslint-plugin": "^5.37.0",
-        "@typescript-eslint/parser": "^5.37.0",
+        "@typescript-eslint/eslint-plugin": "^5.38.0",
+        "@typescript-eslint/parser": "^5.38.0",
         "eslint": "^8.23.1",
         "jest": "^29.0.3",
         "nodemon": "^2.0.20",
-        "npm-check-updates": "^16.1.3",
+        "npm-check-updates": "^16.2.1",
         "replace": "^1.2.1",
         "standard-version": "^9.5.0",
         "ts-jest": "^29.0.1",

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

@@ -1,6 +1,9 @@
 import { ILogger } from '@mojaloop/logging-bc-public-types-lib';
 import {
     DomainEvent,
+    BulkTransfersCallbackReceivedDmEvt,
+    ProcessBulkTransfersCallbackCmdEvt,
+    IProcessBulkTransfersCallbackCmdEvtData,
 } from '@mojaloop/sdk-scheme-adapter-private-shared-lib';
 import { IDomainEventHandlerOptions } from '../../types';
 
@@ -9,8 +12,28 @@ export async function handleBulkTransfersCallbackReceived(
     options: IDomainEventHandlerOptions,
     logger: ILogger,
 ): Promise<void> {
+    const bulkTransfersCallbackReceivedMessage
+        = BulkTransfersCallbackReceivedDmEvt.CreateFromDomainEvent(message);
     try {
+        const processPartyInfoCallbackMessageData: IProcessBulkTransfersCallbackCmdEvtData = {
+            bulkId: bulkTransfersCallbackReceivedMessage.getKey(),
+            content: {
+                batchId: bulkTransfersCallbackReceivedMessage.batchId,
+                bulkTransferId: bulkTransfersCallbackReceivedMessage.bulkTransferId,
+                bulkTransfersResult: bulkTransfersCallbackReceivedMessage.bulkTransfersResult,
+            },
+            timestamp: Date.now(),
+            headers: bulkTransfersCallbackReceivedMessage.getHeaders(),
+        };
+
+        const processBulkTransfersCallbackMessage
+            = new ProcessBulkTransfersCallbackCmdEvt(processPartyInfoCallbackMessageData);
+
+        await options.commandProducer.sendCommandEvent(processBulkTransfersCallbackMessage);
+
+        logger.info(`Sent command event ${processBulkTransfersCallbackMessage.getName()}`);
+        console.log(processBulkTransfersCallbackMessage);
     } catch (err) {
-        logger.info(`Failed to send command event . ${(err as Error).message}`);
+        logger.info(`Failed to send command event ${ProcessBulkTransfersCallbackCmdEvt.name}. ${(err as Error).message}`);
     }
 }

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

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