Merge branch 'fb/propagateHeaders' into fb/adjustPayloadReturn

# Conflicts:
#	CHANGELOG.md
#	package.json
#	src/config.js
#	src/outbox/eventQueueAsOutbox.js
#	test/asset/outboxProject/srv/service/standard-service.js

Author: soccermax

CHANGELOG.md

@@ -5,20 +5,20 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
-## v1.12.0 - 2025-XX-XX
+## v1.11.1 - 2025-11-13
 
 ### Added
 
-- fully support cds.queued
+- [Event Configuration] `propagateHeaders` allows forwarding headers from the original CDS context to the outbox call by specifying their names in an array.
 
-## v1.11.0 - 2025-07-09
+## v1.11.0 - 2025-10-16
 
 ### Added
 
 - Added `authInfo` to cds.User as CDS 9.3 deprecated `tokenInfo`.
 - Disable the automatic processing of suspended tenants. Can be turned off using `disableProcessingOfSuspendedTenants`.
 
-## v1.10.10 - 2025-07-09
+## v1.10.10 - 2025-07-10
 
 ### Fixed
 

docs/configure-event/index.md

@@ -48,6 +48,7 @@ they should be processed.
 | load                          | Indicates the load of the event, affecting the processing concurrency.                                                                                                                                                                                                                                                                                                                                                                    | 1               |
 | retryAttempts                 | Number of retry attempts for failed events. Set to `-1` for infinite retries.                                                                                                                                                                                                                                                                                                                                                             | 3               |
 | processAfterCommit            | Indicates whether an event is processed immediately after the transaction, in which the event was written, has been committed.                                                                                                                                                                                                                                                                                                            | true            |
+| propagateHeaders              | Specifies which headers from the original CDS context should be forwarded to the outbox call. Provide an array of header names to propagate.                                                                                                                                                                                                                                                                                              | []              |
 | parallelEventProcessing       | Number of events of the same type and subType that can be processed in parallel. The maximum limit is 10.                                                                                                                                                                                                                                                                                                                                 | 1               |
 | transactionMode               | Specifies the transaction mode for the event. For allowed values, refer to [Transaction Handling](/event-queue/transaction-handling/#transaction-modes).                                                                                                                                                                                                                                                                                  | isolated        |
 | selectMaxChunkSize            | Number of events selected in a single batch. Set `checkForNextChunk` to `true` if you want to check for more available events after processing a batch.                                                                                                                                                                                                                                                                                   | 100             |

src/config.js

@@ -60,6 +60,7 @@ const ALLOWED_EVENT_OPTIONS_AD_HOC = [
   "processAfterCommit",
   "checkForNextChunk",
   "retryFailedAfter",
+  "propagateHeaders",
   "retryOpenAfter",
   "multiInstanceProcessing",
   "kind",
@@ -251,6 +252,7 @@ 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,
     });
@@ -436,6 +438,7 @@ class Config {
     event._appInstancesMap = event.appInstances
       ? Object.fromEntries(new Map(event.appInstances.map((a) => [a, true])))
       : null;
+    event.propagateHeaders = event.propagateHeaders ?? [];
   }
 
   #basicEventValidation(event) {

src/outbox/eventQueueAsOutbox.js

@@ -51,14 +51,16 @@ function outboxed(srv, customOpts) {
       customOpts || {}
     );
     config.addCAPOutboxEventBase(srv.name, outboxOpts);
