saga pattern start

Author: Max Gruenfelder

src/EventQueueProcessorBase.js

@@ -43,6 +43,7 @@ class EventQueueProcessorBase {
   #currentKeepAlivePromise = Promise.resolve();
   #etagMap;
   #namespace;
+  #nextSagaEvents;
 
   constructor(context, eventType, eventSubType, config) {
     this.__context = context;
@@ -468,6 +469,10 @@ class EventQueueProcessorBase {
         return result;
       }, {});
 
+      if (!Object.values(updateData).length) {
+        return;
+      }
+
       for (const { ids, data } of Object.values(updateData)) {
         if (!("status" in data)) {
           // TODO: Can this still happen?
@@ -522,6 +527,11 @@ class EventQueueProcessorBase {
             .where("ID IN", ids)
         );
       }
+
+      if (this.#nextSagaEvents?.length) {
+        await tx.run(INSERT.into(this.#config.tableNameEventQueue).entries(this.#nextSagaEvents));
+        this.#nextSagaEvents = [];
+      }
     });
   }
 
@@ -1390,6 +1400,10 @@ class EventQueueProcessorBase {
   get allowedFieldsEventHandler() {
     return ALLOWED_FIELDS_FOR_UPDATE;
   }
+
+  set nextSagaEvents(value) {
+    this.#nextSagaEvents = value;
+  }
 }
 
 module.exports = EventQueueProcessorBase;

src/outbox/EventQueueGenericOutboxHandler.js

@@ -14,6 +14,8 @@ const EVENT_QUEUE_ACTIONS = {
   EXCEEDED: "eventQueueRetriesExceeded",
   CLUSTER: "eventQueueCluster",
   CHECK_AND_ADJUST: "eventQueueCheckAndAdjustPayload",
+  SAGA_SUCCESS: "#succeeded",
+  SAGA_FAILED: "#failed",
 };
 
 class EventQueueGenericOutboxHandler extends EventQueueBaseClass {
@@ -260,7 +262,7 @@ class EventQueueGenericOutboxHandler extends EventQueueBaseClass {
   async checkEventAndGeneratePayload(queueEntry) {
     const payload = await super.checkEventAndGeneratePayload(queueEntry);
     const { event } = payload;
-    const handlerName = this.#checkHandlerExists(EVENT_QUEUE_ACTIONS.CHECK_AND_ADJUST, event);
+    const handlerName = this.#checkHandlerExists({ eventQueueFn: EVENT_QUEUE_ACTIONS.CHECK_AND_ADJUST, event });
     if (!handlerName) {
       return payload;
     }
@@ -281,7 +283,7 @@ class EventQueueGenericOutboxHandler extends EventQueueBaseClass {
 
   async hookForExceededEvents(exceededEvent) {
     const { event } = exceededEvent.payload;
-    const handlerName = this.#checkHandlerExists(EVENT_QUEUE_ACTIONS.EXCEEDED, event);
+    const handlerName = this.#checkHandlerExists({ eventQueueFn: EVENT_QUEUE_ACTIONS.EXCEEDED, event });
     if (!handlerName) {
       return await super.hookForExceededEvents(exceededEvent);
     }
@@ -299,14 +301,23 @@ class EventQueueGenericOutboxHandler extends EventQueueBaseClass {
     return await super.beforeProcessingEvents();
   }
 
-  #checkHandlerExists(eventQueueFn, event) {
-    const specificHandler = this.__onHandlers[[eventQueueFn, event].join(".")];
+  #checkHandlerExists({ eventQueueFn, event, saga } = {}) {
+    if (eventQueueFn) {
+      const specificHandler = this.__onHandlers[[eventQueueFn, event].join(".")];
+      if (specificHandler) {
+        return specificHandler;
+      }
+
+      const genericHandler = this.__onHandlers[eventQueueFn];
+      return genericHandler ?? null;
+    }
+
+    const specificHandler = this.__onHandlers[[event, saga].join("/")];
     if (specificHandler) {
       return specificHandler;
     }
 
-    const genericHandler = this.__onHandlers[eventQueueFn];
-    return genericHandler ?? null;
+    return this.__onHandlers[saga];
   }
 
   async processPeriodicEvent(processContext, key, queueEntry) {
@@ -319,7 +330,7 @@ class EventQueueGenericOutboxHandler extends EventQueueBaseClass {
   #buildDispatchData(context, payload, { key, queueEntries } = {}) {
     const { useEventQueueUser } = this.eventConfig;
     const userId = useEventQueueUser ? config.userId : payload.contextUser;
-    const reg = payload._fromSend ? new cds.Request(payload) : new cds.Event(payload);
+    const reg = payload.das_fromSend ? new cds.Request(payload) : new cds.Event(payload);
     const invocationFn = payload._fromSend ? "send" : "emit";
     delete reg._fromSend;
     delete reg.contextUser;
@@ -344,7 +355,9 @@ class EventQueueGenericOutboxHandler extends EventQueueBaseClass {
       const { userId, invocationFn, reg } = this.#buildDispatchData(processContext, payload, { key, queueEntries });
       await this.#setContextUser(processContext, userId, reg);
       const result = await this.__srvUnboxed.tx(processContext)[invocationFn](reg);
-      return this.#determineResultStatus(result, queueEntries);
+      const statusTuple = this.#determineResultStatus(result, queueEntries);
+      await this.#publishFollowupEvents(processContext, reg, statusTuple);
+      return statusTuple;
     } catch (err) {
       this.logger.error("error processing outboxed service call", err, {
         serviceName: this.eventSubType,
@@ -359,11 +372,41 @@ class EventQueueGenericOutboxHandler extends EventQueueBaseClass {
     }
   }
 
+  async #publishFollowupEvents(processContext, req, statusTuple) {
+    const succeeded = this.#checkHandlerExists({ event: req.event, saga: EVENT_QUEUE_ACTIONS.SAGA_SUCCESS });
+    const failed = this.#checkHandlerExists({ event: req.event, saga: EVENT_QUEUE_ACTIONS.SAGA_FAILED });
+
+    if (!succeeded && !failed) {
+      return;
+    }
+
+    // NOTE: required for #failed because tx is rolledback and new events would not be commmited!
+    const tx = cds.tx(processContext);
+    const nextEvents = tx._eventQueue?.events;
+
+    if (nextEvents?.length) {
+      tx._eventQueue.events = [];
+    }
+
+    for (const [id, result] of statusTuple) {
+      if (succeeded && result.status === EventProcessingStatus.Done) {
+        (await this.__srv.tx(processContext)).send(succeeded, result.nextData ?? req.data);
+      }
+
+      if (failed && result.status === EventProcessingStatus.Error) {
+        (await this.__srv.tx(processContext)).send(failed, result.nextData ?? req.data);
+      }
+    }
+
+    this.nextSagaEvents = tx._eventQueue.events;
+    tx._eventQueue.events = nextEvents ?? [];
+  }
+
   #determineResultStatus(result, queueEntries) {
     const validStatusValues = Object.values(EventProcessingStatus);
     const validStatus = validStatusValues.includes(result);
     if (validStatus) {
-      return queueEntries.map((queueEntry) => [queueEntry.ID, result]);
+      return queueEntries.map((queueEntry) => [queueEntry.ID, { status: result }]);
     }
 
     if (result instanceof Object && !Array.isArray(result)) {

test/asset/outboxProject/srv/service/saga-service.js

@@ -0,0 +1,50 @@
+"use strict";
+
+const cds = require("@sap/cds");
+
+class StandardService extends cds.Service {
+  async init() {
+    await super.init();
+    this.on("saga", (req) => {
+      cds.log(this.name).info(req.event, {
+        data: req.data,
+        user: req.user.id,
+        error: req.data.error,
+      });
+
+      return {
+        status: req.data.status ?? 2,
+        ...(req.data.nextData && { nextData: req.data.nextData }),
+        ...(req.data.errorMessage && { error: new Error(req.data.errorMessage) }),
+      };
+    });
+
+    this.on("saga/#succeeded", (req) => {
+      cds.log(this.name).info(req.event, {
+        data: req.data,
+        user: req.user.id,
+        error: req.data.error,
+      });
+
+      return {
+        status: req.data.status ?? 2,
+        ...(req.data.errorMessage && { error: new Error(req.data.errorMessage) }),
+      };
+    });
+
+    this.on("saga/#failed", (req) => {
+      cds.log(this.name).info(req.event, {
+        data: req.data,
+        user: req.user.id,
+        error: req.data.error,
+      });
+
+      return {
+        status: 2,
+        ...(req.data.errorMessage && { error: new Error(req.data.errorMessage) }),
+      };
+    });
+  }
+}
+
+module.exports = StandardService;

test/eventQueueOutbox.test.js

@@ -162,6 +162,13 @@ cds.env.requires.Namespace = {
   },
 };
 
+cds.env.requires.Saga = {
+  impl: path.join(basePath, "srv/service/saga-service.js"),
+  queued: {
+    kind: "persistent-queue",
+  },
+};
+
 cds.env.requires["sapafcsdk.scheduling.ProviderService"] = {
   impl: path.join(basePath, "srv/service/standard-service.js"),
   outbox: {
@@ -2470,6 +2477,77 @@ describe("event-queue outbox", () => {
         expect(loggerMock.callsLengths().error).toEqual(0);
       });
     });
+
+    describe("saga pattern", () => {
+      describe("event specific handlers", () => {
+        it("if succeeded handler exists and event is green, trigger next event", async () => {
+          const service = await cds.connect.to("Saga");
+          await service.send("saga", {});
+          await commitAndOpenNew();
+          await processEventQueue(tx.context, "CAP_OUTBOX", service.name);
+          const [done, next] = await testHelper.selectEventQueueAndReturn(tx, {
+            expectedLength: 2,
+            additionalColumns: ["payload"],
+          });
+          expect(JSON.parse(done.payload)).toMatchObject({ event: "saga" });
+          expect(done.status).toEqual(EventProcessingStatus.Done);
+
+          expect(JSON.parse(next.payload)).toMatchObject({ event: "saga/#succeeded" });
+          expect(next.status).toEqual(EventProcessingStatus.Done);
+          expect(loggerMock.callsLengths().error).toEqual(0);
+        });
+
+        it("if failed handler exists and event is red, trigger next event", async () => {
+          const service = await cds.connect.to("Saga");
+          await service.send("saga", { status: EventProcessingStatus.Error });
+          await commitAndOpenNew();
+          await processEventQueue(tx.context, "CAP_OUTBOX", service.name);
+          const [done, next] = await testHelper.selectEventQueueAndReturn(tx, {
+            expectedLength: 2,
+            additionalColumns: ["payload", "lastAttemptTimestamp"],
+          });
+          expect(JSON.parse(done.payload)).toMatchObject({ event: "saga" });
+          expect(done.status).toEqual(EventProcessingStatus.Error);
+
+          expect(JSON.parse(next.payload)).toMatchObject({ event: "saga/#failed" });
+          expect(next.status).toEqual(EventProcessingStatus.Done);
+          expect(loggerMock.callsLengths().error).toEqual(0);
+        });
+
+        it.skip("how to deal with specific event configuration srv.actionName", async () => {
+          const service = await cds.connect.to("Saga");
+          await service.send("saga");
+          await commitAndOpenNew();
+          await processEventQueue(tx.context, "CAP_OUTBOX", service.name);
+          const [done, next] = await testHelper.selectEventQueueAndReturn(tx, {
+            expectedLength: 2,
+            additionalColumns: ["payload", "status"],
+          });
+          expect(JSON.parse(done.payload)).toMatchObject({});
+          expect(JSON.parse(done.payload)).toMatchObject({});
+          expect(loggerMock.callsLengths().error).toEqual(0);
+        });
+      });
+
+      describe("provide next data", () => {
+        it("failed handler with next data", async () => {
+          const service = await cds.connect.to("Saga");
+          await service.send("saga", { status: EventProcessingStatus.Error, nextData: { newData: "dummyData" } });
+          await commitAndOpenNew();
+          await processEventQueue(tx.context, "CAP_OUTBOX", service.name);
+          const [done, next] = await testHelper.selectEventQueueAndReturn(tx, {
+            expectedLength: 2,
+            additionalColumns: ["payload"],
+          });
+          expect(JSON.parse(done.payload)).toMatchObject({ event: "saga" });
+          expect(done.status).toEqual(EventProcessingStatus.Error);
+
+          expect(JSON.parse(next.payload)).toMatchObject({ event: "saga/#failed", data: { newData: "dummyData" } });
+          expect(next.status).toEqual(EventProcessingStatus.Done);
+          expect(loggerMock.callsLengths().error).toEqual(0);
+        });
+      });
+    });
   });
 
   const commitAndOpenNew = async () => {