suaveplan/packages/api-stack/api/api-plugin-sse/tests/integration/full-stream.test.ts

import { describe, expect, it } from "@suaveplan/testing/runner";
import { Hono } from "hono";
import { createRingBuffer } from "../../src/buffer/ring-buffer.js";
import { createInProcessFanout } from "../../src/fanout/in-process.js";
import { parseSseStream } from "../../src/formatter/parse.js";
import { createSseRoute } from "../../src/middleware/sse-handler.js";
import { createSseService } from "../../src/services/service.js";

/**
 * End-to-end integration coverage. A real Hono app, a real fetch
 * round-trip (via `app.fetch` over a `Request`), and a real ReadableStream
 * client reader that decodes the SSE wire format using `parseSseStream`.
 */
describe("api-plugin-sse — full-stream integration", () => {
    it("end-to-end fetch subscriber receives three ordered events", {
        timeout: 10_000,
    }, async () => {
        let n = 0;
        const service = createSseService({
            buffer: createRingBuffer(100),
            fanout: createInProcessFanout(),
            idGenerator: () => `evt-${++n}`,
            maxConnectionsPerPrincipal: 16,
            maxTotalConnections: 100,
        });
        const app = new Hono();
        app.get(
            "/sse/:channel",
            createSseRoute(service, {
                heartbeatIntervalMs: 60_000,
                queueHighWaterMark: 64,
                overflowPolicy: "drop-oldest",
            })
        );

        const controller = new AbortController();
        const res = await app.request(
            new Request("http://test.local/sse/notifications", {
                signal: controller.signal,
            })
        );
        expect(res.status).toBe(200);

        // Broadcast on a small delay so the subscriber has installed.
        setTimeout(() => {
            service.broadcast("notifications", { data: { n: 1 } });
            service.broadcast("notifications", { data: { n: 2 } });
            service.broadcast("notifications", { data: { n: 3 } });
        }, 10);

        const body = res.body;
        if (body === null) throw new Error("no body");
        const reader = body.getReader();
        const collected = [] as { id: string; data: unknown }[];
        for await (const event of parseSseStream(reader)) {
            collected.push({
                id: event.id,
                data: JSON.parse(event.data as string),
            });
            if (collected.length >= 3) break;
        }
        controller.abort();
        await reader.cancel().catch(() => undefined);

        // service.subscribe consumes one id for the connection handle
        // (evt-1), then broadcasts consume evt-2/3/4 for the events.
        expect(collected.map((c) => c.id)).toEqual(["evt-2", "evt-3", "evt-4"]);
        expect(collected.map((c) => (c.data as { n: number }).n)).toEqual([
            1, 2, 3,
        ]);
        await service.close();
    });

    it("reconnect with Last-Event-ID skips delivered events", {
        timeout: 10_000,
    }, async () => {
        let n = 0;
        const service = createSseService({
            buffer: createRingBuffer(100),
            fanout: createInProcessFanout(),
            idGenerator: () => `evt-${++n}`,
            maxConnectionsPerPrincipal: 16,
            maxTotalConnections: 100,
        });
        const app = new Hono();
        app.get(
            "/sse/:channel",
            createSseRoute(service, {
                heartbeatIntervalMs: 60_000,
                queueHighWaterMark: 64,
                overflowPolicy: "drop-oldest",
            })
        );

        // Pre-load three broadcasts into the ring buffer.
        service.broadcast("c", { data: { n: 1 } });
        service.broadcast("c", { data: { n: 2 } });
        service.broadcast("c", { data: { n: 3 } });

        const controller = new AbortController();
        const res = await app.request(
            new Request("http://test.local/sse/c", {
                headers: { "Last-Event-ID": "evt-2" },
                signal: controller.signal,
            })
        );
        const body = res.body;
        if (body === null) throw new Error("no body");
        const reader = body.getReader();
        const collected = [] as { id: string }[];
        for await (const event of parseSseStream(reader)) {
            collected.push({ id: event.id });
            if (collected.length >= 1) break;
        }
        controller.abort();
        await reader.cancel().catch(() => undefined);

        expect(collected.map((c) => c.id)).toEqual(["evt-3"]);
        await service.close();
    });

    it("heartbeat tick via injected interval triggers a comment frame", {
        timeout: 10_000,
    }, async () => {
        const service = createSseService({
            buffer: createRingBuffer(10),
            fanout: createInProcessFanout(),
            idGenerator: () => "x",
            maxConnectionsPerPrincipal: 16,
            maxTotalConnections: 100,
        });
        const app = new Hono();
        let fire: (() => void) | undefined;
        app.get(
            "/sse/:channel",
            createSseRoute(service, {
                heartbeatIntervalMs: 30_000,
                queueHighWaterMark: 64,
                overflowPolicy: "drop-oldest",
                setIntervalImpl: (fn) => {
                    fire = fn;
                    return { mock: true };
                },
                clearIntervalImpl: () => {
                    fire = undefined;
                },
            })
        );
        const controller = new AbortController();
        const res = await app.request(
            new Request("http://t/sse/c", { signal: controller.signal })
        );
        const body = res.body;
        if (body === null) throw new Error("no body");
        const reader = body.getReader();
        // Allow stream install
        await new Promise((r) => setTimeout(r, 10));
        fire?.();
        const { value } = await reader.read();
        const decoder = new TextDecoder();
        const text = decoder.decode(value ?? new Uint8Array());
        expect(text).toContain(": heartbeat");
        controller.abort();
        await reader.cancel().catch(() => undefined);
        await service.close();
    });
});