-    const specificSettings = config.getCdsOutboxEventSpecificConfig(srv.name, req.event);
-    if (specificSettings) {
+    const hasSpecificSettings = !!config.getCdsOutboxEventSpecificConfig(srv.name, req.event);
+    if (hasSpecificSettings) {
       outboxOpts = config.addCAPOutboxEventSpecificAction(srv.name, req.event);
     }
-
-    const namespace = (specificSettings ?? outboxOpts).namespace ?? config.namespace;
+    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;
     if (["persistent-outbox", "persistent-queue"].includes(outboxOpts.kind)) {
-      await _mapToEventAndPublish(context, srv.name, req, !!specificSettings, namespace);
+      await _mapToEventAndPublish(context, subType , req, eventHeaders, namespace);
       return;
     }
     context.on("succeeded", async () => {
@@ -84,32 +86,43 @@ function unboxed(srv) {
   return srv[UNBOXED] || srv;
 }
 
-const _mapToEventAndPublish = async (context, name, req, actionSpecific, namespace) => {
+const getPropagatedHeaders = (config, req) => {
+  const propagateHeaders = config.propagateHeaders.reduce((headers, headerName) => {
+    if (headerName in req.tx.context.headers) {
+      headers[headerName] = req.tx.context.headers[headerName];
+    }
+    return headers;
+  }, {});
+  return Object.assign(propagateHeaders, req.headers);
+};
+
+const _mapToEventAndPublish = async (context, subType, req, eventHeaders, namespace) => {
   const eventQueueSpecificValues = {};
   for (const header in req.headers ?? {}) {
     for (const field of EVENT_QUEUE_SPECIFIC_FIELDS) {
       if (header.toLocaleLowerCase() === `x-eventqueue-${field.toLocaleLowerCase()}`) {
         eventQueueSpecificValues[field] = req.headers[header];
-        delete req.headers[header];
+        delete eventHeaders[header];
         break;
       }
     }
   }
+
   const event = {
     contextUser: context.user.id,
     ...(req._fromSend || (req.reply && { _fromSend: true })), // send or emit
     ...(req.inbound && { inbound: req.inbound }),
     ...(req.event && { event: req.event }),
     ...(req.data && { data: req.data }),
-    ...(req.headers && { headers: req.headers }),
+    ...(eventHeaders && { headers: eventHeaders }),
     ...(req.query && { query: req.query }),
   };
 
   await publishEvent(
     cds.tx(context),
     {
       type: CDS_EVENT_TYPE,
-      subType: actionSpecific ? [name, req.event].join(".") : name,
+      subType,
       payload: JSON.stringify(event),
       namespace: eventQueueSpecificValues.namespace ?? namespace,
       ...eventQueueSpecificValues,

test/__snapshots__/eventQueueOutbox.test.js.snap

@@ -41,6 +41,7 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true ad-hoc events overwrite se
   "namespace": "default",
   "parallelEventProcessing": 5,
   "priority": "medium",
+  "propagateHeaders": [],
   "retryAttempts": 20,
   "selectMaxChunkSize": 10,
   "subType": "NotificationServicePeriodic.action",
@@ -66,6 +67,7 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true ad-hoc events overwrite se
   "namespace": "default",
   "parallelEventProcessing": 5,
   "priority": "medium",
+  "propagateHeaders": [],
   "retryAttempts": 20,
   "selectMaxChunkSize": 10,
   "subType": "NotificationServicePeriodic.action",
@@ -91,6 +93,7 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true custom options should win
   "namespace": "default",
   "parallelEventProcessing": 5,
   "priority": "medium",
+  "propagateHeaders": [],
   "retryAttempts": 20,
   "selectMaxChunkSize": 10,
   "subType": "NotificationServiceOutboxedByConfig",
@@ -128,6 +131,7 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true map config to event-queue
   "namespace": "default",
   "parallelEventProcessing": 5,
   "priority": "medium",
+  "propagateHeaders": [],
   "retryAttempts": 20,
   "selectMaxChunkSize": 10,
   "subType": "NotificationService",
@@ -142,6 +146,7 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true option to use eventQueue.u
     "subject": "subject",
     "to": "to",
   },
+  "headers": {},
   "subType": "NotificationServiceOutboxedByConfigUserId",
   "user": "dummyTestUser",
 }
@@ -163,6 +168,7 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true periodic events inherit co
   "load": 60,
   "namespace": "default",
   "priority": "medium",
+  "propagateHeaders": [],
   "subType": "NotificationServicePeriodic.main",
   "transactionMode": "isolated",
   "type": "CAP_OUTBOX_PERIODIC",
@@ -197,6 +203,7 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true periodic events inherit co
   "load": 60,
   "namespace": "default",
   "priority": "medium",
+  "propagateHeaders": [],
   "subType": "NotificationServicePeriodic.main",
   "transactionMode": "alwaysRollback",
   "type": "CAP_OUTBOX_PERIODIC",
@@ -222,6 +229,7 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true periodic events inherit co
   "load": 60,
   "namespace": "default",
   "priority": "medium",
+  "propagateHeaders": [],
   "subType": "NotificationServicePeriodic.main",
   "transactionMode": "isolated",
   "type": "CAP_OUTBOX_PERIODIC",
@@ -309,6 +317,7 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true req.data should be stored
     "subject": "subject",
     "to": "to",
   },
+  "headers": {},
   "subType": "NotificationService",
   "user": "testUser",
 }
@@ -334,6 +343,7 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true req.data should be stored
     "subject": "subject",
     "to": "to",
   },
+  "headers": {},
   "subType": "NotificationService",
   "user": "testUser",
 }
@@ -385,6 +395,7 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true should store correct user
     "subject": "subject",
     "to": "to",
   },
+  "headers": {},
   "subType": "NotificationService",
   "user": "badman",
 }
@@ -407,6 +418,7 @@ exports[`event-queue outbox monkeyPatchCAPOutbox=true should work for outboxed s
   "namespace": "default",
   "parallelEventProcessing": 5,
   "priority": "medium",
+  "propagateHeaders": [],
   "retryAttempts": 20,
   "selectMaxChunkSize": 10,
   "subType": "NotificationServiceOutboxedByConfig",

test/__snapshots__/initialize.test.js.snap

@@ -16,6 +16,7 @@ exports[`initialize read yaml config file 1`] = `
     "namespace": "default",
     "parallelEventProcessing": 5,
     "priority": "medium",
+    "propagateHeaders": [],
     "subType": "Task",
     "type": "Notifications",
   },
