fix redis namespaces

Author: soccermax

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "@cap-js-community/event-queue",
-  "version": "2.0.0-beta.2",
+  "version": "2.0.0-beta.3",
   "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

@@ -784,8 +784,8 @@ class Config {
     };
   }
 
-  get redisNamespace() {
-    return `${[REDIS_PREFIX, this.#namespace].filter((a) => a).join("##")}`;
+  redisNamespace(addPublishNamespace = true) {
+    return addPublishNamespace ? `${[REDIS_PREFIX, this.#namespace].filter((a) => a).join("##")}` : REDIS_PREFIX;
   }
 
   set insertEventsBeforeCommit(value) {

src/config.js

@@ -115,6 +115,8 @@ const initialize = async (options = {}) => {
   registerCdsShutdown();
   logger.info("event queue initialized", {
     registerAsEventProcessor: config.registerAsEventProcessor,
+    namespace: config.namespace,
+    processingNamespaces: config.processingNamespaces,
     processEventsAfterPublish: config.processEventsAfterPublish,
     multiTenancyEnabled: config.isMultiTenancy,
     redisEnabled: config.redisEnabled,

src/initialize.js

@@ -195,7 +195,7 @@ const _acquireLockDB = async (
 };
 
 const _generateKey = (context, tenantScoped, key) => {
-  const keyParts = [config.redisNamespace];
+  const keyParts = [config.redisNamespace()];
   tenantScoped && keyParts.push(context.tenant);
   keyParts.push(key);
   return `${keyParts.join("##")}`;

src/shared/distributedLock.js

@@ -13,13 +13,13 @@ const createMainClientAndConnect = async () => {
 
 const subscribeRedisChannel = async (channel, subscribeHandler) => {
   const redisClient = RedisClient.create(REDIS_CLIENT_NAME);
-  const channelWithNamespace = [config.redisNamespace, channel].join("##");
+  const channelWithNamespace = [config.redisNamespace(false), channel].join("##");
   return await redisClient.subscribeChannel(config.redisOptions, channelWithNamespace, subscribeHandler);
 };
 
-const publishMessage = async (channel, message) => {
+const publishMessage = async (channel, message, { addNamespace = true } = {}) => {
   const redisClient = RedisClient.create(REDIS_CLIENT_NAME);
-  const channelWithNamespace = [config.redisNamespace, channel].join("##");
+  const channelWithNamespace = [config.redisNamespace(addNamespace), channel].join("##");
   return await redisClient.publishMessage(config.redisOptions, channelWithNamespace, message);
 };
 

src/shared/redis/client.js

@@ -15,7 +15,7 @@ const attachRedisUnsubscribeHandler = () => {
       try {
         const { tenantId } = JSON.parse(messageData);
         cds.log(COMPONENT_NAME).info("received unsubscribe broadcast event", { tenantId });
-        this.executeUnsubscribeHandlers(tenantId);
+        config.executeUnsubscribeHandlers(tenantId);
       } catch (err) {
         cds.log(COMPONENT_NAME).error("could not parse unsubscribe broadcast event", err, {
           messageData,
@@ -27,9 +27,11 @@ const attachRedisUnsubscribeHandler = () => {
 
 const handleUnsubscribe = (tenantId) => {
   if (config.redisEnabled) {
-    client.publishMessage(REDIS_OFFBOARD_TENANT_CHANNEL, JSON.stringify({ tenantId })).catch((error) => {
-      cds.log(COMPONENT_NAME).error(`publishing tenant unsubscribe failed. tenantId: ${tenantId}`, error);
-    });
+    client
+      .publishMessage(REDIS_OFFBOARD_TENANT_CHANNEL, JSON.stringify({ tenantId }), { addNamespace: false })
+      .catch((error) => {
+        cds.log(COMPONENT_NAME).error(`publishing tenant unsubscribe failed. tenantId: ${tenantId}`, error);
+      });
   } else {
     config.executeUnsubscribeHandlers(tenantId);
   }