refactor and add tests

Author: soccermax

docs/setup/index.md

@@ -50,7 +50,7 @@ already provides a robust foundation.
 
 ```json
 {
-  "cds": { "eventQueue": { "useAsCAPOutbox": true } }
+  "cds": { "eventQueue": { "useAsCAPQueue": true } }
 }
 ```
 
@@ -62,7 +62,7 @@ such as the configuration file path, event processing behavior, load balancing,
 The table includes the parameter name, a description of its purpose, and the default value if not specified.
 
 | Name                                 | Description                                                                                                                                                                                                                                                                                                                                                                                                                                 | Default        | Can be changed at runtime |
-| :----------------------------------- | :------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | :------------- | :------------------------ |
+|:-------------------------------------| :------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | :------------- | :------------------------ |
 | configFilePath                       | Path to the configuration file.                                                                                                                                                                                                                                                                                                                                                                                                             | null           | no                        |
 | events                               | Options to allow events in the configuration. E.g. via cds-env.                                                                                                                                                                                                                                                                                                                                                                             | {}             | no                        |
 | periodicEvents                       | Options to allow periodicEvents in the configuration. E.g. via cds-env                                                                                                                                                                                                                                                                                                                                                                      | {}             | no                        |
@@ -72,7 +72,7 @@ The table includes the parameter name, a description of its purpose, and the def
 | runInterval [ms]                     | The interval in milliseconds at which the runner runs.                                                                                                                                                                                                                                                                                                                                                                                      | 25 _ 60 _ 1000 | yes                       |
 | updatePeriodicEvents                 | Whether or not to update periodic events.                                                                                                                                                                                                                                                                                                                                                                                                   | true           | no                        |
 | thresholdLoggingEventProcessing [ms] | Threshold after how many milliseconds the processing of a event or periodic event is logged for observability.                                                                                                                                                                                                                                                                                                                              | 50             | yes                       |
