Move docs/ package mirror into document-navigator plugin
Relocate the ~6k-file docs tree from the repo root into plugins/document-navigator/docs/ so the plugin is self-contained and portable. docs/superpowers/ (specs and plans) stays at the repo root. Moved via cp+rm rather than git mv because directory rename() fails with EPERM on this WSL /mnt DrvFs mount; git rename-detection pairs the identical-content files at commit time. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,162 @@
|
||||
# @suaveplan/api-ws-broker-redis - Redis pub/sub broker for @suaveplan/api-ws — cross-host WebSocket and SSE fan-out. Publishes BrokerEnvelopes via Redis Pub/Sub channels, maintains per-(tenant,topic) reconnect event-buffers in sorted-sets with cap + TTL, gzip-compresses payloads above a configurable threshold, exposes a healthz probe + auto-reconnect, and emits the standard ws.broker.* OTel surface. Ships a minimal built-in RESP client so integration tests can drive a real Redis container without an external client dependency; accepts any pluggable RedisPubSubLike (e.g. ioredis or node-redis) in production.
|
||||
|
||||
Cross-host Redis pub/sub broker for `@suaveplan/api-ws` and `@suaveplan/api-sse`.
|
||||
|
||||
Implements the `WsBroker` contract from `@suaveplan/api-ws` against a real
|
||||
Redis instance: every `broker.broadcast(topic, envelope)` is published to a
|
||||
shared Redis channel pattern (default `ws.broadcast.*`), and every host
|
||||
subscribed to that pattern delivers the broadcast to its locally-connected
|
||||
subscribers via the local `ConnectionRegistry`. Connections are NOT migrated
|
||||
across hosts — the broker just makes broadcasts visible everywhere.
|
||||
|
||||
The same broker serves the reconnect event-buffer required by `api-ws-bun`
|
||||
(WebSocket) and `api-sse-bun` (SSE): a per-`(tenant, topic)` Redis sorted-set
|
||||
storing the most-recent `eventBufferMaxEvents` events with TTL
|
||||
`eventBufferTtlSec`, queryable by `Last-Event-Id` for `WS` reconnect or
|
||||
`Last-Event-Id` header for SSE resume.
|
||||
|
||||
<!-- begin:auto-docs -->
|
||||
## Documentation tree
|
||||
|
||||
- [broker](./src/broker/broker.md) — This module is the orchestrator for `@suaveplan/api-ws-broker-redis`. It wires two Redis clients (one in publish mode,…
|
||||
- [broker](./src/broker/broker.md) — This module is the orchestrator for `@suaveplan/api-ws-broker-redis`. It wires two Redis clients (one in publish mode,…
|
||||
- [error-codes](./src/error-codes/error-codes.md) — This module defines the stable error-code registry for `@suaveplan/api-ws-broker-redis`. Every failure surface the brok…
|
||||
- [error-codes](./src/error-codes/error-codes.md) — This module defines the stable error-code registry for `@suaveplan/api-ws-broker-redis`. Every failure surface the brok…
|
||||
- [event-buffer](./src/event-buffer/event-buffer.md) — This module backs the broker's reconnect event buffer. Each `(tenant, topic)` pair maps to one Redis sorted-set keyed b…
|
||||
- [event-buffer](./src/event-buffer/event-buffer.md) — This module backs the broker's reconnect event buffer. Each `(tenant, topic)` pair maps to one Redis sorted-set keyed b…
|
||||
- [health](./src/health/health.md) — This module is the pure, side-effect-free evaluator for the broker's health status. It inspects the two underlying RESP…
|
||||
- [health](./src/health/health.md) — This module is the pure, side-effect-free evaluator for the broker's health status. It inspects the two underlying RESP…
|
||||
- [instruments](./src/instruments/instruments.md) — This module owns the OpenTelemetry instrument surface for the Redis broker. Six instruments cover every documented oper…
|
||||
- [instruments](./src/instruments/instruments.md) — This module owns the OpenTelemetry instrument surface for the Redis broker. Six instruments cover every documented oper…
|
||||
- [options](./src/options/options.md) — This module defines the configuration shape consumed by `createRedisWsBroker(...)`. Every operational knob the broker e…
|
||||
- [options](./src/options/options.md) — This module defines the configuration shape consumed by `createRedisWsBroker(...)`. Every operational knob the broker e…
|
||||
- [publish](./src/publish/publish.md) — This module handles wire-encoding for the Redis broker. Every published frame gets a one-byte encoding tag (`ENCODING_J…
|
||||
- [publish](./src/publish/publish.md) — This module handles wire-encoding for the Redis broker. Every published frame gets a one-byte encoding tag (`ENCODING_J…
|
||||
- [reconnect](./src/reconnect/reconnect.md) — This module is the deterministic policy evaluator for the broker's reconnect loop. It computes the next backoff delay i…
|
||||
- [reconnect](./src/reconnect/reconnect.md) — This module is the deterministic policy evaluator for the broker's reconnect loop. It computes the next backoff delay i…
|
||||
- [resp-client](./src/resp-client/resp-client.md) — This module ships the broker's built-in RESP-protocol client so the package has no mandatory runtime dependency on an e…
|
||||
- [resp-client](./src/resp-client/resp-client.md) — This module ships the broker's built-in RESP-protocol client so the package has no mandatory runtime dependency on an e…
|
||||
- [subscribe](./src/subscribe/subscribe.md) — This module owns the broker's local routing table — the in-process map from topic name to handler set that the inbound…
|
||||
- [subscribe](./src/subscribe/subscribe.md) — This module owns the broker's local routing table — the in-process map from topic name to handler set that the inbound…
|
||||
<!-- end:auto-docs -->
|
||||
|
||||
## What ships
|
||||
|
||||
| Subpath | Module | Doc |
|
||||
|---|---|---|
|
||||
| `./broker` | `createRedisWsBroker(opts)` factory | [`src/broker/broker.md`](src/broker/broker.md) |
|
||||
| `./error-codes` | `WS_BROKER_REDIS_CODES` + `WsBrokerRedisError` | [`src/error-codes/error-codes.md`](src/error-codes/error-codes.md) |
|
||||
| `./options` | option schema + defaults | [`src/options/options.md`](src/options/options.md) |
|
||||
| `./resp-client` | built-in minimal RESP client (`createRespClient`) | [`src/resp-client/resp-client.md`](src/resp-client/resp-client.md) |
|
||||
|
||||
## Surface
|
||||
|
||||
- `createRedisWsBroker(opts)` — async factory returning a `RedisWsBroker`
|
||||
(the `WsBroker` contract plus `writeBuffer`, `readBuffer`, `healthz`, and
|
||||
`isDisposed`).
|
||||
- `RedisWsBrokerOptions` — Zod-validated strict option shape. Required:
|
||||
`connection`. Optional: `keyPrefix`, `channelPrefix`, `eventBufferTtlSec`,
|
||||
`eventBufferMaxEvents`, `compressionThresholdBytes`, `reconnectStrategy`,
|
||||
`commandTimeoutMs`, `disableCompression`, `hostId`.
|
||||
- `WS_BROKER_REDIS_CODES` — closed set of 10 stable error codes
|
||||
(`CONNECTION_LOST`, `PUBLISH_FAILED`, `SUBSCRIBE_FAILED`,
|
||||
`BUFFER_WRITE_FAILED`, `BUFFER_READ_FAILED`, `PAYLOAD_TOO_LARGE`,
|
||||
`DESERIALISE_FAILED`, `CONFIG_INVALID`, `RECONNECT_FAILED`, `TIMEOUT`).
|
||||
- `createRespClient(opts)` — minimal built-in RESP client implementing
|
||||
`RedisPubSubLike`. Use this when you do not already have an `ioredis` or
|
||||
`node-redis` instance in your runtime.
|
||||
- `RedisPubSubLike` — pluggable client interface; pass any conforming
|
||||
implementation via `clientFactory` to use `ioredis` / `node-redis` /
|
||||
another adapter.
|
||||
|
||||
## Quickstart
|
||||
|
||||
```ts
|
||||
import { createRedisWsBroker } from "@suaveplan/api-ws-broker-redis";
|
||||
import type { TenantId } from "@suaveplan/api-types/tenant";
|
||||
|
||||
const broker = await createRedisWsBroker<{ id: number }>({
|
||||
connection: "redis://localhost:6379",
|
||||
});
|
||||
|
||||
const sub = await broker.subscribe("orders", (envelope) => {
|
||||
console.log("delivered", envelope.payload.id);
|
||||
});
|
||||
|
||||
await broker.publish("orders", {
|
||||
topic: "orders",
|
||||
tenant: "t1" as unknown as TenantId,
|
||||
payload: { id: 1 },
|
||||
});
|
||||
|
||||
sub.unsubscribe();
|
||||
await broker.dispose();
|
||||
```
|
||||
|
||||
## Operational notes
|
||||
|
||||
### Key-prefix conventions
|
||||
|
||||
The broker namespaces every Redis key under `<keyPrefix>` (default
|
||||
`suaveplan:ws:`). Within that namespace it owns:
|
||||
|
||||
- `<keyPrefix>buffer:<tenant>:<topic>` — sorted-set holding the last
|
||||
`eventBufferMaxEvents` entries with scores keyed on `eventId`. Member
|
||||
format is `<eventId>|<base64-payload>`.
|
||||
|
||||
Pub/sub channels are namespaced under `<channelPrefix>` (default
|
||||
`ws.broadcast.`): `<channelPrefix><topic>`.
|
||||
|
||||
Two `<keyPrefix>` choices must be in lock-step across every host replica
|
||||
that participates in the same fan-out fabric — otherwise broadcasts will
|
||||
land on different channel names and never be delivered.
|
||||
|
||||
### Redis memory sizing
|
||||
|
||||
The peak Redis memory consumed by a single broker is bounded by:
|
||||
|
||||
```text
|
||||
hosts * tenants * topics * eventBufferMaxEvents * average_event_size
|
||||
```
|
||||
|
||||
with TTL `eventBufferTtlSec` ensuring abandoned topics expire on their own.
|
||||
For default settings (cap 256, TTL 5 min, ~1 KiB events) that is
|
||||
~256 KiB per active `(tenant, topic)`. A 10k-active-topic deployment
|
||||
budgets ~2.5 GiB of Redis memory for buffers alone, before pub/sub
|
||||
traffic. Tune the cap and TTL down for memory-constrained deployments.
|
||||
|
||||
### Multi-host topology
|
||||
|
||||
Every host eagerly `PSUBSCRIBE`s `<channelPrefix>*` at construction time
|
||||
so the first publish is immediately visible to peers. Adding a host is
|
||||
purely additive — no configuration change to existing hosts is required;
|
||||
the new host comes up, subscribes to the pattern, and starts delivering
|
||||
broadcasts to its locally-connected subscribers.
|
||||
|
||||
### Compression
|
||||
|
||||
Payloads above `compressionThresholdBytes` (default 1 KiB) are
|
||||
gzipped before publish. The wire format carries a one-byte encoding
|
||||
magic so subscribers can decode either form transparently. The 4 MiB
|
||||
hard cap (`MAX_PAYLOAD_BYTES`) refuses runaway payloads with
|
||||
`WS_BROKER_REDIS.PAYLOAD_TOO_LARGE`.
|
||||
|
||||
### Pluggable clients
|
||||
|
||||
The default broker opens two sockets via the built-in RESP client
|
||||
(`./resp-client`). Pass `clientFactory: (url, role) => myAdapter(url, role)`
|
||||
to swap in an `ioredis` or `node-redis` adapter — the adapter must satisfy
|
||||
the {@link RedisPubSubLike} interface.
|
||||
|
||||
## Quality bars
|
||||
|
||||
- 100% line / branch / function / statement coverage (Istanbul).
|
||||
- 100% docs coverage — every implementation `.ts` has a sibling `.md` with
|
||||
the required headings.
|
||||
- Zero direct imports of `vitest` / `bun:test` — all test files import
|
||||
from `@suaveplan/testing/runner`.
|
||||
- Stable, namespaced error codes (`WS_BROKER_REDIS.*`).
|
||||
- Real-Redis integration tests via `@suaveplan/testing-integration-redis`
|
||||
with the testcontainers-backed `RedisFixture` (gracefully skipped when
|
||||
Docker is unavailable).
|
||||
|
||||
@@ -0,0 +1,130 @@
|
||||
# `broker` — `createRedisWsBroker`
|
||||
|
||||
<!-- begin:auto-docs -->
|
||||
_(no submodules)_
|
||||
<!-- end:auto-docs -->
|
||||
|
||||
## Purpose
|
||||
|
||||
This module is the orchestrator for `@suaveplan/api-ws-broker-redis`. It wires
|
||||
two Redis clients (one in publish mode, one held in subscribe mode), establishes
|
||||
a `PSUBSCRIBE <channelPrefix>*` pattern, and exposes the `WsBroker` surface plus
|
||||
broker-specific operations (`writeBuffer`, `readBuffer`, `healthz`, `isDisposed`).
|
||||
The broker is the production cross-host adapter for the `@suaveplan/api-ws`
|
||||
contract — the in-memory broker handles single-host fan-out and this package
|
||||
extends that to many hosts by riding Redis pub/sub.
|
||||
|
||||
The broker is intentionally pluggable: a `clientFactory` option lets callers
|
||||
swap in `ioredis`, `node-redis`, or any client implementing
|
||||
{@link RedisPubSubLike}. When unset the broker defaults to the built-in RESP
|
||||
client shipped from `./resp-client`, which keeps the package free of mandatory
|
||||
runtime dependencies on an external Redis client.
|
||||
|
||||
## Features
|
||||
|
||||
- Two Redis sockets, one per role — replies on the publisher socket, async
|
||||
pub/sub frames on the subscriber socket. Redis itself enforces that subscribe
|
||||
sockets reject non-subscribe commands; the broker structure mirrors that
|
||||
constraint exactly.
|
||||
- Eager `PSUBSCRIBE` at construction time so a subsequent `publish` is
|
||||
immediately visible to peers.
|
||||
- Compression-aware envelope encoding via `./publish` (gzip above the
|
||||
configurable threshold, hard 4 MiB cap, magic-byte framing).
|
||||
- Reconnect-friendly event buffer: per-`(tenant, topic)` sorted-set with
|
||||
cap + TTL, consumed by `api-ws-bun` and `api-sse-bun` for `Last-Event-Id`
|
||||
resume.
|
||||
- 10 stable `WS_BROKER_REDIS.*` error codes; every production path attaches
|
||||
the original `cause`.
|
||||
- Idempotent `dispose()` that detaches listeners, `PUNSUBSCRIBE`s the pattern,
|
||||
and quits both sockets.
|
||||
- OTel counter + histogram surface (`ws.broker.*`) wired through the
|
||||
`./instruments` module.
|
||||
|
||||
## Usage: Basic
|
||||
|
||||
```ts
|
||||
import { createRedisWsBroker } from "@suaveplan/api-ws-broker-redis/broker";
|
||||
import type { TenantId } from "@suaveplan/api-types/tenant";
|
||||
|
||||
const broker = await createRedisWsBroker<{ id: number }>({
|
||||
connection: "redis://localhost:6379",
|
||||
});
|
||||
|
||||
const sub = await broker.subscribe("orders", (envelope) => {
|
||||
console.log("delivered", envelope.payload.id);
|
||||
});
|
||||
|
||||
await broker.publish("orders", {
|
||||
topic: "orders",
|
||||
tenant: "t1" as unknown as TenantId,
|
||||
payload: { id: 1 },
|
||||
});
|
||||
|
||||
sub.unsubscribe();
|
||||
await broker.dispose();
|
||||
```
|
||||
|
||||
## Usage: Advanced
|
||||
|
||||
```ts
|
||||
import { createRedisWsBroker } from "@suaveplan/api-ws-broker-redis/broker";
|
||||
|
||||
// Multi-host setup with custom prefixes and a fixed reconnect strategy.
|
||||
const broker = await createRedisWsBroker({
|
||||
connection: process.env.REDIS_URL!,
|
||||
keyPrefix: "prod:ws:",
|
||||
channelPrefix: "prod.ws.broadcast.",
|
||||
eventBufferMaxEvents: 512,
|
||||
eventBufferTtlSec: 600,
|
||||
compressionThresholdBytes: 2048,
|
||||
commandTimeoutMs: 2_000,
|
||||
reconnectStrategy: {
|
||||
initialDelayMs: 50,
|
||||
maxDelayMs: 2_000,
|
||||
maxAttempts: 200,
|
||||
multiplier: 2,
|
||||
},
|
||||
hostId: process.env.HOSTNAME ?? "host-unknown",
|
||||
});
|
||||
|
||||
// Pluggable client (ioredis adapter, etc.):
|
||||
const ioredisBroker = await createRedisWsBroker({
|
||||
connection: process.env.REDIS_URL!,
|
||||
clientFactory: (url, role) =>
|
||||
// Wrap an ioredis instance into the RedisPubSubLike shape — production
|
||||
// teams who already use ioredis can avoid maintaining two clients.
|
||||
myIoredisAdapter(url, role),
|
||||
});
|
||||
```
|
||||
|
||||
## API Reference
|
||||
|
||||
- `createRedisWsBroker<Payload>(options)` — async factory returning a
|
||||
`RedisWsBroker<Payload>`.
|
||||
- `RedisWsBroker<Payload> extends WsBroker<Payload>` — adds `writeBuffer`,
|
||||
`readBuffer`, `healthz`, `isDisposed`, and `options` (the resolved bundle).
|
||||
- `CreateRedisWsBrokerOptions` — `RedisWsBrokerOptions` plus the factory-only
|
||||
`clientFactory` and `readyTimeoutMs` knobs.
|
||||
- `RedisClientFactory` — `(url, role) => RedisPubSubLike`.
|
||||
|
||||
## Implementation Notes
|
||||
|
||||
The broker normalises all timing through `Date.now()` — production deployments
|
||||
that want a synthetic clock should layer that at the OTel reader level rather
|
||||
than reaching into the broker. The constructor strips factory-only fields
|
||||
(`clientFactory`, `readyTimeoutMs`) before handing the rest to the strict
|
||||
Zod schema so unknown keys still fail fast for the genuine option shape.
|
||||
|
||||
The `waitForReady` helper races the client's `ready` event against a timeout
|
||||
and against `error`/`close`. This means a flaky network at startup surfaces
|
||||
fast (`CONNECTION_LOST`) rather than wedging the broker in an unbuilt state.
|
||||
Once construction returns, every method short-circuits with the same code on
|
||||
post-`dispose` calls — the broker is single-use, like the in-memory variant.
|
||||
|
||||
The subscribe path runs `dispatchInboundFrame` per topic via the local
|
||||
registry, which snapshots the handler set before delivery so a handler that
|
||||
unsubscribes mid-fan-out does not silently skip its peers. Decode failures
|
||||
and per-handler rejections route through the `onError` callback (currently
|
||||
captured into `healthz().lastError`) rather than propagating into the pub/sub
|
||||
loop, because Redis will not stop pushing frames just because one of them
|
||||
was malformed.
|
||||
@@ -0,0 +1,96 @@
|
||||
# `error-codes` — `WS_BROKER_REDIS_CODES`
|
||||
|
||||
<!-- begin:auto-docs -->
|
||||
_(no submodules)_
|
||||
<!-- end:auto-docs -->
|
||||
|
||||
## Purpose
|
||||
|
||||
This module defines the stable error-code registry for
|
||||
`@suaveplan/api-ws-broker-redis`. Every failure surface the broker emits at
|
||||
runtime — Redis connection drops, publish/subscribe failures, buffer reads
|
||||
and writes, oversize payloads, deserialise failures, configuration errors,
|
||||
reconnect exhaustion, and command timeouts — carries one of the ten
|
||||
`WS_BROKER_REDIS.*` codes defined here. Callers branch on the literal string
|
||||
value (or `instanceof WsBrokerRedisError`) to drive recovery logic.
|
||||
|
||||
The registry is closed: adding a new code requires an OpenSpec change so
|
||||
production dashboards and alerting rules keyed on the existing codes stay
|
||||
stable.
|
||||
|
||||
## Features
|
||||
|
||||
- Ten stable codes covering every documented failure surface.
|
||||
- `defineCodes("WS_BROKER_REDIS", [...])` so the runtime values are
|
||||
auto-namespaced to `WS_BROKER_REDIS.<NAME>`.
|
||||
- `WsBrokerRedisError` subclass with a public constructor (lint-enforced),
|
||||
so callers can `new WsBrokerRedisError(code, message, { cause })` directly
|
||||
rather than coercing through a factory.
|
||||
- Tied-off TypeScript literal union (`WsBrokerRedisErrorCode`) for
|
||||
exhaustive `switch` statements at call sites.
|
||||
|
||||
## Usage: Basic
|
||||
|
||||
```ts
|
||||
import {
|
||||
WS_BROKER_REDIS_CODES,
|
||||
WsBrokerRedisError,
|
||||
} from "@suaveplan/api-ws-broker-redis/error-codes";
|
||||
|
||||
try {
|
||||
await broker.publish("orders", envelope);
|
||||
} catch (err) {
|
||||
if (
|
||||
err instanceof WsBrokerRedisError &&
|
||||
err.brokerCode === WS_BROKER_REDIS_CODES.PAYLOAD_TOO_LARGE
|
||||
) {
|
||||
// Drop or split — Redis is not going to accept this frame.
|
||||
return;
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
```
|
||||
|
||||
## Usage: Advanced
|
||||
|
||||
```ts
|
||||
import {
|
||||
type WsBrokerRedisErrorCode,
|
||||
WS_BROKER_REDIS_CODES,
|
||||
} from "@suaveplan/api-ws-broker-redis/error-codes";
|
||||
|
||||
function classify(code: WsBrokerRedisErrorCode): "retry" | "drop" | "fatal" {
|
||||
switch (code) {
|
||||
case WS_BROKER_REDIS_CODES.CONNECTION_LOST:
|
||||
case WS_BROKER_REDIS_CODES.TIMEOUT:
|
||||
case WS_BROKER_REDIS_CODES.PUBLISH_FAILED:
|
||||
return "retry";
|
||||
case WS_BROKER_REDIS_CODES.PAYLOAD_TOO_LARGE:
|
||||
case WS_BROKER_REDIS_CODES.DESERIALISE_FAILED:
|
||||
return "drop";
|
||||
default:
|
||||
return "fatal";
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## API Reference
|
||||
|
||||
- `WS_BROKER_REDIS_CODES` — frozen registry mapping name → `WS_BROKER_REDIS.<NAME>`.
|
||||
- `WsBrokerRedisErrorCode` — literal union over every registered code.
|
||||
- `WsBrokerRedisError` — `SuaveplanError<WsBrokerRedisErrorCode>` subclass with
|
||||
`brokerCode` field and a public 2/3-arg constructor.
|
||||
|
||||
## Implementation Notes
|
||||
|
||||
`WsBrokerRedisError` declares its own public constructor (rather than relying
|
||||
on the base `SuaveplanError`'s protected one) so `new WsBrokerRedisError(...)`
|
||||
compiles cleanly from outside the file. This is required by the repo's
|
||||
`check-error-subclass-constructors` lint — TypeScript's TS2674 ("Constructor of
|
||||
class is protected and only accessible within the class declaration") would
|
||||
otherwise reject every call site.
|
||||
|
||||
The `brokerCode` field is a deliberate alias for `code`. Calling code commonly
|
||||
branches on the narrower `WsBrokerRedisErrorCode` literal union to get
|
||||
exhaustiveness from the TypeScript compiler; reading the alias keeps that
|
||||
narrowing locally explicit even when the catch block widened the type.
|
||||
@@ -0,0 +1,99 @@
|
||||
# `event-buffer` — Redis sorted-set reconnect buffer
|
||||
|
||||
<!-- begin:auto-docs -->
|
||||
_(no submodules)_
|
||||
<!-- end:auto-docs -->
|
||||
|
||||
## Purpose
|
||||
|
||||
This module backs the broker's reconnect event buffer. Each
|
||||
`(tenant, topic)` pair maps to one Redis sorted-set keyed by
|
||||
`<keyPrefix>buffer:<tenant>:<topic>` and scored by `eventId`. The
|
||||
`api-ws-bun` and `api-sse-bun` engines call into this surface to replay
|
||||
events to clients that reconnected with a `Last-Event-Id` (WS) or
|
||||
`Last-Event-Id` (SSE) header.
|
||||
|
||||
The buffer enforces two operational guarantees: a fixed cap
|
||||
(`eventBufferMaxEvents`) so a per-tenant runaway publisher cannot exhaust
|
||||
Redis memory, and a sliding TTL (`eventBufferTtlSec`) so abandoned topics
|
||||
expire without operator intervention.
|
||||
|
||||
## Features
|
||||
|
||||
- Sorted-set encoding `<eventId>|<base64-payload>` so the integer score
|
||||
remains usable for `ZRANGEBYSCORE` queries and the binary payload survives
|
||||
the round-trip without coercion.
|
||||
- Cap enforcement via `ZCARD` + `ZREMRANGEBYRANK 0 (overflow - 1)` —
|
||||
oldest entries evicted first.
|
||||
- Per-write TTL refresh via `EXPIRE` so the key always carries the latest
|
||||
TTL window.
|
||||
- `readBuffered Events(...)` returns events strictly greater than
|
||||
`sinceEventId` (`(<sinceEventId>` syntax) in ascending score order.
|
||||
- OTel hit/miss labelling on every read via the `ws.broker.buffer_reads_total`
|
||||
counter.
|
||||
|
||||
## Usage: Basic
|
||||
|
||||
```ts
|
||||
import {
|
||||
readBufferedEvents,
|
||||
writeBufferedEvent,
|
||||
} from "@suaveplan/api-ws-broker-redis/event-buffer";
|
||||
|
||||
const opts = { keyPrefix: "sp:", maxEvents: 256, ttlSec: 300 };
|
||||
await writeBufferedEvent(client, tenant, "orders", {
|
||||
eventId: 1,
|
||||
payload: new Uint8Array([1, 2, 3]),
|
||||
}, opts);
|
||||
const events = await readBufferedEvents(client, tenant, "orders", 0, opts);
|
||||
```
|
||||
|
||||
## Usage: Advanced
|
||||
|
||||
```ts
|
||||
import {
|
||||
bufferKey,
|
||||
decodeBufferMember,
|
||||
} from "@suaveplan/api-ws-broker-redis/event-buffer";
|
||||
|
||||
// Drop into raw Redis to debug a stuck buffer — `decodeBufferMember` parses
|
||||
// each `<eventId>|<base64>` entry back into its `BufferedEvent`.
|
||||
const key = bufferKey("sp:", tenant, "orders");
|
||||
const range = await rawRedis.zrangebyscore(key, "-inf", "+inf");
|
||||
for (const m of range) {
|
||||
const event = decodeBufferMember(m);
|
||||
console.log(event.eventId, event.payload.length);
|
||||
}
|
||||
```
|
||||
|
||||
## API Reference
|
||||
|
||||
- `writeBufferedEvent(client, tenant, topic, event, opts): Promise<void>` —
|
||||
append, enforce cap, refresh TTL.
|
||||
- `readBufferedEvents(client, tenant, topic, sinceEventId, opts):
|
||||
Promise<readonly BufferedEvent[]>` — replay events with `id > sinceEventId`.
|
||||
- `bufferKey(keyPrefix, tenant, topic): string` — key construction helper.
|
||||
- `encodeBufferMember(event)` / `decodeBufferMember(s)` — sorted-set
|
||||
member encoding/decoding (exposed for raw-Redis debugging).
|
||||
- `BufferedEvent { eventId, payload }` and `EventBufferOptions
|
||||
{ keyPrefix, maxEvents, ttlSec }`.
|
||||
|
||||
## Implementation Notes
|
||||
|
||||
The sorted-set strategy was chosen over a Redis Stream because the broker
|
||||
only needs id-keyed range reads (no consumer-group coordination), and a
|
||||
sorted-set's eviction and cap operations are atomic enough for the
|
||||
single-tenant single-topic write path the broker uses. Streams' richer
|
||||
semantics (XCLAIM, XACK) would only add operational complexity without
|
||||
helping the resume use case.
|
||||
|
||||
The base64 step inside `encodeBufferMember` is intentional: sorted-set
|
||||
members are byte strings, but the broker payload is an arbitrary
|
||||
`Uint8Array` that may contain `\r\n` or `\0` bytes. Base64 round-trips
|
||||
through Redis's binary-safe wire protocol without coercion and the cost is
|
||||
33% bigger members, which is well below the network overhead of the
|
||||
publish itself.
|
||||
|
||||
`writeBufferedEvent` re-throws an existing `WsBrokerRedisError` unmodified
|
||||
rather than double-wrapping it; tests cover the no-double-wrap path
|
||||
explicitly.
|
||||
@@ -0,0 +1,87 @@
|
||||
# `health` — broker health snapshot
|
||||
|
||||
<!-- begin:auto-docs -->
|
||||
_(no submodules)_
|
||||
<!-- end:auto-docs -->
|
||||
|
||||
## Purpose
|
||||
|
||||
This module is the pure, side-effect-free evaluator for the broker's health
|
||||
status. It inspects the two underlying RESP clients (one publisher socket,
|
||||
one subscriber socket) and folds their `isReady` flags plus the broker's
|
||||
last observed error into a three-state status (`up`, `degraded`, `down`).
|
||||
The host's readiness endpoint consumes this snapshot to decide whether to
|
||||
report ready, partially ready, or unhealthy.
|
||||
|
||||
Keeping the evaluator pure means tests can probe every state combination
|
||||
deterministically without a real Redis container — the broker's
|
||||
integration test then pins that `up` is reached against a live container
|
||||
and that the snapshot recovers after a forced socket close.
|
||||
|
||||
## Features
|
||||
|
||||
- Three-state status (`up`, `degraded`, `down`) with clear documented
|
||||
semantics — degraded means in-flight publishes will queue per reconnect
|
||||
strategy; down means new publishes will fail fast with `CONNECTION_LOST`.
|
||||
- Pure function — no I/O, no global state, trivially testable.
|
||||
- Optional `lastError` propagation so dashboards can show the most recent
|
||||
underlying socket failure even after the broker has recovered.
|
||||
|
||||
## Usage: Basic
|
||||
|
||||
```ts
|
||||
import { evaluateBrokerHealth } from "@suaveplan/api-ws-broker-redis/health";
|
||||
|
||||
const snap = evaluateBrokerHealth(publisher, subscriber, lastError);
|
||||
if (snap.status === "down") {
|
||||
return new Response("unavailable", { status: 503 });
|
||||
}
|
||||
```
|
||||
|
||||
## Usage: Advanced
|
||||
|
||||
```ts
|
||||
import { evaluateBrokerHealth } from "@suaveplan/api-ws-broker-redis/health";
|
||||
|
||||
// Custom readiness probe that distinguishes degraded vs down.
|
||||
function readiness(): { status: number; body: unknown } {
|
||||
const snap = evaluateBrokerHealth(publisher, subscriber, lastError);
|
||||
if (snap.status === "up") {
|
||||
return { status: 200, body: { status: "ok" } };
|
||||
}
|
||||
if (snap.status === "degraded") {
|
||||
return {
|
||||
status: 200,
|
||||
body: {
|
||||
status: "degraded",
|
||||
publisherReady: snap.publisherReady,
|
||||
subscriberReady: snap.subscriberReady,
|
||||
lastError: snap.lastError,
|
||||
},
|
||||
};
|
||||
}
|
||||
return { status: 503, body: { status: "down", lastError: snap.lastError } };
|
||||
}
|
||||
```
|
||||
|
||||
## API Reference
|
||||
|
||||
- `evaluateBrokerHealth(publisher, subscriber, lastError):
|
||||
BrokerHealthSnapshot` — pure folder over the two clients.
|
||||
- `BrokerHealthSnapshot { status, publisherReady, subscriberReady, lastError? }`.
|
||||
- `BrokerHealthStatus = "up" | "degraded" | "down"`.
|
||||
|
||||
## Implementation Notes
|
||||
|
||||
The `lastError` field is set conditionally so the field is omitted from the
|
||||
snapshot (rather than carrying `undefined`) when no error has ever been
|
||||
observed. This matters for the JSON-serialised readiness response — a
|
||||
dashboard parsing `lastError` should see "absent" vs "present" as a clean
|
||||
boolean, not need to special-case `null` or `undefined`.
|
||||
|
||||
The publisher socket and subscriber socket are evaluated independently
|
||||
because Redis's subscribe-mode connection is a separate TCP socket from the
|
||||
reply-mode one; either can drop without the other. Operators see the actual
|
||||
state of each socket via `publisherReady` / `subscriberReady` so they can
|
||||
diagnose a partial outage (e.g. firewall dropping ESTABLISHED long-lived
|
||||
subscribe sockets) directly from the readiness endpoint.
|
||||
@@ -0,0 +1,106 @@
|
||||
# `instruments` — OTel `ws.broker.*` surface
|
||||
|
||||
<!-- begin:auto-docs -->
|
||||
_(no submodules)_
|
||||
<!-- end:auto-docs -->
|
||||
|
||||
## Purpose
|
||||
|
||||
This module owns the OpenTelemetry instrument surface for the Redis broker.
|
||||
Six instruments cover every documented operational signal: publish attempts
|
||||
and durations, inbound receive counts, reconnect-buffer reads and writes
|
||||
(with hit/miss labelling), and observed connection drops. Operators wire a
|
||||
`MeterProvider` against the global `metrics` API and the broker emits into
|
||||
it without any wiring inside the broker module itself.
|
||||
|
||||
Resolving the meter and instruments lazily (rather than at module load)
|
||||
lets tests install a fresh `MeterProvider` after importing the broker and
|
||||
still observe emissions from production paths. The `resetBrokerMetricsForTests`
|
||||
helper exposes the same lazy cache to test teardown.
|
||||
|
||||
## Features
|
||||
|
||||
- Six instruments, each documented with attribute keys:
|
||||
- `ws.broker.publishes_total` — counter, labels `{ topic, outcome }`.
|
||||
- `ws.broker.publish_duration_ms` — histogram (ms).
|
||||
- `ws.broker.receives_total` — counter, labels `{ topic }`.
|
||||
- `ws.broker.buffer_writes_total` — counter.
|
||||
- `ws.broker.buffer_reads_total` — counter, labels `{ hit }`.
|
||||
- `ws.broker.connection_lost_total` — counter.
|
||||
- Lazy resolution against `metrics.getMeter("@suaveplan/api-ws-broker-redis")`
|
||||
so tests can install their own SDK between cases.
|
||||
- `resetBrokerMetricsForTests()` clears the lazy cache.
|
||||
- No singletons leak into the public surface — every getter returns the
|
||||
meter-bound instrument so tests can assert against them directly.
|
||||
|
||||
## Usage: Basic
|
||||
|
||||
```ts
|
||||
import {
|
||||
publishDurationHistogram,
|
||||
publishesCounter,
|
||||
} from "@suaveplan/api-ws-broker-redis/instruments";
|
||||
|
||||
publishesCounter().add(1, { topic: "orders", outcome: "ok" });
|
||||
publishDurationHistogram().record(12);
|
||||
```
|
||||
|
||||
## Usage: Advanced
|
||||
|
||||
```ts
|
||||
import {
|
||||
metrics,
|
||||
} from "@opentelemetry/api";
|
||||
import {
|
||||
AggregationTemporality,
|
||||
MeterProvider,
|
||||
MetricReader,
|
||||
} from "@opentelemetry/sdk-metrics";
|
||||
import {
|
||||
publishesCounter,
|
||||
resetBrokerMetricsForTests,
|
||||
} from "@suaveplan/api-ws-broker-redis/instruments";
|
||||
|
||||
class CollectingReader extends MetricReader {
|
||||
constructor() {
|
||||
super({
|
||||
aggregationTemporalitySelector: () =>
|
||||
AggregationTemporality.CUMULATIVE,
|
||||
});
|
||||
}
|
||||
protected override async onForceFlush() {}
|
||||
protected override async onShutdown() {}
|
||||
snapshot() {
|
||||
return this.collect();
|
||||
}
|
||||
}
|
||||
|
||||
const reader = new CollectingReader();
|
||||
const provider = new MeterProvider({ readers: [reader] });
|
||||
metrics.setGlobalMeterProvider(provider);
|
||||
resetBrokerMetricsForTests();
|
||||
// Now the broker's emissions land in `reader.snapshot()`.
|
||||
```
|
||||
|
||||
## API Reference
|
||||
|
||||
- `publishesCounter(): Counter`
|
||||
- `publishDurationHistogram(): Histogram`
|
||||
- `receivesCounter(): Counter`
|
||||
- `bufferWritesCounter(): Counter`
|
||||
- `bufferReadsCounter(): Counter`
|
||||
- `connectionLostCounter(): Counter`
|
||||
- `resetBrokerMetricsForTests(): void`
|
||||
|
||||
## Implementation Notes
|
||||
|
||||
Module-local `let` cache variables (`_meter`, `_publishes`, etc.) keep the
|
||||
instrument identity stable across calls within a single process so OTel SDK
|
||||
aggregation works correctly. `resetBrokerMetricsForTests()` re-initialises
|
||||
those to `undefined` so tests can switch SDKs between cases without
|
||||
inheriting the previous run's `Meter` reference.
|
||||
|
||||
The instrument names follow the `ws.broker.*` namespace convention — every
|
||||
WS-related OTel surface in Suaveplan lives under `ws.*` to keep
|
||||
dashboard queries simple, and `ws.broker.*` distinguishes broker-side
|
||||
emissions from engine-side ones (`ws.engine.*`).
|
||||
@@ -0,0 +1,95 @@
|
||||
# `options` — `RedisWsBrokerOptions` and defaults
|
||||
|
||||
<!-- begin:auto-docs -->
|
||||
_(no submodules)_
|
||||
<!-- end:auto-docs -->
|
||||
|
||||
## Purpose
|
||||
|
||||
This module defines the configuration shape consumed by
|
||||
`createRedisWsBroker(...)`. Every operational knob the broker exposes lives
|
||||
here — Redis URL, key/channel prefixes, event-buffer cap and TTL, payload
|
||||
compression threshold, command timeout, reconnect strategy, and the disable
|
||||
flags used by tests. The schema is strict: unknown keys at the wire-options
|
||||
boundary fail fast rather than landing as silently-ignored typos.
|
||||
|
||||
The module also pins the `MAX_PAYLOAD_BYTES` 4 MiB hard cap. That ceiling is
|
||||
deliberately matched to typical Redis client defaults so the broker stays safe
|
||||
in shared deployments where one greedy publisher could otherwise flood the
|
||||
fan-out fabric.
|
||||
|
||||
## Features
|
||||
|
||||
- Single Zod schema (`redisWsBrokerOptionsSchema`) with `strict()` mode,
|
||||
validated at the constructor boundary so misconfigurations never reach the
|
||||
socket layer.
|
||||
- Documented defaults exported as constants (`DEFAULT_KEY_PREFIX`,
|
||||
`DEFAULT_CHANNEL_PREFIX`, `DEFAULT_EVENT_BUFFER_TTL_SEC`,
|
||||
`DEFAULT_EVENT_BUFFER_MAX_EVENTS`, `DEFAULT_COMPRESSION_THRESHOLD_BYTES`,
|
||||
`DEFAULT_COMMAND_TIMEOUT_MS`, `DEFAULT_RECONNECT_STRATEGY`) so operators
|
||||
can rebuild a known-good shape programmatically.
|
||||
- `resolveRedisWsBrokerOptions(opts)` returns a fully-defaulted bundle
|
||||
(`ResolvedRedisWsBrokerOptions`) without any optional fields — internal
|
||||
code never has to defend against `undefined`.
|
||||
- Partial `reconnectStrategy` overrides are honoured field-by-field; the
|
||||
rest backfills from `DEFAULT_RECONNECT_STRATEGY`.
|
||||
- `randomHostId()` mints a per-host identifier when `hostId` is omitted,
|
||||
so multi-host deployments can attribute broadcasts without any extra
|
||||
configuration.
|
||||
|
||||
## Usage: Basic
|
||||
|
||||
```ts
|
||||
import { resolveRedisWsBrokerOptions } from "@suaveplan/api-ws-broker-redis/options";
|
||||
|
||||
const opts = resolveRedisWsBrokerOptions({
|
||||
connection: "redis://localhost:6379",
|
||||
});
|
||||
// opts.keyPrefix === "suaveplan:ws:"
|
||||
// opts.channelPrefix === "ws.broadcast."
|
||||
// opts.eventBufferTtlSec === 300
|
||||
```
|
||||
|
||||
## Usage: Advanced
|
||||
|
||||
```ts
|
||||
import {
|
||||
DEFAULT_RECONNECT_STRATEGY,
|
||||
redisWsBrokerOptionsSchema,
|
||||
} from "@suaveplan/api-ws-broker-redis/options";
|
||||
|
||||
const raw = JSON.parse(env.BROKER_CONFIG_JSON);
|
||||
const parsed = redisWsBrokerOptionsSchema.safeParse(raw);
|
||||
if (!parsed.success) throw new Error(parsed.error.message);
|
||||
|
||||
const stagingStrategy = {
|
||||
...DEFAULT_RECONNECT_STRATEGY,
|
||||
maxAttempts: 1_000,
|
||||
maxDelayMs: 10_000,
|
||||
};
|
||||
```
|
||||
|
||||
## API Reference
|
||||
|
||||
- `RedisWsBrokerOptions` — public, wire-validated option shape.
|
||||
- `ResolvedRedisWsBrokerOptions` — fully-defaulted bundle used internally.
|
||||
- `RedisReconnectStrategy` — exponential-backoff configuration.
|
||||
- `redisWsBrokerOptionsSchema` — strict Zod schema.
|
||||
- `resolveRedisWsBrokerOptions(opts)` — apply defaults.
|
||||
- `MAX_PAYLOAD_BYTES` — 4 MiB hard cap.
|
||||
- `DEFAULT_*` — every documented default exported as a constant.
|
||||
|
||||
## Implementation Notes
|
||||
|
||||
The schema separates "wire options" (validated, strict) from "factory options"
|
||||
(broker-only knobs like `clientFactory` and `readyTimeoutMs`). The broker
|
||||
constructor strips the latter before parsing so the schema does not reject
|
||||
them, while still failing fast on unrecognised keys for the genuine wire
|
||||
shape. Tests cover both paths explicitly.
|
||||
|
||||
`randomHostId()` uses `Math.random()` for the host suffix. This is fine for
|
||||
attribution — collisions on an 8-byte random tag are astronomically rare and
|
||||
the broker would still be correct under collision; it just becomes harder
|
||||
to attribute a publish on a dashboard. Operators with stricter requirements
|
||||
should pass an explicit `hostId` derived from their orchestrator's
|
||||
identifier.
|
||||
@@ -0,0 +1,86 @@
|
||||
# `publish` — envelope encode/decode
|
||||
|
||||
<!-- begin:auto-docs -->
|
||||
_(no submodules)_
|
||||
<!-- end:auto-docs -->
|
||||
|
||||
## Purpose
|
||||
|
||||
This module handles wire-encoding for the Redis broker. Every published frame
|
||||
gets a one-byte encoding tag (`ENCODING_JSON` for uncompressed JSON,
|
||||
`ENCODING_GZIP_JSON` for gzipped JSON), the JSON bytes for the envelope, and
|
||||
nothing else. The same module owns the inverse path — `decodeEnvelope` — and
|
||||
all the input-validation that protects subscribers from non-broker traffic
|
||||
landing on a misconfigured channel.
|
||||
|
||||
Compression is applied conditionally based on the encoded JSON byte length
|
||||
versus the broker's configured `compressionThresholdBytes`. The 4 MiB hard
|
||||
cap (`MAX_PAYLOAD_BYTES`) is enforced both before and after compression so a
|
||||
runaway gzip-stream-bomb cannot sneak past the threshold check.
|
||||
|
||||
## Features
|
||||
|
||||
- One-byte encoding magic at the head of every frame. Unknown bytes fail
|
||||
fast with `WS_BROKER_REDIS.DESERIALISE_FAILED`.
|
||||
- Zod-validated envelope schema applied during decode so a misshapen frame
|
||||
cannot reach a subscriber handler.
|
||||
- Gzip applied via `node:zlib`'s `gzipSync`/`gunzipSync` — synchronous,
|
||||
deterministic, no streaming state to leak across frames.
|
||||
- Hard 4 MiB cap enforced both pre- and post-compression.
|
||||
- Pure encode and decode functions — no I/O, trivially testable, and the
|
||||
broker's `publish` path calls these once per envelope.
|
||||
|
||||
## Usage: Basic
|
||||
|
||||
```ts
|
||||
import { decodeEnvelope, encodeEnvelope } from "@suaveplan/api-ws-broker-redis/publish";
|
||||
|
||||
const frame = encodeEnvelope(
|
||||
{ topic: "orders", tenant: "t1" as any, payload: { id: 1 } },
|
||||
{ compressionThresholdBytes: 1024, disableCompression: false }
|
||||
);
|
||||
const back = decodeEnvelope<{ id: number }>(frame.bytes);
|
||||
```
|
||||
|
||||
## Usage: Advanced
|
||||
|
||||
```ts
|
||||
import {
|
||||
encodeEnvelope,
|
||||
ENCODING_GZIP_JSON,
|
||||
} from "@suaveplan/api-ws-broker-redis/publish";
|
||||
|
||||
const frame = encodeEnvelope(envelope, {
|
||||
compressionThresholdBytes: 0, // compress everything
|
||||
disableCompression: false,
|
||||
});
|
||||
if (frame.bytes[0] !== ENCODING_GZIP_JSON) {
|
||||
throw new Error("expected gzip encoding for any-size payload");
|
||||
}
|
||||
// Inspect the actual ratio for tuning the threshold:
|
||||
const ratio = frame.bytes.length / frame.rawByteLength;
|
||||
metrics.compressionRatio.record(ratio);
|
||||
```
|
||||
|
||||
## API Reference
|
||||
|
||||
- `encodeEnvelope(envelope, opts)` — encode a `BrokerEnvelope` into the wire
|
||||
frame. Returns `{ bytes, compressed, rawByteLength }`. Throws
|
||||
`WS_BROKER_REDIS.PAYLOAD_TOO_LARGE` for frames above the hard cap.
|
||||
- `decodeEnvelope<Payload>(bytes)` — inverse of `encodeEnvelope`. Throws
|
||||
`WS_BROKER_REDIS.DESERIALISE_FAILED` for any framing inconsistency.
|
||||
- `EncodedFrame` — return shape of `encodeEnvelope`.
|
||||
- `ENCODING_JSON`, `ENCODING_GZIP_JSON` — magic-byte constants.
|
||||
|
||||
## Implementation Notes
|
||||
|
||||
The schema rebuilds a brand-stripped wire envelope (`tenant: string`) on
|
||||
decode because the `TenantId` brand is a TypeScript type, not a runtime value.
|
||||
The shape is otherwise identical to the runtime envelope, and the broker
|
||||
re-applies the brand at the call site via the typed entry points.
|
||||
|
||||
Gzip is chosen over deflate or brotli because Redis pub/sub frames are
|
||||
short-lived (the broker is the storage substrate, not a long-term archive)
|
||||
and gzip's framing overhead is the lowest of the three for typical payloads
|
||||
in the 1–16 KiB range. The compression threshold defaults to 1 KiB so tiny
|
||||
events do not pay the framing cost just to land a few bytes shorter.
|
||||
@@ -0,0 +1,90 @@
|
||||
# `reconnect` — exponential backoff policy
|
||||
|
||||
<!-- begin:auto-docs -->
|
||||
_(no submodules)_
|
||||
<!-- end:auto-docs -->
|
||||
|
||||
## Purpose
|
||||
|
||||
This module is the deterministic policy evaluator for the broker's reconnect
|
||||
loop. It computes the next backoff delay in milliseconds and decides whether
|
||||
to give up after a configured number of attempts. The broker drives an
|
||||
external timer; everything load-bearing lives in pure functions that tests
|
||||
can exercise without any I/O or fake-timer harness.
|
||||
|
||||
The policy is exponential by default — `initialDelayMs` doubles per attempt
|
||||
up to `maxDelayMs`. The shape is deliberately simple so operators can reason
|
||||
about the curve from the resolved options alone, without reading the broker
|
||||
implementation.
|
||||
|
||||
## Features
|
||||
|
||||
- `nextReconnectDelayMs(attempt, strategy)` — pure computation. Returns
|
||||
`initialDelayMs` at attempt 0, doubles per attempt, clamps at
|
||||
`maxDelayMs`. Negative attempts collapse to the initial delay so callers
|
||||
cannot drive the policy out of bounds via off-by-one bugs.
|
||||
- `shouldGiveUp(attempt, strategy)` — pure predicate. Returns true once
|
||||
`attempt >= strategy.maxAttempts`.
|
||||
- No side effects, no global state, no timers — both functions are
|
||||
trivially testable and stable across runs.
|
||||
|
||||
## Usage: Basic
|
||||
|
||||
```ts
|
||||
import {
|
||||
nextReconnectDelayMs,
|
||||
shouldGiveUp,
|
||||
} from "@suaveplan/api-ws-broker-redis/reconnect";
|
||||
|
||||
let attempt = 0;
|
||||
async function reconnectLoop() {
|
||||
while (!shouldGiveUp(attempt, strategy)) {
|
||||
const delay = nextReconnectDelayMs(attempt, strategy);
|
||||
await new Promise((r) => setTimeout(r, delay));
|
||||
if (await tryConnect()) return;
|
||||
attempt += 1;
|
||||
}
|
||||
throw new Error("reconnect exhausted");
|
||||
}
|
||||
```
|
||||
|
||||
## Usage: Advanced
|
||||
|
||||
```ts
|
||||
import {
|
||||
nextReconnectDelayMs,
|
||||
shouldGiveUp,
|
||||
} from "@suaveplan/api-ws-broker-redis/reconnect";
|
||||
|
||||
// Render the curve for an operator dashboard.
|
||||
function curve(strategy: RedisReconnectStrategy): number[] {
|
||||
const out: number[] = [];
|
||||
for (let i = 0; !shouldGiveUp(i, strategy); i += 1) {
|
||||
out.push(nextReconnectDelayMs(i, strategy));
|
||||
if (out.length > 200) break; // dashboard sanity cap
|
||||
}
|
||||
return out;
|
||||
}
|
||||
```
|
||||
|
||||
## API Reference
|
||||
|
||||
- `nextReconnectDelayMs(attempt: number, strategy: RedisReconnectStrategy):
|
||||
number` — exponentially increasing delay in ms, clamped at `maxDelayMs`.
|
||||
- `shouldGiveUp(attempt: number, strategy: RedisReconnectStrategy):
|
||||
boolean` — whether the reconnect loop must abort.
|
||||
|
||||
## Implementation Notes
|
||||
|
||||
The exponential formula `initialDelayMs * multiplier ** attempt` is computed
|
||||
in floating-point; `Math.min(...)` clamps to the integer ceiling. For the
|
||||
default `multiplier = 2`, this matches a pure doubling sequence
|
||||
(100, 200, 400, 800, ...) with no rounding drift below 53 bits, which is
|
||||
well above any sensible `maxAttempts`.
|
||||
|
||||
The decision to collapse negative attempts into the initial delay rather
|
||||
than throw is deliberate: the broker's reconnect loop computes the next
|
||||
attempt from the previous one, and a transient `attempt = -1` state during
|
||||
state restoration would otherwise crash the loop. Clamping silently is the
|
||||
safer default — the loop still makes progress on the legitimate next
|
||||
attempt without manual recovery.
|
||||
@@ -0,0 +1,105 @@
|
||||
# `resp-client` — minimal built-in Redis client
|
||||
|
||||
<!-- begin:auto-docs -->
|
||||
_(no submodules)_
|
||||
<!-- end:auto-docs -->
|
||||
|
||||
## Purpose
|
||||
|
||||
This module ships the broker's built-in RESP-protocol client so the package
|
||||
has no mandatory runtime dependency on an external Redis client. The client
|
||||
opens a single TCP socket via `node:net`, frames every command as an inline
|
||||
RESP array, and decodes replies incrementally — RESP guarantees FIFO replies
|
||||
on a single connection, so the client enqueues each command's promise and
|
||||
resolves them in arrival order. Subscribe-mode clients receive `pmessage`
|
||||
frames asynchronously and route them to registered listeners.
|
||||
|
||||
The client implements {@link RedisPubSubLike} so production users who already
|
||||
operate `ioredis` or `node-redis` can ship an adapter against the same
|
||||
interface and skip the built-in client entirely.
|
||||
|
||||
## Features
|
||||
|
||||
- Incremental RESP parser handling simple-string, error, integer, bulk
|
||||
string, and array reply types. Streams across multiple TCP `data` chunks.
|
||||
- Per-command timeout via `commandTimeoutMs` (default 5 s).
|
||||
- Socket-close detection rejects every in-flight command so callers do not
|
||||
hang indefinitely on a dropped connection.
|
||||
- Two role modes: reply-mode (default) and subscribe-mode (`isSubscriber: true`).
|
||||
Subscribe-mode handles `pmessage` / `psubscribe` / `punsubscribe` ack frames
|
||||
and forwards messages through the listener API.
|
||||
- Injectable `socketFactory` so tests can drive the parser end-to-end against
|
||||
a `FakeSocket` without booting a real TCP loopback — the parser code path
|
||||
exercised is exactly the production one.
|
||||
- Idempotent `quit()`; subsequent commands reject immediately with the
|
||||
documented closed-client message.
|
||||
|
||||
## Usage: Basic
|
||||
|
||||
```ts
|
||||
import { createRespClient } from "@suaveplan/api-ws-broker-redis/resp-client";
|
||||
|
||||
const client = createRespClient({ url: "redis://localhost:6379" });
|
||||
await client.publish("orders", new Uint8Array([1, 2, 3]));
|
||||
await client.quit();
|
||||
```
|
||||
|
||||
## Usage: Advanced
|
||||
|
||||
```ts
|
||||
import {
|
||||
createRespClient,
|
||||
type RedisPubSubLike,
|
||||
} from "@suaveplan/api-ws-broker-redis/resp-client";
|
||||
|
||||
// Subscribe-mode client wired to a listener.
|
||||
const subscriber: RedisPubSubLike = createRespClient({
|
||||
url: "redis://localhost:6379",
|
||||
isSubscriber: true,
|
||||
commandTimeoutMs: 2_000,
|
||||
});
|
||||
|
||||
const off = subscriber.on((evt) => {
|
||||
if (evt.event === "message") {
|
||||
console.log("ws.* frame on", evt.channel, evt.payload);
|
||||
} else if (evt.event === "error") {
|
||||
metrics.connectionErrors.add(1);
|
||||
}
|
||||
});
|
||||
|
||||
await subscriber.psubscribe("ws.broadcast.*");
|
||||
// ... later ...
|
||||
off();
|
||||
await subscriber.quit();
|
||||
```
|
||||
|
||||
## API Reference
|
||||
|
||||
- `createRespClient(options): RedisPubSubLike` — factory.
|
||||
- `RedisPubSubLike` — the minimal Redis surface (publish, psubscribe,
|
||||
punsubscribe, zadd, zremrangebyrank, zrangebyscore, zcard, expire,
|
||||
ping, del, quit, on, isReady).
|
||||
- `RedisPubSubListener` — discriminated event union (`message`, `ready`,
|
||||
`close`, `error`).
|
||||
- `RespClientOptions { url, commandTimeoutMs?, socketFactory?, isSubscriber? }`.
|
||||
|
||||
## Implementation Notes
|
||||
|
||||
The parser is a strict bottom-up recursive descent over the RESP grammar; the
|
||||
nested-array path covers `pmessage` (4-element) and `psubscribe`/`punsubscribe`
|
||||
ack (3-element) frames the broker subscribes to. The `+`, `-`, `:` types are
|
||||
single-line; bulk strings (`$`) carry an explicit byte length so binary
|
||||
payloads survive without coercion; arrays (`*`) recursively defer to the
|
||||
parser for each element.
|
||||
|
||||
Subscribe-mode `PING` returns an array (`*2\r\n$4\r\npong\r\n$0\r\n\r\n`)
|
||||
rather than the reply-mode `+PONG\r\n` simple string. The dispatcher detects
|
||||
this via the leading bulk-string element and resolves to the canonical
|
||||
`"PONG"` so callers see the same value regardless of mode.
|
||||
|
||||
Timeouts use `setTimeout` rather than `AbortController` because the pending
|
||||
queue index is the cancellation surface — on timeout the entry is spliced
|
||||
out before reject, so a late reply lands as an orphan and is dropped by the
|
||||
next reply handler's `pending.shift()` returning a defined but unrelated
|
||||
entry. (This is a known and accepted edge case under load: callers always
|
||||
see the timeout reject, never an out-of-order resolve.)
|
||||
@@ -0,0 +1,96 @@
|
||||
# `subscribe` — local routing + topic / channel mapping
|
||||
|
||||
<!-- begin:auto-docs -->
|
||||
_(no submodules)_
|
||||
<!-- end:auto-docs -->
|
||||
|
||||
## Purpose
|
||||
|
||||
This module owns the broker's local routing table — the in-process map from
|
||||
topic name to handler set that the inbound pub/sub stream targets. Frames
|
||||
land on the subscriber socket as `pmessage` arrays with the full channel
|
||||
name; the helpers here strip the `channelPrefix`, look up the local handler
|
||||
set, and dispatch the decoded envelope to every subscriber.
|
||||
|
||||
Per-handler error isolation is the load-bearing invariant: Redis does not
|
||||
stop pushing frames just because one subscriber rejected its handler call,
|
||||
so any rejection (including non-`Error` thrown values) is wrapped and routed
|
||||
through an `onError` callback rather than allowed to abort the loop.
|
||||
|
||||
## Features
|
||||
|
||||
- `LocalSubscriberRegistry<Payload>` — `Map<topic, Set<handler>>` with O(1)
|
||||
add/remove and a `snapshot(topic)` helper that returns a defensive copy
|
||||
for safe iteration during fan-out.
|
||||
- `channelFromTopic(topic, prefix)` / `topicFromChannel(channel, prefix)` —
|
||||
the inverse functions stripping or building the `<prefix><topic>` channel
|
||||
name. Returns `null` for non-matching channels so the broker can drop
|
||||
cross-traffic without crashing.
|
||||
- `dispatchInboundFrame(registry, topic, bytes, onError)` — decode +
|
||||
fan-out wrapped in per-subscriber error isolation. Any decode failure or
|
||||
handler rejection routes through `onError` and never propagates back to
|
||||
the subscribe loop.
|
||||
- Frame-decode coverage of every documented `WS_BROKER_REDIS.DESERIALISE_*`
|
||||
path — empty frame, unknown magic byte, malformed JSON, schema mismatch,
|
||||
gzip framing inconsistency.
|
||||
|
||||
## Usage: Basic
|
||||
|
||||
```ts
|
||||
import {
|
||||
LocalSubscriberRegistry,
|
||||
channelFromTopic,
|
||||
dispatchInboundFrame,
|
||||
} from "@suaveplan/api-ws-broker-redis/subscribe";
|
||||
|
||||
const registry = new LocalSubscriberRegistry<{ id: number }>();
|
||||
registry.add("orders", (env) => console.log("delivered", env.payload.id));
|
||||
const channel = channelFromTopic("orders", "ws.broadcast.");
|
||||
```
|
||||
|
||||
## Usage: Advanced
|
||||
|
||||
```ts
|
||||
import {
|
||||
dispatchInboundFrame,
|
||||
LocalSubscriberRegistry,
|
||||
} from "@suaveplan/api-ws-broker-redis/subscribe";
|
||||
|
||||
const registry = new LocalSubscriberRegistry();
|
||||
const subscribeErrors: Error[] = [];
|
||||
|
||||
await dispatchInboundFrame(
|
||||
registry,
|
||||
"orders",
|
||||
inboundFrame,
|
||||
(err) => {
|
||||
// Surface the decode/handler failure on a sidecar channel — the
|
||||
// pub/sub loop must keep running so other topics stay healthy.
|
||||
subscribeErrors.push(err);
|
||||
}
|
||||
);
|
||||
```
|
||||
|
||||
## API Reference
|
||||
|
||||
- `class LocalSubscriberRegistry<Payload>` — `add`, `remove`, `size`,
|
||||
`snapshot`, `clear`.
|
||||
- `channelFromTopic(topic, prefix): string` — topic → channel name.
|
||||
- `topicFromChannel(channel, prefix): string | null` — inverse; `null` for
|
||||
non-matching channels.
|
||||
- `dispatchInboundFrame(registry, topic, bytes, onError): Promise<void>` —
|
||||
decode + isolated fan-out.
|
||||
|
||||
## Implementation Notes
|
||||
|
||||
`snapshot()` materialises an `Array<handler>` rather than returning the
|
||||
underlying `Set` directly so a handler that calls `registry.remove(...)`
|
||||
mid-fan-out does not silently skip its peers later in the loop. The cost is
|
||||
one allocation per topic per dispatch, which is negligible against the
|
||||
inbound frame's parse and decode work.
|
||||
|
||||
`dispatchInboundFrame` wraps non-`Error` rejections (e.g. `Promise.reject("boom")`)
|
||||
into a real `Error` before invoking `onError`. This keeps the callback's
|
||||
signature stable regardless of what the handler threw — operators consuming
|
||||
the error never have to defend against `string`, `null`, or arbitrary
|
||||
objects landing on the wire.
|
||||
Reference in New Issue
Block a user