release: WRNexusJS 0.8.0

packages/pubsub/src/brokers.ts

@@ -0,0 +1,59 @@
+import type { Handler, PubSubDriver } from "./index.ts";
+
+export interface NatsClient {
+  publish(subject: string, data: Uint8Array): void | Promise<void>;
+  subscribe(subject: string, handler: (data: Uint8Array, subject: string) => void): () => void;
+  close?(): void | Promise<void>;
+}
+
+export function natsDriver(client: NatsClient): PubSubDriver {
+  const encoder = new TextEncoder();
+  const decoder = new TextDecoder();
+  return {
+    publish(topic, message) {
+      return client.publish(topic, encoder.encode(JSON.stringify(message)));
+    },
+    subscribe(pattern, handler) {
+      const subject =
+        pattern === "*" ? ">" : pattern.endsWith(":*") ? `${pattern.slice(0, -2)}:>` : pattern;
+      return client.subscribe(subject, (data, topic) => {
+        const raw = decoder.decode(data);
+        let value: unknown = raw;
+        try {
+          value = JSON.parse(raw);
+        } catch {
+          /* raw broker payload */
+        }
+        void handler(value, topic);
+      });
+    },
+    close: () => client.close?.(),
+  };
+}
+
+export interface KafkaClient {
+  publish(topic: string, value: string): void | Promise<void>;
+  subscribe(pattern: string, handler: (value: string, topic: string) => void): () => void;
+  close?(): void | Promise<void>;
+}
+
+/** Kafka adapter contract; consumer-group/rebalance policy remains owned by the selected client. */
+export function kafkaDriver(client: KafkaClient): PubSubDriver {
+  return {
+    publish(topic, message) {
+      return client.publish(topic, JSON.stringify(message));
+    },
+    subscribe(pattern, handler: Handler) {
+      return client.subscribe(pattern, (raw, topic) => {
+        let value: unknown = raw;
+        try {
+          value = JSON.parse(raw);
+        } catch {
+          /* raw broker payload */
+        }
+        void handler(value, topic);
+      });
+    },
+    close: () => client.close?.(),
+  };
+}

packages/pubsub/src/index.ts

