feat: implement zero-dependency real-time event bus in AuthSdk

Added a lightweight, zero-dependency event bus to the `AuthSdk` to listen for Valkey RESP3 push invalidation events and emit them to registered listeners.
The implementation properly captures errors from synchronous and asynchronous listeners, ensuring they do not crash the primary SDK listener loop. Extracted the Valkey event processor to improve hermetic testability.

Co-authored-by: mrteye <1945243+mrteye@users.noreply.github.com>

sdk/mod.test.ts

@@ -9,3 +9,80 @@ Deno.test("AuthSdk - initializes with config", () => {
   assertEquals(typeof sdk.validateSession, "function");
   assertEquals(typeof sdk.requireAuth, "function");
 });
+
+Deno.test("AuthSdk - Event Bus: registers, receives, and unregisters events", () => {
+  const sdk = createAuthSdk({
+    authApiUrl: "http://localhost:8000",
+    timeoutMs: 3000,
+  });
+
+  const receivedTokens: string[] = [];
+  const handler = (token: string) => {
+    receivedTokens.push(token);
+  };
+
+  sdk.on("invalidate", handler);
+
+  // Simulate Valkey push event
+  (sdk as any)._handleValkeyPush(["invalidate", ["token1", "token2"]]);
+
+  assertEquals(receivedTokens, ["token1", "token2"]);
+
+  sdk.off("invalidate", handler);
+
+  // Simulate another push event
+  (sdk as any)._handleValkeyPush(["invalidate", ["token3"]]);
+
+  // The tokens should not be added because handler was unregistered
+  assertEquals(receivedTokens, ["token1", "token2"]);
+});
+
+Deno.test("AuthSdk - Event Bus: failing async handlers do not crash bus", async () => {
+  const sdk = createAuthSdk({
+    authApiUrl: "http://localhost:8000",
+    timeoutMs: 3000,
+  });
+
+  const receivedTokens: string[] = [];
+
+  const failingHandlerAsync = async (token: string) => {
+    throw new Error(`Simulated async failure for ${token}`);
+  };
+
+  const failingHandlerSync = (token: string) => {
+    throw new Error(`Simulated sync failure for ${token}`);
+  };
+
+  const successHandler = (token: string) => {
+    receivedTokens.push(token);
+  };
+
+  sdk.on("invalidate", failingHandlerAsync);
+  sdk.on("invalidate", failingHandlerSync);
+  sdk.on("invalidate", successHandler);
+
+  // Stub console.error to avoid test noise
+  const originalConsoleError = console.error;
+  let loggedErrors = 0;
+  console.error = (...args: any[]) => {
+    const msg = args.join(" ");
+    if (msg.includes("Error in 'invalidate' listener")) {
+      loggedErrors++;
+    }
+  };
+
+  try {
+    (sdk as any)._handleValkeyPush(["invalidate", ["token4"]]);
+
+    // Give microtasks a chance to process the async rejection
+    await new Promise((resolve) => setTimeout(resolve, 0));
+
+    // The successful handler should have run despite the failures
+    assertEquals(receivedTokens, ["token4"]);
+
+    // Both the sync and async errors should have been caught and logged
+    assertEquals(loggedErrors, 2);
+  } finally {
+    console.error = originalConsoleError;
+  }
+});

sdk/mod.ts

@@ -6,6 +6,16 @@ import { createClient } from "npm:@connectrpc/connect@^1.4.0";
 import { createConnectTransport } from "npm:@connectrpc/connect-node@^1.4.0";
 import { AuthService } from "./gen/auth_connect.ts";
 
+/**
+ * Supported event types for the Auth SDK.
+ */
+export type AuthSdkEvent = "invalidate";
+
+/**
+ * Handler type for invalidation events.
+ */
+export type InvalidationHandler = (token: string) => void | Promise<void>;
+
 /**
  * Configuration options for the Auth SDK.
  */
@@ -58,6 +68,7 @@ export class AuthSdk {
   private l1Cache: Map<string, SessionData>;
   private valkeyClient: Redis | null = null;
   private grpcClient: any;
+  private listeners: Map<string, Set<InvalidationHandler>> = new Map();
 
   constructor(config: AuthSdkConfig) {
     this.config = config;
@@ -84,6 +95,78 @@ export class AuthSdk {
     this.grpcClient = createClient(AuthService, transport);
   }
 
+  /**
+   * Registers a callback to be invoked when a specific event occurs.
+   *
+   * @param event The event name (e.g., "invalidate").
+   * @param handler The callback function.
+   */
+  on(event: AuthSdkEvent, handler: InvalidationHandler): void {
+    let eventListeners = this.listeners.get(event);
+    if (!eventListeners) {
+      eventListeners = new Set();
+      this.listeners.set(event, eventListeners);
+    }
+    eventListeners.add(handler);
+  }
+
+  /**
+   * Unregisters a previously registered callback.
+   *
+   * @param event The event name (e.g., "invalidate").
+   * @param handler The callback function to remove.
+   */
+  off(event: AuthSdkEvent, handler: InvalidationHandler): void {
+    const eventListeners = this.listeners.get(event);
+    if (eventListeners) {
+      eventListeners.delete(handler);
+      if (eventListeners.size === 0) {
+        this.listeners.delete(event);
+      }
+    }
+  }
+
+  /**
+   * Internal method to emit an event safely to all registered listeners.
+   */
+  private emit(event: AuthSdkEvent, token: string): void {
+    const eventListeners = this.listeners.get(event);
+    if (eventListeners) {
+      for (const handler of eventListeners) {
+        try {
+          const result = handler(token);
+          if (result instanceof Promise) {
+            result.catch((err) => {
+              console.error("[AuthSdk] Error in 'invalidate' listener:", err);
+            });
+          }
+        } catch (err) {
+          console.error("[AuthSdk] Error in 'invalidate' listener:", err);
+        }
+      }
+    }
+  }
+
+  /**
+   * Internal handler for RESP3 push invalidation messages.
+   * Exposed internally/to tests via the Valkey event listener.
+   */
+  _handleValkeyPush(msg: unknown): void {
+    if (
+      Array.isArray(msg) && msg.length >= 2 && msg[0] === "invalidate"
+    ) {
+      const keysToInvalidate = msg[1];
+      if (Array.isArray(keysToInvalidate)) {
+        for (const key of keysToInvalidate) {
+          // SIDE EFFECT: Delete the invalidated key from the local Map
+          this.l1Cache.delete(key);
+          // Emit invalidate event for real-time consumers
+          this.emit("invalidate", key);
+        }
+      }
+    }
+  }
+
   private initValkeyClient() {
     this.valkeyClient = new Redis(this.config.valkeyUrl!, {
       enableOfflineQueue: false,
@@ -102,19 +185,7 @@ export class AuthSdk {
     });
 
     // Listen for RESP3 push invalidation messages
-    this.valkeyClient.on("push", (msg: unknown) => {
+    this.valkeyClient.on("push", (msg: unknown) => this._handleValkeyPush(msg));
-      if (
-        Array.isArray(msg) && msg.length >= 2 && msg[0] === "invalidate"
-      ) {
-        const keysToInvalidate = msg[1];
-        if (Array.isArray(keysToInvalidate)) {
-          for (const key of keysToInvalidate) {
-            // SIDE EFFECT: Delete the invalidated key from the local Map
-            this.l1Cache.delete(key);
-          }
-        }
-      }
-    });
 
     this.valkeyClient.on("error", (err: unknown) => {
       console.error("Valkey SDK Client error:", err);