first commit

packages/pubsub/README.md

@@ -0,0 +1,129 @@
+# @wrnexus/pubsub
+
+> Topic-based publish/subscribe with a pluggable driver — in-process by default, Redis for cross-process messaging.
+
+Part of the **WrNexus** framework — an SSR-first, Bun-native full-stack web framework.
+
+## Overview
+
+`@wrnexus/pubsub` is a small server-side pub/sub bus. You publish messages to a
+topic and subscribe with topic patterns; handlers fire for matching topics. The
+default driver keeps everything in-process, and you can swap in the Redis driver
+(`@wrnexus/pubsub/redis`) to fan messages out across processes or hosts. It also
+backs `@wrnexus/core`'s realtime bridge for horizontal scaling.
+
+## Installation
+
+```bash
+bun add @wrnexus/pubsub
+```
+
+> Private package — the machine must be authenticated to the `wrnexus` npm org
+> (a read token in `~/.npmrc`). Requires **Bun** (Node is not supported).
+
+## API
+
+### `createPubSub(driver?): PubSub`
+
+Creates a bus over a driver. Defaults to `memoryDriver()` (in-process).
+
+```ts
+interface PubSub {
+  publish<T = unknown>(topic: string, message: T): Promise<void>;
+  subscribe<T = unknown>(pattern: string, handler: Handler<T>): () => void;
+}
+
+type Handler<T = unknown> = (message: T, topic: string) => void | Promise<void>;
+```
+
+- `publish(topic, message)` — resolves once the driver has dispatched the message.
+- `subscribe(pattern, handler)` — returns an unsubscribe function.
+
+### Pattern matching
+
+Subscription patterns match in three ways:
+
+- **Exact** — `"order:created"` matches only that topic.
+- **Prefix** — `"order:*"` matches any topic starting with `"order:"`.
+- **Everything** — `"*"` matches all topics.
+
+### `memoryDriver(): PubSubDriver`
+
+The default in-process driver. Handlers are invoked synchronously (fire-and-forget
+for async handlers) whenever a published topic matches a registered pattern.
+
+```ts
+interface PubSubDriver {
+  publish(topic: string, message: unknown): void | Promise<void>;
+  subscribe(pattern: string, handler: Handler): () => void;
+}
+```
+
+### `@wrnexus/pubsub/redis` — `redisDriver(url?)`
+
+A cross-process driver backed by Redis. It speaks RESP over a raw TCP socket via
+`Bun.connect`, so it adds **no npm dependency**. `url` defaults to `$REDIS_URL`,
+then `redis://localhost:6379`. The URL may carry a password and a database index
+(e.g. `redis://:secret@host:6379/2`).
+
+```ts
+function redisDriver(url?: string): PubSubDriver & { close(): void };
+```
+
+- Exact topics use Redis `SUBSCRIBE`; wildcard patterns (`ns:*`, `*`) use
+  `PSUBSCRIBE`, whose glob semantics line up with this library's matching.
+- Messages are JSON-stringified on publish and `JSON.parse`d on receipt; a payload
+  that isn't valid JSON is delivered as the raw string.
+- `close()` tears down both the subscriber and publisher connections.
+
+### RESP codec (internal)
+
+`redis.ts` uses a minimal RESP implementation exported from `resp.ts`
+(`encodeCommand`, `parseReply`, `concat`, and the `RespValue` type). These are
+implementation details of the Redis driver, not part of the public package entry.
+
+## Usage
+
+In-process (default):
+
+```ts
+import { createPubSub } from "@wrnexus/pubsub";
+
+const bus = createPubSub();
+
+const off = bus.subscribe("order:*", (msg, topic) => {
+  console.log(topic, msg);
+});
+
+await bus.publish("order:created", { id: 7 });
+
+off(); // unsubscribe
+```
+
+Cross-process with Redis:
+
+```ts
+import { createPubSub } from "@wrnexus/pubsub";
+import { redisDriver } from "@wrnexus/pubsub/redis";
+
+const driver = redisDriver("redis://localhost:6379");
+const bus = createPubSub(driver);
+
+bus.subscribe("order:*", (msg, topic) => {
+  // received on any app process subscribed to this pattern
+});
+
+await bus.publish("order:created", { id: 7 });
+
+// on shutdown
+driver.close();
+```
+
+## Requirements / Notes
+
+- **Bun-only.** The Redis driver depends on `Bun.connect`; it throws
+  `redisDriver requires the Bun runtime (Bun.connect).` outside Bun. The default
+  in-memory driver has no runtime dependencies.
+- The Redis driver reads `REDIS_URL` from the environment when no `url` is passed.
+- Backs [`@wrnexus/core`](../core)'s realtime bridge for horizontal scaling.
+- No external npm dependencies — the Redis client is a self-contained RESP codec.

