268 lines
19 KiB
Markdown
268 lines
19 KiB
Markdown
<!-- markdownlint-disable MD024 -->
|
||
|
||
# Capability Spec Delta: api-plugin-sse (add-api-plugin-sse)
|
||
|
||
**Package**: `@suaveplan/api-plugin-sse`
|
||
**Status**: PROPOSED
|
||
|
||
> Every requirement below carries a `**Verified by:**` line at authoring time. No exceptions.
|
||
|
||
## ADDED Requirements
|
||
|
||
### Requirement: createSsePlugin returns a Zod-validated PluginDescriptor
|
||
|
||
The package SHALL export `createSsePlugin(options): PluginDescriptor<SsePluginConfig>` that validates `options` against `SsePluginOptionsSchema` at the factory boundary. Invalid input SHALL raise `SseError(SSE_INVALID_INPUT)` carrying the `ZodError` as `cause`. The returned descriptor SHALL declare `name: "sse"`, the resolved `configSchema`, and the four lifecycle hooks (`onRegister`, `onInit`, `onReady`, `onClose`) of which at least `onInit` and `onClose` are populated.
|
||
|
||
#### Scenario: factory produces a valid descriptor
|
||
|
||
- **WHEN** `createSsePlugin({})` is called with defaults
|
||
- **THEN** the returned descriptor satisfies `PluginDescriptor<SsePluginConfig>` with `name === "sse"` AND `onInit` / `onClose` defined
|
||
|
||
#### Scenario: invalid options raise SSE_INVALID_INPUT
|
||
|
||
- **WHEN** `createSsePlugin({ heartbeatIntervalMs: -1 } as never)` is called
|
||
- **THEN** `SseError(SSE_INVALID_INPUT)` is thrown with `cause instanceof ZodError`
|
||
|
||
**Verified by:** `src/plugin/plugin.test.ts::"createSsePlugin returns a valid PluginDescriptor"` AND `src/plugin/plugin.test.ts::"invalid options raise SSE_INVALID_INPUT"`
|
||
|
||
---
|
||
|
||
### Requirement: onInit registers sseService in ServiceRegistry and mounts default route
|
||
|
||
The descriptor's `onInit(ctx)` hook SHALL: (1) resolve `app` from `ctx`'s ServiceRegistry — when absent throw `SseError(SSE_SERVICE_NOT_AVAILABLE)` listing `app` as missing; (2) construct a `RingBuffer`, an `InProcessFanout`, and an `SseService`; (3) call `ctx.registerService("sseService", service)` so downstream plugins / route handlers consume it through the canonical registry; (4) when `options.attachDefaultRoute !== false`, mount `GET <routePrefix>/:channel` (default `/sse/:channel`) wired to `createSseRoute(service, ...)`. `onClose` SHALL call `service.close()` to terminate every open connection.
|
||
|
||
#### Scenario: missing app raises SSE_SERVICE_NOT_AVAILABLE
|
||
|
||
- **WHEN** the plugin initializes against a context whose `getService("app")` returns `undefined`
|
||
- **THEN** `SseError(SSE_SERVICE_NOT_AVAILABLE)` is thrown with `context.missing === ["app"]`
|
||
|
||
#### Scenario: sseService is registered and the default route is mounted
|
||
|
||
- **WHEN** the plugin initializes against a context with a Hono `app` registered
|
||
- **THEN** `ctx.getService("sseService")` returns the `SseService` AND a `GET /sse/<channel>` request returns `Content-Type: text/event-stream`
|
||
|
||
#### Scenario: attachDefaultRoute: false skips mounting
|
||
|
||
- **WHEN** the plugin initializes with `attachDefaultRoute: false`
|
||
- **THEN** `ctx.getService("sseService")` returns the service AND no route is registered under `routePrefix`
|
||
|
||
#### Scenario: onClose terminates every open connection
|
||
|
||
- **WHEN** the plugin's `onClose` runs after a subscriber is active
|
||
- **THEN** `service.stats().activeConnections === 0` after the hook resolves
|
||
|
||
**Verified by:** `src/plugin/on-init.test.ts::"missing app raises SSE_SERVICE_NOT_AVAILABLE"` AND `src/plugin/on-init.test.ts::"registers sseService and mounts default route"` AND `src/plugin/on-init.test.ts::"attachDefaultRoute false skips route registration"` AND `src/plugin/on-init.test.ts::"onClose terminates active connections"`
|
||
|
||
---
|
||
|
||
### Requirement: SseService exposes broadcast / subscribe / close / stats
|
||
|
||
The package SHALL register `sseService` in the ServiceRegistry with the typed surface `{ broadcast(channel, event): void; subscribe(args): SubscriptionHandle; close(channel?): Promise<void>; stats(): SseStats }`. `broadcast` SHALL append the event to the channel's ring buffer with an id supplied by the configured `idGenerator` (when the event has no id) AND publish it through the `Fanout` so every active subscriber to that channel receives it. `subscribe` SHALL: (a) assign a connection id; (b) enforce `maxTotalConnections` (throw `SseError(SSE_TOO_MANY_CONNECTIONS)`) and `maxConnectionsPerPrincipal` (throw `SseError(SSE_PER_PRINCIPAL_LIMIT_EXCEEDED)`); (c) when `lastEventId` is supplied, replay the buffered events strictly after that id before forwarding live events; (d) return a `SubscriptionHandle` exposing `id`, `events` (an `AsyncIterableIterator<SseEvent>`), and `close()`. `close(channel?)` SHALL terminate every subscription (optionally restricted to a single channel) and resolve once their iterators have completed. `stats()` SHALL return `{ activeConnections: number; perChannel: Record<string, number> }`.
|
||
|
||
#### Scenario: broadcast appends to buffer and fans out to subscribers
|
||
|
||
- **WHEN** a subscriber is active on channel `notifications` and `service.broadcast("notifications", { data: { kind: "ping" } })` is called
|
||
- **THEN** the subscriber's iterator yields an `SseEvent` with `data.kind === "ping"` AND the buffer's `since("notifications", "")` includes that event
|
||
|
||
#### Scenario: subscribe with Last-Event-ID replays missed events
|
||
|
||
- **WHEN** broadcasts produce ids `1`, `2`, `3` on a channel and a new subscriber connects with `lastEventId: "1"`
|
||
- **THEN** the iterator yields events `2` and `3` in order before any live event
|
||
|
||
#### Scenario: maxConnectionsPerPrincipal enforces per-principal cap
|
||
|
||
- **WHEN** `principalId: "u1"` already holds the configured cap and another `subscribe` is attempted with the same principalId
|
||
- **THEN** `SseError(SSE_PER_PRINCIPAL_LIMIT_EXCEEDED)` is thrown with `context.principalId === "u1"`
|
||
|
||
#### Scenario: maxTotalConnections enforces global cap
|
||
|
||
- **WHEN** `activeConnections === maxTotalConnections` and another `subscribe` is attempted
|
||
- **THEN** `SseError(SSE_TOO_MANY_CONNECTIONS)` is thrown with `context.max` carrying the cap
|
||
|
||
#### Scenario: close terminates active connections
|
||
|
||
- **WHEN** `service.close("notifications")` is called with two subscribers on that channel
|
||
- **THEN** both iterators complete AND `service.stats().perChannel["notifications"]` is `undefined` or `0`
|
||
|
||
**Verified by:** `src/services/service.test.ts::"broadcast appends to buffer and fans out to subscribers"` AND `src/services/service.test.ts::"subscribe with Last-Event-ID replays missed events"` AND `src/services/service.test.ts::"maxConnectionsPerPrincipal enforces per-principal cap"` AND `src/services/service.test.ts::"maxTotalConnections enforces global cap"` AND `src/services/service.test.ts::"close terminates active connections"`
|
||
|
||
---
|
||
|
||
### Requirement: Hono SSE handler streams text/event-stream with heartbeats, abort, and resume
|
||
|
||
`createSseRoute(service, options)` SHALL return a Hono `Handler` that: (a) reads `:channel` from the path; missing → throw `SseError(SSE_CHANNEL_REQUIRED)`; (b) reads `Last-Event-ID` from request headers; (c) calls `service.subscribe({ channel, lastEventId, principalId: options.resolvePrincipalId?.(c), signal: c.req.raw.signal })`; (d) returns a Hono `streamSSE` response whose body writes one `id: <id>\nevent: <event>\ndata: <json>\n\n` frame per event AND a `: heartbeat\n\n` comment on the configured `heartbeatIntervalMs`; (e) on stream abort (client disconnect), unsubscribes by calling `handle.close()`; (f) on backpressure detection (`controller.desiredSize === 0` for longer than one heartbeat tick), drops or buffers per `overflowPolicy` and increments `sse.queue_overflow_total{policy}`. The response status SHALL be `200` with `Content-Type: text/event-stream; charset=utf-8`, `Cache-Control: no-cache, no-transform`, `X-Accel-Buffering: no`, and `Connection: keep-alive`.
|
||
|
||
#### Scenario: handler streams framed events to a real fetch client
|
||
|
||
- **WHEN** a real Hono app with the plugin mounted serves `GET /sse/notifications` and a `fetch` client subscribes, then `service.broadcast("notifications", { data: { n: 1 } })`, `{ data: { n: 2 } }`, `{ data: { n: 3 } }` run
|
||
- **THEN** the client receives three `SseEvent`s in order with ids `1`, `2`, `3` (or whatever the configured idGenerator emits) AND the response carries the SSE content-type headers above
|
||
|
||
#### Scenario: Last-Event-ID resumes from the ring buffer
|
||
|
||
- **WHEN** the client reconnects with `Last-Event-ID: 2` after the same broadcasts
|
||
- **THEN** the client receives only event `3` before live events begin
|
||
|
||
#### Scenario: heartbeats fire on the configured interval
|
||
|
||
- **WHEN** `heartbeatIntervalMs` elapses on an open connection with a FakeClock-driven timer
|
||
- **THEN** the client observes a `: heartbeat\n\n` comment frame AND `sse.heartbeats_total` increments by one
|
||
|
||
#### Scenario: abort signal closes the subscription
|
||
|
||
- **WHEN** the client `AbortController.abort()`s mid-stream
|
||
- **THEN** `service.stats().activeConnections` drops by one within one event-loop tick
|
||
|
||
**Verified by:** `src/middleware/sse-handler.test.ts::"streams framed events to a real fetch client"` AND `src/middleware/sse-handler.test.ts::"Last-Event-ID resumes from the ring buffer"` AND `src/middleware/sse-handler.test.ts::"heartbeats fire on the configured interval"` AND `src/middleware/sse-handler.test.ts::"abort signal closes the subscription"` AND `tests/integration/full-stream.test.ts::"end-to-end fetch subscriber receives three ordered events"` AND `tests/integration/full-stream.test.ts::"reconnect with Last-Event-ID skips delivered events"`
|
||
|
||
---
|
||
|
||
### Requirement: SSE wire-format serializer and streaming parser
|
||
|
||
The package SHALL export `formatSseFrame(event: SseEvent): string` that serialises an event into the SSE wire format (`id: <id>\n`, optional `event: <name>\n`, optional `retry: <ms>\n`, `data: <serialised>\n\n`). `data` SHALL be JSON-stringified when not already a string; multi-line string `data` SHALL be split into one `data:` line per source line. The package SHALL also export `parseSseStream(reader)` — an async iterator that decodes a `ReadableStreamDefaultReader<Uint8Array>` into `SseEvent`s, used by the integration test client and by any consumer that needs to parse SSE.
|
||
|
||
#### Scenario: formatSseFrame emits id, event, data with correct delimiters
|
||
|
||
- **WHEN** `formatSseFrame({ id: "42", event: "ping", data: { n: 1 } })` is called
|
||
- **THEN** the result equals `"id: 42\nevent: ping\ndata: {\"n\":1}\n\n"`
|
||
|
||
#### Scenario: multi-line string data is split into multiple data: lines
|
||
|
||
- **WHEN** `formatSseFrame({ id: "1", data: "a\nb" })` is called
|
||
- **THEN** the result contains `"data: a\ndata: b\n"`
|
||
|
||
#### Scenario: parseSseStream yields events for each frame
|
||
|
||
- **WHEN** a stream emits two frames separated by a blank line
|
||
- **THEN** the iterator yields two `SseEvent`s with the parsed `id`, `event`, and `data` fields
|
||
|
||
**Verified by:** `src/formatter/format.test.ts::"emits id, event, data with correct delimiters"` AND `src/formatter/format.test.ts::"multi-line string data splits into multiple data lines"` AND `src/formatter/parse.test.ts::"yields events for each frame in the stream"`
|
||
|
||
---
|
||
|
||
### Requirement: Bounded per-channel ring buffer powers replay
|
||
|
||
`createRingBuffer(retentionPerChannel)` SHALL return an object with `append(channel: string, event: SseEvent): void` and `since(channel: string, lastEventId: string): readonly SseEvent[]`. The buffer SHALL retain at most `retentionPerChannel` events per channel (oldest evicted on overflow). `since` SHALL return all events whose ids strictly follow `lastEventId` (oldest first) and an empty array when `lastEventId` is not found in the buffer (so a stale client starts fresh rather than receiving the entire buffer).
|
||
|
||
#### Scenario: append + since returns events after the cursor
|
||
|
||
- **WHEN** `append("c", { id: "1", data: {} })`, `append("c", { id: "2", data: {} })`, `append("c", { id: "3", data: {} })` and `since("c", "1")` is called
|
||
- **THEN** the returned array contains events with ids `2` and `3` in order
|
||
|
||
#### Scenario: retention rolls over
|
||
|
||
- **WHEN** `retention = 2` and three events are appended
|
||
- **THEN** `since("c", "")` returns the latest two events only
|
||
|
||
#### Scenario: unknown lastEventId returns empty
|
||
|
||
- **WHEN** `since("c", "non-existent")` is called against a non-empty buffer
|
||
- **THEN** the returned array is empty
|
||
|
||
**Verified by:** `src/buffer/ring-buffer.test.ts::"append + since returns events after the cursor"` AND `src/buffer/ring-buffer.test.ts::"retention rolls over"` AND `src/buffer/ring-buffer.test.ts::"unknown lastEventId returns empty"`
|
||
|
||
---
|
||
|
||
### Requirement: Fanout interface and in-process implementation
|
||
|
||
The package SHALL export a `Fanout` interface — `publish(channel: string, event: SseEvent): void` and `subscribe(channel: string, listener: (event: SseEvent) => void): () => void` — and a concrete `InProcessFanout` implementation. `subscribe` SHALL return an unsubscribe function; calling it removes the listener and prevents future deliveries.
|
||
|
||
#### Scenario: publish delivers to every subscribed listener for the channel
|
||
|
||
- **WHEN** two listeners are subscribed on `"a"` and one on `"b"`, then `publish("a", event)` fires
|
||
- **THEN** both `"a"` listeners receive the event AND the `"b"` listener does not
|
||
|
||
#### Scenario: unsubscribe stops further deliveries
|
||
|
||
- **WHEN** a listener subscribes, then its returned unsubscribe is called, then `publish` runs
|
||
- **THEN** the listener does not receive the event
|
||
|
||
**Verified by:** `src/fanout/in-process.test.ts::"publish delivers to every listener for the channel"` AND `src/fanout/in-process.test.ts::"unsubscribe stops further deliveries"`
|
||
|
||
---
|
||
|
||
### Requirement: Zod schemas validate options and event payloads at the boundary
|
||
|
||
The package SHALL export `SsePluginOptionsSchema` (Zod) validating every field of `SsePluginOptions` (positive integers for limits, positive `heartbeatIntervalMs`, `overflowPolicy` enum, optional functions). The package SHALL export `SseEventSchema` (Zod) validating the `SseEvent` shape used by callers of `service.broadcast`. Invalid input to either schema SHALL flow through `SseError(SSE_INVALID_INPUT)`.
|
||
|
||
#### Scenario: SsePluginOptionsSchema rejects non-positive heartbeat
|
||
|
||
- **WHEN** `SsePluginOptionsSchema.safeParse({ heartbeatIntervalMs: 0 })` is called
|
||
- **THEN** the result has `success === false` with an issue on `heartbeatIntervalMs`
|
||
|
||
#### Scenario: SseEventSchema accepts arbitrary JSON-serialisable data
|
||
|
||
- **WHEN** `SseEventSchema.safeParse({ id: "1", data: { nested: { arr: [1, 2] } } })` is called
|
||
- **THEN** `success === true`
|
||
|
||
**Verified by:** `src/schemas/options.test.ts::"rejects non-positive heartbeat"` AND `src/schemas/event.test.ts::"accepts arbitrary JSON data"`
|
||
|
||
---
|
||
|
||
### Requirement: SseError registry with eight stable codes thrown from production paths
|
||
|
||
The package SHALL declare `SSE_CODES = defineCodes("SSE", [...])` covering `SSE_UNAUTHENTICATED`, `SSE_TOO_MANY_CONNECTIONS`, `SSE_PER_PRINCIPAL_LIMIT_EXCEEDED`, `SSE_CHANNEL_REQUIRED`, `SSE_SERVICE_NOT_AVAILABLE`, `SSE_INVALID_INPUT`, `SSE_QUEUE_OVERFLOW`, `SSE_CLOSED`. Every code SHALL be reachable from at least one production call site (asserted by the spec-conformance test).
|
||
|
||
#### Scenario: every code is thrown by a production path
|
||
|
||
- **WHEN** the spec-conformance suite exercises the documented trigger for each code
|
||
- **THEN** an `SseError` instance is thrown carrying that code
|
||
|
||
**Verified by:** `src/error-codes/error-codes.test.ts::"every SSE_* code is reachable from a production path"` AND `src/api-plugin-sse.spec-conformance.test.ts::"every SSE_* code is thrown from src/"`
|
||
|
||
---
|
||
|
||
### Requirement: OpenTelemetry instruments are emitted from production paths
|
||
|
||
The package SHALL register OTel instruments via `getSseInstruments()` and emit each from production code: `sse.active_connections` (UpDownCounter — +1 on subscribe, −1 on close), `sse.broadcasts_total{channel}` (Counter on broadcast), `sse.heartbeats_total` (Counter on heartbeat tick), `sse.message_bytes_total` (Counter on every framed write), `sse.queue_overflow_total{policy}` (Counter on overflow), `sse.connection_duration_ms` (Histogram on close). The conformance test SHALL install a real in-memory `MeterProvider`, exercise each path, and assert the metric was recorded.
|
||
|
||
#### Scenario: subscribe + close moves active_connections up and back down
|
||
|
||
- **WHEN** a subscription is created and then closed against a real `MeterProvider`
|
||
- **THEN** the collected metrics show `sse.active_connections` increased by 1 then decreased by 1
|
||
|
||
#### Scenario: broadcast increments broadcasts_total{channel}
|
||
|
||
- **WHEN** `service.broadcast("notifications", event)` is called
|
||
- **THEN** the collected metrics show `sse.broadcasts_total` with attribute `channel === "notifications"` incremented by 1
|
||
|
||
**Verified by:** `src/otel/instruments.test.ts::"subscribe + close emits active_connections"` AND `src/otel/instruments.test.ts::"broadcast emits broadcasts_total"`
|
||
|
||
---
|
||
|
||
### Requirement: Backpressure honors queueHighWaterMark and overflowPolicy
|
||
|
||
The Hono SSE handler SHALL track per-connection write-queue depth. When the writable controller's `desiredSize` falls to zero or the in-memory backlog exceeds `queueHighWaterMark`, the handler SHALL apply `overflowPolicy`: `"drop-oldest"` removes the oldest queued event before enqueueing the new one; `"drop-newest"` discards the incoming event. In both cases, `sse.queue_overflow_total{policy}` SHALL increment.
|
||
|
||
#### Scenario: drop-oldest evicts the oldest queued event
|
||
|
||
- **WHEN** `queueHighWaterMark = 2`, `overflowPolicy = "drop-oldest"`, and three events arrive faster than the controller drains
|
||
- **THEN** the third event lands in the queue AND the first is dropped AND `sse.queue_overflow_total{policy: "drop-oldest"}` increments
|
||
|
||
**Verified by:** `src/middleware/sse-handler.test.ts::"drop-oldest evicts the oldest queued event under backpressure"`
|
||
|
||
---
|
||
|
||
### Requirement: ContextVariableMap untouched; subscriber identity flows via options.resolvePrincipalId
|
||
|
||
The package SHALL NOT augment Hono's global `ContextVariableMap`. Subscriber principal identity SHALL be resolved via `options.resolvePrincipalId?: (c: Context) => string | undefined` passed to `createSseRoute`, so consumers wiring `@suaveplan/api-plugin-auth` upstream can read `c.get("user")?.id` and feed it in without this plugin depending on the auth package.
|
||
|
||
#### Scenario: resolvePrincipalId is invoked once per subscription
|
||
|
||
- **WHEN** a request reaches the handler and `resolvePrincipalId` is configured
|
||
- **THEN** the resolver is called with the Hono `Context` and the returned value is passed to `service.subscribe` as `principalId`
|
||
|
||
**Verified by:** `src/middleware/sse-handler.test.ts::"resolvePrincipalId is invoked once per subscription"`
|
||
|
||
---
|
||
|
||
### Requirement: Subpath exports per feature module
|
||
|
||
The package SHALL declare subpath exports in `package.json` so each feature module is independently importable: `.`, `./plugin/plugin`, `./services/service`, `./middleware/sse-handler`, `./formatter/format`, `./formatter/parse`, `./buffer/ring-buffer`, `./fanout/in-process`, `./schemas/options`, `./schemas/event`, `./error-codes/error-codes`, `./otel/instruments`, `./types/types`, `./registry/service-registry`.
|
||
|
||
#### Scenario: dep-graph test enumerates expected subpaths
|
||
|
||
- **WHEN** `tests/dep-graph.test.ts` reads `package.json`'s `exports` map
|
||
- **THEN** the list of `./*` entries matches the documented set exactly
|
||
|
||
**Verified by:** `tests/dep-graph.test.ts::"exports map advertises every feature subpath"`
|