API index and search · Build metadata

Source snapshot

packages/api-server/src/streams.ts

1// LM stream checkpointing — the two-plane response path (designs-implemented/LM-STREAM-CHECKPOINT-DESIGN.md).2//3// A model response arrives as hundreds of tiny deltas per second. Every one wants to be on a screen4// immediately; none wants to be a durable write. So the response runs on TWO planes sharing ONE5// monotone coordinate (`seq`, in UTF-16 code units of the response text):6//7//   - the LIVE plane (this module): process-local, volatile, every delta, straight to subscribers;8//   - the DURABLE plane: the app's own tables, advanced at coarse boundaries — N chars, T ms, an9//     explicit `flush()`, and always at close.10//11// The contract both planes stand on (§0):12//13//   S (splice)  durableText(f) ++ concat(frames delivered from f) == producedText, for every join14//               offset f and every join time.15//   P (prefix)  durableText is always a PREFIX of producedText, and durableSeq is monotone.16//17// P is what makes S mechanical — two prefixes of one string merge by taking the longer (see18// {@link spliceStreamText}). Everything else here is bookkeeping in service of those two.19//20// TWO THINGS THIS DELIBERATELY IS NOT (both were wrong in the first cut of this module):21//22//   1. Checkpoints are SYSTEM WRITES, not mutation envelopes. `lmid` is client-authored and exists23//      to release a client's optimistic rebase point; a server-authored checkpoint has no24//      prediction to release, and CDC → IVM fanout needs no envelope at all. Checkpoints therefore25//      go through `outsideSql` with no clientID/mid — the same rule the realtime lifecycle writes26//      follow ("a system write must never advance an lmid").27//   2. A checkpoint INSERTS A CHUNK ROW; it never appends to a growing column. A normalized `edit`28//      carries BOTH the old and the new full row, so advancing a `body` column would re-stream the29//      whole message twice per checkpoint — O(n²) wire to deliver n characters. One row per chunk30//      makes each checkpoint a single `add` carrying only its own slice.31//32// Durability is never overclaimed (§0, the RINDLE-REALTIME §8.1 "ack honesty" rule): a `chunk` frame33// says only "produced", a `durable` frame says "committed", an `end` frame says "sealed". Losing34// this plane entirely costs latency and the un-checkpointed tail — never consistency: a client with35// no live stream still sees the message grow through IVM at checkpoint granularity, which is why36// `absent`/`stale` are ordinary answers rather than errors.3738// Type-only (erased at runtime) — no import cycle with ./index.ts.39import { STREAM_STATUS_STREAMING, frameResumePoint } from "@rindle/client";40import type { StreamFrame, StreamStatus } from "@rindle/client";41import type { SqlStatement } from "@rindle/daemon-client";4243import type { Authorizer, ServerSql, SqlDialect } from "./index.ts";4445// ------------------------------------------------------------------------------- the wire46//47// The frame shapes and the two pure reassembly functions live in `@rindle/client` — BOTH tiers need48// them, and a browser must never pull this package to read a stream. Re-exported here so the49// server-side import path is unchanged.5051export {52  STREAM_STATUS_STREAMING,53  assembleDurableText,54  frameResumePoint,55  spliceStreamText,56} from "@rindle/client";57export type { StreamFrame, StreamStatus } from "@rindle/client";5859// ------------------------------------------------------------------------------- the durable seam6061/** What a checkpoint hands the durable plane when the app supplies its own {@link StreamCommit}. */62export type StreamCommitInput =63  /** The pointer is marked live. The app's own mutator created the row (it owns `chatId`, `role`,64   *  the model name…); this only flips it to `streaming`. */65  | { kind: "open"; streamId: string; meta: unknown; hostId?: string; startedAt: number }66  /** A prefix advance carrying ONLY its own slice. `text.length === seq - from`, and `from` is the67   *  last append the PLANE saw confirmed — so appends are contiguous in the fault-free run, but an68   *  append that committed while its ack was lost makes the next one RE-COVER (its `from` lags what69   *  the app already applied). See {@link StreamCommit} for the two ways to stay idempotent. */70  | { kind: "append"; streamId: string; from: number; seq: number; text: string }71  /** The seal. `body` is the producer's retained text and `bodyFrom` its absolute start offset:72   *  `bodyFrom === 0` — and `body` is the WHOLE response — unless the app opted into trimming via73   *  `retainChars` (`commit` mode only). A compacting app requires `bodyFrom === 0` and writes74   *  `body` wholesale; a non-compacting app appends the outstanding tail —75   *  `body.slice(from - bodyFrom)` — as a final chunk. `seq === bodyFrom + body.length`, always. */76  | { kind: "close"; streamId: string; from: number; seq: number; body: string; bodyFrom: number; status: StreamStatus; error?: string };7778/** The escape hatch: persist the checkpoint however the app likes. Retried on throw (§3.3), so it79 *  must be idempotent under a repeated `(from, seq)` — AND under a RE-COVER: an append whose write80 *  committed but whose ack was lost leaves the plane believing less than the store holds, so the81 *  next append's `(from, seq)` can OVERLAP text already applied. Apply only the unseen suffix82 *  (`text.slice(applied - from)` when `applied > from`), or better, return `{seq: applied}` — the83 *  authoritative applied length — and the plane resynchronizes instead of re-covering at all.84 *  Return `{cancelRequested: true}` to tell the producer the reader asked it to stop (§6). */85export type StreamCommit = (86  input: StreamCommitInput,87) => Promise<void | { cancelRequested?: boolean; seq?: number }>;8889/** Which columns of the app's own tables the plane reads and writes. Every entry has a default90 *  except `cancel`, `error`, and `host`, which are opt-in BY NAMING: the plane never emits SQL91 *  against a column the app did not ask it to use. */92export interface StreamColumns {93  /** The message row's primary key, matched against `streamId`. Default `id`. */94  key?: string;95  /** The compacted response text. Default `body`. */96  body?: string;97  /** {@link STREAM_STATUS_STREAMING} then a {@link StreamStatus}. Default `status`. */98  status?: string;99  /** Total durable length — `length(body) + Σ chunk lengths`. The CAS column (§3.2). Default `seq`. */100  seq?: string;101  /** Opt-in (§6): a truthy value here stops the generation at the next checkpoint. No default —102   *  naming it is what turns cancellation on. */103  cancel?: string;104  /** Opt-in: where a failed generation's message is recorded. No default. */105  error?: string;106  /** Opt-in: where the open write records this producer's identity ({@link RindleStreamOptions.hostId}).107   *  Naming it upgrades the row-level single-flight guard from check-then-act to a true108   *  compare-and-swap — the open write turns conditional and a read-back names the winner (§5.1) —109   *  and gives multi-instance subscribe routing a column to read (§4). No default. */110  host?: string;111  /** Chunk primary key; the plane writes the deterministic `"<streamId>:<seq>"`. Default `id`. */112  chunkKey?: string;113  /** Chunk → message reference. Default `streamId`. */114  chunkStream?: string;115  /** The chunk's END offset — the ordering key. Default `seq`. */116  chunkSeq?: string;117  /** The chunk's slice of the response. Default `text`. */118  chunkText?: string;119}120121/**122 * The app's tables (§5). The app authors and migrates BOTH — the message row is unambiguously123 * app-owned (it has `chatId`, `role`, token counts) and the chunk row must be reachable from the124 * app's own query as a `related` subquery, which a Rindle system table would make awkward. The plane125 * only needs to be told where things live. Use {@link streamChunkTableDdl} for the chunk table's126 * migration.127 *128 * The message table carries WHATEVER ELSE the app wants; the plane's hard requirements are only:129 *130 * | mapped column | requirement | why |131 * | --- | --- | --- |132 * | `key`    | UNIQUE (normally the pk) | every checkpoint targets one row by it |133 * | `seq`    | integer, **NOT NULL DEFAULT 0** | the CAS column — nothing matches NULL (§3.2) |134 * | `body`   | text, empty at open | compaction overwrites it with the whole response |135 * | `status` | text accepting `streaming`/`complete`/`cancelled`/`error`/`interrupted` | a CHECK136 *              constraint that omits one of these turns a seal into an infra failure |137 * | `cancel` | truthy-readable, if mapped | read on the checkpoint round-trip (§6) |138 * | `error`  | nullable text, if mapped | compaction writes `NULL` when there is no error |139 * | `host`   | text, if mapped | written at open with the producer's token; the read-back decides the open race (§5.1) |140 *141 * `body` is PLANE-OWNED and always a bare string. Rich content (an array of content blocks, tool142 * calls, attachments) belongs in SIBLING columns the app's own mutators write — `flush()` orders the143 * text before them. For genuinely multi-block streaming, point `message` at a per-BLOCK table144 * instead: `streamId` is just an app key, so one stream per block needs nothing from this plane.145 */146export interface StreamTables {147  /** The app's message table. Must already contain the row when {@link StreamPlane.open} runs. */148  message: string;149  /** The append-only chunk table. */150  chunks: string;151  columns?: StreamColumns;152}153154export type StreamCheckpointTarget = { tables: StreamTables } | { commit: StreamCommit };155156// ------------------------------------------------------------------------------- configuration157158/** When a checkpoint fires — the FIRST of these wins (§3.1). Checkpoints are serialized, so a slow159 *  store degrades to fewer, larger checkpoints, never to a queue of them. */160export interface StreamCheckpointPolicy {161  /** Produced-but-uncommitted characters that force a checkpoint. Default 512. */162  chars?: number;163  /** Milliseconds since the last checkpoint that force one. Default 750. */164  intervalMs?: number;165  /** Retries for a failing commit before the slice is left for the next trigger. Default 3. */166  retries?: number;167}168169export interface AuthorizeStreamInput<User> {170  user: User;171  streamId: string;172  /** Where the subscriber claims to be. */173  from: number;174  /** The `meta` this stream was opened with — `undefined` when the stream is not hosted here, which175   *  is precisely when the app must decide from `streamId` and its own durable state. */176  meta: unknown;177  request?: unknown;178}179180/**181 * Optional cross-process transport for the LIVE plane182 * (designs-implemented/LM-STREAM-RELAY-DESIGN.md). Both methods are independently optional; which183 * ones you implement is which topology you built — an addressing adapter (a Durable Object named by184 * `streamId`, `fly-replay`) implements only `attach`; a broadcast adapter (Redis pub/sub, NATS)185 * mirrors with `publish` and subscribes with `attach`; a log adapter (Redis Streams, Kafka) appends186 * and replays. Never consulted for the durable plane — checkpoints are unaffected by any of this.187 *188 * The plane does not trust what `attach` yields: frames are run through the conform pass189 * ({@link StreamRelayConform}) and any contract violation downgrades the subscription to `stale`,190 * which already means "you are on the durable plane now". A broken relay costs a reader smooth191 * tokens, never corrupted text — and can never reach the producer.192 */193export interface StreamRelay {194  /** Producer side: every frame this process's producer fans out (`chunk`, `durable`, and the195   *  terminal `end`), mirrored outward. MUST NOT block; a throw or rejected promise is caught and196   *  routed to {@link RindleStreamOptions.onRelayError} — a relay outage may cost the live leg,197   *  never the generation. Returned promises are observed but never awaited. */198  publish?(streamId: string, frame: StreamFrame): void | PromiseLike<void>;199  /** Subscriber side: this process is not hosting `streamId`. Return a frame source, or `undefined`200   *  for `absent` — exactly the no-relay answer. Consulted only AFTER `authorize` has passed, and201   *  only on a live-plane miss (a local stream always wins). The plane closes the source202   *  (`return()`) when the reader disconnects. An adapter that cannot serve `from` (pub/sub has no203   *  history) yields `stale` and stops — the reader converges on the durable plane (§5). */204  attach?(streamId: string, from: number): Promise<AsyncIterable<StreamFrame> | undefined>;205}206207export interface StreamRelayErrorInfo {208  streamId: string;209  /** Where it failed: mirroring a frame out (`publish`), dialing the adapter (`attach`), or210   *  consuming/conforming its frames (`frames`). */211  phase: "publish" | "attach" | "frames";212}213214export interface RindleStreamOptions<User> {215  /** Where checkpoints land: the app's tables (the default path) or a raw `commit` callback. */216  checkpoint: StreamCheckpointTarget;217  /** REQUIRED. Subscribing to a stream is reading someone's chat, so there is no default-allow.218   *  Runs BEFORE existence is checked, so a denial cannot be used to probe for stream ids. */219  authorize: Authorizer<AuthorizeStreamInput<User>>;220  policy?: StreamCheckpointPolicy;221  /** This process's identity — it must be UNIQUE per producer process, because the open CAS trusts222   *  it to distinguish rivals (§5.1). In `tables` mode, map {@link StreamColumns.host} and the open223   *  write persists it on the message row (so the app can route later subscribers to the hosting224   *  instance, §4) and uses it as the single-flight token; setting it WITHOUT a mapped `host` column225   *  is refused at construction. In `commit` mode it rides the `open` input. When a `host` column is226   *  mapped and no hostId is given, a random per-plane token is used — the CAS still holds, routing227   *  just has no stable name to read. */228  hostId?: string;229  /** Slack retained BELOW `durableSeq` so a client whose IVM view lags a checkpoint can still join230   *  without a `stale` round trip. Text at or above `durableSeq` is never trimmed. Default 64 KiB.231   *  `commit`-mode only — REFUSED (a construction-time `TypeError`) in `tables` mode, where232   *  compaction needs the whole produced text at close (§3.4), so the buffer is retained in full. */233  retainChars?: number;234  /** How long a sealed stream stays joinable before eviction. Default 30s. */235  lingerMs?: number;236  /** Per-subscriber frame queue cap; overflow drops that subscriber with `stale` (§4). Default 1024.237   *  Relayed readers reuse the same bound: a slow reader on a relayed stream costs itself the live238   *  leg exactly as a local one does. */239  maxQueuedFrames?: number;240  /** A checkpoint that exhausted its retries. The stream keeps streaming — this is a durability241   *  stall, not a stream stall — so the error must not vanish. Absent ⇒ `console.error`. */242  onCheckpointError?: (err: unknown, info: { streamId: string; from: number; seq: number }) => void;243  /** Cross-process transport for the live plane ({@link StreamRelay}). Without one, a subscriber244   *  that lands on a process not hosting its stream gets `absent` and reads the durable plane at245   *  checkpoint granularity — correct, just chunky. */246  relay?: StreamRelay;247  /** Bound on `relay.attach`: a hung adapter yields `absent`, not a hung HTTP request. An248   *  addressing adapter MAY deliberately spend this window waiting out a subscribe that races its249   *  own kick. Default 2000. */250  relayAttachTimeoutMs?: number;251  /** A diagnostic, never a control path: relay failures (a throwing or rejecting `publish`, a failed252   *  or timed-out `attach`, a conform violation in the frames) land here, wrapped so a throwing hook253   *  cannot reach the plane. The reader-facing outcome is always the same legal `absent`/`stale`.254   *  Absent ⇒ `console.error`. */255  onRelayError?: (err: unknown, info: StreamRelayErrorInfo) => void;256}257258// ------------------------------------------------------------------------------- producer handle259260export interface OpenStreamInput<User> {261  user: User;262  /** The app's message-row id: the durable pointer AND the live plane's key. The row must already263   *  exist (the app's own mutator wrote it, alongside the user's prompt) with `seq = 0`. */264  streamId: string;265  /** Opaque app payload, passed through to a `commit` callback. Unused in `tables` mode. */266  meta?: unknown;267  request?: unknown;268}269270/** The producer's handle (§2). One writer per stream, by construction. */271export interface StreamHandle {272  readonly streamId: string;273  /** Total produced code units. */274  readonly seq: number;275  /** Total code units the store has committed. Never exceeds {@link seq} (contract P). */276  readonly durableSeq: number;277  /** True once a checkpoint round-trip has seen the reader's cancel flag (§6). `pump` stops on it;278   *  a hand-rolled generation loop should check it. */279  readonly cancelled: boolean;280  /** Append a delta: fanned to subscribers synchronously, checkpointed on policy. */281  push(text: string): void;282  /** Force a checkpoint and resolve once the store holds every character produced so far. This is283   *  the ORDERING primitive: `await flush()` before writing a discrete row (a tool call, a stop284   *  reason) so the text precedes it in the store (§1). Rejects if the checkpoint cannot commit. */285  flush(): Promise<number>;286  /** Drain a delta iterable into the stream (the shape every LLM SDK's text stream already has),287   *  stopping early — and closing the iterator, which aborts the underlying request — once the288   *  reader has cancelled. */289  pump(deltas: AsyncIterable<string>): Promise<void>;290  /** Seal the stream: `cancelled` if the reader asked it to stop, else `complete`. In `tables` mode291   *  this is the compaction (§3.4) — it writes the whole body and drops the chunk rows in one292   *  transaction, so it also REPAIRS any checkpoint that failed along the way. */293  close(): Promise<void>;294  /** Seal `error` at whatever was produced. Never throws for the reason it is sealing. */295  fail(error: unknown): Promise<void>;296}297298export interface SubscribeStreamInput<User> {299  user: User;300  streamId: string;301  /** "I already have this many characters" — from the client's IVM view, or a `Last-Event-ID`.302   *  A non-negative integer; default 0. */303  from?: number;304  request?: unknown;305}306307export interface StreamSubscription {308  readonly streamId: string;309  /** Terminates after exactly one of `end` / `stale` / `absent`. */310  readonly frames: AsyncIterable<StreamFrame>;311  /** Detach early (a disconnected client). Idempotent. */312  close(): void;313}314315// ------------------------------------------------------------------------------- defaults316317const DEFAULT_CHECKPOINT_CHARS = 512;318const DEFAULT_CHECKPOINT_INTERVAL_MS = 750;319const DEFAULT_CHECKPOINT_RETRIES = 3;320const DEFAULT_RETAIN_CHARS = 64 * 1024;321const DEFAULT_LINGER_MS = 30_000;322const DEFAULT_MAX_QUEUED_FRAMES = 1024;323const DEFAULT_RELAY_ATTACH_TIMEOUT_MS = 2000;324/** Retry backoff base — 50ms, 100ms, 200ms, … A checkpoint failure is usually a blip at the write325 *  authority; the text is safe in the buffer meanwhile, so there is nothing to rush. */326const RETRY_BACKOFF_MS = 50;327328/** A SCHEDULING timer — the checkpoint cadence, the linger window, the SSE keep-alive. It never329 *  holds the process open: none of them is work in flight, and a stream plane must not keep a CLI330 *  alive. `unref` is Node-only — Workers' `setTimeout` has no such method, hence the guard. */331function timer(fn: () => void, ms: number): ReturnType<typeof setTimeout> {332  const t = setTimeout(fn, ms);333  (t as { unref?: () => void }).unref?.();334  return t;335}336337/** The retry backoff, and deliberately NOT unref'd: a checkpoint mid-retry IS work in flight, and a338 *  process that exits during it drops text the caller is still awaiting. */339function delay(ms: number): Promise<void> {340  return new Promise((resolve) => setTimeout(resolve, ms));341}342343function errText(err: unknown): string {344  return String((err as Error)?.message ?? err);345}346347/** A per-plane stand-in for {@link RindleStreamOptions.hostId} when only the open CAS — not348 *  subscribe routing — needs a token. */349function randomOpenToken(): string {350  const c = (globalThis as { crypto?: { randomUUID?: () => string } }).crypto;351  return c?.randomUUID?.() ?? `open-${Math.random().toString(36).slice(2)}`;352}353354// ------------------------------------------------------------------------------- mapped-table SQL355356/** Every mapped column, defaults applied. `cancel`/`error` stay `undefined` unless named. */357export interface ResolvedStreamColumns {358  key: string;359  body: string;360  status: string;361  seq: string;362  cancel?: string;363  error?: string;364  host?: string;365  chunkKey: string;366  chunkStream: string;367  chunkSeq: string;368  chunkText: string;369}370371export function resolveStreamColumns(columns: StreamColumns | undefined): ResolvedStreamColumns {372  return {373    key: columns?.key ?? "id",374    body: columns?.body ?? "body",375    status: columns?.status ?? "status",376    seq: columns?.seq ?? "seq",377    ...(columns?.cancel !== undefined ? { cancel: columns.cancel } : {}),378    ...(columns?.error !== undefined ? { error: columns.error } : {}),379    ...(columns?.host !== undefined ? { host: columns.host } : {}),380    chunkKey: columns?.chunkKey ?? "id",381    chunkStream: columns?.chunkStream ?? "streamId",382    chunkSeq: columns?.chunkSeq ?? "seq",383    chunkText: columns?.chunkText ?? "text",384  };385}386387/** Quote an identifier for both dialects (`"x"` is standard in SQLite and Postgres alike). An388 *  embedded quote is refused rather than escaped: a mapping is app configuration, and a name that389 *  needs escaping is a mistake worth failing loudly on. */390function q(identifier: string): string {391  if (identifier.includes('"')) throw new TypeError(`invalid stream column/table name: ${JSON.stringify(identifier)}`);392  return `"${identifier}"`;393}394395/** The chunk row's deterministic id: a replayed checkpoint collides with itself and is absorbed by396 *  `ON CONFLICT DO NOTHING` — idempotency without an envelope or a dedup ledger (§3.3). */397export function streamChunkId(streamId: string, seq: number): string {398  return `${streamId}:${seq}`;399}400401/**402 * The chunk table's DDL, for the app's migration. The app owns the message table (this only states403 * the three columns the plane needs on it); the chunk table is entirely protocol-shaped, so it is404 * generated rather than hand-written.405 */406export function streamChunkTableDdl(tables: StreamTables, dialect: SqlDialect): string[] {407  const c = resolveStreamColumns(tables.columns);408  const text = dialect.name === "postgres" ? "text" : "TEXT";409  const int = dialect.name === "postgres" ? "bigint" : "INTEGER";410  return [411    `CREATE TABLE IF NOT EXISTS ${q(tables.chunks)} (` +412      `${q(c.chunkKey)} ${text} PRIMARY KEY, ` +413      `${q(c.chunkStream)} ${text} NOT NULL, ` +414      `${q(c.chunkSeq)} ${int} NOT NULL, ` +415      `${q(c.chunkText)} ${text} NOT NULL)`,416    // The read path is always "this message's chunks, in order" — and compaction deletes by stream.417    `CREATE INDEX IF NOT EXISTS ${q(`${tables.chunks}_stream_seq`)} ` +418      `ON ${q(tables.chunks)} (${q(c.chunkStream)}, ${q(c.chunkSeq)})`,419  ];420}421422/** The plane's SQL, one place, both dialects. */423class MappedTableSql {424  private readonly tables: StreamTables;425  private readonly cols: ResolvedStreamColumns;426  private readonly dialect: SqlDialect;427428  constructor(tables: StreamTables, dialect: SqlDialect) {429    this.tables = tables;430    this.cols = resolveStreamColumns(tables.columns);431    this.dialect = dialect;432  }433434  private p(i: number): string {435    return this.dialect.placeholder(i);436  }437438  /** The open probe: the row must exist and be empty. A generation is never a resume — regenerating439   *  is a NEW message id — so an already-advanced row is a bug worth refusing loudly. */440  probeOpen(): string {441    const { key, seq, status } = this.cols;442    return `SELECT ${q(seq)} AS seq, ${q(status)} AS status FROM ${q(this.tables.message)} WHERE ${q(key)} = ${this.p(1)}`;443  }444445  /** The open write. With a `host` column mapped this is the CAS half of the single-flight guard:446   *  set the status AND this producer's token only if no producer holds the row, verified by the447   *  read-back in {@link probeHost}. Without one it is the plain flip the probe already vetted. */448  markStreaming(streamId: string, hostToken: string): SqlStatement {449    const { key, status, host } = this.cols;450    if (host !== undefined) {451      return {452        sql:453          `UPDATE ${q(this.tables.message)} SET ${q(status)} = ${this.p(1)}, ${q(host)} = ${this.p(2)} ` +454          `WHERE ${q(key)} = ${this.p(3)} AND ${q(status)} <> ${this.p(4)}`,455        params: [STREAM_STATUS_STREAMING, hostToken, streamId, STREAM_STATUS_STREAMING],456      };457    }458    return {459      sql: `UPDATE ${q(this.tables.message)} SET ${q(status)} = ${this.p(1)} WHERE ${q(key)} = ${this.p(2)}`,460      params: [STREAM_STATUS_STREAMING, streamId],461    };462  }463464  /** The read-back half of the open CAS, or `undefined` when no `host` column is mapped. */465  probeHost(): string | undefined {466    const { host, key } = this.cols;467    if (host === undefined) return undefined;468    return `SELECT ${q(host)} AS host FROM ${q(this.tables.message)} WHERE ${q(key)} = ${this.p(1)}`;469  }470471  /** ONE transaction: insert the slice, then CAS the message's durable length. BOTH statements are472   *  gated on the same `seq = :from` predicate: the CAS refuses to apply twice or out of order, and473   *  the guarded insert makes a STALE RE-COVER a no-op — after a committed-but-ack-lost append, the474   *  next slice's `from` lags the row, and an unguarded insert would land an OVERLAPPING chunk row475   *  (a different deterministic id, so `ON CONFLICT` alone cannot absorb it) and corrupt the476   *  assembled text until compaction. The `ON CONFLICT` clause stays as the belt under that477   *  guard's braces. Which case actually happened is decided by the read-back ({@link probeState}),478   *  never assumed (§3.2/§3.3). */479  append(streamId: string, from: number, seq: number, text: string): SqlStatement[] {480    const c = this.cols;481    return [482      {483        sql:484          `INSERT INTO ${q(this.tables.chunks)} ` +485          `(${q(c.chunkKey)}, ${q(c.chunkStream)}, ${q(c.chunkSeq)}, ${q(c.chunkText)}) ` +486          `SELECT ${this.p(1)}, ${this.p(2)}, ${this.p(3)}, ${this.p(4)} ` +487          `WHERE EXISTS (SELECT 1 FROM ${q(this.tables.message)} ` +488          `WHERE ${q(c.key)} = ${this.p(5)} AND ${q(c.seq)} = ${this.p(6)}) ` +489          `ON CONFLICT DO NOTHING`,490        params: [streamChunkId(streamId, seq), streamId, seq, text, streamId, from],491      },492      {493        sql:494          `UPDATE ${q(this.tables.message)} SET ${q(c.seq)} = ${this.p(1)} ` +495          `WHERE ${q(c.key)} = ${this.p(2)} AND ${q(c.seq)} = ${this.p(3)}`,496        params: [seq, streamId, from],497      },498    ];499  }500501  /** Compaction (§3.4): write the whole body, seal the status, drop every chunk — atomically. Also502   *  the repair path: whatever the chunks did or did not capture, the body ends up correct. */503  compact(streamId: string, body: string, status: StreamStatus, error: string | undefined): SqlStatement[] {504    const c = this.cols;505    const sets = [`${q(c.body)} = ${this.p(1)}`, `${q(c.seq)} = ${this.p(2)}`, `${q(c.status)} = ${this.p(3)}`];506    const params: Array<string | number | null> = [body, body.length, status];507    if (c.error !== undefined) {508      sets.push(`${q(c.error)} = ${this.p(params.length + 1)}`);509      params.push(error ?? null);510    }511    params.push(streamId);512    return [513      {514        sql: `UPDATE ${q(this.tables.message)} SET ${sets.join(", ")} WHERE ${q(c.key)} = ${this.p(params.length)}`,515        params,516      },517      {518        sql: `DELETE FROM ${q(this.tables.chunks)} WHERE ${q(c.chunkStream)} = ${this.p(1)}`,519        params: [streamId],520      },521    ];522  }523524  get hasCancel(): boolean {525    return this.cols.cancel !== undefined;526  }527528  /** The post-append read-back: the row's authoritative length — what CONFIRMS an append applied529   *  (and absorbs a committed-but-ack-lost one) — plus the cancel flag when mapped (§6). */530  probeState(): string {531    const { cancel, key, seq } = this.cols;532    const picked = cancel === undefined ? `${q(seq)} AS seq` : `${q(seq)} AS seq, ${q(cancel)} AS cancel`;533    return `SELECT ${picked} FROM ${q(this.tables.message)} WHERE ${q(key)} = ${this.p(1)}`;534  }535}536537// ------------------------------------------------------------------------------- subscribers538539/** One attached reader. Frames queue; a reader that stops draining is bounded and then DROPPED with540 *  `stale` rather than allowed to pin the producer's memory — it resolves that by rejoining. */541class Subscriber {542  private queue: StreamFrame[] = [];543  private waiter?: (r: IteratorResult<StreamFrame>) => void;544  private done = false;545  private readonly cap: number;546  private readonly onDetach: (s: Subscriber) => void;547548  constructor(cap: number, onDetach: (s: Subscriber) => void) {549    this.cap = cap;550    this.onDetach = onDetach;551  }552553  /** @returns false when the queue overflowed (caller drops this subscriber). */554  offer(frame: StreamFrame): boolean {555    if (this.done) return true;556    if (this.waiter) {557      const w = this.waiter;558      this.waiter = undefined;559      w({ value: frame, done: false });560      return true;561    }562    if (this.queue.length >= this.cap) return false;563    this.queue.push(frame);564    return true;565  }566567  /** Deliver a terminal frame (bypassing the cap — it is the last one) and end the iterator. */568  finish(frame: StreamFrame): void {569    if (this.done) return;570    if (this.waiter) {571      const w = this.waiter;572      this.waiter = undefined;573      this.done = true;574      w({ value: frame, done: false });575      return;576    }577    this.queue.push(frame);578    this.done = true;579  }580581  close(): void {582    this.done = true;583    this.queue.length = 0;584    if (this.waiter) {585      const w = this.waiter;586      this.waiter = undefined;587      w({ value: undefined as never, done: true });588    }589    this.onDetach(this);590  }591592  frames(): AsyncIterable<StreamFrame> {593    const self = this;594    return {595      [Symbol.asyncIterator](): AsyncIterator<StreamFrame> {596        return {597          next(): Promise<IteratorResult<StreamFrame>> {598            const queued = self.queue.shift();599            if (queued !== undefined) return Promise.resolve({ value: queued, done: false });600            if (self.done) {601              self.onDetach(self);602              return Promise.resolve({ value: undefined as never, done: true });603            }604            return new Promise((resolve) => {605              self.waiter = resolve;606            });607          },608          return(): Promise<IteratorResult<StreamFrame>> {609            self.close();610            return Promise.resolve({ value: undefined as never, done: true });611          },612        };613      },614    };615  }616}617618// ------------------------------------------------------------------------------- the relay conform pass619620/**621 * One frame source arriving over a relay, conformed to the CP §4 contract622 * (designs-implemented/LM-STREAM-RELAY-DESIGN.md §4).623 *624 * An adapter is app code talking to Redis or a socket, and its frames feed `spliceStreamText` on a625 * browser — so the plane does not trust them. This pass enforces the frame invariants against the626 * prefix actually delivered and downgrades EVERY violation to a legal `stale` and nothing else:627 * `stale` already means "you are on the durable plane now, the store is the whole truth", so a628 * broken relay costs a reader smooth tokens, never corrupted text — and cannot wedge a producer.629 *630 * Replayed spans (a reconnecting adapter re-delivering what it already sent) are ABSORBED rather631 * than punished — deduping against the delivered prefix is what makes reconnect-replay safe without632 * every adapter hand-rolling it. Spans that overlap the prefix but extend past it pass through633 * whole: the client splices at the frame's own offset, so an exact overlap re-covers and appends.634 *635 * Pure state, no I/O, no plane: `feed` maps one incoming frame to 0-2 outgoing frames (a missing636 * `open` is synthesized at the join offset); `end`/`fail` close out a source that finished or threw637 * without a terminal. After a terminal, every method returns `[]`.638 */639export class StreamRelayConform {640  private readonly streamId: string;641  /** The requested join offset — the synthesized `open`'s position, and where the prefix starts. */642  private readonly from: number;643  private readonly onViolation: ((reason: string) => void) | undefined;644  /** End of the delivered prefix. */645  private pos: number;646  private lastDurable = 0;647  private opened = false;648  private done = false;649650  constructor(streamId: string, from: number, onViolation?: (reason: string) => void) {651    this.streamId = streamId;652    this.from = from;653    this.pos = from;654    this.onViolation = onViolation;655  }656657  feed(frame: StreamFrame): StreamFrame[] {658    if (this.done) return [];659    switch (frame.type) {660      case "open": {661        if (this.opened) return []; // a reconnecting adapter's second open: absorbed662        if (663          frame.streamId !== this.streamId ||664          !Number.isInteger(frame.from) ||665          !Number.isInteger(frame.seq) ||666          !Number.isInteger(frame.durableSeq) ||667          frame.from < 0 ||668          frame.seq < frame.from669        ) {670          return this.violate(`relay open for ${JSON.stringify(frame.streamId)} at ${frame.from} is malformed`);671        }672        // An open PAST the requested offset is the adapter saying it cannot serve `from` (pub/sub673        // has no history, §5): delivering it would leave a hole in the middle of the response, so674        // the honest answer is the durable plane.675        if (frame.from > this.from) {676          return this.violate(`relay open at ${frame.from} cannot serve the requested ${this.from}`);677        }678        this.opened = true;679        this.pos = frame.from;680        return [frame];681      }682      case "chunk": {683        if (684          !Number.isInteger(frame.from) ||685          !Number.isInteger(frame.seq) ||686          frame.from < 0 ||687          typeof frame.text !== "string" ||688          frame.text.length !== frame.seq - frame.from689        ) {690          return this.violate(`relay chunk ${frame.from}→${frame.seq} does not span exactly its offsets`);691        }692        if (frame.from > this.pos) {693          return this.violate(`relay chunk at ${frame.from} leaves a gap after ${this.pos}`);694        }695        if (frame.seq <= this.pos) return []; // entirely within the delivered prefix (a replay): absorbed696        const out = this.opened ? [] : [this.synthOpen()];697        this.pos = frame.seq;698        out.push(frame);699        return out;700      }701      case "durable": {702        // Purely informational to a reader (the hook ignores it; SSE uses it as a resume id), so a703        // claim that rewinds — or outruns what this subscription has SEEN produced, which P forbids704        // — is dropped rather than downgraded.705        if (!Number.isInteger(frame.seq) || frame.seq < this.lastDurable || frame.seq > this.pos) return [];706        const out = this.opened ? [] : [this.synthOpen()];707        this.lastDurable = frame.seq;708        out.push(frame);709        return out;710      }711      case "end": {712        if (!Number.isInteger(frame.seq) || frame.seq < 0) {713          return this.violate(`relay end at ${String(frame.seq)} is malformed`);714        }715        // An `end` whose durable length outruns the delivered prefix means the adapter LOST text716        // (durable never exceeds produced), so the reader is short and must not be told it saw717        // everything. Downgrading keeps `end` meaning the same thing relayed as local: the whole718        // produced text arrived.719        if (frame.seq > this.pos) {720          return this.violate(`relay end at ${frame.seq} outruns the ${this.pos} characters delivered`);721        }722        this.done = true;723        const out = this.opened ? [] : [this.synthOpen()];724        out.push(frame);725        return out;726      }727      case "stale":728      case "absent": {729        // Legal bare — the local plane's own floor/eviction answers carry no `open` either.730        this.done = true;731        return [frame];732      }733    }734  }735736  /** The source completed without a terminal (a truncated relay): the reader falls back. */737  end(): StreamFrame[] {738    if (this.done) return [];739    this.onViolation?.("relay source ended without a terminal frame");740    return this.terminate();741  }742743  /** The source threw mid-iteration, or the plane is dropping a reader that stopped draining:744   *  a bare `stale` at the delivered position. */745  fail(): StreamFrame[] {746    return this.done ? [] : this.terminate();747  }748749  /** A synthesized join, for an adapter that (correctly, in broadcast mode) never mirrors the750   *  per-subscriber `open`: positioned at the requested offset, which the reader asked from because751   *  its durable view already holds it. */752  private synthOpen(): StreamFrame {753    this.opened = true;754    return { type: "open", streamId: this.streamId, from: this.from, seq: this.from, durableSeq: this.from, ended: false };755  }756757  private terminate(): StreamFrame[] {758    this.done = true;759    return [{ type: "stale", floorSeq: this.pos, durableSeq: this.lastDurable }];760  }761762  private violate(reason: string): StreamFrame[] {763    this.onViolation?.(reason);764    return this.terminate();765  }766}767768// ------------------------------------------------------------------------------- the live stream769770interface Waiter {771  at: number;772  resolve: (seq: number) => void;773  reject: (err: unknown) => void;774}775776class LiveStream<User> {777  /** Retained text; `buf[0]` sits at offset {@link bufFrom}. */778  private buf = "";779  private bufFrom = 0;780  seq = 0;781  durableSeq = 0;782  ended = false;783  cancelled = false;784785  readonly streamId: string;786  readonly user: User;787  readonly meta: unknown;788  private readonly plane: StreamPlane<User>;789790  private readonly subs = new Set<Subscriber>();791  private sealing?: { status: StreamStatus; error?: string };792  private draining = false;793  private forced = false;794  /** The last drain round left a slice uncommitted: wait out the interval instead of re-attempting795   *  immediately, so a down store costs a retry cadence and not a hot loop. */796  private stalled = false;797  private tick: ReturnType<typeof setTimeout> | null = null;798  private waiters: Waiter[] = [];799  private sealed?: { resolve: () => void; reject: (e: unknown) => void; promise: Promise<void> };800  private sealError: unknown;801802  constructor(streamId: string, user: User, meta: unknown, plane: StreamPlane<User>) {803    this.streamId = streamId;804    this.user = user;805    this.meta = meta;806    this.plane = plane;807  }808809  // ---- producer side810811  push(text: string): void {812    if (this.sealing || this.ended) throw new Error(`stream ${this.streamId} is closed`);813    if (text.length === 0) return;814    const from = this.seq;815    this.buf += text;816    this.seq += text.length;817    this.fanout({ type: "chunk", from, seq: this.seq, text });818    if (this.seq - this.durableSeq >= this.plane.chars) this.kick(true);819    else this.arm();820  }821822  flush(): Promise<number> {823    const at = this.seq;824    if (this.durableSeq >= at) return Promise.resolve(this.durableSeq);825    // After a failed seal the shortfall is PERMANENT (the stream ended; nothing will retry it), so a826    // late flush rejects honestly instead of quietly re-running the seal.827    if (this.ended) {828      return Promise.reject(829        new Error(`stream ${this.streamId} ended with only ${this.durableSeq} of ${this.seq} characters durable`),830      );831    }832    const p = new Promise<number>((resolve, reject) => {833      this.waiters.push({ at, resolve, reject });834    });835    this.kick(true);836    return p;837  }838839  /** Seal the stream. Resolves once the terminal checkpoint has committed; rejects if it could not840   *  (the stream still ENDS — subscribers always get their `end` frame — the caller just learns the841   *  store is short of what was produced). */842  seal(status: StreamStatus, error?: string): Promise<void> {843    // Already sealed: hand back the SAME settled promise (already marked handled below), so a second844    // `close()`/`fail()` can neither re-seal nor mint a stray rejection.845    if (this.ended) return this.sealed?.promise ?? Promise.resolve();846    if (!this.sealing) {847      this.sealing = { status, error };848      let resolve!: () => void;849      let reject!: (e: unknown) => void;850      const promise = new Promise<void>((res, rej) => {851        resolve = res;852        reject = rej;853      });854      // A `fail()` nobody awaited must not become an unhandled rejection at process level. This855      // marks the promise handled WITHOUT consuming it — a caller that does await still sees the856      // rejection, because `await` attaches its own handler to the same promise.857      promise.catch(() => {});858      this.sealed = { resolve, reject, promise };859      this.disarm();860      this.kick(true);861    }862    return this.sealed!.promise;863  }864865  // ---- checkpoint driving866867  private arm(): void {868    if (this.tick || this.seq === this.durableSeq) return;869    this.tick = timer(() => {870      this.tick = null;871      this.kick(true);872    }, this.plane.intervalMs);873  }874875  private disarm(): void {876    if (this.tick) clearTimeout(this.tick);877    this.tick = null;878  }879880  private kick(force: boolean): void {881    if (force) this.forced = true;882    if (this.draining) return;883    this.draining = true;884    void this.drain();885  }886887  private async drain(): Promise<void> {888    this.stalled = false;889    try {890      for (;;) {891        // Sealing goes STRAIGHT to the seal: it writes the whole body (§3.4), so committing the892        // outstanding tail as one more chunk first would be pure waste — and the seal repairs any893        // earlier checkpoint that failed, which a tail commit could not.894        if (this.sealing) {895          await this.commitSeal();896          break;897        }898        const behind = this.seq - this.durableSeq;899        if (behind > 0 && (this.forced || behind >= this.plane.chars)) {900          // Consume the force BEFORE the round, not after: a `flush()` that lands while this round901          // is already in flight re-sets it and gets a round of its OWN — clearing it afterwards902          // would swallow that request and leave the flush to the interval cadence.903          this.forced = false;904          this.disarm();905          const failure = await this.commitTail();906          if (failure) {907            this.stalled = true;908            // A seal requested WHILE this checkpoint was in flight outranks the retry cadence: the909            // seal writes the whole body anyway, so it both supersedes and repairs the failed slice.910            // Backing off here instead would make `close()` wait out `intervalMs` for nothing. An911            // explicit re-force during the failed round likewise earns one immediate re-attempt —912            // someone is waiting on it — while an unforced failure stalls to the cadence.913            if (!this.sealing && !this.forced) break;914          }915          continue;916        }917        this.forced = false;918        break;919      }920    } finally {921      this.draining = false;922      if (!this.ended) {923        // A pending seal ALWAYS proceeds at once — including out of a stall (see the loop above);924        // otherwise a stall waits out the interval, and deltas that landed mid-commit may already925        // meet the threshold and deserve an immediate round.926        if (this.sealing) this.kick(false);927        else if (this.stalled) this.arm();928        else if (this.seq - this.durableSeq >= this.plane.chars) this.kick(false);929        else this.arm();930      }931    }932  }933934  /** @returns the failure, or `undefined` when the slice committed. */935  private async commitTail(): Promise<{ err: unknown } | undefined> {936    const from = this.durableSeq;937    const to = this.seq;938    const text = this.buf.slice(from - this.bufFrom, to - this.bufFrom);939    let out: { cancelRequested?: boolean; seq?: number } | void;940    try {941      out = await this.plane.commit({ kind: "append", streamId: this.streamId, from, seq: to, text });942    } catch (err) {943      // The store may hold MORE than we believe — an append that committed while its ack was lost,944      // somewhere in the retry chain. The read-back is the truth, and adopting it is what keeps the945      // NEXT slice contiguous instead of overlapping (§3.3).946      const truth = await this.plane.probeDurableSeq(this.streamId);947      if (truth !== undefined && truth > this.durableSeq) this.advanceDurable(truth);948      this.plane.reportCheckpointError(err, { streamId: this.streamId, from, seq: to });949      // Waiters that asked for THIS prefix learn it did not land; later waiters keep waiting for a950      // later attempt. A durability failure is never silent.951      this.rejectWaiters(to, err);952      return { err };953    }954    if (out?.cancelRequested) this.cancelled = true;955    // `durableSeq` advances to what the store CONFIRMED — the mapped path reads it back, a `commit`956    // callback may report it — never to a plane-side assumption. A confirmation short of `to` means957    // this slice's `from` was stale (a prior ack loss the read-back could not see at the time): the958    // guarded statements made it a no-op, and the next round re-covers from the confirmed offset.959    const confirmed = out?.seq ?? to;960    if (confirmed > this.durableSeq) this.advanceDurable(confirmed);961    if (confirmed < to) {962      const err = new Error(963        `stream ${this.streamId}: the store confirmed ${confirmed} of ${to} — re-covering from there`,964      );965      this.plane.reportCheckpointError(err, { streamId: this.streamId, from, seq: to });966      this.rejectWaiters(to, err);967      return { err };968    }969    return undefined;970  }971972  /** The one place durable progress is recorded: position, `durable` frame, buffer trim, waiters. */973  private advanceDurable(seq: number): void {974    this.durableSeq = seq;975    this.fanout({ type: "durable", seq });976    this.trim();977    this.resolveWaiters();978  }979980  private async commitSeal(): Promise<void> {981    const seal = this.sealing!;982    const from = this.durableSeq;983    try {984      await this.plane.commit({985        kind: "close",986        streamId: this.streamId,987        from,988        seq: this.seq,989        // A trimmed buffer (only reachable in `commit`-callback mode, where the app opted into990        // trimming) cannot supply the whole body: `bodyFrom` says where the retained text starts,991        // so the app can still append exactly the outstanding tail (`body.slice(from - bodyFrom)`).992        body: this.buf,993        bodyFrom: this.bufFrom,994        status: seal.status,995        ...(seal.error !== undefined ? { error: seal.error } : {}),996      });997      this.durableSeq = this.seq;998    } catch (err) {999      this.sealError = err;1000      this.plane.reportCheckpointError(err, { streamId: this.streamId, from, seq: this.seq });1001    }1002    this.ended = true;1003    this.disarm();1004    this.rejectWaiters(1005      Number.POSITIVE_INFINITY,1006      new Error(`stream ${this.streamId} ended with only ${this.durableSeq} of ${this.seq} characters durable`),1007    );1008    this.resolveWaiters();1009    const frame: StreamFrame = {1010      type: "end",1011      seq: this.durableSeq,1012      status: seal.status,1013      ...(seal.error !== undefined ? { error: seal.error } : {}),1014    };1015    // The terminal reaches the relay too (it bypasses `fanout` locally only to bypass the cap).1016    this.plane.publishRelay(this.streamId, frame);1017    for (const sub of [...this.subs]) sub.finish(frame);1018    this.plane.retire(this.streamId);1019    if (this.sealError) this.sealed?.reject(this.sealError);1020    else this.sealed?.resolve();1021  }10221023  /** Keep everything at or above `durableSeq`, plus a slack window below it so a client whose IVM1024   *  view lags one checkpoint can still join without a `stale` round trip (§4). A `retainChars` of1025   *  `Infinity` (the mapped-table default — compaction needs the whole text) never trims. */1026  private trim(): void {1027    if (!Number.isFinite(this.plane.retainChars)) return;1028    const floor = Math.max(0, this.durableSeq - this.plane.retainChars);1029    if (floor <= this.bufFrom) return;1030    this.buf = this.buf.slice(floor - this.bufFrom);1031    this.bufFrom = floor;1032  }10331034  private resolveWaiters(): void {1035    if (this.waiters.length === 0) return;1036    const still: Waiter[] = [];1037    for (const w of this.waiters) {1038      if (w.at <= this.durableSeq) w.resolve(this.durableSeq);1039      else still.push(w);1040    }1041    this.waiters = still;1042  }10431044  private rejectWaiters(upTo: number, err: unknown): void {1045    if (this.waiters.length === 0) return;1046    const still: Waiter[] = [];1047    for (const w of this.waiters) {1048      if (w.at <= upTo && w.at > this.durableSeq) w.reject(err);1049      else still.push(w);1050    }1051    this.waiters = still;1052  }10531054  // ---- subscriber side10551056  private fanout(frame: StreamFrame): void {1057    // Every frame local subscribers get, the relay gets — including when nobody local is attached1058    // (a broadcast relay's whole point). Wrapped so an outage costs the live leg, never this stream.1059    this.plane.publishRelay(this.streamId, frame);1060    for (const sub of [...this.subs]) {1061      if (!sub.offer(frame)) {1062        // Bounded, then dropped: a reader that stopped draining costs itself a rejoin, never the1063        // producer's memory.1064        sub.finish({ type: "stale", floorSeq: this.bufFrom, durableSeq: this.durableSeq });1065        this.subs.delete(sub);1066      }1067    }1068  }10691070  /** Attach a reader at `from`. Runs in ONE synchronous block — snapshot, replay, register — so no1071   *  delta can slip between the replay and the live tail (the gap-free half of contract S). */1072  subscribe(from: number): StreamSubscription {1073    const sub = new Subscriber(this.plane.maxQueuedFrames, (s) => this.subs.delete(s));1074    // Floored as well as clamped: a fractional `from` (a hand-built subscribe input) would otherwise1075    // produce a replay chunk whose `text.length !== seq - from`, breaking the frame invariant.1076    const at = Math.min(Math.max(Math.floor(from), 0), this.seq);1077    if (at < this.bufFrom) {1078      sub.finish({ type: "stale", floorSeq: this.bufFrom, durableSeq: this.durableSeq });1079      return this.wrap(sub);1080    }1081    sub.offer({ type: "open", streamId: this.streamId, from: at, seq: this.seq, durableSeq: this.durableSeq, ended: this.ended });1082    if (this.seq > at) {1083      sub.offer({ type: "chunk", from: at, seq: this.seq, text: this.buf.slice(at - this.bufFrom) });1084    }1085    if (this.durableSeq > at) sub.offer({ type: "durable", seq: this.durableSeq });1086    if (this.ended) {1087      const seal = this.sealing ?? { status: "interrupted" as StreamStatus };1088      sub.finish({1089        type: "end",1090        seq: this.durableSeq,1091        status: seal.status,1092        ...(seal.error !== undefined ? { error: seal.error } : {}),1093      });1094      return this.wrap(sub);1095    }1096    this.subs.add(sub);1097    return this.wrap(sub);1098  }10991100  private wrap(sub: Subscriber): StreamSubscription {1101    return { streamId: this.streamId, frames: sub.frames(), close: () => sub.close() };1102  }11031104  /** Drop every reader without sealing (process teardown — `drainStreams` seals first). */1105  detachAll(): void {1106    for (const sub of [...this.subs]) sub.close();1107    this.subs.clear();1108    this.disarm();1109  }1110}11111112// ------------------------------------------------------------------------------- the plane11131114/** What the plane needs from the api-server to write a checkpoint: the backend's OUTSIDE-transaction1115 *  SQL surface (`batch` is one transaction on every backend) and its dialect. Deliberately narrow so1116 *  `streams.ts` never imports the server (no cycle). */1117export interface StreamSqlSink {1118  readonly dialect: SqlDialect;1119  readonly sql: ServerSql;1120}11211122export class StreamPlane<User> {1123  readonly chars: number;1124  readonly intervalMs: number;1125  readonly retries: number;1126  readonly retainChars: number;1127  readonly lingerMs: number;1128  readonly maxQueuedFrames: number;1129  readonly relayAttachTimeoutMs: number;11301131  private readonly live = new Map<string, LiveStream<User>>();1132  /** Live relayed subscriptions, so teardown ({@link closeSync}) releases their drivers too. */1133  private readonly relayed = new Set<Subscriber>();1134  private readonly opts: RindleStreamOptions<User>;1135  private readonly sink: StreamSqlSink | undefined;1136  private readonly mapped: MappedTableSql | undefined;1137  private readonly tables: StreamTables | undefined;1138  /** The open CAS token (§5.1): `hostId`, or a random per-plane stand-in when only the CAS — not1139   *  routing — needs it. */1140  private readonly openToken: string;11411142  constructor(opts: RindleStreamOptions<User>, sink?: StreamSqlSink) {1143    this.opts = opts;1144    this.sink = sink;1145    this.chars = opts.policy?.chars ?? DEFAULT_CHECKPOINT_CHARS;1146    this.intervalMs = opts.policy?.intervalMs ?? DEFAULT_CHECKPOINT_INTERVAL_MS;1147    this.retries = opts.policy?.retries ?? DEFAULT_CHECKPOINT_RETRIES;1148    this.lingerMs = opts.lingerMs ?? DEFAULT_LINGER_MS;1149    this.maxQueuedFrames = opts.maxQueuedFrames ?? DEFAULT_MAX_QUEUED_FRAMES;1150    this.relayAttachTimeoutMs = opts.relayAttachTimeoutMs ?? DEFAULT_RELAY_ATTACH_TIMEOUT_MS;1151    this.openToken = opts.hostId ?? randomOpenToken();1152    // Relay misconfiguration is refused loudly (the room-profile rule), never ignored.1153    if (opts.relay !== undefined && opts.relay.publish === undefined && opts.relay.attach === undefined) {1154      throw new TypeError("streams.relay implements neither publish nor attach — which topology is this? (LM-STREAM-RELAY §3.1)");1155    }1156    if (opts.relay === undefined && (opts.relayAttachTimeoutMs !== undefined || opts.onRelayError !== undefined)) {1157      throw new TypeError("streams.relayAttachTimeoutMs/onRelayError do nothing without streams.relay");1158    }1159    if ("tables" in opts.checkpoint) {1160      if (!sink) throw new TypeError("streams.checkpoint.tables needs a SQL-capable mutation backend");1161      this.tables = opts.checkpoint.tables;1162      this.mapped = new MappedTableSql(opts.checkpoint.tables, sink.dialect);1163      // Options this mode cannot honour are refused loudly (the room-profile rule), never ignored.1164      if (opts.retainChars !== undefined) {1165        throw new TypeError(1166          "streams.retainChars is `commit`-mode only — mapped tables retain the whole text for compaction (§3.4)",1167        );1168      }1169      if (opts.hostId !== undefined && resolveStreamColumns(opts.checkpoint.tables.columns).host === undefined) {1170        throw new TypeError("streams.hostId does nothing in tables mode until columns.host names where to persist it");1171      }1172      // Compaction writes the WHOLE body at close, so the buffer is retained in full. `retainChars`1173      // is honoured only where the app owns persistence and may not need the tail.1174      this.retainChars = Number.POSITIVE_INFINITY;1175    } else {1176      this.retainChars = opts.retainChars ?? DEFAULT_RETAIN_CHARS;1177    }1178  }11791180  /** Open a stream on an EXISTING message row (§5): the app's own mutator wrote it, alongside the1181   *  user's prompt, so the pointer is already durable and every client's query already shows the1182   *  message. This verifies it and flips it to `streaming`. */1183  async open(input: OpenStreamInput<User>): Promise<StreamHandle> {1184    if (this.live.has(input.streamId)) {1185      throw new Error(`stream ${input.streamId} is already open on this host`);1186    }1187    const stream = new LiveStream<User>(input.streamId, input.user, input.meta, this);1188    this.live.set(input.streamId, stream);1189    try {1190      if (this.mapped) await this.assertOpenable(input.streamId);1191      await this.commit({1192        kind: "open",1193        streamId: input.streamId,1194        meta: input.meta ?? null,1195        ...(this.opts.hostId !== undefined ? { hostId: this.opts.hostId } : {}),1196        startedAt: Date.now(),1197      });1198    } catch (err) {1199      this.live.delete(input.streamId);1200      throw err;1201    }1202    return {1203      streamId: stream.streamId,1204      get seq() {1205        return stream.seq;1206      },1207      get durableSeq() {1208        return stream.durableSeq;1209      },1210      get cancelled() {1211        return stream.cancelled;1212      },1213      push: (text) => stream.push(text),1214      flush: () => stream.flush(),1215      pump: async (deltas) => {1216        // `for await` closes the iterator on `break`, which is what aborts the SDK's underlying1217        // request — the whole point of honouring cancellation here rather than in the caller.1218        for await (const d of deltas) {1219          stream.push(d);1220          if (stream.cancelled) break;1221        }1222      },1223      close: () => stream.seal(stream.cancelled ? "cancelled" : "complete"),1224      fail: (error) => stream.seal("error", errText(error)),1225    };1226  }12271228  /**1229   * The read-only precondition on the app's message row, checked ONCE per `open` (§5.1). Read-only on1230   * purpose: it is a decision about the app's data, so re-deciding it per write attempt would let a1231   * lost ack turn a success into a refusal.1232   *1233   * The `streaming` check is the **single-flight guard**, and it lives at the ROW rather than in this1234   * process's map because the thing it defends against is distributed: the kick that starts a1235   * generation is an at-least-once effect (a retried mutation envelope re-runs its post-commit code,1236   * §10.5), so a second kick can land on another instance, where the in-memory map is empty. Two1237   * producers on one `streamId` would interleave: both CAS the same length, one stalls, and whichever1238   * closes last overwrites the body with ITS buffer. Cheap to refuse; expensive to debug.1239   *1240   * This probe alone is check-then-act — it and the open write are separate round trips, so two1241   * SIMULTANEOUS kicks can both pass it. Mapping a `host` column closes that window: the open write1242   * turns conditional and {@link verifyOpenWinner}'s read-back names the winner.1243   */1244  private async assertOpenable(streamId: string): Promise<void> {1245    const cols = resolveStreamColumns(this.tables!.columns);1246    const rows = await this.sink!.sql.query<{ seq: number | null; status: string | null }>(1247      this.mapped!.probeOpen(),1248      [streamId],1249    );1250    const row = rows[0];1251    if (!row) {1252      throw new StreamOpenRefused(1253        `stream ${streamId}: no row in "${this.tables!.message}" — the app's own mutator must write the message ` +1254          `(with its chatId/role/…) BEFORE opening the stream on it`,1255      );1256    }1257    // A NULL length is refused as loudly as an advanced one, and for a subtler reason: it is the CAS1258    // column, and `WHERE seq = 0` can never match NULL. A nullable column would let every chunk1259    // insert land while its length CAS silently matched nothing — the one failure mode this plane1260    // cannot detect (`batch` reports no row count) and one compaction would paper over (§5.1).1261    if (typeof row.seq !== "number") {1262      throw new StreamOpenRefused(1263        `stream ${streamId}: "${this.tables!.message}"."${cols.seq}" is ${row.seq === null ? "NULL" : typeof row.seq} — ` +1264          `the length column must be NOT NULL DEFAULT 0 (it is compare-and-swapped on every checkpoint, and no ` +1265          `comparison matches NULL)`,1266      );1267    }1268    if (row.seq !== 0) {1269      throw new StreamOpenRefused(1270        `stream ${streamId}: the message row already holds ${row.seq} characters — a regeneration is a NEW message ` +1271          `id, never a resume of an advanced row`,1272      );1273    }1274    if (row.status === STREAM_STATUS_STREAMING) {1275      throw new StreamOpenRefused(1276        `stream ${streamId}: "${cols.status}" is already "${STREAM_STATUS_STREAMING}" — another producer holds this ` +1277          `stream. Two producers on one stream interleave their checkpoints; if the first one's host died, the ` +1278          `sweeper marking it "interrupted" is what releases the row (§7)`,1279      );1280    }1281  }12821283  async subscribe(input: SubscribeStreamInput<User>): Promise<StreamSubscription> {1284    const stream = this.live.get(input.streamId);1285    const from = input.from ?? 0;1286    // Authorize BEFORE existence is consulted: a denial must not double as an existence oracle.1287    const verdict = await this.opts.authorize({1288      user: input.user,1289      streamId: input.streamId,1290      from,1291      meta: stream === undefined ? undefined : stream.meta,1292      request: input.request,1293    });1294    if (verdict === false) throw new StreamForbidden(input.streamId);1295    if (!stream) {1296      // The relay is consulted only on a live-plane miss, and only after `authorize` passed — a1297      // denial must not become an existence probe against the relay either. A local stream always1298      // wins: the producer's own readers keep the lowest-latency path.1299      const relayed = await this.attachRelay(input.streamId, from);1300      return relayed ?? oneFrame(input.streamId, { type: "absent" });1301    }1302    return stream.subscribe(from);1303  }13041305  /** The subscribe-miss leg (LM-STREAM-RELAY §3): ask the app's relay for the frames of a stream1306   *  this process is not hosting. `undefined` — no relay, no `attach`, the adapter declined, timed1307   *  out, or threw — is `absent`, exactly today's answer. */1308  private async attachRelay(streamId: string, from: number): Promise<StreamSubscription | undefined> {1309    const relay = this.opts.relay;1310    if (relay?.attach === undefined) return undefined;1311    // Floored like the local join, but clamped only below: the producer's length is not known here.1312    const at = Number.isFinite(from) ? Math.max(Math.floor(from), 0) : 0;1313    let source: AsyncIterable<StreamFrame> | undefined;1314    try {1315      source = await this.boundedAttach(relay, streamId, at);1316    } catch (err) {1317      this.reportRelayError(err, { streamId, phase: "attach" });1318      return undefined;1319    }1320    if (source === undefined) return undefined;1321    return this.relaySubscription(streamId, at, source);1322  }13231324  /** `attach`, bounded by {@link RindleStreamOptions.relayAttachTimeoutMs}: a hung adapter yields1325   *  `absent`, not a hung HTTP request. A source that resolves after the deadline is closed, not1326   *  leaked. */1327  private boundedAttach(1328    relay: StreamRelay,1329    streamId: string,1330    from: number,1331  ): Promise<AsyncIterable<StreamFrame> | undefined> {1332    // `async` wrapping so a synchronously-throwing adapter is an attach failure, not a plane throw.1333    const attempt = (async () => relay.attach!(streamId, from))();1334    return new Promise((resolve, reject) => {1335      let late = false;1336      const t = timer(() => {1337        late = true;1338        reject(new Error(`stream ${streamId}: relay.attach timed out after ${this.relayAttachTimeoutMs}ms`));1339      }, this.relayAttachTimeoutMs);1340      attempt.then(1341        (source) => {1342          clearTimeout(t);1343          if (!late) return resolve(source);1344          closeFrameSource(source); // too late to serve the reader; don't leak the channel1345        },1346        (err) => {1347          clearTimeout(t);1348          if (!late) reject(err);1349        },1350      );1351    });1352  }13531354  /** Wrap an adapter's frame source as a plane subscription: conform every frame (LM-STREAM-RELAY1355   *  §4), bound the reader with the same queue cap as a local one (§7), and tear the adapter down1356   *  when either side lets go. The driver never throws into the plane: adapter failures become one1357   *  `stale`. */1358  private relaySubscription(streamId: string, from: number, source: AsyncIterable<StreamFrame>): StreamSubscription {1359    const conform = new StreamRelayConform(streamId, from, (reason) =>1360      this.reportRelayError(new Error(reason), { streamId, phase: "frames" }),1361    );1362    let closed = false;1363    let signalClose!: () => void;1364    const closedP = new Promise<void>((resolve) => (signalClose = resolve));1365    const closedTag = closedP.then(() => "closed" as const);1366    const release = (): void => {1367      if (!closed) {1368        closed = true;1369        signalClose();1370      }1371    };1372    const sub = new Subscriber(this.maxQueuedFrames, (s) => {1373      this.relayed.delete(s);1374      release();1375    });1376    this.relayed.add(sub);1377    let it: AsyncIterator<StreamFrame> | undefined;1378    /** @returns false once the subscription finished (terminal delivered, or the reader dropped). */1379    const deliver = (frames: StreamFrame[]): boolean => {1380      for (const frame of frames) {1381        if (frame.type === "end" || frame.type === "stale" || frame.type === "absent") {1382          sub.finish(frame);1383          return false;1384        }1385        if (!sub.offer(frame)) {1386          // The same bound as a local reader: a relayed subscriber that stops draining costs1387          // itself the live leg, never unbounded memory.1388          const [stale] = conform.fail();1389          if (stale) sub.finish(stale);1390          return false;1391        }1392      }1393      return true;1394    };1395    void (async () => {1396      try {1397        // Iterator construction is adapter code too: keep a throwing factory inside the same1398        // stale/report/cleanup boundary as a throwing `next()`.1399        it = source[Symbol.asyncIterator]();1400        for (;;) {1401          // Raced rather than awaited bare: a reader disconnect must release this driver even when1402          // the adapter never yields another frame (its own `return()` may be queued behind the1403          // pending `next()` forever).1404          const res = await Promise.race([it.next(), closedTag]);1405          if (res === "closed") return;1406          if (res.done) {1407            deliver(conform.end());1408            return;1409          }1410          if (!deliver(conform.feed(res.value))) return;1411        }1412      } catch (err) {1413        this.reportRelayError(err, { streamId, phase: "frames" });1414        deliver(conform.fail());1415      } finally {1416        release();1417        // If construction itself threw there is no iterator to return, and asking the source to1418        // construct a second one during cleanup could repeat side effects or throw again.1419        closeFrameSource(undefined, it);1420      }1421    })();1422    return { streamId, frames: sub.frames(), close: () => sub.close() };1423  }14241425  /** Mirror one producer frame outward (LM-STREAM-RELAY §3). Never blocks or breaks the producer:1426   *  a throw or rejected promise is reported and swallowed — a relay outage may cost relayed1427   *  readers the live leg, never the generation or its checkpoints. */1428  publishRelay(streamId: string, frame: StreamFrame): void {1429    const relay = this.opts.relay;1430    if (relay?.publish === undefined) return;1431    try {1432      const published = relay.publish(streamId, frame);1433      if (published !== undefined) {1434        void Promise.resolve(published).catch((err) =>1435          this.reportRelayError(err, { streamId, phase: "publish" }),1436        );1437      }1438    } catch (err) {1439      this.reportRelayError(err, { streamId, phase: "publish" });1440    }1441  }14421443  reportRelayError(err: unknown, info: StreamRelayErrorInfo): void {1444    // Same discipline as reportCheckpointError: a diagnostic must never take its caller down.1445    try {1446      if (this.opts.onRelayError) this.opts.onRelayError(err, info);1447      else console.error(`[rindle api-server] stream ${info.streamId}: relay ${info.phase} failed:`, err);1448    } catch (hookErr) {1449      console.error(`[rindle api-server] stream ${info.streamId}: onRelayError itself threw:`, hookErr);1450    }1451  }14521453  /** Seal every live stream `interrupted`. In mapped-table mode the seal IS the compaction, so a1454   *  graceful drain loses nothing PRODUCED; the status still says the response was cut short rather1455   *  than claiming completion (§5). Wire it to SIGTERM. */1456  async drainStreams(): Promise<void> {1457    await Promise.allSettled([...this.live.values()].map((s) => s.seal("interrupted")));1458  }14591460  /** Teardown: drop readers and timers WITHOUT a durable write (that is `drainStreams`). */1461  closeSync(): void {1462    for (const s of this.live.values()) s.detachAll();1463    this.live.clear();1464    for (const s of [...this.relayed]) s.close();1465    this.relayed.clear();1466  }14671468  /** A sealed stream stays joinable for the linger window, so a subscribe that races the last token1469   *  still gets `end` (and the tail it missed) rather than a bare `absent`. */1470  retire(streamId: string): void {1471    timer(() => {1472      const entry = this.live.get(streamId);1473      if (entry?.ended) {1474        this.live.delete(streamId);1475        entry.detachAll();1476      }1477    }, this.lingerMs);1478  }14791480  reportCheckpointError(err: unknown, info: { streamId: string; from: number; seq: number }): void {1481    // Reporting runs INSIDE the drain loop, so a throwing hook would take the loop down with it and1482    // wedge the stream — the one failure mode a diagnostic must never cause.1483    try {1484      if (this.opts.onCheckpointError) this.opts.onCheckpointError(err, info);1485      else console.error(`[rindle api-server] stream ${info.streamId}: checkpoint ${info.from}→${info.seq} failed:`, err);1486    } catch (hookErr) {1487      console.error(`[rindle api-server] stream ${info.streamId}: onCheckpointError itself threw:`, hookErr);1488    }1489  }14901491  /** Drive ONE checkpoint, retrying on failure. Every statement it emits is idempotent under replay1492   *  — the chunk insert dedups on its deterministic id, the length CAS refuses to apply twice, the1493   *  compaction is a whole-row overwrite — so "retry until it sticks" needs no dedup ledger and no1494   *  `lmid` (§3.3). */1495  async commit(input: StreamCommitInput): Promise<{ cancelRequested?: boolean; seq?: number } | void> {1496    let lastErr: unknown;1497    for (let attempt = 0; attempt <= this.retries; attempt++) {1498      if (attempt > 0) await delay(RETRY_BACKOFF_MS * 2 ** (attempt - 1));1499      try {1500        return await this.commitOnce(input);1501      } catch (err) {1502        if (err instanceof StreamOpenRefused) throw err; // a settled verdict, not a blip1503        lastErr = err;1504      }1505    }1506    throw lastErr;1507  }15081509  private async commitOnce(input: StreamCommitInput): Promise<{ cancelRequested?: boolean; seq?: number } | void> {1510    const target = this.opts.checkpoint;1511    if ("commit" in target) return (await target.commit(input)) ?? undefined;1512    const sql = this.sink!.sql;1513    const mapped = this.mapped!;1514    switch (input.kind) {1515      case "open": {1516        // The PRECONDITION is checked once, in `assertOpenable`, before this write — never here.1517        // Re-reading it per attempt would make a lost ack on the write below refuse its own success1518        // (the row would already say `streaming`).1519        await sql.batch([mapped.markStreaming(input.streamId, this.openToken)]);1520        await this.verifyOpenWinner(input.streamId);1521        return undefined;1522      }1523      case "append": {1524        try {1525          await sql.batch(mapped.append(input.streamId, input.from, input.seq, input.text));1526        } catch (err) {1527          // The batch may have COMMITTED with its ack lost. The read-back is the truth: a row at1528          // (or past) this slice's end means it landed, and the throw was only the reply.1529          const truth = await this.readAppendState(input.streamId).catch(() => undefined);1530          if (truth === undefined || truth.seq < input.seq) throw err;1531          return truth;1532        }1533        // The read-back CONFIRMS the append (the guarded statements report no row count). If it1534        // throws, the retry is safe: a replay of an applied slice no-ops on the guard and confirms1535        // on its own read-back.1536        return await this.readAppendState(input.streamId);1537      }1538      case "close": {1539        await sql.batch(mapped.compact(input.streamId, input.body, input.status, input.error));1540        return undefined;1541      }1542    }1543  }15441545  /** The read-back half of the open CAS, run only when a `host` column is mapped. The probe in1546   *  `assertOpenable` and the open write are separate round trips, so bare check-then-act leaves a1547   *  window where two simultaneous kicks both pass the probe; the conditional `markStreaming`1548   *  matches nothing when a rival got there first, and whichever token the row now holds names the1549   *  winner — a true CAS with no row counts needed. A replayed open (lost ack) reads back its OWN1550   *  token and proceeds. */1551  private async verifyOpenWinner(streamId: string): Promise<void> {1552    const probe = this.mapped!.probeHost();1553    if (probe === undefined) return;1554    const rows = await this.sink!.sql.query<{ host: unknown }>(probe, [streamId]);1555    if (rows[0]?.host !== this.openToken) {1556      throw new StreamOpenRefused(1557        `stream ${streamId}: another producer won the open race (the row's host is ` +1558          `${JSON.stringify(rows[0]?.host ?? null)}) — two producers on one stream interleave their checkpoints, ` +1559          `so the loser stands down`,1560      );1561    }1562  }15631564  /** The post-append read-back (§3.2): the row's authoritative length — which both CONFIRMS an1565   *  append (the guarded batch reports no row count) and absorbs a committed-but-ack-lost one —1566   *  plus the reader's cancel flag when mapped (§6), riding the same indexed point read. This read1567   *  is LOAD-BEARING: a failure here fails the append attempt (retried, then stalled), because1568   *  claiming durability the store did not confirm is the one dishonesty this plane refuses. */1569  private async readAppendState(streamId: string): Promise<{ seq: number; cancelRequested?: boolean }> {1570    const rows = await this.sink!.sql.query<{ seq: unknown; cancel?: unknown }>(this.mapped!.probeState(), [1571      streamId,1572    ]);1573    const row = rows[0];1574    if (row === undefined || typeof row.seq !== "number") {1575      throw new Error(`stream ${streamId}: the read-back found no usable message row`);1576    }1577    return { seq: row.seq, ...(this.mapped!.hasCancel ? { cancelRequested: Boolean(row.cancel) } : {}) };1578  }15791580  /** The store's word on how much is durable — for resynchronizing after a FAILED append, where a1581   *  lost ack may have left the store ahead of the plane. `undefined` in `commit` mode (no readable1582   *  authority) or when the read itself fails (the next attempt retries the resync too). */1583  async probeDurableSeq(streamId: string): Promise<number | undefined> {1584    if (!this.mapped) return undefined;1585    try {1586      return (await this.readAppendState(streamId)).seq;1587    } catch {1588      return undefined;1589    }1590  }1591}15921593/** The open probe's verdict: the message row is missing or already advanced. Not retried — it is a1594 *  settled statement about the app's data, and retrying re-asks a question already answered. */1595export class StreamOpenRefused extends Error {1596  constructor(message: string) {1597    super(message);1598    this.name = "StreamOpenRefused";1599  }1600}16011602/** Refused by {@link RindleStreamOptions.authorize}. Distinct from the api-server's own1603 *  `RindleApiError` so this module stays importable without the server (no cycle); the server1604 *  translates it to a 403 at the handler seam. */1605export class StreamForbidden extends Error {1606  readonly streamId: string;16071608  constructor(streamId: string) {1609    super(`stream ${streamId}: forbidden`);1610    this.name = "StreamForbidden";1611    this.streamId = streamId;1612  }1613}16141615/** Best-effort adapter teardown (the `for await` discipline, by hand): never awaited into the1616 *  plane — a hung `return()` must not hold anything — and a synchronously-throwing one is the1617 *  adapter's bug, not the plane's problem. */1618function closeFrameSource(source: AsyncIterable<StreamFrame> | undefined, it?: AsyncIterator<StreamFrame>): void {1619  try {1620    const iter = it ?? source?.[Symbol.asyncIterator]();1621    void Promise.resolve(iter?.return?.()).catch(() => {});1622  } catch {1623    // ignored — see above1624  }1625}16261627function oneFrame(streamId: string, frame: StreamFrame): StreamSubscription {1628  let taken = false;1629  return {1630    streamId,1631    frames: {1632      [Symbol.asyncIterator](): AsyncIterator<StreamFrame> {1633        return {1634          next(): Promise<IteratorResult<StreamFrame>> {1635            if (taken) return Promise.resolve({ value: undefined as never, done: true });1636            taken = true;1637            return Promise.resolve({ value: frame, done: false });1638          },1639        };1640      },1641    },1642    close: () => {1643      taken = true;1644    },1645  };1646}16471648// ------------------------------------------------------------------------------- SSE transport16491650/** Headers for the SSE response. `x-accel-buffering` is the nginx-family opt-out — without it a1651 *  buffering proxy holds the tokens and hands the user a paragraph at a time. */1652export const STREAM_SSE_HEADERS: Record<string, string> = {1653  "content-type": "text/event-stream; charset=utf-8",1654  "cache-control": "no-cache, no-transform",1655  connection: "keep-alive",1656  "x-accel-buffering": "no",1657};16581659/** Pull a subscribe request out of a fetch-style GET: `?streamId=…&from=…`, with `Last-Event-ID`1660 *  winning over an explicit `from` (a reconnecting `EventSource` knows better than its own URL —1661 *  the URL is the ORIGINAL join point, the header is where it actually got to). */1662export function streamRequestFromHttp(req: {1663  url: string;1664  headers: { get(name: string): string | null };1665}): { streamId: string; from: number } {1666  const url = new URL(req.url);1667  const streamId = url.searchParams.get("streamId") ?? "";1668  const lastEventId = req.headers.get("last-event-id");1669  const raw = lastEventId ?? url.searchParams.get("from") ?? "0";1670  const from = Number.parseInt(raw, 10);1671  return { streamId, from: Number.isFinite(from) && from > 0 ? from : 0 };1672}16731674/**1675 * Encode a subscription as an SSE body. Each positional frame carries `id: <seq>`, so a browser1676 * `EventSource` that drops the connection resumes at exactly the right offset with no application1677 * code — its own `Last-Event-ID` header is the `from` of the next subscribe ({@link1678 * streamRequestFromHttp}).1679 *1680 * The reader must close the `EventSource` on the `end` frame: `EventSource` reconnects on ANY close,1681 * including a clean one.1682 */1683export function streamFramesToSse(1684  sub: StreamSubscription,1685  opts?: { keepAliveMs?: number },1686): ReadableStream<Uint8Array> {1687  const encoder = new TextEncoder();1688  const keepAliveMs = opts?.keepAliveMs ?? 15_000;1689  let ping: ReturnType<typeof setTimeout> | null = null;1690  let cancelled = false;1691  let iter: AsyncIterator<StreamFrame>;1692  // PULL-based on purpose: a frame is taken from the subscription only when the consumer has1693  // demand, so a slow HTTP client backs frames up in the SUBSCRIBER's bounded queue — where1694  // `maxQueuedFrames` drops it with `stale` (§4) — instead of unboundedly in this stream's own.1695  return new ReadableStream<Uint8Array>({1696    start(controller) {1697      iter = sub.frames[Symbol.asyncIterator]();1698      const beat = (): void => {1699        if (cancelled) return;1700        // A comment line: keeps intermediaries from reaping an idle connection, costs 8 bytes.1701        controller.enqueue(encoder.encode(": ping\n\n"));1702        ping = timer(beat, keepAliveMs);1703      };1704      ping = timer(beat, keepAliveMs);1705    },1706    async pull(controller) {1707      const next = await iter.next();1708      if (cancelled) return;1709      if (next.done) {1710        if (ping) clearTimeout(ping);1711        sub.close();1712        controller.close();1713        return;1714      }1715      const id = frameResumePoint(next.value);1716      const head = id === undefined ? "" : `id: ${id}\n`;1717      controller.enqueue(encoder.encode(`${head}data: ${JSON.stringify(next.value)}\n\n`));1718    },1719    cancel() {1720      cancelled = true;1721      if (ping) clearTimeout(ping);1722      sub.close();1723    },1724  });1725}1726