Refactor on custom event

package.json

@@ -34,6 +34,7 @@
   "dependencies": {
     "@myunisoft/redis": "github:MyUnisoft/redis#v3.4.2",
     "@openally/mutex": "^1.0.0",
+    "@openally/result": "^1.2.1",
     "ajv": "^8.12.0",
     "pino-pretty": "^10.3.1",
     "ts-pattern": "^4.3.0"

src/class/eventManagement/dispatcher/transaction-handler.class.ts

@@ -361,8 +361,6 @@ export class TransactionHandler<T extends GenericEvent = GenericEvent> {
           inactiveIncomerTransactionStore.deleteTransaction(relatedHandlerTransaction.redisMetadata.transactionId)
       ]));
 
-      // To publish later when concerned Incomer
-
       this.logger.debug(this.standardLogFn(
         dispatcherTransaction as unknown as StandardLogOpts<T>
       )("Spread transaction has been backup"));

src/class/eventManagement/incomer.class.ts

@@ -11,6 +11,7 @@ import {
 import { pino } from "pino";
 import { P, match } from "ts-pattern";
 import { ValidateFunction } from "ajv";
+import { Result } from "@openally/result";
 
 // Import Internal Dependencies
 import {
@@ -74,6 +75,8 @@ function isIncomerChannelMessage<T extends GenericEvent = GenericEvent>(value:
   return value.name !== "APPROVEMENT";
 }
 
+export type EventCallbackResponse = Result<"Resolved" | "NotResolved", string>;
+
 export type IncomerOptions<T extends GenericEvent = GenericEvent> = {
   /* Service name */
   name: string;
@@ -86,7 +89,7 @@ export type IncomerOptions<T extends GenericEvent = GenericEvent> = {
     eventsValidationFn?: eventsValidationFn<T>;
     customValidationCbFn?: customValidationCbFn<T>;
   };
-  eventCallback: (message: CallBackEventMessage<T>) => void;
+  eventCallback: (message: CallBackEventMessage<T>) => Promise<EventCallbackResponse>;
   dispatcherInactivityOptions?: {
     /* max interval between received ping before considering dispatcher off */
     maxPingInterval?: number;
@@ -102,7 +105,7 @@ export class Incomer <
 > extends EventEmitter {
   readonly name: string;
   readonly prefix: Prefix | undefined;
-  readonly eventCallback: (message: CallBackEventMessage<T>) => void;
+  readonly eventCallback: (message: CallBackEventMessage<T>) => Promise<EventCallbackResponse>;
 
   public dispatcherConnectionState = false;
   public baseUUID = randomUUID();
@@ -642,18 +645,23 @@ export class Incomer <
 
     const formattedTransaction = await store.setTransaction(transaction);
 
-    await Promise.all([
-      this.eventCallback({ ...event, eventTransactionId } as unknown as CallBackEventMessage<T>),
-      store.updateTransaction(formattedTransaction.redisMetadata.transactionId, {
+    const callbackResult = await this.eventCallback({ ...event, eventTransactionId } as unknown as CallBackEventMessage<T>);
+
+    if (callbackResult.ok) {
+      await store.updateTransaction(formattedTransaction.redisMetadata.transactionId, {
         ...formattedTransaction,
         redisMetadata: {
           ...formattedTransaction.redisMetadata,
           resolved: true
         }
-      } as Transaction<"incomer">)
-    ]);
+      } as Transaction<"incomer">);
+
+      this.logger.info(this.standardLogFn(logData)("Resolved Custom event"));
+
+      return;
+    }
 
-    this.logger.info(this.standardLogFn(logData)("Resolved Custom event"));
+    this.logger.info(this.standardLogFn(logData)(`Callback error reason: ${callbackResult.val}`));
   }
 
   private async handleApprovement(message: DispatcherApprovementMessage) {

src/class/store/incomer.class.ts

@@ -103,7 +103,7 @@ export class IncomerStore extends KVPeer<RegisteredIncomer> {
     const incomer = await this.getValue(incomerKey);
 
     if (!incomer) {
-      throw new Error("Cannot find the Incomer");
+      throw new Error(`Cannot find the Incomer ${incomerKey}`);
     }
 
     this.setValue({ key: incomerKey, value: { ...incomer, lastActivity: Date.now() } });

test/UT/class/eventManagement/dispatcher.spec.ts

@@ -30,7 +30,7 @@ const mockedLoggerInfo = jest.spyOn(logger, "info");
 const mockedHandleDispatcherMessages = jest.spyOn(Dispatcher.prototype as any, "approveIncomer");
 const mockedHandleIncomerMessages = jest.spyOn(Dispatcher.prototype as any, "handleCustomEvents");
 const mockedPing = jest.spyOn(Dispatcher.prototype as any, "ping");
-const mockedCheckLastActivity = jest.spyOn(Dispatcher.prototype as any, "checkLastActivityIntervalFn");
+const mockedCheckLastActivity = jest.spyOn(Dispatcher.prototype as any, "checkLastActivity");
 const mockedHandleInactiveIncomer =  jest.spyOn(TransactionHandler.prototype, "resolveInactiveIncomerTransactions");
 
 const mockedSetTransaction = jest.spyOn(TransactionStore.prototype, "setTransaction");

test/UT/class/eventManagement/handle-inactive-no-backup.spec.ts

@@ -182,24 +182,25 @@ describe("Publishing/exploiting a custom event & inactive incomer", () => {
       });
 
       await concernedIncomer.initialize();
+    });
 
+    test("expect the second incomer to have handle the event by retaking the main Transaction", async() => {
       await concernedIncomer.publish(event);
 
       await timers.setTimeout(1_000);
 
       expect(handlerTransaction).toBeDefined();
 
+      await secondConcernedIncomer.initialize();
       await concernedIncomer.close();
 
+      jest.spyOn(dispatcher as any, "checkLastActivity").mockImplementation(async(opts: any) => {
+        // do nothing
+      });
       dispatcher["subscriber"]!.on("message", (channel, message) => dispatcher["handleMessages"](channel, message));
+      secondConcernedIncomer["subscriber"]!.on("message", (channel, message) => secondConcernedIncomer["handleMessages"](channel, message));
 
-      await secondConcernedIncomer.initialize();
-
-      await timers.setTimeout(10_000);
-    });
-
-    test("expect the second incomer to have handle the event by retaking the main Transaction", async() => {
-      await timers.setTimeout(18_000);
+      await timers.setTimeout(5_000);
 
       const mockCalls = mockedSetTransaction.mock.calls.flat();