packages/pubsub/package.json

@@ -0,0 +1,11 @@
+{
+  "name": "@wrnexus/pubsub",
+  "version": "0.2.12",
+  "private": true,
+  "type": "module",
+  "main": "src/index.ts",
+  "exports": {
+    ".": "./src/index.ts",
+    "./redis": "./src/redis.ts"
+  }
+}

packages/pubsub/src/index.ts

@@ -0,0 +1,63 @@
+/**
+ * @wrnexus/pubsub — topic-based publish/subscribe with a pluggable driver.
+ * The default is in-process; swap in a Redis/NATS driver for cross-instance
+ * messaging (it also backs @wrnexus/core's realtime bridge).
+ *
+ *   const bus = createPubSub();
+ *   const off = bus.subscribe("order:*", (msg, topic) => {...});
+ *   await bus.publish("order:created", { id: 7 });
+ *
+ * Subscriptions match exact topics, "ns:*" prefixes, and "*" (everything).
+ */
+
+export type Handler<T = unknown> = (message: T, topic: string) => void | Promise<void>;
+
+export interface PubSubDriver {
+  publish(topic: string, message: unknown): void | Promise<void>;
+  subscribe(pattern: string, handler: Handler): () => void;
+}
+
+export interface PubSub {
+  publish<T = unknown>(topic: string, message: T): Promise<void>;
+  subscribe<T = unknown>(pattern: string, handler: Handler<T>): () => void;
+}
+
+function patternMatches(pattern: string, topic: string): boolean {
+  if (pattern === "*" || pattern === topic) return true;
+  if (pattern.endsWith(":*")) return topic.startsWith(pattern.slice(0, -1)); // "post:" prefix
+  return false;
+}
+
+/** In-process pub/sub driver (default). */
+export function memoryDriver(): PubSubDriver {
+  const subs = new Map<string, Set<Handler>>();
+  return {
+    publish(topic, message) {
+      for (const [pattern, handlers] of subs) {
+        if (!patternMatches(pattern, topic)) continue;
+        for (const handler of handlers) void handler(message, topic);
+      }
+    },
+    subscribe(pattern, handler) {
+      let set = subs.get(pattern);
+      if (!set) subs.set(pattern, (set = new Set()));
+      set.add(handler);
+      return () => {
+        set!.delete(handler);
+        if (!set!.size) subs.delete(pattern);
+      };
+    },
+  };
+}
+
+/** Create a pub/sub bus over a driver (in-memory by default). */
+export function createPubSub(driver: PubSubDriver = memoryDriver()): PubSub {
+  return {
+    async publish(topic, message) {
+      await driver.publish(topic, message);
+    },
+    subscribe(pattern, handler) {
+      return driver.subscribe(pattern, handler as Handler);
+    },
+  };
+}

packages/pubsub/src/redis.ts

