# `routes/progress-sse` — Server-Sent Events progress push _(no submodules)_ ## Purpose Exposes `GET /imports/:id/progress` returning a long-lived `text/event-stream` so clients can subscribe to a job's real-time progress without polling. The handler is driven entirely off the existing `ImportService.bus` — three subscriptions (`import.progress`, `import.completed`, `import.failed`) are wired into a `ReadableStream` and detached on terminal events or client abort. This module was deferred from the original `add-api-plugin-import` migration's `Excluded` list and ships independently via the `add-import-sse-progress` change. ## Features - Three SSE event kinds — `progress` (per `import.progress` tick), `complete` (terminal, on `import.completed`), `failed` (terminal, on `import.failed`). - Initial frame snapshot from `service.getJob` so clients see current state before the next bus tick. - Owner-scoped — foreign-owner / missing-job lookups return HTTP 404 RFC 7807 problem+json (`IMPORT.NOT_FOUND`); anonymous requests return HTTP 401 via the existing `requireAuth()` middleware. - Per-job filtering on every subscription callback so a long-lived connection cannot leak progress for other jobs. - `AbortSignal` lifecycle — client disconnect detaches every listener and emits `import.sse.disconnected{reason: "client"}`. - OTel observability — `import.sse.connected`, `import.sse.events_sent{kind}`, `import.sse.disconnected{reason}`. - Pure stream — no polling timer, no background interval, no heartbeat (job runs are short-lived so proxy-keepalive concerns are handled by upstream LB tuning, not by the handler). ## Usage: Basic Example ```ts import { createImportRoutes, ImportService } from "@suaveplan/api-plugin-import"; import { Hono } from "hono"; const service = new ImportService(); const host = new Hono(); host.use("*", async (c, next) => { c.set("user", { id: "alice" }); await next(); }); host.route("/imports", createImportRoutes({ service })); // Client side: const es = new EventSource("/imports/job-1/progress"); es.addEventListener("progress", (e) => console.log(JSON.parse(e.data))); es.addEventListener("complete", () => es.close()); es.addEventListener("failed", () => es.close()); ``` ## Usage: Advanced Example ```ts import { createProgressSseHandler, ImportService, } from "@suaveplan/api-plugin-import"; import { Hono } from "hono"; const service = new ImportService(); const app = new Hono(); app.use("*", async (c, next) => { c.set("user", { id: "alice" }); await next(); }); // Mount stand-alone (without the rest of `createImportRoutes`). app.get("/jobs/:id/progress", createProgressSseHandler({ service })); // Drive the service from a worker; every emitted progress event is // fanned to every connected client whose `:id` matches. const controller = new AbortController(); const res = await app.request( new Request("http://t/jobs/job-1/progress", { signal: controller.signal }) ); const reader = res.body!.getReader(); const decoder = new TextDecoder(); let buf = ""; while (true) { const { value, done } = await reader.read(); if (done) break; buf += decoder.decode(value, { stream: true }); const frames = buf.split("\n\n"); buf = frames.pop() ?? ""; for (const frame of frames) console.log(frame); } ``` ## API Reference - `createProgressSseHandler(deps: ProgressSseDeps): (c: Context) => Response` - `formatProgressFrame(kind: ProgressSseEventKind, seq: number, payload: ImportProgress): string` - `interface ProgressSseDeps { service: ImportService }` - `type ProgressSseEventKind = "progress" | "complete" | "failed"` ## Implementation Notes - The handler is registered via `app.get("/:id/progress", ...)` rather than `app.openapi(...)` because `c.body(ReadableStream, ...)` does not conform to a static OpenAPI response schema — the parent router's `app.use("/*", requireAuth())` and `app.onError(...)` still apply. - Subscriptions are filtered by `payload.jobId === jobId` so the bus cannot leak foreign-job progress to a connection. The router's authoritative ownership check happens once via `service.getJob` at connection-open time; subsequent leak prevention is structural. - The terminal `complete` / `failed` branch in `start` handles the edge case where a job reached terminal state between `getJob` and the bus subscription — without it, a client would be stuck on the initial `progress` frame forever. - The OTel counters are read through `getImportInstruments()` rather than cached at module scope so `resetImportInstruments()` (used by test setup) rebinds correctly. - The response headers (`Cache-Control: no-cache, no-transform`, `X-Accel-Buffering: no`, `Connection: keep-alive`) match the `@suaveplan/api-plugin-sse` handler so reverse proxies (nginx, cloudflare) do not buffer or coalesce the stream.