# @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;
  close(): Promise<void>;
}

type Handler<T = unknown> = (message: T, topic: string) => void | Promise<void>;
```

- `publish(topic, message)` — resolves once the driver and in-memory async handlers finish.
- `subscribe(pattern, handler)` — returns an unsubscribe function.
- `close()` — idempotently rejects new work, clears local subscriptions, and closes the driver.

### 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, options?: RedisDriverOptions): 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.
- Lost sockets reconnect with bounded exponential backoff and active subscriptions
  are replayed. `maxPending` bounds unavailable-connection writes (default 1000);
  `reconnectDelayMs` and `reconnectMaxDelayMs` tune recovery (100ms/5000ms).

### 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 (also closes the driver)
await bus.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.