@@ -0,0 +1,182 @@
+/**
+ * Redis-backed pub/sub driver — lets pub/sub messages cross **processes**, so
+ * multiple WrNexus app runs (or apps on different domains) can communicate. Uses a
+ * self-contained RESP client over a raw TCP socket (`Bun.connect`), so it adds no
+ * npm dependency. It also backs @wrnexus/core's realtime bridge for horizontal
+ * scaling.
+ *
+ *   import { createPubSub } from "@wrnexus/pubsub";
+ *   import { redisDriver } from "@wrnexus/pubsub/redis";
+ *
+ *   const bus = createPubSub(redisDriver("redis://localhost:6379"));
+ *   bus.subscribe("order:*", (msg, topic) => { ... });   // any app process receives it
+ *   await bus.publish("order:created", { id: 7 });
+ *
+ * Pattern mapping: an exact topic uses Redis SUBSCRIBE; a wildcard pattern
+ * (`ns:*` or `*`) uses PSUBSCRIBE, whose glob semantics match this library's.
+ */
+
+import type { Handler, PubSubDriver } from "./index.ts";
+import { concat, encodeCommand, parseReply, type RespValue } from "./resp.ts";
+
+interface RedisConn {
+  send(bytes: Uint8Array): void;
+  close(): void;
+}
+
+interface ParsedUrl {
+  host: string;
+  port: number;
+  password?: string;
+  db?: number;
+  tls: boolean;
+}
+
+function parseUrl(url: string): ParsedUrl {
+  const u = new URL(url);
+  if (u.protocol !== "redis:" && u.protocol !== "rediss:") {
+    throw new TypeError("Redis URL must use redis:// or rediss://");
+  }
+  const path = u.pathname.replace(/^\//, "");
+  const port = u.port ? Number(u.port) : 6379;
+  const db = path ? Number(path) : undefined;
+  if (!Number.isInteger(port) || port < 1 || port > 65_535)
+    throw new TypeError("Redis URL has an invalid port");
+  if (db !== undefined && (!Number.isInteger(db) || db < 0))
+    throw new TypeError("Redis URL database must be a non-negative integer");
+  return {
+    host: u.hostname || "localhost",
+    port,
+    password: u.password ? decodeURIComponent(u.password) : undefined,
+    db,
+    tls: u.protocol === "rediss:",
+  };
+}
+
+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 {
+  let socket: { write(data: Uint8Array): void; end(): void } | null = null;
+  const pending: Uint8Array[] = [];
+  const maxPending = 1000;
+  let buf: Uint8Array<ArrayBufferLike> = new Uint8Array(0);
+
+  // 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;
+      },
+      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;
+      },
+    },
+  });
+
+  return {
+    send(bytes) {
+      if (socket) socket.write(bytes);
+      else {
+        if (pending.length >= maxPending) {
+          throw new Error("Redis connection is unavailable and its pending write queue is full");
+        }
+        pending.push(bytes);
+      }
+    },
+    close() {
+      socket?.end();
+    },
+  };
+}
+
+/**
+ * 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 } {
+  const cfg = parseUrl(url ?? getEnv("REDIS_URL") ?? "redis://localhost:6379");
+  const subs = new Map<string, Set<Handler>>();
+  const redisSubscribed = new Set<string>();
+
+  const dispatch = (key: string, topic: string, payload: string) => {
+    const handlers = subs.get(key);
+    if (!handlers) return;
+    let message: unknown = payload;
+    try {
+      message = JSON.parse(payload);
+    } catch {
+      /* not JSON — deliver the raw string */
+    }
+    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);
+
+  return {
+    publish(topic, message) {
+      const payload = typeof message === "string" ? message : JSON.stringify(message);
+      pubConn.send(encodeCommand(["PUBLISH", topic, payload]));
+    },
+    subscribe(pattern, handler) {
+      let set = subs.get(pattern);
+      if (!set) subs.set(pattern, (set = new Set()));
+      set.add(handler);
+      if (!redisSubscribed.has(pattern)) {
+        redisSubscribed.add(pattern);
+        subConn.send(encodeCommand([isPattern(pattern) ? "PSUBSCRIBE" : "SUBSCRIBE", pattern]));
+      }
+      return () => {
+        set!.delete(handler);
+        if (!set!.size) {
+          subs.delete(pattern);
+          redisSubscribed.delete(pattern);
+          subConn.send(
+            encodeCommand([isPattern(pattern) ? "PUNSUBSCRIBE" : "UNSUBSCRIBE", pattern]),
+          );
+        }
+      };
+    },
+    close() {
+      subConn.close();
+      pubConn.close();
+    },
+  };
+}
+
+function getEnv(key: string): string | undefined {
+  return (globalThis as { process?: { env?: Record<string, string | undefined> } }).process?.env?.[
+    key
+  ];
+}

packages/pubsub/src/resp.ts