@@ -33,6 +34,7 @@ exports[`initialize read yaml config file 1`] = `
     "namespace": "default",
     "parallelEventProcessing": 1,
     "priority": "medium",
+    "propagateHeaders": [],
     "subType": "CommunicationSystem",
     "type": "BusinessLogs",
   },

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

@@ -10,6 +10,7 @@ class NotificationService extends cds.Service {
         data: req.data,
         user: req.user.id,
         subType: req.eventQueue?.processor.eventSubType,
+        headers: req.headers,
       });
     });
 

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

@@ -9,6 +9,7 @@ class StandardService extends cds.Service {
       cds.log(this.name).info(req.event, {
         data: req.data,
         user: req.user.id,
+        headers: req.headers,
       });
     });
 
@@ -19,6 +20,37 @@ class StandardService extends cds.Service {
       });
     });
 
+    this.on("callNextOutbox", async (req) => {
+      const outboxedService = cds.outboxed(this).tx(req);
+      await outboxedService.send(
+        "main",
+        {
+          to: "to",
+          subject: "subject",
+          body: "body",
+        },
+        {
+          "x-eventqueue-startAfter": new Date(),
+        }
+      );
+    });
+
+    this.on("callNextOutboxMix", async (req) => {
+      const outboxedService = cds.outboxed(this).tx(req);
+      await outboxedService.send(
+        "main",
+        {
+          to: "to",
+          subject: "subject",
+          body: "body",
+        },
+        {
+          customHeader: 456,
+          myNextHeader: 123,
+        }
+      );
+    });
+
     this.on("plainStatus", (req) => {
       cds.log(this.name).info(req.event, {
         data: req.data,

test/eventQueueOutbox.test.js

@@ -98,6 +98,7 @@ cds.env.requires.StandardService = {
   impl: path.join(basePath, "srv/service/standard-service.js"),
   outbox: {
     kind: "persistent-outbox",
+    propagateHeaders: ["authId", "customHeader"],
     events: {
       timeBucketAction: {
         timeBucket: "*/60 * * * * *",
@@ -672,6 +673,62 @@ describe("event-queue outbox", () => {
       expect(loggerMock.callsLengths().error).toEqual(0);
     });
 
+    it("should correctly propagate headers: send/emit event from CAP service with existing headers", async () => {
+      const service = await cds.connect.to("StandardService");
+      const outboxedService = cds.outboxed(service).tx(context);
+      const headers = { customHeader: 123, authId: 123 };
+      await outboxedService.send(
+        "callNextOutbox",
+        {
+          to: "to",
+          subject: "subject",
+          body: "body",
+        },
+        { ...headers, xyz: 123 }
+      );
+      await commitAndOpenNew();
+      await testHelper.selectEventQueueAndExpectOpen(tx, { expectedLength: 1 });
+      expect(loggerMock).not.sendFioriActionCalled();
+      await processEventQueue(tx.context, "CAP_OUTBOX", service.name);
+      await commitAndOpenNew();
+      await processEventQueue(tx.context, "CAP_OUTBOX", service.name);
+      await testHelper.selectEventQueueAndExpectDone(tx, { expectedLength: 2 });
+      expect(loggerMock).actionCalled("main", {
+        headers,
+      });
+      expect(loggerMock.callsLengths().error).toEqual(0);
+    });
+
+    it("should correctly propagate headers: should mix headers from both calls", async () => {
+      const service = await cds.connect.to("StandardService");
+      const outboxedService = cds.outboxed(service).tx(context);
+      const headers = { customHeader: 123, authId: 123 };
+      await outboxedService.send(
+        "callNextOutboxMix",
+        {
+          to: "to",
+          subject: "subject",
+          body: "body",
+        },
+        { ...headers, xyz: 123 }
+      );
+      await commitAndOpenNew();
+      await testHelper.selectEventQueueAndExpectOpen(tx, { expectedLength: 1 });
+      expect(loggerMock).not.sendFioriActionCalled();
+      await processEventQueue(tx.context, "CAP_OUTBOX", service.name);
+      await commitAndOpenNew();
+      await processEventQueue(tx.context, "CAP_OUTBOX", service.name);
+      await testHelper.selectEventQueueAndExpectDone(tx, { expectedLength: 2 });
+      expect(loggerMock).actionCalled("main", {
+        headers: {
+          customHeader: 456,
+          authId: 123,
+          myNextHeader: 123,
+        },
+      });
+      expect(loggerMock.callsLengths().error).toEqual(0);
+    });
+
     describe("not connected service should lazily connect and create configuration", () => {
       beforeEach(() => {
         eventQueue.config.removeEvent("CAP_OUTBOX", "NotificationService");
@@ -1996,7 +2053,7 @@ expect.extend({
     };
   },
   actionCalled: (loggerMock, actionName, properties = {}) => {
-    const call = loggerMock.calls().info.find((c) => c[0] === actionName);
+    const call = loggerMock.calls().info.find((c) => actionName === c[0]);
     if (!call) {
       return {
         message: () => `action not called! name: ${actionName}`,