@@ -15,11 +15,14 @@ export type Handler<T = unknown> = (message: T, topic: string) => void | Promise
 export interface PubSubDriver {
   publish(topic: string, message: unknown): void | Promise<void>;
   subscribe(pattern: string, handler: Handler): () => void;
+  close?(): void | Promise<void>;
 }
 
 export interface PubSub {
   publish<T = unknown>(topic: string, message: T): Promise<void>;
   subscribe<T = unknown>(pattern: string, handler: Handler<T>): () => void;
+  /** Stop new work, remove subscriptions, and close the backing driver. */
+  close(): Promise<void>;
 }
 
 function patternMatches(pattern: string, topic: string): boolean {
@@ -31,14 +34,19 @@ function patternMatches(pattern: string, topic: string): boolean {
 /** In-process pub/sub driver (default). */
 export function memoryDriver(): PubSubDriver {
   const subs = new Map<string, Set<Handler>>();
+  let closed = false;
   return {
-    publish(topic, message) {
+    async publish(topic, message) {
+      if (closed) throw new Error("WRN-PUBSUB-CLOSED: pubsub driver is closed");
+      const pending: Promise<void>[] = [];
       for (const [pattern, handlers] of subs) {
         if (!patternMatches(pattern, topic)) continue;
-        for (const handler of handlers) void handler(message, topic);
+        for (const handler of handlers) pending.push(Promise.resolve(handler(message, topic)));
       }
+      await Promise.all(pending);
     },
     subscribe(pattern, handler) {
+      if (closed) throw new Error("WRN-PUBSUB-CLOSED: pubsub driver is closed");
       let set = subs.get(pattern);
       if (!set) subs.set(pattern, (set = new Set()));
       set.add(handler);
@@ -47,19 +55,35 @@ export function memoryDriver(): PubSubDriver {
         if (!set!.size) subs.delete(pattern);
       };
     },
+    close() {
+      closed = true;
+      subs.clear();
+    },
   };
 }
 
 /** Create a pub/sub bus over a driver (in-memory by default). */
 export function createPubSub(driver: PubSubDriver = memoryDriver()): PubSub {
+  let closed = false;
   return {
     async publish(topic, message) {
+      if (closed) throw new Error("WRN-PUBSUB-CLOSED: pubsub is closed");
+      if (!topic.trim()) throw new TypeError("pubsub topic cannot be empty");
       await driver.publish(topic, message);
     },
     subscribe(pattern, handler) {
+      if (closed) throw new Error("WRN-PUBSUB-CLOSED: pubsub is closed");
+      if (!pattern.trim()) throw new TypeError("pubsub pattern cannot be empty");
       return driver.subscribe(pattern, handler as Handler);
     },
+    async close() {
+      if (closed) return;
+      closed = true;
+      await driver.close?.();
+    },
   };
 }
 export { createResilientPubSub, PresenceChannel } from "./resilient.ts";
 export type { MessageEnvelope, ResilientPubSubOptions, PresenceMember } from "./resilient.ts";
+export { natsDriver, kafkaDriver } from "./brokers.ts";
+export type { NatsClient, KafkaClient } from "./brokers.ts";

packages/pubsub/src/redis.ts

@@ -32,6 +32,15 @@ interface ParsedUrl {
   tls: boolean;
 }
 
+export interface RedisDriverOptions {
+  /** Maximum writes buffered while Redis is unavailable. Default 1000. */
+  maxPending?: number;
+  /** Initial reconnect delay. Doubles up to reconnectMaxDelayMs. Default 100. */
+  reconnectDelayMs?: number;
+  /** Maximum reconnect delay. Default 5000. */
+  reconnectMaxDelayMs?: number;
+}
+
 function parseUrl(url: string): ParsedUrl {
   const u = new URL(url);
   if (u.protocol !== "redis:" && u.protocol !== "rediss:") {
@@ -59,58 +68,101 @@ const isPattern = (p: string): boolean => p.includes("*");
  * Open a Redis TCP connection. `onReply` receives every parsed top-level reply
  * (used by the subscriber connection to dispatch message/pmessage pushes).
  */
-function connect(cfg: ParsedUrl, onReply?: (value: RespValue) => void): RedisConn {
+function connect(
+  cfg: ParsedUrl,
+  options: Required<RedisDriverOptions>,
+  onReply?: (value: RespValue) => void,
+  onReconnect?: () => void,
+): RedisConn {
   let socket: { write(data: Uint8Array): void; end(): void } | null = null;
   const pending: Uint8Array[] = [];
-  const maxPending = 1000;
   let buf: Uint8Array<ArrayBufferLike> = new Uint8Array(0);
+  let closed = false;
+  let connecting = false;
+  let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
+  let reconnectAttempt = 0;
+  let openedOnce = false;
 
   // Bun.connect is available in the Bun runtime.
   const Bun = (globalThis as { Bun?: { connect: (opts: unknown) => Promise<unknown> } }).Bun;
   if (!Bun?.connect) throw new Error("redisDriver requires the Bun runtime (Bun.connect).");
 
-  void Bun.connect({
-    hostname: cfg.host,
-    port: cfg.port,
-    tls: cfg.tls,
-    socket: {
-      open(s: { write(data: Uint8Array): void; end(): void }) {
-        socket = s;
-        if (cfg.password) s.write(encodeCommand(["AUTH", cfg.password]));
-        if (cfg.db) s.write(encodeCommand(["SELECT", String(cfg.db)]));
-        for (const p of pending) s.write(p);
-        pending.length = 0;
+  const scheduleReconnect = () => {
+    if (closed || reconnectTimer) return;
+    const delay = Math.min(
+      options.reconnectMaxDelayMs,
+      options.reconnectDelayMs * 2 ** reconnectAttempt++,
+    );
+    reconnectTimer = setTimeout(() => {
+      reconnectTimer = null;
+      start();
+    }, delay);
+  };
+
+  const start = () => {
+    if (closed || connecting) return;
+    connecting = true;
+    void Bun.connect({
+      hostname: cfg.host,
+      port: cfg.port,
+      tls: cfg.tls,
+      socket: {
+        open(s: { write(data: Uint8Array): void; end(): void }) {
+          connecting = false;
+          socket = s;
+          reconnectAttempt = 0;
+          if (cfg.password) s.write(encodeCommand(["AUTH", cfg.password]));
+          if (cfg.db) s.write(encodeCommand(["SELECT", String(cfg.db)]));
+          if (openedOnce) onReconnect?.();
+          openedOnce = true;
+          for (const p of pending) s.write(p);
+          pending.length = 0;
+        },
+        data(_s: unknown, chunk: Uint8Array) {
+          buf = buf.length ? concat([buf, chunk]) : chunk;
+          for (;;) {
+            const r = parseReply(buf);
+            if (!r) break;
+            buf = buf.slice(r.next);
+            onReply?.(r.value);
+          }
+        },
+        error() {
+          socket = null;
+          connecting = false;
+          scheduleReconnect();
+        },
+        close() {
+          socket = null;
+          connecting = false;
+          scheduleReconnect();
+        },
       },
-      data(_s: unknown, chunk: Uint8Array) {
-        buf = buf.length ? concat([buf, chunk]) : chunk;
-        for (;;) {
-          const r = parseReply(buf);
-          if (!r) break;
-          buf = buf.slice(r.next);
-          onReply?.(r.value);
-        }
-      },
-      error() {
-        /* connection error — writes silently no-op until reconnect */
-      },
-      close() {
-        socket = null;
-      },
-    },
-  });
+    }).catch(() => {
+      connecting = false;
+      scheduleReconnect();
+    });
+  };
+  start();
 
   return {
     send(bytes) {
       if (socket) socket.write(bytes);
       else {
-        if (pending.length >= maxPending) {
+        if (closed) throw new Error("Redis pubsub connection is closed");
+        if (pending.length >= options.maxPending) {
           throw new Error("Redis connection is unavailable and its pending write queue is full");
         }
         pending.push(bytes);
       }
     },
     close() {
+      closed = true;
+      if (reconnectTimer) clearTimeout(reconnectTimer);
+      reconnectTimer = null;
+      pending.length = 0;
       socket?.end();
+      socket = null;
     },
   };
 }
@@ -119,8 +171,22 @@ function connect(cfg: ParsedUrl, onReply?: (value: RespValue) => void): RedisCon
  * A cross-process pub/sub driver backed by Redis. `url` defaults to
  * `$REDIS_URL` or `redis://localhost:6379`.
  */
-export function redisDriver(url?: string): PubSubDriver & { close(): void } {
+export function redisDriver(
+  url?: string,
+  options: RedisDriverOptions = {},
+): PubSubDriver & { close(): void } {
   const cfg = parseUrl(url ?? getEnv("REDIS_URL") ?? "redis://localhost:6379");
+  const resolved = {
+    maxPending: positiveInteger(options.maxPending ?? 1000, "Redis maxPending"),
+    reconnectDelayMs: positiveInteger(options.reconnectDelayMs ?? 100, "Redis reconnectDelayMs"),
+    reconnectMaxDelayMs: positiveInteger(
+      options.reconnectMaxDelayMs ?? 5000,
+      "Redis reconnectMaxDelayMs",
+    ),
+  };
+  if (resolved.reconnectMaxDelayMs < resolved.reconnectDelayMs) {
+    throw new RangeError("Redis reconnectMaxDelayMs must be >= reconnectDelayMs");
+  }
   const subs = new Map<string, Set<Handler>>();
   const redisSubscribed = new Set<string>();
 
@@ -136,13 +202,22 @@ export function redisDriver(url?: string): PubSubDriver & { close(): void } {
     for (const handler of handlers) void handler(message, topic);
   };
 
-  const subConn = connect(cfg, (value) => {
-    if (!Array.isArray(value)) return;
-    const kind = value[0];
-    if (kind === "message") dispatch(String(value[1]), String(value[1]), String(value[2]));
-    else if (kind === "pmessage") dispatch(String(value[1]), String(value[2]), String(value[3]));
-  });
-  const pubConn = connect(cfg);
+  const subConn = connect(
+    cfg,
+    resolved,
+    (value) => {
+      if (!Array.isArray(value)) return;
+      const kind = value[0];
+      if (kind === "message") dispatch(String(value[1]), String(value[1]), String(value[2]));
+      else if (kind === "pmessage") dispatch(String(value[1]), String(value[2]), String(value[3]));
+    },
+    () => {
+      for (const pattern of redisSubscribed) {
+        subConn.send(encodeCommand([isPattern(pattern) ? "PSUBSCRIBE" : "SUBSCRIBE", pattern]));
+      }
+    },
+  );
+  const pubConn = connect(cfg, resolved);
 
   return {
     publish(topic, message) {
@@ -175,6 +250,13 @@ export function redisDriver(url?: string): PubSubDriver & { close(): void } {
   };
 }
 
+function positiveInteger(value: number, label: string): number {
+  if (!Number.isInteger(value) || value < 1) {
+    throw new RangeError(`${label} must be a positive integer`);
+  }
+  return value;
+}
+
 function getEnv(key: string): string | undefined {
   return (globalThis as { process?: { env?: Record<string, string | undefined> } }).process?.env?.[
     key

packages/pubsub/src/resilient.ts

@@ -26,9 +26,11 @@ export function createResilientPubSub(
   if (!Number.isFinite(retryDelayMs) || retryDelayMs < 0) {
     throw new RangeError("pubsub retryDelayMs must be a non-negative number");
   }
+  let closed = false;
 
   return {
     async publish(topic, message) {
+      if (closed) throw new Error("WRN-PUBSUB-CLOSED: pubsub is closed");
       if (!topic.trim()) throw new TypeError("pubsub topic cannot be empty");
       const envelope: MessageEnvelope = {
         id: crypto.randomUUID(),
@@ -55,6 +57,7 @@ export function createResilientPubSub(
     },
 
     subscribe<T>(pattern: string, handler: Handler<T>) {
+      if (closed) throw new Error("WRN-PUBSUB-CLOSED: pubsub is closed");
       if (!pattern.trim()) throw new TypeError("pubsub pattern cannot be empty");
       return driver.subscribe(pattern, async (value, topic) => {
         const envelope = value as MessageEnvelope<T>;
@@ -65,6 +68,11 @@ export function createResilientPubSub(
         }
       });
     },
+    async close() {
+      if (closed) return;
+      closed = true;
+      await driver.close?.();
+    },
   };
 }