@@ -0,0 +1,92 @@
+/**
+ * Minimal RESP (REdis Serialization Protocol) codec — just enough to drive
+ * Redis pub/sub over a raw TCP socket, with no external client dependency.
+ *
+ * Encodes commands as arrays of bulk strings, and streaming-parses the replies
+ * we care about: simple strings (+), errors (-), integers (:), bulk strings ($),
+ * and arrays (*) — which is what SUBSCRIBE/PSUBSCRIBE confirmations and
+ * message/pmessage pushes are made of.
+ */
+
+export type RespValue = string | number | null | RespValue[];
+
+const encoder = new TextEncoder();
+const decoder = new TextDecoder();
+
+/** Encode a command (`["PUBLISH", topic, payload]`) as a RESP array of bulk strings. */
+export function encodeCommand(args: string[]): Uint8Array {
+  let head = `*${args.length}\r\n`;
+  const parts: Uint8Array[] = [];
+  for (const arg of args) {
+    const bytes = encoder.encode(arg);
+    head += `$${bytes.length}\r\n`;
+    parts.push(encoder.encode(head), bytes, encoder.encode("\r\n"));
+    head = "";
+  }
+  return concat(parts);
+}
+
+/** Concatenate byte arrays. */
+export function concat(parts: Uint8Array[]): Uint8Array {
+  let len = 0;
+  for (const p of parts) len += p.length;
+  const out = new Uint8Array(len);
+  let off = 0;
+  for (const p of parts) {
+    out.set(p, off);
+    off += p.length;
+  }
+  return out;
+}
+
+function indexOfCRLF(buf: Uint8Array, from: number): number {
+  for (let i = from; i + 1 < buf.length; i++) {
+    if (buf[i] === 13 && buf[i + 1] === 10) return i;
+  }
+  return -1;
+}
+
+/**
+ * Parse one RESP reply from `buf` at `off`. Returns the value and the offset just
+ * past it, or `null` if the buffer doesn't yet hold a complete reply (caller
+ * should wait for more bytes).
+ */
+export function parseReply(buf: Uint8Array, off = 0): { value: RespValue; next: number } | null {
+  if (off >= buf.length) return null;
+  const type = buf[off];
+  const lineEnd = indexOfCRLF(buf, off + 1);
+  if (lineEnd === -1) return null;
+  const line = decoder.decode(buf.subarray(off + 1, lineEnd));
+  const after = lineEnd + 2;
+
+  switch (type) {
+    case 43: // '+' simple string
+    case 45: // '-' error
+      return { value: line, next: after };
+    case 58: // ':' integer
+      return { value: Number(line), next: after };
+    case 36: {
+      // '$' bulk string
+      const len = Number(line);
+      if (len === -1) return { value: null, next: after };
+      if (after + len + 2 > buf.length) return null;
+      return { value: decoder.decode(buf.subarray(after, after + len)), next: after + len + 2 };
+    }
+    case 42: {
+      // '*' array
+      const count = Number(line);
+      if (count === -1) return { value: null, next: after };
+      const arr: RespValue[] = [];
+      let cur = after;
+      for (let i = 0; i < count; i++) {
+        const r = parseReply(buf, cur);
+        if (!r) return null;
+        arr.push(r.value);
+        cur = r.next;
+      }
+      return { value: arr, next: cur };
+    }
+    default:
+      return null;
+  }
+}

packages/pubsub/test/pubsub.test.ts

@@ -0,0 +1,37 @@
+import { test, expect } from "bun:test";
+import { createPubSub } from "../src/index.ts";
+
+test("publish reaches exact + wildcard subscribers", async () => {
+  const bus = createPubSub();
+  const got: string[] = [];
+  bus.subscribe("order:created", (m: { id: number }) => {
+    got.push(`exact:${m.id}`);
+  });
+  bus.subscribe("order:*", (_m, topic) => {
+    got.push(`prefix:${topic}`);
+  });
+  bus.subscribe("*", (_m, topic) => {
+    got.push(`all:${topic}`);
+  });
+
+  await bus.publish("order:created", { id: 7 });
+  await bus.publish("user:login", { id: 1 });
+
+  expect(got).toContain("exact:7");
+  expect(got).toContain("prefix:order:created");
+  expect(got).toContain("all:order:created");
+  expect(got).toContain("all:user:login");
+  expect(got).not.toContain("prefix:user:login"); // order:* doesn't match user:*
+});
+
+test("unsubscribe stops delivery", async () => {
+  const bus = createPubSub();
+  let n = 0;
+  const off = bus.subscribe("t", () => {
+    n++;
+  });
+  await bus.publish("t", 1);
+  off();
+  await bus.publish("t", 2);
+  expect(n).toBe(1);
+});