-| useAsCAPOutbox                       | Uses the event-queue as the [outbox](https://cap.cloud.sap/docs/node.js/outbox) of CAP. Outbox calls are stored and processed in the event-queue instead of the outbox of CAP.                                                                                                                                                                                                                                                              | false          | no                        |
+| useAsCAPQueue/useAsCAPOutbox         | Uses the event-queue as the [outbox](https://cap.cloud.sap/docs/node.js/outbox) of CAP. Outbox calls are stored and processed in the event-queue instead of the outbox of CAP.                                                                                                                                                                                                                                                              | false          | no                        |
 | userId                               | User id for all created cds contexts. This influences the value for updated managed database fields like createdBy and modifiedBy.                                                                                                                                                                                                                                                                                                          | false          | yes                       |
 | cleanupLocksAndEventsForDev          | Deletes all semantic locks and sets all events that are in progress to error during server start. This is used to clean up leftovers from server crashes or restarts during processing.                                                                                                                                                                                                                                                     | true           | no                        |
 | insertEventsBeforeCommit             | If enabled, this feature allows events (including those for outboxed services) to be inserted in bulk using the before commit handler. This is performed to improve performance by mass inserting events instead of single insert operations. This can be disabled by the parameter `skipInsertEventsBeforeCommit` in the function publishEvent.                                                                                            | true           | yes                       |

docs/use-as-cap-outbox/index.md

@@ -24,14 +24,14 @@ transaction modes, load balancing, and others, with outboxed CDS services.
 
 ## How to enable the event-queue as outbox mechanism for CAP
 
-The initialization parameter `useAsCAPOutbox` enables the event-queue to act as a CAP outbox. To set this parameter,
+The initialization parameter `useAsCAPQueue` enables the event-queue to act as a CAP outbox. To set this parameter,
 refer to the [setup](/event-queue/setup/#initialization-parameters) part of the documentation. This is the only
 configuration needed to enable the event-queue as a CAP outbox.
 
 ```json
 {
   "cds": {
-    "eventQueue": { "useAsCAPOutbox": true }
+    "eventQueue": { "useAsCAPQueue": true }
   }
 }
 ```

example-cap-server/package.json

@@ -32,7 +32,7 @@
     "eventQueue": {
       "plugin": true,
       "runInterval": 300000,
-      "useAsCAPOutbox": true,
+      "useAsCAPQueue": true,
       "thresholdLoggingEventProcessing": 5000,
       "cleanupLocksAndEventsForDev": true,
       "events": {

src/EventQueueProcessorBase.js

@@ -24,7 +24,6 @@ const SELECT_LIMIT_EVENTS_PER_TICK = 100;
 const TRIES_FOR_EXCEEDED_EVENTS = 3;
 const EVENT_START_AFTER_HEADROOM = 3 * 1000;
 const SUFFIX_PERIODIC = "_PERIODIC";
-const DEFAULT_RETRY_AFTER = 5 * 60 * 1000;
 
 const ALLOWED_FIELDS_FOR_UPDATE = ["status", "startAfter"];
 
@@ -40,8 +39,6 @@ class EventQueueProcessorBase {
   #eventConfig;
   #isPeriodic;
   #lastSuccessfulRunTimestamp;
-  #retryFailedAfter;
-  #retryOpenAfter;
   #keepAliveRunner;
   #currentKeepAlivePromise = Promise.resolve();
   #etagMap;
@@ -69,8 +66,6 @@ class EventQueueProcessorBase {
     if (this.__parallelEventProcessing > LIMIT_PARALLEL_EVENT_PROCESSING) {
       this.__parallelEventProcessing = LIMIT_PARALLEL_EVENT_PROCESSING;
     }
-    this.#retryFailedAfter = this.#eventConfig.retryFailedAfter ?? DEFAULT_RETRY_AFTER;
-    this.#retryOpenAfter = this.#eventConfig.retryOpenAfter ?? DEFAULT_RETRY_AFTER;
     this.__concurrentEventProcessing = this.#eventConfig.multiInstanceProcessing;
     this.__retryAttempts = this.#isPeriodic ? 1 : this.#eventConfig.retryAttempts ?? DEFAULT_RETRY_ATTEMPTS;
     this.__selectMaxChunkSize = this.#eventConfig.selectMaxChunkSize ?? SELECT_LIMIT_EVENTS_PER_TICK;
@@ -456,10 +451,7 @@ class EventQueueProcessorBase {
         return result;
       }, {});
 
-      // TODO: schedule open events???
       for (const { ids, data } of Object.values(updateData)) {
-        let startAfter;
-
         if (!("status" in data)) {
           this.logger.error("can't find status value in return value of event-processing. Setting event to done", {
             ids,
@@ -474,14 +466,29 @@ class EventQueueProcessorBase {
           delete this.__notCommitedStatusMap[id];
         }
 
-        if ([EventProcessingStatus.Error, EventProcessingStatus.Open].includes(data.status)) {
-          startAfter = new Date(Date.now() + this.#retryFailedAfter);
+        if (![EventProcessingStatus.Open, EventProcessingStatus.Error].includes(data.status)) {
+          delete data.startAfter;
+        }
+
+        if (data.startAfter) {
+          data.startAfter = this.#normalizeDate(data.startAfter);
+        }
+
+        if (!data.startAfter && [EventProcessingStatus.Error, EventProcessingStatus.Open].includes(data.status)) {
+          data.startAfter = new Date(
+            Date.now() +
+              (EventProcessingStatus.Error ? this.#eventConfig.retryFailedAfter : this.#eventConfig.retryOpenAfter)
+          );
+        }
+
+        // TODO: check that startAfter is only allowed for tbp events
+        if (data.startAfter) {
           this.#eventSchedulerInstance.scheduleEvent(
             this.__context.tenant,
             this.#eventType,
             this.#eventSubType,
             this.#namespace,
-            startAfter
+            data.startAfter
           );
         }
 
@@ -490,14 +497,28 @@ class EventQueueProcessorBase {
             .set({
               ...data,
               lastAttemptTimestamp: ts,
-              ...(data.status === EventProcessingStatus.Error ? { startAfter: startAfter.toISOString() } : {}),
             })
             .where("ID IN", ids)
         );
       }
     });
   }
 
+  #normalizeDate(value) {
+    if (value instanceof Date) {
+      return value;
+    }
+
+    if (typeof value === "string" || typeof value === "number") {
+      const date = new Date(value);
+      if (!isNaN(date)) {
+        return date;
+      }
+    }
+
+    return null;
+  }
+
   #ensureEveryQueueEntryHasStatus() {
     this.__queueEntries.forEach((queueEntry) => {
       if (
@@ -529,10 +550,6 @@ class EventQueueProcessorBase {
         return;
       }
 
-      if (!status) {
-        return;
-      }
-
       this.logger.error("Not allowed event status returned. Only Open, Done, Error is allowed!", {
         eventType: this.#eventType,
         eventSubType: this.#eventSubType,

src/config.js

@@ -24,6 +24,7 @@ const CAP_EVENT_TYPE = "CAP_OUTBOX";
 const CAP_PARALLEL_DEFAULT = 5;
 const CAP_MAX_ATTEMPTS_DEFAULT = 5;
 const DELETE_TENANT_BLOCK_AFTER_MS = 5 * 60 * 1000;
+const DEFAULT_RETRY_AFTER = 5 * 60 * 1000;
 const PRIORITIES = Object.values(Priorities);
 const UTC_DEFAULT = false;
 const USE_CRON_TZ_DEFAULT = true;
@@ -252,7 +253,6 @@ class Config {
       selectMaxChunkSize: config.selectMaxChunkSize ?? config.chunkSize,
       parallelEventProcessing: config.parallelEventProcessing ?? (config.parallel && CAP_PARALLEL_DEFAULT),
       retryAttempts: config.retryAttempts ?? config.maxAttempts ?? CAP_MAX_ATTEMPTS_DEFAULT,
-      propagateHeaders: config.propagateHeaders ?? [],
       namespace,
       ...config,
     });
@@ -439,6 +439,8 @@ class Config {
       ? Object.fromEntries(new Map(event.appInstances.map((a) => [a, true])))
       : null;
     event.propagateHeaders = event.propagateHeaders ?? [];
+    event.retryFailedAfter = event.retryFailedAfter ?? DEFAULT_RETRY_AFTER;
+    event.retryOpenAfter = event.retryOpenAfter ?? DEFAULT_RETRY_AFTER;
   }
 
   #basicEventValidation(event) {
@@ -755,13 +757,24 @@ class Config {
   }
 
   set useAsCAPOutbox(value) {
+    if (this.#useAsCAPOutbox) {
+      return;
+    }
     this.#useAsCAPOutbox = value;
   }
 
   get useAsCAPOutbox() {
     return this.#useAsCAPOutbox;
   }
 
+  set useAsCAPQueue(value) {
+    this.useAsCAPOutbox = value;
+  }
+
+  get useAsCAPQueue() {
+    return this.useAsCAPOutbox;
+  }
+
   set userId(value) {
     this.#userId = value;
   }

src/initialize.js

@@ -35,6 +35,7 @@ const CONFIG_VARS = [
   ["updatePeriodicEvents", true],
   ["thresholdLoggingEventProcessing", 50],
   ["useAsCAPOutbox", false],
+  ["useAsCAPQueue", false],
   ["userId", null],
   ["cleanupLocksAndEventsForDev", false],
   ["redisOptions", {}],
@@ -65,6 +66,7 @@ const CONFIG_VARS = [
  * @param {boolean} [options.updatePeriodicEvents=true] - Automatically update periodic events.
  * @param {number} [options.thresholdLoggingEventProcessing=50] - Threshold for logging event processing time (in milliseconds).
  * @param {boolean} [options.useAsCAPOutbox=false] - Use the event queue as a CAP Outbox.
+ * @param {boolean} [options.useAsCAPQueue=false] - Use the event queue as a CAP Outbox.
  * @param {string} [options.userId=null] - ID of the user initiating the process.
  * @param {boolean} [options.cleanupLocksAndEventsForDev=false] - Cleanup locks and events for development environments.
  * @param {Object} [options.redisOptions={}] - Configuration options for Redis.

src/outbox/EventQueueGenericOutboxHandler.js

@@ -367,6 +367,40 @@ class EventQueueGenericOutboxHandler extends EventQueueBaseClass {
       return queueEntries.map((queueEntry) => [queueEntry.ID, EventProcessingStatus.Done]);
     }
 
+    const [firstEntry] = result;
+    if (Array.isArray(firstEntry)) {
+      const [, innerResult] = firstEntry;
+      if (innerResult instanceof Object) {
+        return result;
+      } else {
+        return result.map(([id, status]) => {
+          return [id, { status }];
+        });
+      }
+    } else if (firstEntry instanceof Object) {
+      return result.reduce((result, entry) => {
+        let { ID } = entry;
+
+        if (!ID) {
+          if (queueEntries.length > 1) {
+            throw new Error(
+              "The CAP handler return value does not match the event-queue specification. Please check the documentation"
+            );
+          } else {
+            ID = queueEntries[0].ID;
+          }
+        }
+
+        delete entry.ID;
+        if (!("status" in entry)) {
+          entry.status = EventProcessingStatus.Done;
+        }
+
+        result.push([ID, entry]);
+        return result;
+      }, []);
+    }
+
     const valid = !result.some((entry) => {
       const [, status] = entry;
       return !validStatusValues.includes(status);

src/outbox/eventQueueAsOutbox.js

@@ -56,11 +56,11 @@ function outboxed(srv, customOpts) {
       outboxOpts = config.addCAPOutboxEventSpecificAction(srv.name, req.event);
     }
     const subType = hasSpecificSettings ? [srv.name, req.event].join(".") : srv.name;
-    outboxOpts = config.getEventConfig(CDS_EVENT_TYPE, subType);
-    const eventHeaders = getPropagatedHeaders(outboxOpts, req);
     const namespace = outboxOpts.namespace ?? config.namespace;
+    outboxOpts = config.getEventConfig(CDS_EVENT_TYPE, subType, namespace);
+    const eventHeaders = getPropagatedHeaders(outboxOpts, req);
     if (["persistent-outbox", "persistent-queue"].includes(outboxOpts.kind)) {
-      await _mapToEventAndPublish(context, subType , req, eventHeaders, namespace);
+      await _mapToEventAndPublish(context, subType, req, eventHeaders, namespace);
       return;
     }
     context.on("succeeded", async () => {

test/__snapshots__/eventQueueOutbox.test.js.snap

@@ -43,6 +43,8 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true ad-hoc events overwrite se
   "priority": "medium",
   "propagateHeaders": [],
   "retryAttempts": 20,
+  "retryFailedAfter": 300000,
+  "retryOpenAfter": 300000,
   "selectMaxChunkSize": 10,
   "subType": "NotificationServicePeriodic.action",
   "transactionMode": "isolated",
@@ -69,6 +71,8 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true ad-hoc events overwrite se
   "priority": "medium",
   "propagateHeaders": [],
   "retryAttempts": 20,
+  "retryFailedAfter": 300000,
+  "retryOpenAfter": 300000,
   "selectMaxChunkSize": 10,
   "subType": "NotificationServicePeriodic.action",
   "transactionMode": "isolated",
@@ -95,6 +99,8 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true custom options should win
   "priority": "medium",
   "propagateHeaders": [],
   "retryAttempts": 20,
+  "retryFailedAfter": 300000,
+  "retryOpenAfter": 300000,
   "selectMaxChunkSize": 10,
   "subType": "NotificationServiceOutboxedByConfig",
   "transactionMode": "alwaysCommit",
@@ -133,6 +139,8 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true map config to event-queue
   "priority": "medium",
   "propagateHeaders": [],
   "retryAttempts": 20,
+  "retryFailedAfter": 300000,
+  "retryOpenAfter": 300000,
   "selectMaxChunkSize": 10,
   "subType": "NotificationService",
   "type": "CAP_OUTBOX",
@@ -169,6 +177,8 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true periodic events inherit co
   "namespace": "default",
   "priority": "medium",
   "propagateHeaders": [],
+  "retryFailedAfter": 300000,
+  "retryOpenAfter": 300000,
   "subType": "NotificationServicePeriodic.main",
   "transactionMode": "isolated",
   "type": "CAP_OUTBOX_PERIODIC",
@@ -204,6 +214,8 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true periodic events inherit co
   "namespace": "default",
   "priority": "medium",
   "propagateHeaders": [],
+  "retryFailedAfter": 300000,
+  "retryOpenAfter": 300000,
   "subType": "NotificationServicePeriodic.main",
   "transactionMode": "alwaysRollback",
   "type": "CAP_OUTBOX_PERIODIC",
@@ -230,6 +242,8 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true periodic events inherit co
   "namespace": "default",
   "priority": "medium",
   "propagateHeaders": [],
+  "retryFailedAfter": 300000,
+  "retryOpenAfter": 300000,
   "subType": "NotificationServicePeriodic.main",
   "transactionMode": "isolated",
   "type": "CAP_OUTBOX_PERIODIC",
@@ -420,6 +434,8 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true should work for outboxed s
   "priority": "medium",
   "propagateHeaders": [],
   "retryAttempts": 20,
+  "retryFailedAfter": 300000,
+  "retryOpenAfter": 300000,
   "selectMaxChunkSize": 10,
   "subType": "NotificationServiceOutboxedByConfig",
   "transactionMode": "alwaysRollback",