WRNexusJS/packages/pubsub/src/redis.ts

/**
 * 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;
}

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:") {
    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,
  options: Required<RedisDriverOptions>,
  onReply?: (value: RespValue) => void,
  onReconnect?: () => void,
): RedisConn {
  let socket: { write(data: Uint8Array): void; end(): void } | null = null;
  const pending: Uint8Array[] = [];
  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).");

  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();
        },
      },
    }).catch(() => {
      connecting = false;
      scheduleReconnect();
    });
  };
  start();

  return {
    send(bytes) {
      if (socket) socket.write(bytes);
      else {
        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;
    },
  };
}

/**
 * A cross-process pub/sub driver backed by Redis. `url` defaults to
 * `$REDIS_URL` or `redis://localhost:6379`.
 */
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>();

  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,
    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) {
      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 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
  ];
}