From c3acc66b3a78646dcd384b080b144abb4065c7d3 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 28 Aug 2026 19:00:23 +0000 Subject: [PATCH] Teach the conformance harness to run change-feed fixtures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the SSE streaming dispatch helper (SSEDecoder + openChangeFeed) the change-feed fixtures need, since they pin an ordered stream of frames rather than a request/response pair. Verified against a throwaway SSE test app built with Hono's own streamSSE, proving the open/mutate-while- open/collect/timeout mechanics work before there's a real GET /changes to point them at. changeFeedFixtures and changeFeedSequenceFixtures are imported and given their own coverage-gated describe blocks, entirely skipped for now — no src/routes/changes.ts exists yet (#82). The two sequence fixtures stay skipped even after #82 lands, since resume support is #84's work. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_014u78zif6NgggtFVbiSZ2co --- tests/changeFeedClient.ts | 157 +++++++++++++++++++++++++++++ tests/conformance.test.ts | 43 ++++++++ tests/lib/changeFeedClient.test.ts | 127 +++++++++++++++++++++++ 3 files changed, 327 insertions(+) create mode 100644 tests/changeFeedClient.ts create mode 100644 tests/lib/changeFeedClient.test.ts diff --git a/tests/changeFeedClient.ts b/tests/changeFeedClient.ts new file mode 100644 index 0000000..c121adb --- /dev/null +++ b/tests/changeFeedClient.ts @@ -0,0 +1,157 @@ +/** + * A client-side reader for the change feed's SSE wire format + * (docs/spec/wire-format.md § Change feed). Hono's own `hono/streaming` + * ships an SSE *writer* (`streamSSE`) but nothing to consume one — that + * half is what this module provides, so tests/conformance.test.ts can + * drive changeFeedFixtures once GET /changes exists (#82): open a + * connection, decode frames as they arrive, dispatch the fixture's + * `activity` mutations while it's still open, and collect what the + * connection produces. + */ +import type { Hono } from 'hono'; +import type { AppEnv } from '../src/app.js'; + +export type DecodedFrame = { + /** + * SSE `id:`, when the wire line was present on *this* frame — not the + * persisted "last event ID" a spec-compliant EventSource client carries + * forward across events with no id of their own. changeFeedFixtures pins + * the literal per-frame wire content (a `reset` frame carries no id even + * immediately after a `record` frame that did), so this decoder resets + * per frame rather than inheriting. + */ + id?: string; + event: string; + data: unknown; +}; + +/** Incremental SSE decoder: feed it raw chunks of the stream's text as they arrive. */ +export class SSEDecoder { + private buffer = ''; + private eventName: string | undefined; + private id: string | undefined; + private dataLines: string[] = []; + private sawField = false; + + /** Decode one chunk, returning any frames it completed. */ + push(text: string): DecodedFrame[] { + this.buffer += text; + const frames: DecodedFrame[] = []; + let newlineIndex: number; + while ((newlineIndex = this.buffer.indexOf('\n')) !== -1) { + const rawLine = this.buffer.slice(0, newlineIndex); + this.buffer = this.buffer.slice(newlineIndex + 1); + const line = rawLine.endsWith('\r') ? rawLine.slice(0, -1) : rawLine; + if (line === '') { + const frame = this.dispatch(); + if (frame) frames.push(frame); + continue; + } + if (line.startsWith(':')) continue; // comment line — a keepalive + const colonIndex = line.indexOf(':'); + const field = colonIndex === -1 ? line : line.slice(0, colonIndex); + let value = colonIndex === -1 ? '' : line.slice(colonIndex + 1); + if (value.startsWith(' ')) value = value.slice(1); + this.sawField = true; + if (field === 'id') this.id = value; + else if (field === 'event') this.eventName = value; + else if (field === 'data') this.dataLines.push(value); + // 'retry' and any other field: not needed here, ignored. + } + return frames; + } + + private dispatch(): DecodedFrame | undefined { + if (!this.sawField) return undefined; // a blank line with nothing since the last dispatch + const event = this.eventName ?? 'message'; + const id = this.id; + const dataText = this.dataLines.join('\n'); + this.eventName = undefined; + this.id = undefined; + this.dataLines = []; + this.sawField = false; + let data: unknown = dataText; + if (dataText) { + try { + data = JSON.parse(dataText); + } catch { + // Leave as raw text — a fixture asserting on parsed JSON fails loudly. + } + } + return { ...(id !== undefined && { id }), event, data }; + } +} + +export type ChangeFeedConnection = { + status: number; + /** Frames decoded so far, in arrival order. Mutated in place as more arrive. */ + frames: DecodedFrame[]; + /** Resolves with the first `count` frames once they've arrived, or rejects after timeoutMs. */ + waitForFrames(count: number, timeoutMs?: number): Promise; + /** Cancels the underlying reader and stops decoding. */ + close(): Promise; +}; + +export type OpenChangeFeedOpts = { + token?: string; + /** Additional headers, e.g. Last-Event-ID. */ + headers?: Record; +}; + +/** + * Opens an SSE connection against a Hono test app. `app.request()` resolves + * as soon as the response's headers are sent — the body stays open and + * readable while the server keeps writing — so a mutation dispatched via + * the ordinary `req()` helper while a connection from this function is + * still open reaches the same in-process stack and can produce frames on + * it, exactly as changeFeedFixtures' `activity` expects. + */ +export async function openChangeFeed( + app: Hono, + path: string, + opts: OpenChangeFeedOpts = {}, +): Promise { + const headers: Record = { Accept: 'text/event-stream' }; + if (opts.token) headers['Authorization'] = `Bearer ${opts.token}`; + Object.assign(headers, opts.headers); + + const res = await app.request(path, { headers }); + const reader = res.body?.getReader(); + const decoder = new SSEDecoder(); + const textDecoder = new TextDecoder(); + const frames: DecodedFrame[] = []; + let closed = false; + + const pump = (async () => { + if (!reader) return; + while (!closed) { + const { done, value } = await reader.read(); + if (done) return; + frames.push(...decoder.push(textDecoder.decode(value, { stream: true }))); + } + })(); + // A rejected pump would otherwise surface as an unhandled rejection the + // moment close() cancels the reader mid-read. + pump.catch(() => {}); + + async function waitForFrames(count: number, timeoutMs = 2000): Promise { + const start = Date.now(); + while (frames.length < count) { + if (Date.now() - start > timeoutMs) { + throw new Error( + `Timed out after ${timeoutMs}ms waiting for ${count} frame(s); got ${frames.length}: ${JSON.stringify(frames)}`, + ); + } + await new Promise((resolve) => setTimeout(resolve, 10)); + } + return frames.slice(0, count); + } + + async function close(): Promise { + closed = true; + if (reader) await reader.cancel().catch(() => {}); + await pump; + } + + return { status: res.status, frames, waitForFrames, close }; +} diff --git a/tests/conformance.test.ts b/tests/conformance.test.ts index c1de9e8..3b3ef48 100644 --- a/tests/conformance.test.ts +++ b/tests/conformance.test.ts @@ -40,6 +40,8 @@ import { authChallengeFixtures, authTokenFixtures, authSequenceFixtures, + changeFeedFixtures, + changeFeedSequenceFixtures, AUTH_FIXTURE_DID, AUTH_FIXTURE_NONCE, } from '@haverstack/conformance-fixtures'; @@ -986,6 +988,47 @@ describe('error response fixtures', () => { }); }); +// ------------------------------------------------------- +// Change feed (#78): no src/routes/changes.ts yet, so every fixture below +// is necessarily skipped rather than dispatched. This block's value is the +// coverage gate — it fails loudly the moment core adds, removes, or +// renames a change-feed fixture nobody updated these SKIPPED reasons for. +// +// A change-feed fixture isn't a request/response pair like every other +// block here: it pins an ordered stream of frames a connection sees, +// optionally across mutations made while it's open. tests/changeFeedClient.ts +// carries the streaming dispatch helper (SSEDecoder + openChangeFeed) that +// #82 dispatches these fixtures through, once GET /changes exists to +// dispatch them against. +// ------------------------------------------------------- + +describe('changeFeed fixtures', () => { + // GET /changes doesn't exist yet — land the dispatch block in #82. + const SKIPPED = new Set(changeFeedFixtures.map((f) => f.name)); + + test('coverage', () => { + assertCoverage( + changeFeedFixtures.map((f) => f.name), + new Set(), + SKIPPED, + ); + }); +}); + +describe('changeFeed sequence fixtures', () => { + // Both stay skipped even once #82 lands GET /changes — resume support + // (Last-Event-ID / ?since=, the per-session buffer) is #84's work. + const SKIPPED = new Set(changeFeedSequenceFixtures.map((f) => f.name)); + + test('coverage', () => { + assertCoverage( + changeFeedSequenceFixtures.map((f) => f.name), + new Set(), + SKIPPED, + ); + }); +}); + // ------------------------------------------------------- // Auth: DID challenge-response handshake (#53) // ------------------------------------------------------- diff --git a/tests/lib/changeFeedClient.test.ts b/tests/lib/changeFeedClient.test.ts new file mode 100644 index 0000000..1c74f9e --- /dev/null +++ b/tests/lib/changeFeedClient.test.ts @@ -0,0 +1,127 @@ +import { describe, test, expect } from 'vitest'; +import { Hono } from 'hono'; +import { streamSSE } from 'hono/streaming'; +import { EventEmitter } from 'node:events'; +import { SSEDecoder, openChangeFeed } from '../changeFeedClient.js'; +import type { AppEnv } from '../../src/app.js'; + +describe('SSEDecoder', () => { + test('decodes a single frame delivered in one chunk', () => { + const decoder = new SSEDecoder(); + const frames = decoder.push('id: AA3f1R\nevent: record\ndata: {"kind":"created"}\n\n'); + expect(frames).toEqual([{ id: 'AA3f1R', event: 'record', data: { kind: 'created' } }]); + }); + + test('decodes multiple frames from one chunk', () => { + const decoder = new SSEDecoder(); + const frames = decoder.push( + 'event: ready\ndata: {"seq":"AA3f1Q"}\n\nid: AA3f1R\nevent: record\ndata: {"kind":"created"}\n\n', + ); + expect(frames).toEqual([ + { event: 'ready', data: { seq: 'AA3f1Q' } }, + { id: 'AA3f1R', event: 'record', data: { kind: 'created' } }, + ]); + }); + + test('decodes a frame split across chunks, mid-line', () => { + const decoder = new SSEDecoder(); + expect(decoder.push('event: rea')).toEqual([]); + expect(decoder.push('dy\ndata: {"se')).toEqual([]); + const frames = decoder.push('q":"AA3f1Q"}\n\n'); + expect(frames).toEqual([{ event: 'ready', data: { seq: 'AA3f1Q' } }]); + }); + + test('ignores comment lines (keepalives)', () => { + const decoder = new SSEDecoder(); + const frames = decoder.push(': keepalive\n\nevent: ready\ndata: {}\n\n'); + expect(frames).toEqual([{ event: 'ready', data: {} }]); + }); + + test('a control frame with no id decodes with id absent, even right after a frame that had one', () => { + const decoder = new SSEDecoder(); + const frames = decoder.push( + 'id: AA3f1R\nevent: record\ndata: {}\n\nevent: reset\ndata: {"reason":"not_supported"}\n\n', + ); + expect(frames[0]!.id).toBe('AA3f1R'); + expect(frames[1]!.id).toBeUndefined(); + }); + + test('an unrecognized event name is preserved, not dropped', () => { + const decoder = new SSEDecoder(); + const frames = decoder.push('event: something-new\ndata: {}\n\n'); + expect(frames).toEqual([{ event: 'something-new', data: {} }]); + }); + + test('non-JSON data is preserved as raw text rather than throwing', () => { + const decoder = new SSEDecoder(); + const frames = decoder.push('event: ready\ndata: not-json\n\n'); + expect(frames).toEqual([{ event: 'ready', data: 'not-json' }]); + }); +}); + +/** + * A throwaway SSE server, unrelated to this repo's real change feed + * endpoint (#82 hasn't landed it), that exists purely to prove + * openChangeFeed's own mechanics: a connection opens, a mutation dispatched + * against a *second*, concurrent request reaches the still-open stream, and + * waitForFrames/close behave correctly around that. + */ +function buildSseTestApp() { + const emitter = new EventEmitter(); + const app = new Hono(); + app.get('/sse-test', (c) => { + return streamSSE(c, async (stream) => { + await stream.writeSSE({ event: 'ready', data: '{}' }); + for (let i = 0; i < 2; i++) { + const payload = await new Promise<{ id: string; data: unknown }>((resolve) => { + emitter.once('mutate', resolve); + }); + await stream.writeSSE({ + event: 'record', + id: payload.id, + data: JSON.stringify(payload.data), + }); + } + }); + }); + app.post('/mutate', async (c) => { + const body = await c.req.json(); + emitter.emit('mutate', body); + return c.json({ ok: true }); + }); + return app; +} + +describe('openChangeFeed', () => { + test('captures the opening frame immediately, then a mutation made while open', async () => { + const app = buildSseTestApp(); + const conn = await openChangeFeed(app, '/sse-test'); + try { + expect(conn.status).toBe(200); + await conn.waitForFrames(1); + expect(conn.frames[0]).toEqual({ event: 'ready', data: {} }); + + await app.request('/mutate', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ id: 'AA3f1R', data: { kind: 'created' } }), + }); + + const frames = await conn.waitForFrames(2); + expect(frames[1]).toEqual({ id: 'AA3f1R', event: 'record', data: { kind: 'created' } }); + } finally { + await conn.close(); + } + }); + + test('waitForFrames rejects rather than hanging when too few frames arrive', async () => { + const app = buildSseTestApp(); + const conn = await openChangeFeed(app, '/sse-test'); + try { + await conn.waitForFrames(1); + await expect(conn.waitForFrames(5, 50)).rejects.toThrow(/Timed out/); + } finally { + await conn.close(); + } + }); +});