feat(redis): add channel prefix and error handling

src/event-target-redis.ts

@@ -41,6 +41,7 @@ export type CreateRedisEventTargetArgs = {
     stringify: (message: unknown) => string;
     parse: (message: string) => unknown;
   };
+  prefix?: string;
 };
 
 export function createRedisEventTarget<TEvent extends TypedEvent>(
@@ -49,6 +50,7 @@ export function createRedisEventTarget<TEvent extends TypedEvent>(
   const { publishClient, subscribeClient } = args;
 
   const serializer = args.serializer ?? JSON;
+  const prefix = args.prefix ?? "";
 
   const callbacksForTopic = new Map<string, Set<EventListener>>();
 
@@ -58,7 +60,15 @@ export function createRedisEventTarget<TEvent extends TypedEvent>(
       return;
     }
 
-    const envelope = serializer.parse(message) as EventEnvelope;
+    let envelope: EventEnvelope;
+    try {
+      envelope = serializer.parse(message) as EventEnvelope;
+    } catch {
+      console.warn(
+        `Failed to parse message on channel "${channel}": ${message}`,
+      );
+      return;
+    }
     const event = new CustomEvent(channel, {
       detail: envelope,
     }) as TEvent;
@@ -70,18 +80,20 @@ export function createRedisEventTarget<TEvent extends TypedEvent>(
   (subscribeClient as Redis).on("message", onMessage);
 
   function addCallback(topic: string, callback: EventListener) {
-    let callbacks = callbacksForTopic.get(topic);
+    const prefixedTopic = prefix + topic;
+    let callbacks = callbacksForTopic.get(prefixedTopic);
     if (callbacks === undefined) {
       callbacks = new Set();
-      callbacksForTopic.set(topic, callbacks);
+      callbacksForTopic.set(prefixedTopic, callbacks);
 
-      subscribeClient.subscribe(topic);
+      subscribeClient.subscribe(prefixedTopic);
     }
     callbacks.add(callback);
   }
 
   function removeCallback(topic: string, callback: EventListener) {
-    const callbacks = callbacksForTopic.get(topic);
+    const prefixedTopic = prefix + topic;
+    const callbacks = callbacksForTopic.get(prefixedTopic);
     if (callbacks === undefined) {
       return;
     }
@@ -89,8 +101,8 @@ export function createRedisEventTarget<TEvent extends TypedEvent>(
     if (callbacks.size > 0) {
       return;
     }
-    callbacksForTopic.delete(topic);
-    subscribeClient.unsubscribe(topic);
+    callbacksForTopic.delete(prefixedTopic);
+    subscribeClient.unsubscribe(prefixedTopic);
   }
 
   return {
@@ -103,7 +115,7 @@ export function createRedisEventTarget<TEvent extends TypedEvent>(
     },
     dispatchEvent(event: TEvent) {
       publishClient.publish(
-        event.type,
+        prefix + event.type,
         serializer.stringify(event.detail),
       );
       return true;