Propagate also on cds.context (#443)

* propagate

* beta 2

---------

Co-authored-by: Max Gruenfelder <maximilian.gruenfelder@scheer-group.com>

Author: soccermax

CHANGELOG.md

@@ -10,7 +10,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 ### Added
 
 - [CAP Queue] Add support for defining successor and failed events of event handlers. See documentation for how to use it.
-- [CAP Queue] Allow to propagate cds.context properties (e.g. features). This can be configured per event (`cds.env.requires[<SERVICE>].queued.propagatedContextProperties = ["features"]`)
+- [CAP Queue] Allow to propagate cds.context properties (e.g. features). This can be configured per event (`cds.env.requires[<SERVICE>].queued.propagateContextProperties = ["features"]`)
 
 ### Fixed
 

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "@cap-js-community/event-queue",
-  "version": "2.1.0-beta.1",
+  "version": "2.1.0-beta.2",
   "description": "An event queue that enables secure transactional processing of asynchronous and periodic events, featuring instant event processing with Redis Pub/Sub and load distribution across all application instances.",
   "main": "src/index.js",
   "types": "src/index.d.ts",

package.json

@@ -64,7 +64,7 @@ const ALLOWED_EVENT_OPTIONS_AD_HOC = [
   "checkForNextChunk",
   "retryFailedAfter",
   "propagateHeaders",
-  "propagatedContextProperties",
+  "propagateContextProperties",
   "retryOpenAfter",
   "multiInstanceProcessing",
   "kind",
@@ -518,7 +518,7 @@ class Config {
       ? Object.fromEntries(new Map(event.appInstances.map((a) => [a, true])))
       : null;
     event.propagateHeaders = event.propagateHeaders ?? [];
-    event.propagatedContextProperties = event.propagatedContextProperties ?? [];
+    event.propagateContextProperties = event.propagateContextProperties ?? [];
     event.retryFailedAfter = event.retryFailedAfter ?? DEFAULT_RETRY_AFTER;
     event.retryOpenAfter = event.retryOpenAfter ?? DEFAULT_RETRY_AFTER;
   }

src/config.js

@@ -64,7 +64,7 @@ class EventQueueGenericOutboxHandler extends EventQueueBaseClass {
         }
       } else {
         for (const actionName in genericClusterEvents) {
-          const reg = new cds.Request({
+          const req = new cds.Request({
             event: EVENT_QUEUE_ACTIONS.CLUSTER,
             user: this.context.user,
             eventQueue: {
@@ -77,14 +77,14 @@ class EventQueueGenericOutboxHandler extends EventQueueBaseClass {
                 this.#clusterByDataProperty(actionName, genericClusterEvents[actionName], propertyName, cb),
             },
           });
-          const clusterResult = await this.__srvUnboxed.tx(this.context).send(reg);
+          const clusterResult = await this.__srvUnboxed.tx(this.context).send(req);
           if (this.#validateCluster(clusterResult)) {
             Object.assign(clusterMap, clusterResult);
           } else {
             this.logger.error(
               "cluster result of handler is not valid. Check the documentation for the expected structure. Continuing without clustering!",
               {
-                handler: reg.event,
+                handler: req.event,
                 clusterResult: JSON.stringify(clusterResult),
               }
             );
@@ -95,7 +95,7 @@ class EventQueueGenericOutboxHandler extends EventQueueBaseClass {
     }
 
     for (const actionName in specificClusterEvents) {
-      const reg = new cds.Request({
+      const req = new cds.Request({
         event: `${EVENT_QUEUE_ACTIONS.CLUSTER}.${actionName}`,
         user: this.context.user,
         eventQueue: {
@@ -108,14 +108,14 @@ class EventQueueGenericOutboxHandler extends EventQueueBaseClass {
             this.#clusterByDataProperty(actionName, specificClusterEvents[actionName], propertyName, cb),
         },
       });
-      const clusterResult = await this.__srvUnboxed.tx(this.context).send(reg);
+      const clusterResult = await this.__srvUnboxed.tx(this.context).send(req);
       if (this.#validateCluster(clusterResult)) {
         Object.assign(clusterMap, clusterResult);
       } else {
         this.logger.error(
           "cluster result of handler is not valid. Check the documentation for the expected structure. Continuing without clustering!",
           {
-            handler: reg.event,
+            handler: req.event,
             clusterResult: JSON.stringify(clusterResult),
           }
         );
@@ -267,12 +267,12 @@ class EventQueueGenericOutboxHandler extends EventQueueBaseClass {
       return payload;
     }
 
-    const { reg, userId } = this.#buildDispatchData(this.context, payload, {
+    const { req, userId } = this.#buildDispatchData(payload, {
       queueEntries: [queueEntry],
     });
-    reg.event = handlerName;
-    await this.#setContextUser(this.context, userId, reg);
-    const data = await this.__srvUnboxed.tx(this.context).send(reg);
+    req.event = handlerName;
+    await this.#setContextUser(this.context, userId, req);
+    const data = await this.__srvUnboxed.tx(this.context).send(req);
     if (data) {
       payload.data = data;
       return payload;
@@ -288,12 +288,12 @@ class EventQueueGenericOutboxHandler extends EventQueueBaseClass {
       return await super.hookForExceededEvents(exceededEvent);
     }
 
-    const { reg, userId } = this.#buildDispatchData(this.context, exceededEvent.payload, {
+    const { req, userId } = this.#buildDispatchData(exceededEvent.payload, {
       queueEntries: [exceededEvent],
     });
-    await this.#setContextUser(this.context, userId, reg);
-    reg.event = handlerName;
-    await this.__srvUnboxed.tx(this.context).send(reg);
+    await this.#setContextUser(this.context, userId, req);
+    req.event = handlerName;
+    await this.__srvUnboxed.tx(this.context).send(req);
   }
 
   // NOTE: Currently not exposed to CAP service; we wait for a valid use case
@@ -326,40 +326,47 @@ class EventQueueGenericOutboxHandler extends EventQueueBaseClass {
 
   async processPeriodicEvent(processContext, key, queueEntry) {
     const { actionName } = config.normalizeSubType(this.eventType, this.eventSubType);
-    const reg = new cds.Event({ event: actionName, eventQueue: { processor: this, key, queueEntries: [queueEntry] } });
-    await this.#setContextUser(processContext, config.userId, reg);
-    await this.__srvUnboxed.tx(processContext).emit(reg);
+    const req = new cds.Event({ event: actionName, eventQueue: { processor: this, key, queueEntries: [queueEntry] } });
+    await this.#setContextUser(processContext, config.userId, req);
+    await this.__srvUnboxed.tx(processContext).emit(req);
   }
 
-  #buildDispatchData(context, payload, { key, queueEntries } = {}) {
+  #buildDispatchData(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 req = payload._fromSend ? new cds.Request(payload) : new cds.Event(payload);
     const invocationFn = payload._fromSend ? "send" : "emit";
-    delete reg._fromSend;
-    delete reg.contextUser;
-    reg.eventQueue = { processor: this, key, queueEntries, payload };
-    return { reg, userId, invocationFn };
+    delete req._fromSend;
+    delete req.contextUser;
+    req.eventQueue = { processor: this, key, queueEntries, payload };
+
+    if (this.eventConfig.propagateContextProperties?.length && this.transactionMode === "isolated" && cds.context) {
+      for (const prop of this.eventConfig.propagateContextProperties) {
+        req[prop] && (cds.context[prop] = req[prop]);
+      }
+    }
+
+    return { req, userId, invocationFn };
   }
 
-  async #setContextUser(context, userId, reg) {
+  async #setContextUser(context, userId, req) {
     const authInfo = await common.getAuthContext(context.tenant);
     context.user = new cds.User.Privileged({
       id: userId,
       authInfo,
       tokenInfo: authInfo?.token,
     });
-    if (reg) {
-      reg.user = context.user;
+    if (req) {
+      req.user = context.user;
     }
   }
 
   async processEvent(processContext, key, queueEntries, payload) {
     let statusTuple;
-    const { userId, invocationFn, reg } = this.#buildDispatchData(processContext, payload, { key, queueEntries });
+    const { userId, invocationFn, req } = this.#buildDispatchData(payload, { key, queueEntries });
     try {
-      await this.#setContextUser(processContext, userId, reg);
-      const result = await this.__srvUnboxed.tx(processContext)[invocationFn](reg);
+      await this.#setContextUser(processContext, userId, req);
+      const result = await this.__srvUnboxed.tx(processContext)[invocationFn](req);
       statusTuple = this.#determineResultStatus(result, queueEntries);
     } catch (err) {
       this.logger.error("error processing outboxed service call", err, {
@@ -374,7 +381,7 @@ class EventQueueGenericOutboxHandler extends EventQueueBaseClass {
       ]);
     }
 
-    await this.#publishFollowupEvents(processContext, reg, statusTuple);
+    await this.#publishFollowupEvents(processContext, req, statusTuple);
     return statusTuple;
   }
 

src/outbox/EventQueueGenericOutboxHandler.js

@@ -43,7 +43,7 @@ function outboxed(srv, customOpts) {
 
     const outboxOpts = config.getEventConfig(CDS_EVENT_TYPE, subType, srvConfig.namespace);
     const eventHeaders = getPropagatedHeaders(outboxOpts, req);
-    const contextProperties = getPropagatedContextProperties(outboxOpts, req);
+    const contextProperties = getPropagateContextProperties(outboxOpts, req);
     if (["persistent-outbox", "persistent-queue"].includes(outboxOpts.kind)) {
       await _mapToEventAndPublish(req, srvConfig.namespace, subType, eventHeaders, contextProperties);
       return;
@@ -81,8 +81,8 @@ const getPropagatedHeaders = (config, req) => {
   return Object.assign(propagateHeaders, req.headers);
 };
 
-const getPropagatedContextProperties = (config, req) => {
-  return config.propagatedContextProperties.reduce((properties, name) => {
+const getPropagateContextProperties = (config, req) => {
+  return config.propagateContextProperties.reduce((properties, name) => {
     if (name in req.tx.context) {
       properties[name] = req.tx.context[name];
     }

src/outbox/eventQueueAsOutbox.js

@@ -41,8 +41,8 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true ad-hoc events overwrite se
   "namespace": "default",
   "parallelEventProcessing": 5,
   "priority": "medium",
+  "propagateContextProperties": [],
   "propagateHeaders": [],
-  "propagatedContextProperties": [],
   "retryAttempts": 20,
   "retryFailedAfter": 300000,
   "retryOpenAfter": 300000,
@@ -70,8 +70,8 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true ad-hoc events overwrite se
   "namespace": "default",
   "parallelEventProcessing": 5,
   "priority": "medium",
+  "propagateContextProperties": [],
   "propagateHeaders": [],
-  "propagatedContextProperties": [],
   "retryAttempts": 20,
   "retryFailedAfter": 300000,
   "retryOpenAfter": 300000,
@@ -99,8 +99,8 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true config is not overwritten
   "namespace": "default",
   "parallelEventProcessing": 5,
   "priority": "medium",
+  "propagateContextProperties": [],
   "propagateHeaders": [],
-  "propagatedContextProperties": [],
   "retryAttempts": 20,
   "retryFailedAfter": 300000,
   "retryOpenAfter": 300000,
@@ -141,8 +141,8 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true custom options should win
   "namespace": "default",
   "parallelEventProcessing": 5,
   "priority": "medium",
+  "propagateContextProperties": [],
   "propagateHeaders": [],
-  "propagatedContextProperties": [],
   "retryAttempts": 20,
   "retryFailedAfter": 300000,
   "retryOpenAfter": 300000,
@@ -182,8 +182,8 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true map config to event-queue
   "namespace": "default",
   "parallelEventProcessing": 5,
   "priority": "medium",
+  "propagateContextProperties": [],
   "propagateHeaders": [],
-  "propagatedContextProperties": [],
   "retryAttempts": 20,
   "retryFailedAfter": 300000,
   "retryOpenAfter": 300000,
@@ -222,8 +222,8 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true periodic events inherit co
   "load": 60,
   "namespace": "default",
   "priority": "medium",
+  "propagateContextProperties": [],
   "propagateHeaders": [],
-  "propagatedContextProperties": [],
   "retryFailedAfter": 300000,
   "retryOpenAfter": 300000,
   "subType": "NotificationServicePeriodic.main",
@@ -260,8 +260,8 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true periodic events inherit co
   "load": 60,
   "namespace": "default",
   "priority": "medium",
+  "propagateContextProperties": [],
   "propagateHeaders": [],
-  "propagatedContextProperties": [],
   "retryFailedAfter": 300000,
   "retryOpenAfter": 300000,
   "subType": "NotificationServicePeriodic.main",
@@ -289,8 +289,8 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true periodic events inherit co
   "load": 60,
   "namespace": "default",
   "priority": "medium",
+  "propagateContextProperties": [],
   "propagateHeaders": [],
-  "propagatedContextProperties": [],
   "retryFailedAfter": 300000,
   "retryOpenAfter": 300000,
   "subType": "NotificationServicePeriodic.main",

test/__snapshots__/eventQueueOutbox.test.js.snap

@@ -16,8 +16,8 @@ exports[`initialize read yaml config file 1`] = `
     "namespace": "default",
     "parallelEventProcessing": 5,
     "priority": "medium",
+    "propagateContextProperties": [],
     "propagateHeaders": [],
-    "propagatedContextProperties": [],
     "retryFailedAfter": 300000,
     "retryOpenAfter": 300000,
     "subType": "Task",
@@ -37,8 +37,8 @@ exports[`initialize read yaml config file 1`] = `
     "namespace": "default",
     "parallelEventProcessing": 1,
     "priority": "medium",
+    "propagateContextProperties": [],
     "propagateHeaders": [],
-    "propagatedContextProperties": [],
     "retryFailedAfter": 300000,
     "retryOpenAfter": 300000,
     "subType": "CommunicationSystem",

test/__snapshots__/initialize.test.js.snap

@@ -12,7 +12,7 @@ class StandardService extends cds.Service {
           data: req.data,
           user: req.user.id,
           headers: req.headers,
-          features: req.features,
+          features: cds.context?.features,
         });
       });
     });

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

@@ -148,9 +148,9 @@ cds.env.requires.ContextProperties = {
   impl: path.join(basePath, "srv/service/standard-service.js"),
   queued: {
     kind: "persistent-queue",
-    propagatedContextProperties: ["notExisting"],
+    propagateContextProperties: ["notExisting"],
     events: {
-      main2: { propagatedContextProperties: ["features"] },
+      main2: { propagateContextProperties: ["features"] },
     },
   },
 };
@@ -2481,6 +2481,7 @@ describe("event-queue outbox", () => {
         });
         const payload = JSON.parse(event.payload);
         expect(payload.features).toEqual({ a: true, b: true });
+        await commitAndOpenNew();
         await processEventQueue(tx.context, "CAP_OUTBOX", `${service.name}.main2`);
         expect(loggerMock).actionCalled("main2", { features: { a: true, b: true } });
         await commitAndOpenNew();