Teach the conformance harness to run change-feed fixtures

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<DecodedFrame[]>;
+  /** Cancels the underlying reader and stops decoding. */
+  close(): Promise<void>;
+};
+
+export type OpenChangeFeedOpts = {
+  token?: string;
+  /** Additional headers, e.g. Last-Event-ID. */
+  headers?: Record<string, string>;
+};
+
+/**
+ * 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<AppEnv>,
+  path: string,
+  opts: OpenChangeFeedOpts = {},
+): Promise<ChangeFeedConnection> {
+  const headers: Record<string, string> = { 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<DecodedFrame[]> {
+    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<void> {
+    closed = true;
+    if (reader) await reader.cancel().catch(() => {});
+    await pump;
+  }
+
+  return { status: res.status, frames, waitForFrames, close };
+}

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)
 // -------------------------------------------------------

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<AppEnv>();
+  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();
+    }
+  });
+});