packages/pubsub/test/redis-driver.test.ts

@@ -0,0 +1,33 @@
+import { afterEach, expect, test } from "bun:test";
+import { redisDriver } from "../src/redis.ts";
+
+const bun = globalThis.Bun as typeof Bun & { connect: (options: unknown) => Promise<unknown> };
+const originalConnect = bun.connect;
+afterEach(() => {
+  bun.connect = originalConnect;
+});
+
+test("rediss URLs enable TLS and decode credentials", () => {
+  const options: Array<Record<string, unknown>> = [];
+  bun.connect = ((value: Record<string, unknown>) => {
+    options.push(value);
+    return new Promise(() => {});
+  }) as typeof bun.connect;
+  const driver = redisDriver("rediss://user:p%40ss@redis.example.com:6380/2");
+  expect(options).toHaveLength(2);
+  expect(options[0]).toMatchObject({ hostname: "redis.example.com", port: 6380, tls: true });
+  driver.close();
+});
+
+test("rejects invalid Redis URLs before connecting", () => {
+  expect(() => redisDriver("http://localhost:6379")).toThrow("redis:// or rediss://");
+  expect(() => redisDriver("redis://localhost/not-a-db")).toThrow("database");
+});
+
+test("bounds writes queued while Redis is unavailable", () => {
+  bun.connect = (() => new Promise(() => {})) as typeof bun.connect;
+  const driver = redisDriver("redis://localhost:6379");
+  for (let index = 0; index < 1000; index++) driver.publish("topic", index);
+  expect(() => driver.publish("topic", "overflow")).toThrow("queue is full");
+  driver.close();
+});

packages/pubsub/test/resp.test.ts

@@ -0,0 +1,51 @@
+import { test, expect } from "bun:test";
+import { encodeCommand, parseReply, concat } from "../src/resp.ts";
+
+const enc = new TextEncoder();
+const dec = new TextDecoder();
+
+test("encodeCommand produces a RESP array of bulk strings", () => {
+  const bytes = encodeCommand(["PUBLISH", "order:created", '{"id":7}']);
+  expect(dec.decode(bytes)).toBe(
+    '*3\r\n$7\r\nPUBLISH\r\n$13\r\norder:created\r\n$8\r\n{"id":7}\r\n',
+  );
+});
+
+test("parseReply reads a Redis message push", () => {
+  const wire = enc.encode("*3\r\n$7\r\nmessage\r\n$5\r\ntopic\r\n$5\r\nhello\r\n");
+  const r = parseReply(wire);
+  expect(r).not.toBeNull();
+  expect(r!.value).toEqual(["message", "topic", "hello"]);
+  expect(r!.next).toBe(wire.length);
+});
+
+test("parseReply reads a pmessage push (pattern, channel, payload)", () => {
+  const wire = enc.encode(
+    "*4\r\n$8\r\npmessage\r\n$6\r\norder:*\r\n$13\r\norder:created\r\n$2\r\n{}\r\n".replace(
+      "$6\r\norder:*",
+      "$7\r\norder:*",
+    ),
+  );
+  const r = parseReply(wire)!;
+  expect(r.value).toEqual(["pmessage", "order:*", "order:created", "{}"]);
+});
+
+test("parseReply returns null for an incomplete buffer, then completes", () => {
+  const full = enc.encode("*1\r\n$5\r\nhello\r\n");
+  // Half the bytes → incomplete.
+  expect(parseReply(full.subarray(0, 8))).toBeNull();
+  // Full → parses.
+  expect(parseReply(full)!.value).toEqual(["hello"]);
+});
+
+test("parseReply handles integers and simple strings (subscribe confirmations)", () => {
+  const sub = enc.encode("*3\r\n$9\r\nsubscribe\r\n$5\r\ntopic\r\n:1\r\n");
+  expect(parseReply(sub)!.value).toEqual(["subscribe", "topic", 1]);
+  const ok = enc.encode("+OK\r\n");
+  expect(parseReply(ok)!.value).toBe("OK");
+});
+
+test("concat joins byte chunks in order", () => {
+  const joined = concat([enc.encode("ab"), enc.encode("cd")]);
+  expect(dec.decode(joined)).toBe("abcd");
+});