API index and search · Build metadata

Source snapshot

packages/api-server/src/index.ts

1import { driveMutationAsync, insertCell, insertPlan, isGeneratorMutator, isoTx, toCell } from "@rindle/client";2import type {3  Ast,4  ColType,5  Condition,6  KeyedRow,7  MutationEnvelope,8  MutationOp,9  MutatorCtx,10  NamedQuery,11  Query,12  QueryResultRow,13  Schema,14  SharedMutator,15  SharedMutatorWithArgs,16  ServerWriteTx,17} from "@rindle/client";18import { DaemonHttpError, HttpRindleDaemonClient } from "@rindle/daemon-client";19import { createSqlClient, encodeSqlValue, RindleSqlError } from "@rindle/sql-client";20import type {21  ClientOptions as SqlClientOptions,22  MutationReceipt as SqlMutationReceipt,23  MutationRows as SqlMutationRows,24  SqlClient,25  SqlMutationTransaction,26  SqlSession,27  Statement as PublicSqlStatement,28} from "@rindle/sql-client";29import { compile as compileQueryAst } from "@rindle/query-compiler";30import type { Catalog, ColumnType as QueryColumnType, TableSchema } from "@rindle/query-compiler";31import type {32  ClaimRoomEpochInput,33  ClaimRoomEpochOutput,34  DematerializeInput,35  DematerializeOutput,36  MaterializationPolicy,37  MaterializeInput,38  MaterializeOutput,39  MigrateInput,40  MigrateOutput,41  MutationRejection,42  MutationRejectionOutput,43  MutationSessionBegin,44  MutationSessionBeginOutput,45  MutationSessionExec,46  MutationSessionQuery,47  MutationSessionRef,48  QueryOnceInput,49  QueryOnceOutput,50  RindleDaemonClient,51  RoomLmidsInput,52  RoomLmidsOutput,53  RowChangeTxn,54  RowChangeTxnOutput,55  SqlRead,56  SqlReadOutput,57  SqlStatement,58  SqlTxn,59  SqlTxnOutput,60  StreamMode,61  WireValue,62} from "@rindle/daemon-client";6364import {65  assertLabeledProfilesExist,66  assertUnwindowedFootprint,67  attachRealtimeLabel,68  compileRoomProfiles,69  compileRoomScopeSpecs,70  compileRoomTableSpecs,71  mintRoomDoc,72  queryRealtimeLabel,73  queryResultToAst,74  splitRoomDoc,75} from "./rooms.ts";76import type { RoomProfile, RoomScopeSpec, RoomTableSpec } from "./rooms.ts";77import {78  STREAM_SSE_HEADERS,79  StreamForbidden,80  StreamPlane,81  resolveStreamColumns,82  streamFramesToSse,83  streamRequestFromHttp,84} from "./streams.ts";85import type {86  OpenStreamInput,87  RindleStreamOptions,88  StreamHandle,89  StreamSubscription,90  StreamTables,91  SubscribeStreamInput,92} from "./streams.ts";93// The room lease token (RINDLE-REALTIME §10.1): minted here, verified by the room SHELL against94// its `downstream.tokenKeys` ring — the `/token` subpath is pure WebCrypto (no wasm, no shell).95// Loaded LAZILY at the first mint: `@rindle/room` is an OPTIONAL dependency (see package.json), so96// it is installed transitively — a consumer bundling api-server (Vite/Rollup/esbuild) can resolve97// this dynamic import even when it never uses rooms — yet the mint only runs when98// `realtime.roomTokenKey` is configured. Should the module be genuinely absent (an install that99// skipped the optional dep), the serve decision fail-opens: daemon-served leases plus a one-time100// warning naming the missing module. It is NOT a hard `dependency` because the entire code path is101// optional; optionalDependencies keeps a failed install of it non-fatal.102import type { mintRoomToken as MintRoomToken, scopeSpecsHash as ScopeSpecsHash } from "@rindle/room/token";103let roomTokenModule: { mintRoomToken: typeof MintRoomToken; scopeSpecsHash: typeof ScopeSpecsHash } | undefined;104async function loadRoomTokenModule(): Promise<{ mintRoomToken: typeof MintRoomToken; scopeSpecsHash: typeof ScopeSpecsHash }> {105  if (roomTokenModule === undefined) {106    const m = await import("@rindle/room/token");107    roomTokenModule = { mintRoomToken: m.mintRoomToken, scopeSpecsHash: m.scopeSpecsHash };108  }109  return roomTokenModule;110}111112// Re-export the shared (generator) mutator seam so an app builds its server mutators from ONE import:113// co-locate each body with its arg schema (`shared`), bulk-drive the registry ({@link sharedApiMutators}),114// keeping only server-only authority as explicit overrides (see MUTATORS-ISOMORPHIC).115export { isoTx, shared } from "@rindle/client";116export type { ArgSchema, IsoTx, MutationGen, MutatorCtx, SharedMutator, SharedMutatorWithArgs } from "@rindle/client";117118// The room-profile declaration layer (RINDLE-REALTIME-QUERY-ENABLEMENT §2, slice G-iv-a). The119// compiled-profile shapes stay internal to `./rooms.ts` — G-iv-b consumes them in-package.120// `RoomTableSpec` (G-iv-b) is public: it rides the lease wire (`QueryLeaseResponse.realtime`).121// `RoomScopeSpec` (H-iv-b) is public: it rides the boot wire (`RoomBootResponse.scopes`).122export { queryRealtimeLabel, queryResultToAst } from "./rooms.ts";123export type { RoomProfile, RoomScopeSpec, RoomTableSpec } from "./rooms.ts";124export type { RealtimeQueryLabel } from "@rindle/client";125126// The LM stream plane (designs-implemented/LM-STREAM-CHECKPOINT-DESIGN.md): a model response runs on two planes127// — every delta straight to subscribers, one chunk row per coarse checkpoint to the app's tables —128// joined by one monotone `seq`. The plane itself (`StreamPlane`) stays internal; the api-server owns129// it and feeds it the backend's outside-transaction SQL surface (checkpoints are SYSTEM writes: no130// clientID, no mid, no lmid).131export {132  STREAM_SSE_HEADERS,133  STREAM_STATUS_STREAMING,134  StreamOpenRefused,135  StreamRelayConform,136  assembleDurableText,137  frameResumePoint,138  spliceStreamText,139  streamChunkId,140  streamChunkTableDdl,141  streamFramesToSse,142  streamRequestFromHttp,143} from "./streams.ts";144export type {145  AuthorizeStreamInput,146  OpenStreamInput,147  RindleStreamOptions,148  StreamCheckpointPolicy,149  StreamCheckpointTarget,150  StreamColumns,151  StreamCommit,152  StreamCommitInput,153  StreamFrame,154  StreamHandle,155  StreamRelay,156  StreamRelayErrorInfo,157  StreamStatus,158  StreamSubscription,159  StreamTables,160  SubscribeStreamInput,161} from "./streams.ts";162163export const DEFAULT_RINDLE_API_ROUTES = {164  query: "/api/rindle/query",165  read: "/api/rindle/read",166  mutate: "/api/rindle/mutate",167  // The room write-authority host (RINDLE-REALTIME-DESIGN.md §5.3.1): the room's168  // SOLE flush counterpart — the store's own handler is private ingress behind these.169  applyRowChangeTxn: "/api/rindle/apply-row-change-txn",170  claimRoomEpoch: "/api/rindle/claim-room-epoch",171  roomLmids: "/api/rindle/room-lmids",172  // The DO shell's cold-boot callback (§10.1) — active only when `realtime` is configured.173  roomBoot: "/api/rindle/room-boot",174  // The LM stream subscribe leg (LM-STREAM-CHECKPOINT §4) — active only when `streams` is175  // configured. GET + `EventSource` is the intended shape (`Last-Event-ID` IS the resume offset).176  stream: "/api/rindle/stream",177} as const;178179export type MaybePromise<T> = T | PromiseLike<T>;180181/** Request context supplied by the application after authentication. The HTTP adapter or trusted182 *  caller owns `user`; do not populate it from an unverified request body. */183export interface ApiContext<User> {184  user: User;185  request?: unknown;186}187188export type ApiQueryResult = Ast | Query<any, any, any>;189export type ApiQuery<User, Args> = (ctx: ApiContext<User>, args: Args) => MaybePromise<ApiQueryResult>;190export type ApiQueries<User> = Record<string, ApiQuery<User, any>>;191192export interface RunQueryInput<User> {193  user: User;194  name: string;195  args: unknown;196  query: ApiQuery<User, any>;197  context: ApiContext<User>;198}199200export type RunQuery<User> = (input: RunQueryInput<User>) => MaybePromise<ApiQueryResult>;201202export interface AuthorizeQueryInput<User> {203  user: User;204  name: string;205  args: unknown;206  context: ApiContext<User>;207}208209export interface AuthorizeMutationInput<User> {210  user: User;211  envelope: MutationEnvelope;212  context: ApiContext<User>;213}214215/** An optional request gate. Return `false` or throw to deny; `true` and `undefined` allow.216 *  This gate does not add row predicates or replace access checks inside a mutator. */217export type Authorizer<T> = (input: T) => MaybePromise<boolean | void>;218219export interface MutationContext<User> {220  user: User;221  envelope: MutationEnvelope;222  daemon: RindleDaemonClient;223  request?: unknown;224}225226/** A deliberately narrow raw-SQL facade exposed by the API server. On {@link ServerMutationTx}227 *  it is bound to the open mutation transaction; on {@link MutationScope} each call runs in its228 *  own transaction outside the mutation boundary. Column aliases should be unique: positional229 *  driver rows are keyed by column name, so a duplicate alias is represented by its last value. */230export interface ServerSql {231  /** Queue/execute one statement. A transaction-bound call commits with the surrounding mutation. */232  execute(sql: string, params?: readonly WireValue[]): Promise<void>;233  /** Queue/execute an ordered statement batch. An empty batch is a no-op. On an234   *  outside-transaction surface ({@link MutationScope.sql}, `backend.outsideSql`) the batch MUST235   *  execute as ONE atomic transaction — all statements or none. Every built-in backend does (the236   *  daemon's `execute-sql-txn`, the sql-client's `/v1/sql/batch`, the Postgres plugger's237   *  BEGIN/COMMIT); a custom backend that loops statements without a transaction silently breaks238   *  the stream plane's chunk+CAS and compaction invariants, which ride single `batch` calls. */239  batch(statements: readonly SqlStatement[]): Promise<void>;240  /** Run a read and return rows keyed by their column names. */241  query<Row = Record<string, unknown>>(sql: string, params?: readonly WireValue[]): Promise<Row[]>;242}243244/** The raw-SQL escape hatch for relational/authority statements a keyed op can't express — an245 *  owner-gated cascade, a `NOT EXISTS` dedup. Prefer `tx.sql`; `exec` remains the synchronous246 *  compatibility shorthand for a queued `tx.sql.execute`, and `statements` is the raw write list. */247export interface SqlMutationTx {248  readonly sql: ServerSql;249  exec(sql: string, params?: WireValue[]): void;250  readonly statements: readonly SqlStatement[];251}252253/** The write handle a server mutator runs against — the ASYNC twin of the client's `MutationTx`. It254 *  is both the isomorphic {@link ServerWriteTx} logical surface (insert/update/upsert/insertIgnore/255 *  delete/row, rendered to dialect SQL) AND the legacy {@link SqlMutationTx} raw escape hatch. Both256 *  implementations run reads through the OPEN transaction (read-your-writes): Postgres executes257 *  everything live; the SQL-client and daemon adapters accumulate writes and lazily upgrade to an258 *  interactive mutation session at the first read (DAEMON-INTERACTIVE-TXN-DESIGN.md §5). */259export interface ServerMutationTx extends ServerWriteTx, SqlMutationTx {260  /** Run a full query (a fluent `Query` or its wire `Ast`) INSIDE the open transaction —261   *  read-your-writes, like {@link ServerWriteTx.row} but for arbitrary shapes. Returns the262   *  parsed nested result tree: an array for a plural root, an object or `null` for a `.one()`263   *  root, with cells in their raw SQLite storage-class representations (the same vocabulary264   *  `row` speaks). Remote SQLite backends: compiled by `@rindle/query-compiler`'s sqlite dialect265   *  (bind params, NO casts — §5.4) and executed through the mutation session. Postgres: lands with266   *  the read-compiler catalog integration (POSTGRES-READ-COMPILER-DESIGN.md Phase B). */267  query(q: Ast | Query<any, any, any>): Promise<unknown>;268}269270export type ApiMutatorResult = void | SqlStatement[] | SqlTxn;271/** A server mutator: a plain async function against the live {@link ServerMutationTx}. Two ways to272 *  write one (MUTATORS-ISOMORPHIC): drive it directly (the raw escape hatch — an owner-gated cascade,273 *  a `NOT EXISTS` dedup — plus a returned `SqlStatement[]`/`SqlTxn`), OR delegate to a SHARED generator274 *  (the SAME body the client predicts) via {@link runSharedMutation}, keeping only the server-only275 *  authority (arg parse, principal, policy) in the wrapper. */276export type ApiMutator<User, Args> = (277  tx: ServerMutationTx,278  args: Args,279  ctx: MutationContext<User>,280) => MaybePromise<ApiMutatorResult>;281282// --------------------------------------------------------------------------- scoped (outside-tx) mutators283//284// The tx-form {@link ApiMutator} above runs ENTIRELY inside the transaction. A SCOPED mutator285// (WORK-OUTSIDE-TX) instead controls the boundary itself: it receives a {@link MutationScope}, runs286// server-only code BEFORE opening the one atomic transaction (`scope.transact`), and MAY run code287// AFTER it commits. The outside-tx code is server-only by nature (the client's optimistic prediction288// can't call Stripe), so it lives HERE, never in the isomorphic body — the shared generator stays289// pure and identical on both tiers; server-computed values flow into it through `ctx`, exactly like290// `ctx.user` (undefined/predicted on the client, authoritative here).291292/** Thrown by {@link MutationScope.transact} when the transacted body BUSINESS-rejects: the data293 *  rolled back and `lmid` advanced alone (§2.4). Catch it to COMPENSATE an outside-tx side effect294 *  (refund the charge), then rethrow or return — the mutation's protocol outcome is already sealed295 *  as rejected, so a post-reject throw can't change it. A DB/infra failure is NOT this — it296 *  propagates as the raw driver error (the client retries; `lmid` did not advance). */297export class MutationRejected extends Error {298  readonly reason: string;299  constructor(reason: string) {300    super(reason);301    this.name = "MutationRejected";302    this.reason = reason;303  }304}305306/** The per-mutation server handle a {@link ScopedMutator} runs against. Code before {@link transact}307 *  runs OUTSIDE the transaction; code after a clean `transact` runs AFTER the commit. The308 *  `lmid`-always-advances invariant is the HARNESS's, not the author's: {@link RindleApiServer.pushMutation}309 *  seals the response from this handle's recorded outcome, so an early return, a never-called310 *  `transact`, or a swallowed {@link MutationRejected} still advances `lmid` and never wedges the311 *  client's pending queue. */312export interface MutationScope {313  /** Raw SQL OUTSIDE the mutation transaction. Every call commits independently and therefore may314   *  be observed even if {@link transact} later rejects or fails. Calls may also repeat when an315   *  envelope is retried, so outside writes need their own idempotency key/unique constraint. */316  readonly sql: ServerSql;317  /** Open the ONE atomic write transaction and drive `body` inside it, committing (stamping `lmid`318   *  co-transactionally) on a clean return. MAY be called at most once — a second call throws.319   *320   *  Two forms:321   *   - `transact(sharedMutator, args, ctx)` — drive a SHARED (generator) mutator (the same body the322   *     client predicts); pass the already-parsed `args` and the server `ctx` (fold server-only323   *     values like a charge id into `ctx` here).324   *   - `transact(run)` — a raw callback receiving the live {@link ServerMutationTx} (the escape325   *     hatch: `tx.exec`, logical writes, read-your-writes reads).326   *327   *  A THROW from the body that is not a {@link BackendError} is a BUSINESS rejection: the data rolls328   *  back, `lmid` advances alone, and this method throws {@link MutationRejected} (so surrounding329   *  code can compensate). A {@link BackendError} is INFRA: it propagates (the client retries).330   *331   *  **The callback form RETURNS ITS BODY'S VALUE**, and only on the committed path — a rejection332   *  throws, so there is never a value to act on for a transaction that rolled back. That is what333   *  lets a post-commit effect be decided by a TRANSACTIONAL read instead of a second, racy one334   *  afterwards:335   *336   *  ```ts337   *  const kick = await scope.transact(async (tx) => {338   *    const prior = await tx.row("message", { id: a.assistantMessageId });339   *    if (prior) return undefined;                 // a replayed envelope — do NOT re-fire the effect340   *    tx.insert("message", …);341   *    return { streamId: a.assistantMessageId, history: await readHistory(tx, a.chatId) };342   *  });343   *  if (kick) void startGeneration(kick);          // committed, and decided against committed state344   *  ```345   *346   *  A tx-form mutator cannot do this: its return value is already the logical-write channel347   *  ({@link ApiMutatorResult}). A post-commit effect chosen by a transactional read is exactly what348   *  the scoped form is for. */349  transact<T>(run: (tx: ServerMutationTx) => T | Promise<T>): Promise<T>;350  transact<A, C extends MutatorCtx>(mutator: SharedMutator<A, C>, args: A, ctx: C): Promise<void>;351}352353/** A SCOPED server mutator (WORK-OUTSIDE-TX): server-only code, ONE `scope.transact`, optional354 *  post-commit code. Register it by wrapping in {@link scoped} — the tag the api-server routes on to355 *  hand it a {@link MutationScope} instead of running its whole body inside the transaction. */356export type ScopedMutator<User, Args> = (357  scope: MutationScope,358  args: Args,359  ctx: MutationContext<User>,360) => void | Promise<void>;361362/** A {@link ScopedMutator} tagged by {@link scoped} so the harness invokes it with a363 *  {@link MutationScope}. Typed as a BRANDED tx-form {@link ApiMutator} purely so it registers in the364 *  `mutators` record without widening it to a union (which would break contextual inference for every365 *  plain tx-form entry). Its true runtime shape is `(scope, args, ctx)`; the tag — not the type —366 *  routes it, and it is never actually called as a tx-form mutator. */367export type ScopedApiMutator<User, Args> = ApiMutator<User, Args> & { readonly __rindleScoped: true };368369/** Mark a mutator as SCOPED so the api-server gives it a {@link MutationScope} (author-controlled tx370 *  boundary via `scope.transact`) rather than running its whole body inside the transaction. Register371 *  it alongside the tx-form mutators — it wins by key like any override:372 *373 *  ```ts374 *  mutators: defineApiMutators({375 *    ...sharedApiMutators(sharedMutators, sharedCtx),        // tx-form (common case)376 *    createOrder: scoped(async (scope, raw, ctx) => {        // needs outside-tx work377 *      const args = createOrder.args.parse(raw);378 *      const chargeId = await stripe.charge(args.amount, { idempotencyKey: ctx.envelope.mid }); // outside tx379 *      try {380 *        await scope.transact(createOrder, args, { ...sharedCtx(ctx), chargeId });              // inside tx381 *      } catch (e) {382 *        await stripe.refund(chargeId);                       // compensate — the write rejected383 *        throw e;384 *      }385 *      await sendReceipt(ctx.user);                           // after commit386 *    }),387 *  }),388 *  ```389 */390export function scoped<User, Args>(fn: ScopedMutator<User, Args>): ScopedApiMutator<User, Args> {391  // The brand carries the scoped runtime shape; typing the RETURN as the (branded) tx-form keeps it392  // assignable into `mutators` WITHOUT unioning that record — the harness routes on the brand and393  // never calls it as a tx-form mutator, so the cast is sound.394  return Object.assign(fn, { __rindleScoped: true as const }) as unknown as ScopedApiMutator<User, Args>;395}396397function isScoped<User>(m: ApiMutator<User, any>): m is ScopedApiMutator<User, any> {398  return (m as { __rindleScoped?: boolean }).__rindleScoped === true;399}400401export type ApiMutators<User> = Record<string, ApiMutator<User, any>>;402403export interface QueryLeaseRequest<User> {404  user: User;405  name: string;406  args: unknown;407  request?: unknown;408  /** The browser's stable `clientId` (sent on the query POST) — the anonymous routing-key fallback409   *  when there is no authenticated subject and no session cookie (READ-ROUTER-DESIGN.md §1.5/§2.2).410   *  A routing HINT only, never authorization. */411  clientId?: string;412  /** The browser's opaque follower-affinity ticket (FOLLOWER-AFFINITY-DESIGN.md §3), read off the413   *  query POST and forwarded opaquely on `materialize` so the fleet edge selects the follower414   *  the browser's ws is pinned to (§2, §4) — both legs co-locate. The api-server does NOT verify415   *  it (the fleet does); it holds no signing key. Absent ⇒ single daemon / affinity off. */416  affinity?: string;417}418419/**420 * The room-serve block on a query lease (RINDLE-REALTIME-QUERY-ENABLEMENT §2.1 step 5 / §2.4,421 * slice G-iv-b): present when the named query carries a realtime label naming a configured room422 * profile (302 §5 — declared, not derived; no coverage proof). The G-v client uses it to open the423 * room transport for THIS query beside — never instead of — its daemon session.424 *425 * It is a dedicated block on purpose: the ROOM ws is a SEPARATE connection this query opens beside426 * its daemon session (never a migration of the daemon session — the daemon ws host is fixed and427 * placed by the affinity ticket). A room-served lease's top-level fields are byte-identical to the428 * daemon-served ones.429 */430export interface QueryLeaseRealtime {431  /** The client store's gate/domain key for this room source (`connectSource`) AND the string the432   *  wasm engine's `parse_source_key` accepts: any string other than the reserved `"daemon"`433   *  parses as a room source, and the established convention is `"room:" + doc`434   *  (e.g. `room:document/doc:d1`). */435  sourceKey: string;436  /** Where the client opens the ROOM ws for this query (from `realtime.locateRoom`) — its OWN437   *  connection, distinct from the daemon session's fixed ws host. */438  wsEndpoint: string;439  /** The room's self-authorizing signed lease (`@rindle/room/token`): the APPROVED query AST +440   *  doc + subject, HMAC-signed with `realtime.roomTokenKey` so the room shell's441   *  `downstream.tokenKeys` ring verifies it. The room materializes on first presentation. */442  roomToken: string;443  /** Token expiry (ms epoch) — the client's renewal clock (renewal = a fresh lease). */444  exp: number;445  /** The wire room doc (`"<profile>/<key>"`, minted server-side — never client-derived). */446  doc: string;447  /** Per-footprint-table specs: the §2.2 owned/followed split + §3.2 routing metadata. Since448   *  H-iii each spec also carries `footprintWhere` — the same exact membership predicate the boot449   *  wire ships the room gate (one compiler, `compileRoomScopeSpecs`) — feeding the client's §3450   *  prove-or-slow-path router. Advisory routing metadata, never a credential (§3.2). */451  tables: RoomTableSpec[];452}453454/**455 * One minted SYSTEM-STREAM lease on a query lease's `lifecycle` block (RINDLE-REALTIME-QUERY-456 * ENABLEMENT §4, Slice I-iii): an ordinary daemon materialization over one of the four457 * `_rindle_*` lifecycle system tables (registered by the daemon's `enable_realtime_lifecycle` —458 * `rust/rindle-replica/src/mutations.rs`), attachable by the EXISTING client subscribe path459 * (present `leaseToken` on a `subscribe` frame, exactly like the primary lease). The identity460 * fields (`scope`/`doc`/`clientId`) document the minted AST's predicate — the client keys its461 * retains and its release-time row filters on them.462 */463export interface QueryLeaseLifecycleLease {464  /** Which system table this lease's subscription serves. */465  table: string;466  leaseToken: string;467  /** DOORBELL only: the §4.1 occupancy scope — the wire room doc (`"<profile>/<key>"`). */468  scope?: string;469  /** FENCE entries only: the room doc the predicate is scoped to. */470  doc?: string;471  /** FENCE ledger/outcome entries: present iff the predicate was ALSO client-scoped (the lease472   *  request carried `clientId`). Absent ⇒ doc-only predicate — the client filters to its own473   *  rows regardless (defense in depth). */474  clientId?: string;475}476477/** The §4 lifecycle block on a query lease (Slice I-iii): present only when BOTH the realtime478 *  `lifecycle` config is on AND the query is realtime-labeled. `doorbell` rides EVERY labeled479 *  lease (occupancy is counted whether or not the query is room-served — the 1→2 upgrade trigger480 *  needs solo watchers subscribed BEFORE any room exists, §4.1); `fence` rides only a ROOM-SERVED481 *  lease (the §4.2/§7.1/§3.3 downgrade surfaces are meaningful only where a room domain exists).482 *  The §4.2 fence VALUE (`finalFlushSeq`) is deliberately NOT here — it arrives with the I-v483 *  downgrade response; I-iii only stands up the streams. */484export interface QueryLeaseLifecycle {485  doorbell: QueryLeaseLifecycleLease;486  fence?: QueryLeaseLifecycleLease[];487}488489/** The §4.2 downgrade fence on a query lease (Slice I-v): rides a labeled reply whose §4.1490 *  occupancy gate CLOSED (so there is NO `realtime` block) when the server could drain the room.491 *  A SIBLING of `realtime`, never nested inside it — block-ABSENCE is the downgrade signal, and492 *  the fence rides alongside that absence. Its `finalFlushSeq` is the room's last COMMITTED flush493 *  seq; the client's frozen room ghost holds visible until the daemon plane has provably absorbed494 *  it (`_rindle_room_watermark(doc) ≥ finalFlushSeq`). Absent from every non-downgrade reply. */495export interface QueryLeaseRealtimeFence {496  /** The retiring room source's gate/domain key — `"room:" + doc`, matching what the room-served497   *  lease's {@link QueryLeaseRealtime.sourceKey} carried. */498  sourceKey: string;499  doc: string;500  finalFlushSeq: number;501}502503export interface QueryLeaseResponse {504  leaseToken: string;505  materializationId: string;506  queryKey?: string;507  reused?: boolean;508  /** The public daemon/fleet WebSocket endpoint. With the unified `rindle` connection this is509   *  derived from the same ingress URL (or `rindle.wsUrl`) so a browser can open its transport from510   *  this lease and needs no application-authored runtime-config route. */511  wsEndpoint?: string;512  /** Fresh opaque follower-placement ticket minted with this follower-local lease. The optimistic513   *  client offers it on the WebSocket opened at {@link wsEndpoint}, pinning both legs to the same514   *  follower even though the first connection is created only after this response. */515  affinity?: string;516  /** The room-serve block (G-iv-b) — see {@link QueryLeaseRealtime}. Absent ⇒ the lease is517   *  byte-identical to the legacy daemon-served shape. */518  realtime?: QueryLeaseRealtime;519  /** The §4.2 downgrade fence (Slice I-v) — see {@link QueryLeaseRealtimeFence}. Present only on a520   *  labeled reply whose occupancy gate closed AND `realtime.lifecycle.drainRoom` could drain the521   *  room; absent otherwise (including on every room-served reply). */522  realtimeFence?: QueryLeaseRealtimeFence;523  /** The §4 lifecycle system-stream block (Slice I-iii) — see {@link QueryLeaseLifecycle}.524   *  Minted ONLY under the opt-in `realtime.lifecycle` config; absent ⇒ byte-identical to the525   *  pre-lifecycle response. */526  lifecycle?: QueryLeaseLifecycle;527}528529/** A one-shot SSR read of a named query (SSR-DESIGN.md §6): same `(name, args)` surface as a lease,530 *  but the daemon serializes the current view ONCE and registers no subscriber. */531export interface QueryReadRequest<User> {532  user: User;533  name: string;534  args: unknown;535  request?: unknown;536  /** The browser's stable `clientId` — the anonymous routing-key fallback (see537   *  {@link QueryLeaseRequest.clientId}). Lets the SSR read co-locate on the follower the booting538   *  client's first subscribe will hit (READ-ROUTER-DESIGN.md §2.4). */539  clientId?: string;540  /** The browser's opaque follower-affinity ticket — see {@link QueryLeaseRequest.affinity}.541   *  Forwarded on the one-shot `query` so an SSR read lands on the same pinned follower. */542  affinity?: string;543}544545/** The assembled (nested-by-name) first-paint snapshot the server-side Store seeds + dehydrates546 *  (SSR-DESIGN.md §3.3). `rows` hydrate without an engine; `cvMin` is their watermark baseline. */547export interface QueryReadResponse {548  rows: Array<{ cols: Record<string, unknown>; [rel: string]: unknown }>;549  cvMin?: number;550  queryKey?: string;551}552553/** The context a {@link MutationBackend} needs to run one mutation inside its transaction. */554export interface MutationRunInput {555  envelope: MutationEnvelope;556  /** Schema-derived render metadata (from {@link RindleApiServerOptions.schema}). A logical op on a557   *  table absent here throws loudly — never a silent dropped write. `{}` when no schema is set. */558  render: RenderIndex;559  /** Invoke the (authorized) mutator against the backend-provided tx. A THROW that is NOT a560   *  {@link BackendError} is a BUSINESS rejection (roll the data back, then advance `lmid` alone,561   *  §2.4); a {@link BackendError} (a DB-layer failure) is INFRA and rejects the returned promise. */562  run(tx: ServerMutationTx): Promise<void>;563}564565/** The result of {@link MutationBackend.runMutation}: either the data+lmid committed together, or a566 *  business rejection whose data was rolled back but whose `lmid` still advanced (§2.4). */567export type MutationOutcome =568  | { accepted: true; output: SqlTxnOutput }569  | { accepted: false; reason: string; output?: unknown };570571/**572 * Where a mutation runs and who stamps `lmid` — the seam that makes the mutator authoring surface573 * backend-agnostic (`BYO-POSTGRES-LMID-CONTRACT-DESIGN.md` §6; MUTATORS-ISOMORPHIC plan). Three574 * implementations ship: {@link sqlBackend} is the preferred managed-SQL path; {@link daemonBackend}575 * keeps the private control-plane compatibility path; and {@link postgresBackend} runs a real576 * interactive PG transaction with confirmation riding the CDC loop down.577 *578 * The load-bearing invariant is that a mutation ALWAYS advances the client's `last_mutation_id`:579 * - `runMutation` runs the mutator inside the backend's transaction; on success it advances `lmid`580 *   to `envelope.mid` in the SAME atomic unit (OPTIMISTIC-WRITES-DESIGN.md §8.2); on a BUSINESS581 *   rejection it rolls the data back but STILL advances `lmid` (§2.4 — else the client's pending582 *   queue never drains and the optimistic stack wedges).583 * - `reject` is the pre-flight path (unknown mutator, failed authorization): NO data, `lmid` alone.584 * - An infrastructure failure THROWS from `runMutation` — never a user rejection (the client585 *   retries; mid dedup absorbs any applied prefix).586 */587export interface MutationBackend {588  /** The SQL dialect this backend renders logical ops to (drives placeholder style). */589  readonly dialect: SqlDialect;590  /** Optional raw-SQL surface outside the mutation transaction. Built-in backends provide it;591   *  custom backends may omit it, in which case `scope.sql` fails as an infrastructure error. */592  readonly outsideSql?: ServerSql;593  runMutation(input: MutationRunInput): Promise<MutationOutcome>;594  reject(input: { envelope: MutationEnvelope; reason: string }): Promise<unknown>;595}596597export interface PushMutationRequest<User> {598  user: User;599  envelope: MutationEnvelope;600  request?: unknown;601}602603export interface PushMutationsRequest<User> {604  user: User;605  envelopes: MutationEnvelope[];606  request?: unknown;607}608609export type PushMutationResponse =610  | { accepted: true; rejected: false; output: SqlTxnOutput }611  | { accepted: false; rejected: true; reason: string; output?: unknown };612613export interface RindleApiRoutes {614  query: string;615  read: string;616  mutate: string;617  applyRowChangeTxn: string;618  claimRoomEpoch: string;619  roomLmids: string;620  roomBoot: string;621  stream: string;622}623624/** A room-host reply the transport writes VERBATIM (`status` + JSON `body`): the625 *  daemon's fence/conflict/identity semantics ride specific statuses and body shapes626 *  the room's `httpAuthority` decodes, so this endpoint trio can't run through the627 *  throw-on-error result shapes the viewer endpoints use. */628export interface RoomHostResponse {629  status: number;630  body: unknown;631}632633/** A named query to keep permanently materialized (warm with zero subscribers). */634export interface PinnedQuery {635  name: string;636  args?: unknown;637}638639/** The explicit fleet pin fan-out seam (READ-ROUTER-DESIGN.md §4.2). The api-server resolves each640 *  pin's authoritative AST under `pinUser` and hands the ready {@link MaterializeInput}s here; the641 *  implementation (the read router) fans EACH across all live followers. Distinct, on purpose, from642 *  a per-viewer `materialize` (which routes ONE) — a pin-assert always sprays ALL. */643export interface PinFanout {644  assertPins(pins: readonly MaterializeInput[]): Promise<void>;645}646647// --------------------------------------------------------------------------- realtime (room) host648//649// Enabling Rindle Realtime for an app (RINDLE-REALTIME-ENABLEMENT-DESIGN.md §3.1) is ONE named650// opt-in: the `realtime` options block. Its presence activates the flush trio AND `/room-boot`;651// its absence keeps every room endpoint 403 and adds no route. The deprecated bare `authorizeRoom`652// still gates the trio alone (it never activates `/room-boot`).653654/** The room-boot flush leg (enablement §5): where the placed room's write-behind lands and what655 *  credential it presents. `urls` are the trio's ROUTE PATHS (root-relative — the shell resolves656 *  them against the boot call's origin), so an app that overrides `routes` needs no out-of-band657 *  sync; `headers` ride every flush call verbatim (`httpAuthority`'s `headers`). */658export interface RoomBootFlush {659  urls: { apply: string; claim: string; lmids: string };660  headers: Record<string, string>;661}662663/** The `/room-boot` response (RINDLE-REALTIME-DESIGN.md §10.1 — the DO shell's cold-boot664 *  callback): the claimed placement epoch, the room's upstream footprint lease, and the flush665 *  leg. A cold room boots inert and serves nothing until this returns. */666export interface RoomBootResponse {667  epoch: number;668  upstreamLeaseToken: string;669  /** Where the room opens its upstream subscription (a routed deploy's follower). Absent ⇒ the670   *  shell's statically configured rindled ws endpoint. */671  upstreamWsEndpoint?: string;672  /** Fresh opaque follower-placement ticket minted alongside `upstreamLeaseToken`. The DO offers673   *  it with `rindle.v1` on the separate upstream ws so a static fleet endpoint replays to the674   *  exact follower holding that local lease. Absent when daemon affinity is off. */675  upstreamAffinity?: string;676  /** Per-footprint-table scope specs (H-iv-b), compiled from the resolved footprint AST + the677   *  profile's context set (the legacy anonymous profile compiles with an empty context set):678   *  what the shell hands the wasm room's `enableWritesV2` — the §3.3 commit gate. Optional679   *  only for wire compatibility with pre-H-iv-b servers; a shell that doesn't receive them680   *  enables the v1 table-granular write plane exactly as before. */681  scopes?: RoomScopeSpec[];682  flush: RoomBootFlush;683}684685export interface RindleRealtimeOptions<User> {686  /** The boot shell secret (§10.1 — a host binding, never client-derived). Authenticates the687   *  room's `Authorization: Bearer` on `/room-boot` (the default {@link authorizeBoot}) and keys688   *  the DEFAULT epoch-bound flush credential. */689  shellSecret: string;690  /** NAMED room profiles (RINDLE-REALTIME-QUERY-ENABLEMENT-DESIGN.md §2.1): profile name → key691   *  derivation + canonical unwindowed footprint + read-only context tables. The wire room key692   *  for a named profile is `"<profile>/<key>"`; `/room-boot` splits it and resolves the693   *  profile's footprint with the bare key. Compiled + validated LOUDLY at construction (§2.3):694   *  a windowed footprint, a context table missing from the schema/footprint, or a registered695   *  query whose realtime label names a missing profile all throw from `createRindleApiServer`. */696  rooms?: Record<string, RoomProfile<User>>;697  /** doc → the room's approved upstream footprint (§3.1) — an `Ast` or fluent `Query`. MAY698   *  delegate to the named-query registry internally; throw `RindleApiError("not-found", …, 404)`699   *  for a doc that shouldn't exist. The §9 footprint budget belongs here — it runs once per700   *  placement, at lease mint.701   *  @deprecated Prefer named {@link rooms} profiles (RINDLE-REALTIME-QUERY-ENABLEMENT §2.1).702   *  This bare form remains as the single-profile LEGACY alias — the anonymous/default profile:703   *  a doc with no known `"<profile>/"` prefix resolves here, byte-identically to before named704   *  profiles existed (and with none of their construction/boot-time validation). */705  resolveFootprint?: (doc: string, ctx: ApiContext<User>) => MaybePromise<ApiQueryResult>;706  /** Gates the flush trio — `authorizeRoom`, relocated. Default: verify the default flush707   *  credential from the {@link ROOM_FLUSH_CREDENTIAL_HEADER} request header — which requires the708   *  transport to pass its incoming request as `context.request` (Fetch `Request` and node709   *  `IncomingMessage` shapes are both understood). */710  authorize?: Authorizer<ApiContext<User>>;711  /** Gates `/room-boot`. Default: constant-time `Authorization: Bearer` check against712   *  {@link shellSecret} (same `context.request` requirement as {@link authorize}). */713  authorizeBoot?: Authorizer<ApiContext<User>>;714  /** Mint the flush headers a placed room presents on every flush call. Default:715   *  {@link mintRoomFlushCredential} under {@link ROOM_FLUSH_CREDENTIAL_HEADER}. Override this716   *  and {@link authorize} TOGETHER — they are the two ends of one credential. */717  mintFlushHeaders?: (input: { doc: string; epoch: number }) => MaybePromise<Record<string, string>>;718  /** Lease TTL for the room's upstream footprint materialization (defaults to the server-wide719   *  `leaseTtlMs`, else the daemon's default). */720  upstreamLeaseTtlMs?: number;721  /** Static endpoint where rooms open their upstream subscription. In a follower fleet this is the722   *  fleet ws URL; `/room-boot` pairs it with the materialization's fresh placement ticket so the723   *  room lands on the exact follower holding its lease. Absent ⇒ no explicit upstream (the Node724   *  room shell may use its own default; the shipped DO shell requires this endpoint). */725  upstreamWsEndpoint?: string;726  /** Locate (or place) the room serving `doc` and return the ROOM ws endpoint a room-served727   *  lease's client should open (G-iv-b; on the DO shell this is the Worker's room URL). The728   *  endpoint rides the lease's dedicated `realtime.wsEndpoint` — its OWN connection, distinct from729   *  the daemon session's fixed ws host. Absent ⇒ room-serving is OFF: labeled queries serve from730   *  the daemon exactly as today (fail-open). */731  locateRoom?: (doc: string) => MaybePromise<{ wsEndpoint: string }>;732  /** The room lease token signing key (`@rindle/room/token`): `kid` + secret, matching an entry733   *  in the room shell's `downstream.tokenKeys` ring. Required for room-serving (without it a734   *  labeled query fail-opens to the daemon with a one-time warning). A separate secret from735   *  `shellSecret` on purpose — the shell's ring is the client-token trust domain, the shell736   *  secret is the boot/flush trust domain. */737  roomTokenKey?: { kid: string; secret: string };738  /** Room lease token TTL, ms (default 5 minutes — the §4.1 short-TTL backstop; renewal is a739   *  fresh lease through this server, never an extension). */740  roomTokenTtlMs?: number;741  /** Loud-diagnostics sink for the realtime layer (profile compilation warnings + the one-time742   *  per-(query, profile) "not room-served" serve-decision warnings). Defaults to743   *  `console.warn`; injectable for tests. */744  warn?: (message: string) => void;745  /** The §4 upgrade/downgrade lifecycle plane (RINDLE-REALTIME-QUERY-ENABLEMENT §4, Slice746   *  I-iii): PRESENCE of this block is the opt-in — every realtime-labeled lease then747   *  additionally mints the doorbell system lease (occupancy, §4.1) and every ROOM-SERVED lease748   *  the fence bundle (watermark + ledger + outcomes, §4.2/§7.1/§3.3) — see749   *  {@link QueryLeaseLifecycle}. Requires the daemon to have run `enable_realtime_lifecycle`750   *  (the four `_rindle_*` system tables must be registered or the minted materializations fail751   *  — which fail-opens with a one-time warning, never blocking the lease). Absent ⇒ the lease752   *  response is byte-identical to pre-lifecycle. */753  lifecycle?: RindleRealtimeLifecycleOptions;754}755756/** {@link RindleRealtimeOptions.lifecycle}. PRESENCE of the block is the opt-in switch (I-iii);757 *  the fields below are the Slice I-iv occupancy knobs (§4.1, decisions D4/D6/D7). All optional —758 *  `lifecycle: {}` gets the designed defaults. */759export interface RindleRealtimeLifecycleOptions {760  /** D6 (§4.1): the occupancy threshold for room-serving. A labeled lease whose scope counts761   *  FEWER than this many distinct unexpired sessions (the caller's own included) ships WITHOUT762   *  the realtime block — served from the daemon, indistinguishable from an uncovered query —763   *  but WITH the doorbell, so the 1→2 transition wakes it (that is the point: solo docs never764   *  cost room infrastructure). Default **2** (the design's 1→2 trigger). Set `1` to room-serve765   *  solo viewers (the pre-I-iv behavior under lifecycle config). */766  minSessions?: number;767  /** The §9.1 hysteresis window, ms (default **120_000**). Two consumers: (a) the lazy sweep768   *  (D4) keeps expired session rows lingering at least this long past expiry — Slice I-v's769   *  downgrade decision ("no other unexpired row AND the newest other row expired > graceMs770   *  ago") is read FROM those rows, so they must survive to be read; (b) this slice's gate771   *  applies the same hysteresis upward: a scope with an other-session row expired ≤ graceMs772   *  ago keeps room-serving through the window (see `lifecycleOccupancy` — no flap on one773   *  client's brief lapse). §9.1-tunable: raise it for docs where collaborators churn slowly. */774  graceMs?: number;775  /** TTL of an occupancy session row, ms — `expires_at = now + sessionTtlMs` on every labeled776   *  lease mint/renewal (D7: session identity = the request's `clientId`; two tabs are two777   *  sessions iff their clientIds differ). Default = the server's `leaseTtlMs`, else 5 minutes —778   *  matching the room-token renewal cadence (`roomTokenTtlMs`, renewed 30s early), so a779   *  room-attached client's renewals keep its row unexpired; a daemon-attached solo client's row780   *  MAY lapse (it has no renewal timer) and is refreshed by its next doorbell-triggered781   *  re-lease — occupancy converges through the doorbell itself. */782  sessionTtlMs?: number;783  /** The §4.2 downgrade drain hook (Slice I-v). When the occupancy gate CLOSES for a labeled784   *  lease whose scope PLAUSIBLY hosted a room (an other-session row still lingers — never a785   *  never-shared solo doc), the api-server calls this to drain the room's pending write-behind786   *  and learn its last COMMITTED `flush_seq`, then rides the value back on the lease as the787   *  {@link QueryLeaseRealtimeFence}. The deployment wires it to the room shell's / DO's `/drain`788   *  control. Absent ⇒ no fence is attached (the client hits its loud legacy downgrade path);789   *  a throw fails OPEN to the same (a downgrade never blocks the lease). Concurrent drains across790   *  api-server instances are fine — `/drain` is idempotent. */791  drainRoom?: (doc: string) => Promise<{ finalFlushSeq: number }>;792}793794// The DEFAULT flush credential: `rfc1.<b64url payload>.<b64url hmac-sha256>`, payload795// `{v:1, doc, epoch, iat}`. Deliberately EPOCH-bound, not time-bound: the credential's lifecycle796// IS the placement fence (§8.3 — a superseded epoch's flushes 409 at the store no matter what797// credential they carry), and platform revocation is registry suspension, so an `exp` would only798// force spurious re-boots of long-lived rooms. HMAC via WebCrypto (`crypto.subtle`) so the exact799// same code runs in Node and a Cloudflare Worker — no `node:crypto` import (matching800// `@rindle/room`'s token module).801802export const ROOM_FLUSH_CREDENTIAL_HEADER = "x-rindle-room-credential";803804const FLUSH_CREDENTIAL_PREFIX = "rfc1";805806export interface RoomFlushCredentialPayload {807  v: 1;808  doc: string;809  epoch: number;810  iat: number;811}812813function b64url(bytes: Uint8Array): string {814  let bin = "";815  for (const b of bytes) bin += String.fromCharCode(b);816  return btoa(bin).replace(/\+/g, "-").replace(/\//g, "_").replace(/=+$/, "");817}818819// Return type inferred (`Uint8Array<ArrayBuffer>` under TS ≥5.7 libs) — an explicit820// `Uint8Array` annotation widens to `ArrayBufferLike` and fails `crypto.subtle`'s821// `BufferSource` under consumers compiling this source with newer lib types.822function unb64url(s: string) {823  const bin = atob(s.replace(/-/g, "+").replace(/_/g, "/"));824  const out = new Uint8Array(bin.length);825  for (let i = 0; i < bin.length; i++) out[i] = bin.charCodeAt(i);826  return out;827}828829async function flushHmacKey(secret: string, usage: "sign" | "verify") {830  return crypto.subtle.importKey(831    "raw",832    new TextEncoder().encode(secret),833    { name: "HMAC", hash: "SHA-256" },834    false,835    [usage],836  );837}838839/** Sign the default epoch-bound flush credential (`/room-boot` mints one per placement). */840export async function mintRoomFlushCredential(opts: {841  shellSecret: string;842  doc: string;843  epoch: number;844  /** Mint time; defaults to `Date.now()`. Injectable for tests. */845  now?: number;846}): Promise<string> {847  const payload: RoomFlushCredentialPayload = {848    v: 1,849    doc: opts.doc,850    epoch: opts.epoch,851    iat: opts.now ?? Date.now(),852  };853  const body = b64url(new TextEncoder().encode(JSON.stringify(payload)));854  const signed = `${FLUSH_CREDENTIAL_PREFIX}.${body}`;855  const key = await flushHmacKey(opts.shellSecret, "sign");856  const sig = new Uint8Array(857    await crypto.subtle.sign("HMAC", key, new TextEncoder().encode(signed)),858  );859  return `${signed}.${b64url(sig)}`;860}861862/** Verify a flush credential's MAC, then its claims; returns the payload or throws. The MAC is863 *  checked FIRST — no claim is trusted before it passes. */864export async function verifyRoomFlushCredential(865  credential: string,866  shellSecret: string,867): Promise<RoomFlushCredentialPayload> {868  const parts = credential.split(".");869  if (parts.length !== 3 || parts[0] !== FLUSH_CREDENTIAL_PREFIX) {870    throw new Error("not a room flush credential");871  }872  const [, body, sig] = parts;873  const key = await flushHmacKey(shellSecret, "verify");874  const ok = await crypto.subtle.verify(875    "HMAC",876    key,877    unb64url(sig),878    new TextEncoder().encode(`${FLUSH_CREDENTIAL_PREFIX}.${body}`),879  );880  if (!ok) throw new Error("bad signature");881  let payload: RoomFlushCredentialPayload;882  try {883    payload = JSON.parse(new TextDecoder().decode(unb64url(body))) as RoomFlushCredentialPayload;884  } catch {885    throw new Error("malformed payload");886  }887  if (payload.v !== 1) throw new Error("unknown version");888  if (typeof payload.doc !== "string" || payload.doc.length === 0) {889    throw new Error("missing doc");890  }891  if (typeof payload.epoch !== "number") throw new Error("missing epoch");892  return payload;893}894895/** Database connection used by the API server's managed SQL path.896 *897 *  `intMode` defaults to `"number"` because logical Rindle rows use the JSON-safe {@link WireValue}898 *  vocabulary — `"bigint"` does not survive `JSON.stringify`, and `"string"` silently retypes every899 *  integer, breaking arithmetic in mutator bodies. The cost is a HARD BOUND: a mutator read of an900 *  integer outside ±(2^53 − 1) rejects that mutation rather than silently rounding it. Tables with901 *  keys beyond that range (snowflake ids, and so on) must override `intMode` and have their mutators902 *  handle the resulting type. Commit receipts are unaffected — they never decode row values. */903export type RindleDatabaseOptions = Pick<SqlClientOptions, "url" | "authToken" | "fetch" | "intMode">;904905/** The unified fleet connection — one URL, one key. A Rindle is BOTH layers at once: the SQL906 *  database below and the sync/IVM control plane above, served by a single ingress that routes907 *  each request to the right tier. This option derives both from that one origin: the SQL layer908 *  ({@link RindleApiServerOptions.database}) at `url` with `token`, and the control-plane client909 *  ({@link RindleApiServerOptions.daemon}) at the same `url` — so an application configures "a910 *  rindle", not two subsystems.911 *912 *  The derived control-plane client sends the same bearer. A unified ingress is an explicitly913 *  merged customer API-server trust tier: it routes SQL to the master and control requests to a914 *  follower, while the browser still receives neither credential. Deployments that keep those915 *  trust tiers separate pass an explicit {@link RindleApiServerOptions.daemon}; explicit fields916 *  always win over this derivation. */917export interface RindleConnectionOptions {918  /** The single ingress origin. Defaults to `$RINDLE_URL` — exported by `rindle dev` whenever919   *  the rendered topology collapsed read+write onto one ingress. */920  url?: string;921  /** Public subscription WebSocket endpoint returned on query leases. Defaults to {@link url} with922   *  `http:` → `ws:` / `https:` → `wss:`. Override when HTTP and WebSocket ingress differ. */923  wsUrl?: string;924  /** The server-side bearer for both legs of the unified ingress. Defaults to925   *  `$RINDLE_DATABASE_TOKEN` (also a `rindle dev` export); required through one of those channels926   *  unless both legs are configured explicitly. It never reaches the browser. */927  token?: string;928}929930export interface RindleApiServerOptions<User> {931  /** The sync/IVM control-plane client (leases, materialization, rooms). Optional once932   *  {@link rindle} is configured — the api-server then derives an HTTP client against the single933   *  ingress. Pass one explicitly for a tokened production fleet, a split/routed deployment, or a934   *  custom transport; an explicit client wins over the derivation. */935  daemon?: RindleDaemonClient;936  /** The unified connection (one URL, one key) that derives {@link daemon} and {@link database}937   *  from the fleet's single ingress. `rindle: {}` resolves both halves from the `rindle dev`938   *  environment. Any explicitly configured `daemon` / `database` / `sql` / `backend` field takes939   *  precedence over its derived counterpart. */940  rindle?: RindleConnectionOptions;941  /** Preferred managed setup. The API server constructs and owns its SQL client; authoritative942   *  mutators, `tx.sql`, and `scope.sql` use it, while `daemon` remains only the lease/query/943   *  materialization/room control plane. Mutually exclusive with {@link sql} unless `backend`944   *  explicitly replaces both. */945  database?: RindleDatabaseOptions;946  /** Advanced injection/test seam for an already-created SQL session. Most applications should947   *  configure {@link database} and never import `createSqlClient`. When present (and `backend` is948   *  absent), authoritative mutators run through {@link sqlBackend}. */949  sql?: SqlSession;950  /** Where mutations are applied and `lmid` is stamped ({@link MutationBackend}). Default:951   *  managed `sqlBackend` when `database` or `sql` is configured, otherwise the compatibility952   *  `daemonBackend(daemon)`. Pass `postgresBackend(...)` when Postgres is the source of truth. */953  backend?: MutationBackend;954  /** The typed schema (`createSchema`/`refineSchema`). Required only when a mutator uses the LOGICAL955   *  write vocabulary (`tx.insert`/`update`/`upsert`/`insertIgnore`/`delete`/`row`) — it drives the956   *  dialect SQL renderer (column order, pk, quoting). A logical op with no schema configured throws957   *  loudly. Pure raw-`tx.exec` mutators do not need it. */958  schema?: Schema;959  queries?: ApiQueries<User>;960  runQuery?: RunQuery<User>;961  mutators?: ApiMutators<User>;962  /** Optional gate before resolving a named query, for leases and one-shot reads. If omitted,963   *  no gate runs. The query definition still owns row visibility predicates. */964  authorizeQuery?: Authorizer<AuthorizeQueryInput<User>>;965  /** Optional gate before mutation execution. If omitted, no gate runs. Mutators still own966   *  row-level checks and business rules inside their authoritative transaction. */967  authorizeMutation?: Authorizer<AuthorizeMutationInput<User>>;968  /** Rindle Realtime (RINDLE-REALTIME-ENABLEMENT-DESIGN.md §3.1): the ONE named opt-in.969   *  Presence activates the room flush trio AND `/room-boot`; absence keeps every room970   *  endpoint 403. Takes precedence over the deprecated {@link authorizeRoom}. */971  realtime?: RindleRealtimeOptions<User>;972  /** The LM stream plane (LM-STREAM-CHECKPOINT-DESIGN.md): the ONE named opt-in for streaming a973   *  model response live while the durable store only ever sees coarse checkpoints. Presence974   *  activates {@link RindleApiServer.openStream}/{@link RindleApiServer.subscribeStream} and975   *  `/stream`; absence keeps them 403. */976  streams?: RindleStreamOptions<User>;977  /** The room write-authority gate (§5.3.1): validates the caller is a placed room —978   *  the epoch-bound flush credential rides `context.request`, and what it means is979   *  the app's to define. The room endpoints are DISABLED (403) until this is set:980   *  hosting a write authority is an explicit opt-in, never a default.981   *  @deprecated Use {@link realtime} (its `authorize`) — this bare form gates the982   *  flush trio but never activates `/room-boot`. When both are set, `realtime` wins. */983  authorizeRoom?: Authorizer<ApiContext<User>>;984  routes?: Partial<RindleApiRoutes>;985  mode?: StreamMode;986  materializationPolicy?:987    | MaterializationPolicy988    | ((input: QueryLeaseRequest<User>) => MaybePromise<MaterializationPolicy>);989  leaseTtlMs?: number;990  /** Idle TTL (ms) the warm pipeline a one-shot SSR {@link RindleApiServer.readQuery read} leaves991   *  behind is held at (SSR-DESIGN.md §3.4) — it must comfortably cover page-load + client-boot +992   *  the follow-up live `subscribe` so the browser lands on a still-warm pipeline (the warm993   *  handoff). The TTL is NOT part of the dedup key (max-wins), so it only ever extends a shared994   *  query's window. Absent ⇒ the daemon's default idle TTL. */995  readIdleTtlMs?: number;996  subject?: string | ((input: QueryLeaseRequest<User>) => MaybePromise<string | undefined>);997  /** The ANONYMOUS routing key forwarded to the read router (READ-ROUTER-DESIGN.md §2.2) — used by998   *  HRW placement when there is no authenticated `subject`. The router keys on `subject ?? this`;999   *  the resolved value rides `metadata.routingKey` to the daemon. Default: the browser-supplied1000   *  `clientId` (from the query POST body). NOTE: the default cannot co-locate an ANONYMOUS SSR read1001   *  with the booting client — an SSR server can't see the browser's localStorage `clientId`, so the1002   *  two legs compute different keys and §2.4's warm handoff misses (still correct — just an extra1003   *  first-touch materialize). For anonymous SSR co-location, set this to read a server-set session1004   *  cookie from `input.request` (it rides the SSR request AND every browser request). A routing1005   *  HINT only — never authorization. Ignored by a single (unrouted) daemon, which has nothing to1006   *  route. */1007  routingKey?: string | ((input: QueryLeaseRequest<User>) => MaybePromise<string | undefined>);1008  /** The EXPLICIT fleet pin fan-out — when set, {@link RindleApiServer.assertPins} fans each1009   *  resolved pin across ALL live followers through it (a fleet control action over the machine1010   *  list — FOLLOWER-AFFINITY-DESIGN.md §11) instead of materializing each pin once on the (single)1011   *  daemon. A per-viewer `materialize` always routes ONE; a pin-assert always fans ALL — never1012   *  inferred from `policy.kind`. Absent ⇒ single-daemon behavior (one materialize per pin). */1013  pinFanout?: PinFanout;1014  /** Named queries to keep permanently materialized via {@link RindleApiServer.assertPins}.1015   *  Each is materialized with a `pinned` policy (survives zero subscribers) so late joiners1016   *  attach to an already-warm result. Pins are viewer-independent — resolved with `pinUser`. */1017  pinnedQueries?: PinnedQuery[];1018  /** The user context pins resolve under (pins are shared, so they should not depend on a1019   *  per-viewer identity). Defaults to `undefined`. */1020  pinUser?: User;1021  /** Surfaced when a SCOPED mutator ({@link scoped}) throws from code that runs AFTER `scope.transact`1022   *  has already sealed the protocol outcome — a post-commit effect, or a compensation handler running1023   *  after a business rejection. The outcome is fixed (this callback CANNOT change the client's1024   *  response or the `lmid` advance), but the throw must not vanish: a failed refund is real money.1025   *  Absent ⇒ the error is logged to `console.error`. */1026  onScopeError?: (err: unknown, info: { phase: "committed" | "rejected"; envelope: MutationEnvelope }) => void;1027}10281029export interface RindleApiServer<User> {1030  readonly routes: RindleApiRoutes;1031  /** Close the SQL client created from {@link RindleApiServerOptions.database}, and drop every live1032   *  stream's readers and timers WITHOUT a durable write (that is {@link drainStreams}). Injected1033   *  SQL sessions and custom backends remain caller-owned. Idempotent. */1034  close(): void;1035  /** Open an LM stream (LM-STREAM-CHECKPOINT §2): commits the durable POINTER row, then hands back1036   *  the producer handle. A resolved handle means the message already exists for every client's1037   *  query — so a subscriber that arrives before the first token has something to attach to.1038   *  Throws 403 unless {@link RindleApiServerOptions.streams} is configured. */1039  openStream(input: OpenStreamInput<User>): Promise<StreamHandle>;1040  /** Attach a reader at `from` — the same call serves a first-touch subscriber (`from: 0`), a late1041   *  joiner (`from` = the seq its IVM view shows), and a reconnect (`from` = `Last-Event-ID`).1042   *  Terminates with `end`, or with `stale`/`absent` when the client should fall back to the1043   *  durable plane (both are ordinary answers, never errors). */1044  subscribeStream(input: SubscribeStreamInput<User>): Promise<StreamSubscription>;1045  /** Parse a default `{streamId, from?}` subscribe body and run {@link subscribeStream}. For the1046   *  GET + `EventSource` shape, use {@link streamResponse} (or build the body with1047   *  `streamRequestFromHttp(request)` yourself). */1048  handleStreamJson(body: unknown, context: ApiContext<User>): Promise<StreamSubscription>;1049  /** The subscribe route in ONE call: parse a GET (`?streamId=…&from=…`, with `Last-Event-ID`1050   *  winning), authorize + subscribe, and encode the SSE response. A refusal comes back as a JSON1051   *  error `Response` (403 for denied or unconfigured) rather than a throw, so the route body is a1052   *  single expression after authentication. For custom transports, compose1053   *  `streamRequestFromHttp` + {@link subscribeStream} + `streamFramesToSse` instead. */1054  streamResponse(1055    request: { url: string; headers: { get(name: string): string | null } },1056    context: ApiContext<User> & { keepAliveMs?: number },1057  ): Promise<Response>;1058  /** Checkpoint every live stream's outstanding tail, then seal it `interrupted`1059   *  (LM-STREAM-CHECKPOINT §5). Wire it to SIGTERM: without it a rolling deploy drops each1060   *  response's un-checkpointed tail and strands rows saying `streaming` forever. */1061  drainStreams(): Promise<void>;1062  createQueryLease(input: QueryLeaseRequest<User>): Promise<QueryLeaseResponse>;1063  /** (Re-)materialize every `pinnedQueries` entry with a pinned policy. Idempotent — the daemon1064   *  dedupes by canonical query, so a re-assert reuses the existing materialization. Call it at1065   *  startup and whenever the daemon restarts (e.g. from the daemon-client `onBootId` hook), since1066   *  the daemon holds no durable materialization state. No-op when `pinnedQueries` is empty. */1067  assertPins(): Promise<void>;1068  pushMutation(input: PushMutationRequest<User>): Promise<PushMutationResponse>;1069  /** Apply an in-order batch (the client mutation queue's flush). Envelopes run strictly1070   *  sequentially; a rejection still advances the daemon's lmid, so later envelopes in the1071   *  batch stay contiguous and keep applying. A daemon ERROR throws for the whole batch —1072   *  the client retries it and the daemon's mid dedup absorbs the already-applied prefix. */1073  pushMutations(input: PushMutationsRequest<User>): Promise<PushMutationResponse[]>;1074  /** One-shot SSR read (SSR-DESIGN.md §6): resolve `(name, args)` → AST (same authority path as a1075   *  lease, `authorizeQuery` enforced), have the daemon serialize the current view once, and return1076   *  the assembled rows for the loader to seed + dehydrate. Registers NO subscriber — a dropped1077   *  render leaks nothing; the pipeline self-reclaims after the idle TTL ({@link1078   *  RindleApiServerOptions.readIdleTtlMs}) unless the browser's follow-up `subscribe` lands first. */1079  readQuery(input: QueryReadRequest<User>): Promise<QueryReadResponse>;1080  handleQueryJson(body: unknown, context: ApiContext<User>): Promise<QueryLeaseResponse>;1081  /** Parse a default `{name, args}` read body and run {@link readQuery}. */1082  handleReadJson(body: unknown, context: ApiContext<User>): Promise<QueryReadResponse>;1083  /** Accepts `{envelope}` (one) or `{envelopes: [...]}` (an in-order batch → array reply). */1084  handleMutateJson(1085    body: unknown,1086    context: ApiContext<User>,1087  ): Promise<PushMutationResponse | PushMutationResponse[]>;1088  /** The room's flush (§5.3.1): gate on `authorizeRoom`, forward the txn to the write1089   *  authority, and pass the store's verdict through VERBATIM — `200 {applied, cv}`,1090   *  `409 {error:"fenced"|"conflict", …}`, or the loud identity `500`. Write1091   *  `status` + `body` as-is; the room's `httpAuthority` decodes them. */1092  handleApplyRowChangeTxnJson(1093    body: unknown,1094    context: ApiContext<User>,1095  ): Promise<RoomHostResponse>;1096  /** Claim the next placement epoch for a doc (§2.5), same gate + envelope. */1097  handleClaimRoomEpochJson(body: unknown, context: ApiContext<User>): Promise<RoomHostResponse>;1098  /** The room's boot probe (§3.3), same gate + envelope. */1099  handleRoomLmidsJson(body: unknown, context: ApiContext<User>): Promise<RoomHostResponse>;1100  /** The DO shell's cold-boot callback (§10.1; enablement §3.1): authenticate the shell1101   *  secret, resolve the doc's footprint, claim the placement epoch, mint the upstream1102   *  lease and the flush leg. 403 until {@link RindleApiServerOptions.realtime} is set. */1103  handleRoomBootJson(body: unknown, context: ApiContext<User>): Promise<RoomHostResponse>;1104}11051106export type RindleApiErrorCode = "bad-request" | "forbidden" | "not-found" | "rejected";11071108export class RindleApiError extends Error {1109  readonly code: RindleApiErrorCode;1110  readonly status: number;11111112  constructor(code: RindleApiErrorCode, message: string, status: number) {1113    super(message);1114    this.name = "RindleApiError";1115    this.code = code;1116    this.status = status;1117  }1118}11191120/**1121 * A read/write-split `RindleDaemonClient` (READ-ROUTER-DESIGN.md §2.1). Writes1122 * (`executeSqlTxn` / `rejectMutation` / `applyRowChangeTxn` / `migrate`) go to the single1123 * write-master, UNCHANGED and never through the router; reads (`materialize` / `query` /1124 * `dematerialize`, and raw `executeSqlRead`) go to the read router. Raw reads default to a replica1125 * (`consistency:"eventual"`) so they scale off the write-master; pass `consistency:"strong"` on a1126 * read to route it to the master for read-your-writes after a write to the same data. Hand one of1127 * these to {@link createRindleApiServer} as `daemon` to point the reads leg at the router while1128 * writes stay on the master — no placement logic enters the api-server.1129 *1130 * ```ts1131 * const daemon = new SplitDaemonClient(1132 *   new HttpRindleDaemonClient({ baseUrl: MASTER_URL, headers: writeAuth }),  // writes → master1133 *   new HttpRindleDaemonClient({ baseUrl: ROUTER_URL, headers: routerAuth }), // reads  → router1134 * );1135 * ```1136 */1137export class SplitDaemonClient implements RindleDaemonClient {1138  private readonly writes: RindleDaemonClient;1139  private readonly reads: RindleDaemonClient;11401141  constructor(writes: RindleDaemonClient, reads: RindleDaemonClient) {1142    this.writes = writes;1143    this.reads = reads;1144  }11451146  // writes → the single master, never through the router1147  executeSqlTxn(input: SqlTxn): Promise<SqlTxnOutput> {1148    return this.writes.executeSqlTxn(input);1149  }1150  // raw reads → a replica by default (scaled off the master); `consistency:"strong"` opts that1151  // read into the master for read-your-writes after a write to the same data.1152  executeSqlRead(input: SqlRead): Promise<SqlReadOutput> {1153    return input.consistency === "strong"1154      ? this.writes.executeSqlRead(input)1155      : this.reads.executeSqlRead(input);1156  }1157  rejectMutation(input: MutationRejection): Promise<MutationRejectionOutput> {1158    return this.writes.rejectMutation(input);1159  }1160  // Interactive mutation sessions hold the MASTER's write transaction — never a replica's1161  // (DAEMON-INTERACTIVE-TXN-DESIGN.md §4.1; the follower write-fence enforces the same).1162  beginMutationSession(input: MutationSessionBegin): Promise<MutationSessionBeginOutput> {1163    const begin = this.writes.beginMutationSession?.bind(this.writes);1164    if (!begin) return Promise.reject(new Error("the write master lacks mutation sessions"));1165    return begin(input);1166  }1167  execInMutationSession(input: MutationSessionExec): Promise<unknown> {1168    const exec = this.writes.execInMutationSession?.bind(this.writes);1169    if (!exec) return Promise.reject(new Error("the write master lacks mutation sessions"));1170    return exec(input);1171  }1172  queryInMutationSession(input: MutationSessionQuery): Promise<SqlReadOutput> {1173    const query = this.writes.queryInMutationSession?.bind(this.writes);1174    if (!query) return Promise.reject(new Error("the write master lacks mutation sessions"));1175    return query(input);1176  }1177  commitMutationSession(input: MutationSessionRef): Promise<SqlTxnOutput> {1178    const commit = this.writes.commitMutationSession?.bind(this.writes);1179    if (!commit) return Promise.reject(new Error("the write master lacks mutation sessions"));1180    return commit(input);1181  }1182  rollbackMutationSession(input: MutationSessionRef): Promise<unknown> {1183    const rollback = this.writes.rollbackMutationSession?.bind(this.writes);1184    if (!rollback) return Promise.reject(new Error("the write master lacks mutation sessions"));1185    return rollback(input);1186  }1187  applyRowChangeTxn(input: RowChangeTxn): Promise<RowChangeTxnOutput> {1188    return this.writes.applyRowChangeTxn(input);1189  }1190  claimRoomEpoch(input: ClaimRoomEpochInput): Promise<ClaimRoomEpochOutput> {1191    const claim = this.writes.claimRoomEpoch?.bind(this.writes);1192    if (!claim) return Promise.reject(new Error("the write master lacks claimRoomEpoch"));1193    return claim(input);1194  }1195  roomLmids(input: RoomLmidsInput): Promise<RoomLmidsOutput> {1196    const lmids = this.writes.roomLmids?.bind(this.writes);1197    if (!lmids) return Promise.reject(new Error("the write master lacks roomLmids"));1198    return lmids(input);1199  }1200  migrate(input: MigrateInput): Promise<MigrateOutput> {1201    return this.writes.migrate(input);1202  }12031204  // reads → the fleet (one FLEET_URL; the affinity ticket + edge place the follower)1205  materialize(input: MaterializeInput): Promise<MaterializeOutput> {1206    return this.reads.materialize(input);1207  }1208  query(input: QueryOnceInput): Promise<QueryOnceOutput> {1209    return this.reads.query(input);1210  }1211  dematerialize(input: DematerializeInput): Promise<DematerializeOutput> {1212    return this.reads.dematerialize(input);1213  }1214}12151216// --------------------------------------------------------------------------- dialect SQL renderer1217//1218// A logical {@link MutationOp} → dialect `SqlStatement`. The whole per-dialect delta is the1219// PLACEHOLDER STYLE (`?` vs `$n`): identifiers are always double-quoted (SQLite tolerates it, PG1220// requires it — `user` is reserved, camelCase folds), and upserts use portable `ON CONFLICT` (both1221// engines). So `sqliteDialect`/`postgresDialect` differ only in `placeholder`.12221223/** A SQL dialect for the logical mutation renderer. */1224export interface SqlDialect {1225  readonly name: "sqlite" | "postgres";1226  /** Render the i-th (1-based) bind placeholder. sqlite: `?`; postgres: `$i`. */1227  placeholder(oneBased: number): string;1228  /** Optional value coercion hook (e.g. a future SQLite `0/1` boolean). Default: identity. */1229  encodeValue?(v: WireValue, type: ColType): WireValue;1230}12311232export const sqliteDialect: SqlDialect = { name: "sqlite", placeholder: () => "?" };1233export const postgresDialect: SqlDialect = { name: "postgres", placeholder: (i) => `$${i}` };12341235/** Per-table metadata the renderer needs (all reachable from a `TableMeta`). */1236export interface TableRenderMeta {1237  /** Columns in schema (wire) order — the stable INSERT column list + completeness check. */1238  columns: string[];1239  /** Primary-key column NAMES — the WHERE / ON CONFLICT target / SET partition. */1240  pkNames: string[];1241  /** Column name → declared type (only consulted by {@link SqlDialect.encodeValue}). */1242  types: Record<string, ColType>;1243  /** Columns a full insert must name — the non-nullable ones (design 206 §6.2). */1244  required: string[];1245  /** The nullable (omittable-to-null) columns — an omitted one binds `NULL` (design 206 §6.2). */1246  nullable: ReadonlySet<string>;1247}12481249export type RenderIndex = Record<string, TableRenderMeta>;12501251/** Build the {@link RenderIndex} from a typed schema (`schema.tables[name]` is a `TableMeta`). */1252export function buildRenderIndex(schema: Schema): RenderIndex {1253  const out: RenderIndex = {};1254  const tables = (schema as unknown as { tables: Record<string, { columns: Record<string, { type: ColType }>; primaryKey: readonly string[] }> }).tables;1255  for (const name of Object.keys(tables)) {1256    const meta = tables[name];1257    const columns = Object.keys(meta.columns);1258    const types: Record<string, ColType> = {};1259    for (const c of columns) types[c] = meta.columns[c].type;1260    // Same schema-derived plan the client funnel uses (design 206 §6.1), so their required-sets1261    // and null-fills can't drift.1262    const { required, nullable } = insertPlan(schema.tables[name]);1263    out[name] = { columns, pkNames: [...meta.primaryKey], types, required, nullable };1264  }1265  return out;1266}12671268const quoteIdent = (name: string): string => `"${name.replace(/"/g, '""')}"`;12691270function tableMeta(render: RenderIndex, table: string): TableRenderMeta {1271  const meta = render[table];1272  if (!meta) {1273    const known = Object.keys(render);1274    throw new Error(1275      `logical mutator write to unknown table ${JSON.stringify(table)} — did you pass \`schema\` to createRindleApiServer? known tables: ${known.length ? known.join(", ") : "(none — no schema configured)"}`,1276    );1277  }1278  return meta;1279}12801281/** Validate a keyed row against a table: reject unknown columns; require the pk columns; with1282 *  `full`, require every NON-nullable column (a nullable column may be omitted and is filled with1283 *  `NULL`, design 206 §6.2). Mirrors the client `trackingTx.checkColumns` messages so an author sees1284 *  the same error on both tiers. */1285function checkColumns(table: string, obj: KeyedRow, meta: TableRenderMeta, full: boolean): void {1286  const unknown = Object.keys(obj).filter((k) => !meta.columns.includes(k));1287  if (unknown.length) {1288    throw new Error(`unknown column${unknown.length > 1 ? "s" : ""} ${unknown.join(", ")} on ${table} — columns: ${meta.columns.join(", ")}`);1289  }1290  const required = full ? meta.required : meta.pkNames;1291  const missing = required.filter((c) => !(c in obj));1292  if (missing.length) {1293    throw new Error(`missing ${full ? "column" : "primary-key column"}${missing.length > 1 ? "s" : ""} ${missing.join(", ")} on ${table}`);1294  }1295}12961297const encode = (dialect: SqlDialect, v: WireValue, type: ColType): WireValue =>1298  dialect.encodeValue ? dialect.encodeValue(v, type) : v;12991300/** Render one {@link MutationOp} to a `{sql, params}` for the dialect, or `null` for a no-op (an1301 *  `update` whose row names only pk columns — nothing to SET, matching the client's no-op edit). */1302export function renderOp(op: MutationOp, meta: TableRenderMeta, dialect: SqlDialect): SqlStatement | null {1303  const t = quoteIdent(op.table);1304  const params: WireValue[] = [];1305  // Bind a value and return its placeholder at the correct 1-based index (post-push length).1306  // `insertCell` fills NULL for an omitted nullable column on the insert arm (design 206 §6.2);1307  // on update/delete every bound column is guaranteed present, so it is a pass-through there.1308  // `toCell` stringifies a `json` object (a typed mutator passes the parsed object) — a string1309  // passes through, so an author may still pass pre-stringified json.1310  const bind = (c: string, row: KeyedRow): string => {1311    params.push(encode(dialect, toCell(insertCell(row, c), meta.types[c]), meta.types[c]));1312    return dialect.placeholder(params.length);1313  };13141315  if (op.kind === "insert" || op.kind === "insertIgnore" || op.kind === "upsert") {1316    checkColumns(op.table, op.row, meta, true);1317    const cols = meta.columns;1318    const values = cols.map((c) => bind(c, op.row));1319    let sql = `INSERT INTO ${t} (${cols.map(quoteIdent).join(", ")}) VALUES (${values.join(", ")})`;1320    if (op.kind === "insertIgnore") {1321      sql += ` ON CONFLICT (${meta.pkNames.map(quoteIdent).join(", ")}) DO NOTHING`;1322    } else if (op.kind === "upsert") {1323      const nonPk = cols.filter((c) => !meta.pkNames.includes(c));1324      sql += ` ON CONFLICT (${meta.pkNames.map(quoteIdent).join(", ")}) `;1325      sql += nonPk.length1326        ? `DO UPDATE SET ${nonPk.map((c) => `${quoteIdent(c)} = excluded.${quoteIdent(c)}`).join(", ")}`1327        : `DO NOTHING`;1328    }1329    return { sql, params };1330  }13311332  if (op.kind === "update") {1333    checkColumns(op.table, op.row, meta, false);1334    const setCols = meta.columns.filter((c) => !meta.pkNames.includes(c) && c in op.row);1335    if (!setCols.length) return null; // pk-only row → nothing to change (client no-op edit)1336    const setSql = setCols.map((c) => `${quoteIdent(c)} = ${bind(c, op.row)}`);1337    const whereSql = meta.pkNames.map((c) => `${quoteIdent(c)} = ${bind(c, op.row)}`);1338    return { sql: `UPDATE ${t} SET ${setSql.join(", ")} WHERE ${whereSql.join(" AND ")}`, params };1339  }13401341  // delete1342  checkColumns(op.table, op.pk, meta, false);1343  const whereSql = meta.pkNames.map((c) => `${quoteIdent(c)} = ${bind(c, op.pk)}`);1344  return { sql: `DELETE FROM ${t} WHERE ${whereSql.join(" AND ")}`, params };1345}13461347/** Render a point read (`tx.row`) — `SELECT <cols> FROM "T" WHERE <pk>` — for read-your-writes. */1348export function renderPointRead(table: string, pk: KeyedRow, meta: TableRenderMeta, dialect: SqlDialect): SqlStatement {1349  checkColumns(table, pk, meta, false);1350  const params: WireValue[] = [];1351  const whereSql = meta.pkNames.map((c) => {1352    params.push(encode(dialect, pk[c], meta.types[c]));1353    return `${quoteIdent(c)} = ${dialect.placeholder(params.length)}`;1354  });1355  const cols = meta.columns.map(quoteIdent).join(", ");1356  return { sql: `SELECT ${cols} FROM ${quoteIdent(table)} WHERE ${whereSql.join(" AND ")}`, params };1357}13581359/** Map a driver row (column-name keyed) to a {@link KeyedRow} over the table's known columns. */1360function rowToKeyed(row: Record<string, unknown> | undefined, meta: TableRenderMeta): KeyedRow | undefined {1361  if (!row) return undefined;1362  const out: KeyedRow = {};1363  for (const c of meta.columns) out[c] = row[c] as WireValue;1364  return out;1365}13661367// --------------------------------------------------------------------------- backends + server tx13681369/** Thrown (wrapping the driver error) by a server tx's DB calls, so the seam can tell an INFRA1370 *  failure (retry) from a mutator-body throw (business rejection). */1371export class BackendError extends Error {1372  readonly driverError: unknown;1373  constructor(driverError: unknown) {1374    super(driverError instanceof Error ? driverError.message : String(driverError));1375    this.name = "BackendError";1376    this.driverError = driverError;1377  }1378}13791380/** Build the compiler {@link Catalog} for ONE ast from the render index: columns/pk from the1381 *  schema; relationship cardinality from the AST ITSELF — a Rindle relationship is declared at1382 *  the query site (`sub(alias, rel)` / `.one()`), never on the schema, so the alias→cardinality1383 *  map is inherently per-query. `columnTypes` are stubs: the sqlite dialect binds natives and1384 *  never consults them (DAEMON-INTERACTIVE-TXN §5.4 — no casts). */1385function catalogFor(render: RenderIndex, root: Ast): Catalog {1386  const tables: Record<string, TableSchema> = {};1387  const ensure = (table: string): TableSchema => {1388    const existing = tables[table];1389    if (existing) return existing;1390    const meta = tableMeta(render, table);1391    const columnTypes: Record<string, QueryColumnType> = {};1392    for (const c of meta.columns) columnTypes[c] = { type: "text", isEnum: false, isArray: false };1393    return (tables[table] = {1394      columns: [...meta.columns],1395      primaryKey: [...meta.pkNames],1396      columnTypes,1397      relationships: {},1398    });1399  };1400  const walkCondition = (cond: Condition | undefined): void => {1401    if (!cond) return;1402    if (cond.type === "and" || cond.type === "or") {1403      for (const c of cond.conditions) walkCondition(c);1404    } else if (cond.type === "correlatedSubquery") {1405      walkAst(cond.related.subquery);1406    }1407  };1408  const walkAst = (ast: Ast): void => {1409    const t = ensure(ast.table);1410    for (const rel of ast.related ?? []) {1411      const alias = rel.subquery.alias;1412      if (alias != null) t.relationships[alias] = rel.subquery.one === true ? "one" : "many";1413      walkAst(rel.subquery);1414    }1415    walkCondition(ast.where);1416  };1417  walkAst(root);1418  return { tables };1419}14201421/** The control-flow unwind for a begin-absorbed replay (DAEMON-INTERACTIVE-TXN §4.1): thrown1422 *  from the first read so the mutator body stops re-running an already-committed envelope; the1423 *  backend answers with the tx's latched authoritative output. Never surfaces to users. */1424class AbsorbedReplay extends Error {1425  constructor() {1426    super("mutation absorbed by mid dedup at session begin");1427  }1428}14291430const MUTATOR_CONFLICT_MAX_ATTEMPTS = 5;14311432function isRetryableCommitConflict(error: unknown): boolean {1433  if (error instanceof RindleSqlError) {1434    return error.status === 409 && (error.code === "retryable-conflict" || error.code === "TRANSACTION_CONFLICT");1435  }1436  if (!(error instanceof DaemonHttpError) || error.status !== 409) return false;1437  try {1438    const body = JSON.parse(error.body) as { code?: unknown; retryable?: unknown };1439    return body.code === "retryable-conflict" && body.retryable === true;1440  } catch {1441    return false;1442  }1443}14441445async function mutatorConflictBackoff(attempt: number): Promise<void> {1446  const ceiling = Math.min(32, 2 ** attempt);1447  const millis = ceiling + Math.floor(Math.random() * 4);1448  await new Promise<void>((resolve) => setTimeout(resolve, millis));1449}14501451/** SQLite's constraint family: primary result code 19 (`SQLITE_CONSTRAINT`), which every extended1452 *  code (`_UNIQUE`, `_NOTNULL`, `_FOREIGNKEY`, `_CHECK`, `_PRIMARYKEY`, …) carries in its low byte. */1453const SQLITE_CONSTRAINT = 19;14541455/** SQLite renders every constraint violation as `<KIND> constraint failed[: <detail>]`, plus one1456 *  legacy spelling. Matching text is a last resort, used ONLY because the legacy daemon session1457 *  plane flattens deterministic statement errors and genuine infra failures into a single1458 *  `500 {"error":"sqlite: …"}` (`SessionError::Backend` — "poison the session, surface the1459 *  message"), so the message is the only signal it offers. The versioned SQL surface carries1460 *  `sqlite_code` and is matched structurally above. */1461const SQLITE_CONSTRAINT_MESSAGE = /\bconstraint failed\b|\bPRIMARY KEY must be unique\b/i;14621463/** The clean reason to report for a driver refusal — the daemon's flat `{error}` body rather than1464 *  the `rindle daemon request failed: 500 …` envelope wrapped around it. */1465function driverRefusalReason(error: unknown): string {1466  if (error instanceof DaemonHttpError) {1467    try {1468      const body = JSON.parse(error.body) as { error?: unknown; message?: unknown };1469      const text = typeof body.error === "string" ? body.error : body.message;1470      if (typeof text === "string" && text.length > 0) return text;1471    } catch {1472      // Not JSON — fall through to the error's own message.1473    }1474  }1475  return errMessage(error);1476}14771478/** True when the DATABASE definitively refused this write and would refuse it identically on every1479 *  replay: a constraint violation (UNIQUE / NOT NULL / CHECK / FOREIGN KEY / PRIMARY KEY).1480 *1481 *  This is the {@link isUnencodableBind} case one layer down — a deterministic poison message —1482 *  except the database, not the codec, is the one saying no. Left as infra it throws out of1483 *  `runMutation`, the mutate route 500s, and the browser's queue retries the same batch forever:1484 *  the mutation never lands, `lmid` never advances, the prediction never retires, and every LATER1485 *  mutation from that client is stuck behind it. Rejecting instead advances `lmid` alone, so the1486 *  prediction snaps back, the reason reaches `onRejected`, and the queue drains.1487 *1488 *  Deliberately NARROW. A constraint violation is the one class where the failure is both1489 *  DEFINITE (SQLite aborted the statement and rolled the transaction back — nothing committed) and1490 *  DETERMINISTIC (the same rows violate the same index next time). It does NOT cover a 5xx from a1491 *  restarting daemon, a disk error, a timeout, or an auth misconfiguration — those are genuinely1492 *  retryable, and silently dropping every write on a bad token would be far worse than wedging. */1493function isDeterministicWriteRefusal(error: unknown): boolean {1494  if (error instanceof RindleSqlError) {1495    // Structural first: the extended code is authoritative and locale/message independent.1496    if (error.sqliteCode !== undefined) return (error.sqliteCode & 0xff) === SQLITE_CONSTRAINT;1497    return SQLITE_CONSTRAINT_MESSAGE.test(error.message);1498  }1499  if (error instanceof DaemonHttpError) return SQLITE_CONSTRAINT_MESSAGE.test(error.body);1500  return false;1501}15021503/** True when this error is the SQL codec refusing a bind value outright (`undefined`, `Date`, `NaN`,1504 *  a binary view, an out-of-i64 bigint) rather than a transport or database failure. */1505function isUnencodableBind(error: unknown): boolean {1506  return error instanceof RindleSqlError && error.code === "VALUE_UNSUPPORTED";1507}15081509/** Refuse an unencodable bind at the point the MUTATOR supplies it, so it surfaces as a BUSINESS1510 *  rejection (lmid advances, the browser retires its prediction) instead of an infrastructure1511 *  failure. Left as infra it is retried forever against a deterministic mutator, which wedges the1512 *  client's mutation queue behind a poison message.1513 *1514 *  Only the SQL transport needs this: the legacy daemon encoder is JSON, which silently coerces the1515 *  same values (`undefined`/`NaN` -> null, `Date` -> an ISO string). Asserting there would invent a1516 *  failure that the wire does not actually have. */1517function assertEncodableParams(sql: string, params: readonly WireValue[] | undefined): void {1518  if (params === undefined) return;1519  for (let index = 0; index < params.length; index++) {1520    try {1521      encodeSqlValue(params[index] as Parameters<typeof encodeSqlValue>[0]);1522    } catch (error) {1523      if (!isUnencodableBind(error)) throw error;1524      throw new Error(`bind ${index} of \`${sql}\` cannot be stored: ${errMessage(error)}`);1525    }1526  }1527}15281529/** Leading keywords the SQL mutation surface structurally REFUSES inside a mutator's write batch: a1530 *  read (`SELECT`/`EXPLAIN`), transaction control, a connection `PRAGMA`, or DDL. None can begin a1531 *  valid mutation write, so refusing them has no false positives — a `WITH`-prefixed statement is1532 *  deliberately absent because it may resolve to either a read or a write, and the server stays the1533 *  authority for that case. */1534const MUTATION_REFUSED_LEADING_KEYWORDS = new Set([1535  "SELECT", "EXPLAIN", "VALUES",1536  "BEGIN", "COMMIT", "ROLLBACK", "SAVEPOINT", "RELEASE", "END",1537  "PRAGMA", "VACUUM", "ATTACH", "DETACH",1538  "CREATE", "ALTER", "DROP", "REINDEX", "ANALYZE",1539]);15401541/** Refuse a statement whose CLASS the mutation surface rejects, at the point the MUTATOR supplies it,1542 *  so it surfaces as a business rejection instead of a poison. When the batch reaches the transport1543 *  the body has already returned, so a server 400 there is (mis)read as infrastructure and retried1544 *  forever — the same wedge {@link assertEncodableParams} prevents for bind values. Conservative by1545 *  design: it fires only for a leading keyword that can never start a valid write, and leaves every1546 *  ambiguous case (including CTE-prefixed writes) to the server's authoritative classifier. */1547function assertMutationWriteStatement(sql: string): void {1548  const match = /^[\s;]*([a-zA-Z]+)/.exec(sql);1549  if (match === null) return;1550  const keyword = match[1]!.toUpperCase();1551  if (MUTATION_REFUSED_LEADING_KEYWORDS.has(keyword)) {1552    throw new Error(1553      `a mutator write statement cannot begin with ${keyword} (\`${sql}\`); ` +1554        `mutations write rows only — use tx.sql.query(...) for reads and migrations for DDL`,1555    );1556  }1557}15581559interface MutationTransportBegin {1560  handle?: unknown;1561  absorbed?: SqlTxnOutput;1562  read?: SqlReadOutput;1563}15641565/** The mutation-only transport consumed by the API-server transaction harness. Both the legacy1566 *  daemon client and `@rindle/sql-client` adapt to this one shape, so lmid/rejection/lazy-session1567 *  policy is implemented once. */1568interface MutationTransport {1569  readonly interactive: boolean;1570  /** Whether this transport's wire REFUSES values the daemon's JSON encoder coerces. Drives1571   *  {@link assertEncodableParams} — see its docs for why the daemon adapter opts out. */1572  readonly strictValues: boolean;1573  execute(input: { envelope: MutationEnvelope; statements: SqlStatement[] }): Promise<SqlTxnOutput>;1574  reject(input: { envelope: MutationEnvelope; reason: string }): Promise<unknown>;1575  begin(input: {1576    envelope: MutationEnvelope;1577    statements: SqlStatement[];1578    query: SqlStatement;1579  }): Promise<MutationTransportBegin>;1580  exec(handle: unknown, statements: SqlStatement[]): Promise<void>;1581  query(handle: unknown, statement: SqlStatement): Promise<SqlReadOutput>;1582  commit(handle: unknown): Promise<SqlTxnOutput>;1583  rollback(handle: unknown): Promise<void>;1584  readCommitted(statement: SqlStatement): Promise<SqlReadOutput>;1585}15861587function daemonMutationTransport(daemon: RindleDaemonClient): MutationTransport {1588  return {1589    interactive: daemon.beginMutationSession !== undefined,1590    strictValues: false,1591    execute({ envelope, statements }) {1592      const txn: SqlTxn = { statements, clientID: envelope.clientID, mid: envelope.mid };1593      return daemon.executeSqlTxn(txn);1594    },1595    reject({ envelope, reason }) {1596      return daemon.rejectMutation({ clientID: envelope.clientID, mid: envelope.mid, reason });1597    },1598    async begin({ envelope, statements, query }) {1599      if (!daemon.beginMutationSession) throw new Error("the daemon client does not support mutation sessions");1600      const input: MutationSessionBegin = {1601        clientID: envelope.clientID,1602        mid: envelope.mid,1603        statements,1604        query,1605      };1606      const opened = await daemon.beginMutationSession(input);1607      if (opened.absorbed) {1608        const { absorbed: _absorbed, sessionId: _sessionId, read: _read, ...output } = opened;1609        return { absorbed: output as SqlTxnOutput };1610      }1611      return { handle: opened.sessionId, read: opened.read };1612    },1613    async exec(handle, statements) {1614      await daemon.execInMutationSession!({ sessionId: handle as string, statements });1615    },1616    query(handle, statement) {1617      return daemon.queryInMutationSession!({1618        sessionId: handle as string,1619        sql: statement.sql,1620        params: statement.params,1621      });1622    },1623    commit(handle) {1624      return daemon.commitMutationSession!({ sessionId: handle as string });1625    },1626    async rollback(handle) {1627      await daemon.rollbackMutationSession!({ sessionId: handle as string });1628    },1629    readCommitted(statement) {1630      return daemon.executeSqlRead({ sql: statement.sql, params: statement.params });1631    },1632  };1633}16341635function mutationReceiptOutput(receipt: SqlMutationReceipt, clientID: string): SqlTxnOutput {1636  const output: SqlTxnOutput = {1637    applied: receipt.applied,1638    lmid: receipt.lmid,1639    lmidAdvances: [{ clientID, lmid: receipt.lmid }],1640  };1641  if (receipt.commitCursor !== null) output.cursor = receipt.commitCursor;1642  return output;1643}16441645function publicMutationStatement(statement: SqlStatement): PublicSqlStatement {1646  return statement.params === undefined ? { sql: statement.sql } : { sql: statement.sql, args: statement.params };1647}16481649function mutationRowsOutput(rows: SqlMutationRows): SqlReadOutput {1650  return { cols: rows.columns, rows: rows.rows as WireValue[][] };1651}16521653/** Convert the transports' compact positional rows into the ergonomic server-only raw-SQL shape. */1654function keyedSqlRows<Row = Record<string, unknown>>(1655  columns: readonly string[],1656  rows: readonly (readonly unknown[])[],1657): Row[] {1658  // Row objects are keyed by column NAME, so a read that projects the same name twice1659  // (`SELECT parent.status, child.status ...`) would silently keep only the last value — and a1660  // mutator branching on `row.status` would then authorize against the wrong cell. Refuse it loudly1661  // so the collision surfaces as a rejection reason instead of silent, wrong data.1662  const seen = new Set<string>();1663  for (const column of columns) {1664    if (seen.has(column)) {1665      throw new Error(1666        `raw SQL read projects the column name ${JSON.stringify(column)} more than once; ` +1667          `alias them to distinct names (e.g. SELECT a.id AS a_id, b.id AS b_id)`,1668      );1669    }1670    seen.add(column);1671  }1672  return rows.map((cells) => Object.fromEntries(columns.map((column, index) => [column, cells[index]])) as Row);1673}16741675function daemonOutsideSql(daemon: RindleDaemonClient): ServerSql {1676  return {1677    async execute(sql, params = []) {1678      await daemon.executeSqlTxn({ statements: [{ sql, params: [...params] }] });1679    },1680    async batch(statements) {1681      if (statements.length === 0) return;1682      await daemon.executeSqlTxn({1683        statements: statements.map((statement) => ({1684          sql: statement.sql,1685          ...(statement.params !== undefined ? { params: [...statement.params] } : {}),1686        })),1687      });1688    },1689    async query<Row = Record<string, unknown>>(sql: string, params: readonly WireValue[] = []): Promise<Row[]> {1690      const out = await daemon.executeSqlRead({ sql, params: [...params], consistency: "strong" });1691      return keyedSqlRows<Row>(out.cols, out.rows);1692    },1693  };1694}16951696function sqlSessionOutsideSql(sql: SqlSession): ServerSql {1697  return {1698    async execute(text, params = []) {1699      await sql.execute({ sql: text, args: [...params] });1700    },1701    async batch(statements) {1702      if (statements.length === 0) return;1703      await sql.batch(statements.map(publicMutationStatement));1704    },1705    async query<Row = Record<string, unknown>>(text: string, params: readonly WireValue[] = []): Promise<Row[]> {1706      const out = await sql.execute({ sql: text, args: [...params], wantRows: true }, { consistency: "strong" });1707      return keyedSqlRows<Row>(1708        out.result.columns.map((column) => column.name),1709        out.result.rows,1710      );1711    },1712  };1713}17141715function sqlClientMutationTransport(sql: SqlSession): MutationTransport {1716  return {1717    interactive: true,1718    strictValues: true,1719    async execute({ envelope, statements }) {1720      const receipt = await sql.executeMutation({1721        clientId: envelope.clientID,1722        mid: envelope.mid,1723        statements: statements.map(publicMutationStatement),1724      });1725      return mutationReceiptOutput(receipt, envelope.clientID);1726    },1727    async reject({ envelope, reason }) {1728      return mutationReceiptOutput(1729        await sql.rejectMutation({ clientId: envelope.clientID, mid: envelope.mid, reason }),1730        envelope.clientID,1731      );1732    },1733    async begin({ envelope, statements, query }) {1734      const opened = await sql.beginMutation({1735        clientId: envelope.clientID,1736        mid: envelope.mid,1737        statements: statements.map(publicMutationStatement),1738        query: publicMutationStatement(query),1739      });1740      if (opened.absorbed) {1741        return { absorbed: mutationReceiptOutput(opened.receipt, envelope.clientID) };1742      }1743      return {1744        handle: opened.transaction,1745        ...(opened.read !== undefined ? { read: mutationRowsOutput(opened.read) } : {}),1746      };1747    },1748    async exec(handle, statements) {1749      await (handle as SqlMutationTransaction).batch(statements.map(publicMutationStatement));1750    },1751    async query(handle, statement) {1752      return mutationRowsOutput(await (handle as SqlMutationTransaction).query(publicMutationStatement(statement)));1753    },1754    async commit(handle) {1755      const receipt = await (handle as SqlMutationTransaction).commit();1756      const advance = receipt.lmid;1757      // The handle is opened for exactly one client; RemoteLazyTx patches the client id from its1758      // envelope after this call so the legacy MutationBackend receipt remains byte-compatible.1759      return {1760        applied: receipt.applied,1761        cursor: receipt.commitCursor ?? undefined,1762        lmid: advance,1763      };1764    },1765    async rollback(handle) {1766      await (handle as SqlMutationTransaction).rollback();1767    },1768    async readCommitted(statement) {1769      const result = await sql.execute(publicMutationStatement(statement), { consistency: "strong" });1770      return {1771        cols: result.result.columns.map((column) => column.name),1772        rows: result.result.rows as WireValue[][],1773      };1774    },1775  };1776}17771778/**1779 * The remote SQLite server tx (DAEMON-INTERACTIVE-TXN-DESIGN.md §5): ONE authoring surface, two1780 * execution strategies. It starts ACCUMULATING — a pure-write mutator ships one batch to1781 * the selected mutation transport — and LAZILY UPGRADES to an interactive1782 * mutation session at the mutator's first read: `begin` carries the envelope identity, the1783 * accumulated statement prefix (sound to replay — nothing before the first read observed DB1784 * state, §5.2), and the read itself, so a one-read mutator pays exactly one extra round trip.1785 * From then on reads run THROUGH the open transaction (read-your-writes — PG parity, and no1786 * read-then-write race) and writes buffer locally, flushing before the next read/commit: k1787 * reads cost k+2 round trips regardless of write count.1788 *1789 * Begin-time mid dedup can ABSORB the envelope (a redelivery whose commit response was lost):1790 * the replay output is latched on {@link RemoteLazyTx.absorbed} and {@link AbsorbedReplay}1791 * unwinds the body — the latch (not the throw) is authoritative, so a mutator that swallows1792 * the unwind still cannot re-apply (no session opened; buffered writes are never shipped).1793 * A daemon client without session support keeps the LEGACY committed-state point read.1794 */1795class RemoteLazyTx implements ServerMutationTx {1796  /** Pre-upgrade: the accumulated batch/prefix. Post-upgrade: writes buffered for the next flush. */1797  private readonly stmts: SqlStatement[] = [];1798  private readonly render: RenderIndex;1799  private readonly transport: MutationTransport;1800  private readonly envelope: MutationEnvelope;1801  private sessionHandle?: unknown;1802  readonly sql: ServerSql;1803  /** The begin-absorbed replay output (§4.1), latched for the backend. */1804  absorbed?: SqlTxnOutput;18051806  constructor(render: RenderIndex, transport: MutationTransport, envelope: MutationEnvelope) {1807    this.render = render;1808    this.transport = transport;1809    this.envelope = envelope;1810    this.sql = {1811      execute: async (sql, params = []) => {1812        this.exec(sql, [...params]);1813      },1814      batch: async (statements) => {1815        for (const statement of statements) {1816          this.exec(statement.sql, statement.params === undefined ? [] : [...statement.params]);1817        }1818      },1819      query: <Row = Record<string, unknown>>(sql: string, params: readonly WireValue[] = []) =>1820        this.querySql<Row>(sql, params),1821    };1822  }18231824  /** True once the tx upgraded to an interactive session (the backend then commits it). */1825  get session(): boolean {1826    return this.sessionHandle !== undefined;1827  }18281829  get statements(): readonly SqlStatement[] {1830    return this.stmts;1831  }18321833  exec(sql: string, params: WireValue[] = []): void {1834    // Refuse here, INSIDE the mutator body, so the harness reads it as a business rejection. By the1835    // time the statement reaches the transport the body has returned and the throw is infra.1836    if (this.transport.strictValues) {1837      assertMutationWriteStatement(sql);1838      assertEncodableParams(sql, params);1839    }1840    this.stmts.push({ sql, params });1841  }18421843  private push(op: MutationOp): Promise<void> {1844    const rendered = renderOp(op, tableMeta(this.render, op.table), sqliteDialect);1845    if (rendered) {1846      if (this.transport.strictValues) assertEncodableParams(rendered.sql, rendered.params);1847      this.stmts.push(rendered);1848    }1849    return Promise.resolve();1850  }18511852  insert(table: string, row: KeyedRow): Promise<void> {1853    return this.push({ kind: "insert", table, row });1854  }1855  update(table: string, row: KeyedRow): Promise<void> {1856    return this.push({ kind: "update", table, row });1857  }1858  upsert(table: string, row: KeyedRow): Promise<void> {1859    return this.push({ kind: "upsert", table, row });1860  }1861  insertIgnore(table: string, row: KeyedRow): Promise<void> {1862    return this.push({ kind: "insertIgnore", table, row });1863  }1864  delete(table: string, pk: KeyedRow): Promise<void> {1865    return this.push({ kind: "delete", table, pk });1866  }18671868  async row(table: string, pk: KeyedRow): Promise<KeyedRow | undefined> {1869    const read = renderPointRead(table, pk, tableMeta(this.render, table), sqliteDialect);1870    const out = await this.readThroughTxn(read);1871    const cells = out.rows[0];1872    if (!cells) return undefined;1873    const keyed: KeyedRow = {}; // the daemon returns positional cells — zip with `cols`1874    out.cols.forEach((c, i) => (keyed[c] = cells[i]));1875    return keyed;1876  }18771878  /** A full-shape read inside the open transaction (§5.4): compile to ONE SQLite `SELECT`1879   *  (native bind params, no casts — SQLite is the canonical store), ride the session like1880   *  `row` (including the lazy upgrade / begin ride-along), parse the single JSON cell. */1881  async query(q: Ast | Query<any, any, any>): Promise<unknown> {1882    const ast = typeof (q as Query<any, any, any>).ast === "function" ? (q as Query<any, any, any>).ast() : (q as Ast);1883    const compiled = compileQueryAst(ast, catalogFor(this.render, ast), { dialect: "sqlite" });1884    const out = await this.readThroughTxn({ sql: compiled.sql, params: compiled.params as WireValue[] });1885    const cell = out.rows[0]?.[0];1886    if (typeof cell !== "string") return ast.one === true ? null : [];1887    return JSON.parse(cell) as unknown;1888  }18891890  private async querySql<Row>(sql: string, params: readonly WireValue[]): Promise<Row[]> {1891    const out = await this.readThroughTxn({ sql, params: [...params] });1892    return keyedSqlRows<Row>(out.cols, out.rows);1893  }18941895  /** Run one read: upgrade to a session at the first (§5.1), ride the open one after, or fall1896   *  back to the legacy committed-state read when the daemon client lacks sessions. */1897  private async readThroughTxn(read: SqlStatement): Promise<SqlReadOutput> {1898    if (this.absorbed) throw new AbsorbedReplay();1899    // Refuse an unencodable read bind at the mutator boundary, exactly as `exec` does for writes.1900    // A read's parameters are encoded inside the transport, where the throw becomes a BackendError1901    // (infra) that retries the deterministic mutator forever and wedges the client's queue; asserting1902    // here makes it a business rejection instead.1903    if (this.transport.strictValues) assertEncodableParams(read.sql, read.params);1904    if (!this.transport.interactive) {1905      try {1906        return await this.transport.readCommitted(read);1907      } catch (err) {1908        throw new BackendError(err);1909      }1910    }1911    try {1912      if (this.sessionHandle === undefined) {1913        const opened = await this.transport.begin({1914          envelope: this.envelope,1915          statements: this.stmts.splice(0),1916          query: read,1917        });1918        if (opened.absorbed) {1919          this.absorbed = opened.absorbed;1920          throw new AbsorbedReplay();1921        }1922        if (opened.handle === undefined || !opened.read) {1923          throw new Error(`malformed mutate-session begin reply: ${JSON.stringify(opened)}`);1924        }1925        this.sessionHandle = opened.handle;1926        return opened.read;1927      }1928      await this.flush();1929      return await this.transport.query(this.sessionHandle, read);1930    } catch (err) {1931      if (err instanceof AbsorbedReplay || err instanceof BackendError) throw err;1932      throw new BackendError(err);1933    }1934  }19351936  /** Ship buffered writes into the open session, order-preserving; a no-op when none pend. */1937  private async flush(): Promise<void> {1938    if (this.stmts.length === 0) return;1939    await this.transport.exec(this.sessionHandle!, this.stmts.splice(0));1940  }19411942  /** Flush + commit the open session — the daemon stamps lmid co-transactionally (§4.4) and1943   *  answers the same shape `/execute-sql-txn` does. */1944  async commitSession(): Promise<SqlTxnOutput> {1945    try {1946      await this.flush();1947      const output = await this.transport.commit(this.sessionHandle!);1948      if (output.lmid !== undefined && output.lmidAdvances === undefined) {1949        output.lmidAdvances = [{ clientID: this.envelope.clientID, lmid: output.lmid }];1950      }1951      return output;1952    } catch (err) {1953      throw err instanceof BackendError ? err : new BackendError(err);1954    }1955  }19561957  /** Best-effort rollback (the daemon's deadline is the backstop). MUST be awaited before a1958   *  follow-up `/reject-mutation`: that lmid-only commit needs the writer this session holds. */1959  async rollbackSessionQuietly(): Promise<void> {1960    if (this.sessionHandle === undefined) return;1961    const sessionHandle = this.sessionHandle;1962    this.sessionHandle = undefined;1963    try {1964      await this.transport.rollback(sessionHandle);1965    } catch {1966      // Unreachable daemon / already-expired session: the deadline rollback covers it.1967    }1968  }1969}19701971/** The Postgres server tx: a REAL interactive transaction. Logical writes render to `$n` and run1972 *  LIVE against the open txn; raw `exec` runs live too (after `rewrite`); `row` reads the open txn1973 *  (read-your-writes). Ops append to an internally-serialized chain so order holds even when a legacy1974 *  sync mutator does not `await`; the backend drains the chain (`settle`) before the lmid upsert. */1975class PgLiveTx implements ServerMutationTx {1976  private chain: Promise<void> = Promise.resolve();1977  private readonly stmts: SqlStatement[] = [];1978  private readonly q: PgQuery;1979  private readonly render: RenderIndex;1980  private readonly rewrite: (sql: string) => string;1981  readonly sql: ServerSql;19821983  constructor(q: PgQuery, render: RenderIndex, rewrite: (sql: string) => string) {1984    this.q = q;1985    this.render = render;1986    this.rewrite = rewrite;1987    this.sql = {1988      execute: async (sql, params = []) => {1989        this.exec(sql, [...params]);1990        await this.settle();1991      },1992      batch: async (statements) => {1993        for (const statement of statements) {1994          this.exec(statement.sql, statement.params === undefined ? [] : [...statement.params]);1995        }1996        await this.settle();1997      },1998      query: <Row = Record<string, unknown>>(sql: string, params: readonly WireValue[] = []) =>1999        this.querySql<Row>(sql, params),2000    };2001  }20022003  get statements(): readonly SqlStatement[] {2004    return this.stmts;2005  }20062007  settle(): Promise<void> {2008    return this.chain;2009  }20102011  private execLive(sql: string, params: WireValue[]): void {2012    this.chain = this.chain.then(async () => {2013      try {2014        await this.q.exec(sql, params as unknown[]);2015      } catch (err) {2016        throw new BackendError(err);2017      }2018    });2019  }20202021  exec(sql: string, params: WireValue[] = []): void {2022    const stmt = { sql: this.rewrite(sql), params };2023    this.stmts.push(stmt);2024    this.execLive(stmt.sql, params);2025  }20262027  private write(op: MutationOp): Promise<void> {2028    let rendered: SqlStatement | null;2029    try {2030      rendered = renderOp(op, tableMeta(this.render, op.table), postgresDialect);2031    } catch (err) {2032      return Promise.reject(err); // a validation error (business rejection), synchronous shape2033    }2034    if (rendered) this.execLive(rendered.sql, rendered.params ?? []);2035    return this.chain;2036  }20372038  insert(table: string, row: KeyedRow): Promise<void> {2039    return this.write({ kind: "insert", table, row });2040  }2041  update(table: string, row: KeyedRow): Promise<void> {2042    return this.write({ kind: "update", table, row });2043  }2044  upsert(table: string, row: KeyedRow): Promise<void> {2045    return this.write({ kind: "upsert", table, row });2046  }2047  insertIgnore(table: string, row: KeyedRow): Promise<void> {2048    return this.write({ kind: "insertIgnore", table, row });2049  }2050  delete(table: string, pk: KeyedRow): Promise<void> {2051    return this.write({ kind: "delete", table, pk });2052  }2053  async row(table: string, pk: KeyedRow): Promise<KeyedRow | undefined> {2054    const meta = tableMeta(this.render, table);2055    const read = renderPointRead(table, pk, meta, postgresDialect); // validates before draining2056    await this.settle(); // read-your-writes: drain queued writes first2057    let rows: Array<Record<string, unknown>>;2058    try {2059      rows = await this.q.query(read.sql, read.params as unknown[]);2060    } catch (err) {2061      throw new BackendError(err);2062    }2063    return rowToKeyed(rows[0], meta);2064  }20652066  query(): Promise<unknown> {2067    // The compiler's postgres dialect ships (@rindle/query-compiler); what is missing is a source2068    // for the column-TYPE catalog it needs. That dialect branches on the native type for both the2069    // `::text::<type>` filter cast and the temporal projection, so a stub catalog does not fail —2070    // it returns wrong rows. `designs/416-POSTGRES-READ-CATALOG-DESIGN.md` settles where the types2071    // come from (one published Rindle-owned row, read over the mutator's open txn); until it is2072    // built, refusing is the only correct answer.2073    return Promise.reject(2074      new Error(2075        "tx.query is not wired on the Postgres backend yet (designs/416-POSTGRES-READ-CATALOG-DESIGN.md) — use tx.row for point reads meanwhile",2076      ),2077    );2078  }20792080  private async querySql<Row>(sql: string, params: readonly WireValue[]): Promise<Row[]> {2081    await this.settle();2082    try {2083      return (await this.q.query(this.rewrite(sql), [...params])) as Row[];2084    } catch (err) {2085      throw new BackendError(err);2086    }2087  }2088}20892090/** Shared remote-SQL mutation backend. A pure-write mutator remains one request; a read-bearing2091 * mutator lazily upgrades at its first read; accepted effects commit with lmid; business rejection2092 * rolls effects back before an lmid-only commit. Both daemonBackend and sqlBackend use this exact2093 * policy implementation. */2094function remoteMutationBackend(transport: MutationTransport, outsideSql: ServerSql): MutationBackend {2095  return {2096    dialect: sqliteDialect,2097    outsideSql,2098    async runMutation(input) {2099      const { envelope, render, run } = input;2100      for (let attempt = 0; attempt < MUTATOR_CONFLICT_MAX_ATTEMPTS; attempt++) {2101        // Hoisted out of the try so the deterministic-refusal arm below can release the session2102        // before its lmid-only reject commit, exactly as the in-body rejection path does.2103        const tx = new RemoteLazyTx(render, transport, envelope);2104        try {2105          try {2106            await run(tx);2107          } catch (err) {2108            // A begin-absorbed replay: the authoritative outcome already committed — answer it,2109            // whatever the body did with the unwind (§4.1; the latch, not the throw, decides).2110            if (tx.absorbed) return { accepted: true, output: tx.absorbed };2111            if (err instanceof BackendError) {2112              await tx.rollbackSessionQuietly();2113              throw err.driverError; // infra — never a user rejection2114            }2115            const reason = errMessage(err);2116            // Data first, watermark second: rollback releases this session's connection before2117            // the lmid-only rejection commit.2118            await tx.rollbackSessionQuietly();2119            const output = await transport.reject({ envelope, reason });2120            return { accepted: false, reason, output };2121          }2122          if (tx.absorbed) return { accepted: true, output: tx.absorbed };2123          if (tx.session) {2124            try {2125              return { accepted: true, output: await tx.commitSession() };2126            } catch (err) {2127              if (err instanceof BackendError) throw err.driverError;2128              throw err;2129            }2130          }2131          return {2132            accepted: true,2133            output: await transport.execute({ envelope, statements: [...tx.statements] }),2134          };2135        } catch (error) {2136          if (isRetryableCommitConflict(error) && attempt + 1 < MUTATOR_CONFLICT_MAX_ATTEMPTS) {2137            await mutatorConflictBackoff(attempt);2138            continue;2139          }2140          // The database definitively and repeatably refused this write (see2141          // `isDeterministicWriteRefusal`). Throwing here would wedge the client's queue behind a2142          // poison message forever, so demote it to a BUSINESS rejection: roll the session back,2143          // then advance `lmid` alone. Same shape as an authorization refusal — the write does not2144          // land, the prediction retires, `onRejected` carries the database's own reason.2145          if (isDeterministicWriteRefusal(error)) {2146            const reason = driverRefusalReason(error);2147            // Data first, watermark second, as in the in-body rejection path above.2148            await tx.rollbackSessionQuietly();2149            const output = await transport.reject({ envelope, reason });2150            return { accepted: false, reason, output };2151          }2152          throw error;2153        }2154      }2155      throw new Error("unreachable mutator conflict retry loop");2156    },2157    reject({ envelope, reason }) {2158      return transport.reject({ envelope, reason });2159    },2160  };2161}21622163/** Legacy/private-plane adapter. Kept for existing deployments; its mutation policy is shared with2164 * {@link sqlBackend}, so the two transports cannot drift. */2165export function daemonBackend(daemon: RindleDaemonClient): MutationBackend {2166  return remoteMutationBackend(daemonMutationTransport(daemon), daemonOutsideSql(daemon));2167}21682169/** Run API-server mutators through `@rindle/sql-client`'s explicit mutation facade. Query leases,2170 * SSR reads and room control continue to use `daemon`; only authoritative mutation execution moves2171 * to the versioned SQL transport. */2172export function sqlBackend(sql: SqlSession): MutationBackend {2173  return remoteMutationBackend(sqlClientMutationTransport(sql), sqlSessionOutsideSql(sql));2174}21752176/** The query surface a {@link PostgresPlugger} transaction exposes. `exec` runs one statement;2177 *  `query` returns rows keyed by column name (read-your-own-writes inside the txn). */2178export interface PgQuery {2179  exec(sql: string, params?: unknown[]): Promise<void>;2180  query(sql: string, params?: unknown[]): Promise<Array<Record<string, unknown>>>;2181}21822183/** The thin driver adapter that keeps `pg` / `postgres.js` out of this package's dependencies2184 *  (`BYO-POSTGRES-LMID-CONTRACT-DESIGN.md` §6.2): run `fn` inside ONE transaction — commit on2185 *  resolve, roll back on throw. {@link pgPoolPlugger} adapts a node-postgres `Pool`. */2186export interface PostgresPlugger {2187  transaction<T>(fn: (q: PgQuery) => Promise<T>): Promise<T>;2188}21892190export interface PostgresBackendOptions {2191  /** Rewrite each MUTATOR statement's SQL before it runs (never the lmid upsert). The intended2192   *  use is dialect bridging for a dual-topology app whose mutators are written SQLite-style:2193   *  pass {@link questionToDollarParams} to convert `?` placeholders to `$1..$n`. */2194  rewriteSql?: (sql: string) => string;2195}21962197/** The §2.3 upsert, verbatim from the contract: monotonic via GREATEST, keyed by client. The2198 *  identifiers are lowercase so quoting is cosmetic, but quote-everything is the repo's PG rule. */2199const LMID_UPSERT = `INSERT INTO "_rindle_client_mutations" ("client_id", "last_mutation_id")2200VALUES ($1, $2)2201ON CONFLICT ("client_id") DO UPDATE2202  SET "last_mutation_id" = GREATEST("_rindle_client_mutations"."last_mutation_id", EXCLUDED."last_mutation_id")`;22032204/**2205 * The BYO-Postgres {@link MutationBackend} (`BYO-POSTGRES-LMID-CONTRACT-DESIGN.md` §6.3): one PG2206 * transaction runs the mutator's statements and ALWAYS upserts `_rindle_client_mutations` —2207 * the upsert sits outside any acceptance guard by construction, so the §2.4 footgun (a rejection2208 * that forgets to advance `lmid` and wedges the client's pending queue) cannot be written.2209 *2210 * Confirmation does NOT come from this call's response: the lmid row rides the same PG commit2211 * through CDC → relay → follower and reaches the client in the same coherent release as the2212 * data (§8.2 relocated upstream). A rejection's `reason` still returns on the HTTP reply, but2213 * nothing rejection-shaped travels the replication path — the optimistic prediction snaps back2214 * when the advanced `lmid` arrives.2215 */2216export function postgresBackend(plugger: PostgresPlugger, opts: PostgresBackendOptions = {}): MutationBackend {2217  const rewrite = opts.rewriteSql ?? ((sql: string) => sql);2218  // A tagged business rejection escaping `plugger.transaction` — the plugger rolls the data back on2219  // any throw; this marker distinguishes "mutator said no" (reject) from an infra failure (rethrow).2220  class RejectSignal extends Error {}2221  const lmidOnly = (envelope: MutationEnvelope): Promise<SqlTxnOutput> =>2222    plugger.transaction(async (q) => {2223      await q.exec(LMID_UPSERT, [envelope.clientID, envelope.mid]);2224      return { applied: true, lmidAdvances: [{ clientID: envelope.clientID, lmid: envelope.mid }] };2225    });2226  const outsideSql: ServerSql = {2227    async execute(sql, params = []) {2228      await plugger.transaction(async (q) => {2229        await q.exec(rewrite(sql), [...params]);2230      });2231    },2232    async batch(statements) {2233      if (statements.length === 0) return;2234      await plugger.transaction(async (q) => {2235        for (const statement of statements) {2236          await q.exec(rewrite(statement.sql), statement.params === undefined ? [] : [...statement.params]);2237        }2238      });2239    },2240    query<Row = Record<string, unknown>>(sql: string, params: readonly WireValue[] = []): Promise<Row[]> {2241      return plugger.transaction(async (q) => (await q.query(rewrite(sql), [...params])) as Row[]);2242    },2243  };2244  return {2245    dialect: postgresDialect,2246    outsideSql,2247    async runMutation({ envelope, render, run }) {2248      try {2249        const output = await plugger.transaction(async (q) => {2250          const tx = new PgLiveTx(q, render, rewrite);2251          try {2252            await run(tx);2253            await tx.settle(); // drain any un-awaited queued writes before the lmid stamp2254          } catch (err) {2255            if (err instanceof BackendError) throw err; // infra → rollback + propagate2256            throw new RejectSignal(errMessage(err)); // business → rollback data, tag for §2.42257          }2258          // ALWAYS on the accepted path, SAME transaction (§2.2): the lmid upsert commits with data.2259          await q.exec(LMID_UPSERT, [envelope.clientID, envelope.mid]);2260          return { applied: true, lmidAdvances: [{ clientID: envelope.clientID, lmid: envelope.mid }] };2261        });2262        return { accepted: true, output };2263      } catch (err) {2264        if (err instanceof RejectSignal) {2265          // Data rolled back; STILL advance lmid alone (§2.4 — the client's queue must drain).2266          return { accepted: false, reason: err.message, output: await lmidOnly(envelope) };2267        }2268        if (err instanceof BackendError) throw err.driverError; // infra2269        throw err;2270      }2271    },2272    reject: ({ envelope }) => lmidOnly(envelope).then(() => undefined),2273  };2274}22752276/** The slice of a node-postgres `Pool` the plugger needs — structural, so `pg` stays a2277 *  dependency of the APP, never of this package. */2278export interface PgPoolLike {2279  connect(): Promise<{2280    query(sql: string, params?: unknown[]): Promise<{ rows: Array<Record<string, unknown>> }>;2281    release(err?: unknown): void;2282  }>;2283}22842285/** Adapt a node-postgres `Pool` (or anything pool-shaped) to a {@link PostgresPlugger}:2286 *  one client per transaction, `BEGIN`/`COMMIT` bracketing, `ROLLBACK` + rethrow on failure. */2287export function pgPoolPlugger(pool: PgPoolLike): PostgresPlugger {2288  return {2289    async transaction<T>(fn: (q: PgQuery) => Promise<T>): Promise<T> {2290      const client = await pool.connect();2291      try {2292        await client.query("BEGIN");2293        const q: PgQuery = {2294          exec: async (sql, params) => {2295            await client.query(sql, params);2296          },2297          query: async (sql, params) => (await client.query(sql, params)).rows,2298        };2299        const out = await fn(q);2300        await client.query("COMMIT");2301        return out;2302      } catch (err) {2303        try {2304          await client.query("ROLLBACK");2305        } catch {2306          // the connection may already be unusable; the original error is the one that matters2307        }2308        throw err;2309      } finally {2310        client.release();2311      }2312    },2313  };2314}23152316/**2317 * Rewrite SQLite-style `?` positional placeholders to Postgres `$1..$n`, for mutators written2318 * once and run against either backend (pass as {@link PostgresBackendOptions.rewriteSql}).2319 * Skips `'…'` string literals (with `''` escapes), `"…"` quoted identifiers, `--` line comments,2320 * and non-nested C-style block comments. Do not mix `?` and `$n` styles in one statement.2321 */2322export function questionToDollarParams(sql: string): string {2323  let out = "";2324  let n = 0;2325  let i = 0;2326  while (i < sql.length) {2327    const c = sql[i];2328    if (c === "?") {2329      out += `$${++n}`;2330      i += 1;2331    } else if (c === "'" || c === '"') {2332      // consume the quoted span; a doubled quote is an escape inside it2333      const quote = c;2334      let j = i + 1;2335      while (j < sql.length) {2336        if (sql[j] === quote) {2337          if (sql[j + 1] === quote) j += 2;2338          else break;2339        } else {2340          j += 1;2341        }2342      }2343      out += sql.slice(i, j + 1);2344      i = j + 1;2345    } else if (c === "-" && sql[i + 1] === "-") {2346      const end = sql.indexOf("\n", i);2347      const j = end === -1 ? sql.length : end;2348      out += sql.slice(i, j);2349      i = j;2350    } else if (c === "/" && sql[i + 1] === "*") {2351      const end = sql.indexOf("*/", i + 2);2352      const j = end === -1 ? sql.length : end + 2;2353      out += sql.slice(i, j);2354      i = j;2355    } else {2356      out += c;2357      i += 1;2358    }2359  }2360  return out;2361}23622363export function defineApiQueries<User, Q extends ApiQueries<User>>(queries: Q): Q {2364  return queries;2365}23662367/**2368 * Register a list of co-located client {@link NamedQuery `defineQuery`} values as the server's query2369 * surface — the bulk, no-boilerplate counterpart to {@link defineApiQueries}. Each query already2370 * carries its wire `name` and a `resolve` that re-runs its validator on the UNTRUSTED wire args and2371 * builds the authoritative `Query`, so the server just imports every co-located query and hands the2372 * list here. The same validated args build a byte-identical AST on both tiers.2373 *2374 * The query's AUTHORITATIVE {@link ApiContext} is forwarded as `resolve`'s ctx — so a context-scoped2375 * `defineQuery` (e.g. "my issues") is built from the server's trusted principal, never the client's.2376 * A context-free query simply ignores the extra argument.2377 *2378 * Use {@link defineApiQueries} instead (or in addition) only when the server must DIVERGE from the2379 * client — define a server-specific `defineQuery` with the same name and register that.2380 *2381 * ```ts2382 * queries: registerQueries<User>([issuesPageQuery, issueDetailQuery, recentCommentsQuery, usersQuery]),2383 * ```2384 */2385export function registerQueries<User>(queries: readonly NamedQuery<any, any, any>[]): ApiQueries<User> {2386  const out: Record<string, ApiQuery<User, any>> = {};2387  for (const query of queries) {2388    if (Object.prototype.hasOwnProperty.call(out, query.queryName)) {2389      throw new Error(`registerQueries: duplicate query name "${query.queryName}"`);2390    }2391    const wrapped: ApiQuery<User, any> = (ctx, args) => query.resolve(args, ctx);2392    // The §2.1 realtime label survives this seam (read it back with {@link queryRealtimeLabel}) —2393    // the lease path looks up (room profile, args mapping) by query name. Unlabeled queries get2394    // the exact bare wrapper they always did.2395    out[query.queryName] = query.realtime === undefined ? wrapped : attachRealtimeLabel(wrapped, query.realtime);2396  }2397  return out;2398}23992400export function defineApiMutators<User, M extends ApiMutators<User>>(mutators: M): M {2401  return mutators;2402}24032404/**2405 * Bulk-register a SHARED (generator) mutator registry as server mutators — the mutator twin of2406 * {@link registerQueries} (which does the same for co-located `defineQuery` values). Each shared2407 * mutator carries its own arg validator (`shared(schema, gen)`), so this wraps every one with the2408 * UNIVERSAL server triad and nothing else: parse the UNTRUSTED wire args (its `.args`), map the server2409 * {@link MutationContext} to the shared {@link MutatorCtx} principal, and drive the SAME body the2410 * client predicts ({@link runSharedMutation}). The point is that a shared mutator whose server run2411 * adds NO authority beyond that triad needs no hand-written wrapper.2412 *2413 * Server-only AUTHORITY the client cannot predict (a title guard, an owner-gated cascade, a2414 * `NOT EXISTS` dedup) stays an explicit {@link ApiMutator} that OVERRIDES the auto-wrapped default —2415 * spread this first, then the overrides win by key:2416 *2417 * ```ts2418 * mutators: defineApiMutators({2419 *   ...sharedApiMutators(sharedMutators, (ctx) => ({ user: requireUser(ctx.user) })),2420 *   createIssue: withTitleGuard(sharedMutators.createIssue), // + server-only policy2421 *   deleteIssue: async (tx, raw, ctx) => { ... },            // raw owner-gated cascade2422 * }),2423 * ```2424 */2425export function sharedApiMutators<User>(2426  registry: Record<string, SharedMutatorWithArgs<any>>,2427  principal: (ctx: MutationContext<User>) => MutatorCtx,2428): ApiMutators<User> {2429  const out: ApiMutators<User> = {};2430  for (const [name, mutator] of Object.entries(registry)) {2431    out[name] = (tx, raw, ctx) => runSharedMutation(mutator, mutator.args.parse(raw), principal(ctx), tx);2432  }2433  return out;2434}24352436/**2437 * Wrap a SHARED (generator) mutator with a row-level ACCESS GUARD — the multi-tenant authz twin of2438 * {@link sharedApiMutators}. It parses the untrusted wire args, derives the {@link MutatorCtx}2439 * principal (the SAME mapping you pass to `sharedApiMutators`), evaluates `predicate` against the OPEN2440 * mutation txn (so it can READ the rows the write depends on), and throws `forbidden` (403 — the2441 * client's optimistic write snaps back) when access is denied; otherwise it drives the SAME body the2442 * client predicts ({@link runSharedMutation}). Use it for the entries that need server-only authority2443 * the client cannot predict, OVERRIDING the auto-wrapped default (spread `sharedApiMutators(...)`2444 * first, then the guarded overrides win by key):2445 *2446 * ```ts2447 * const principal = (ctx) => ({ user: requireUser(ctx.user) });2448 * mutators: defineApiMutators({2449 *   ...sharedApiMutators(sharedMutators, principal),2450 *   updateSlide: guardMutator(sharedMutators.updateSlide, principal,2451 *     async (tx, a, { user }) =>2452 *       (await tx.query(q.slide.where.id(a.slideId).where(editableBy(user)).one())) != null,2453 *     { message: "not permitted to edit this slide" }),2454 * }),2455 * ```2456 *2457 * The predicate keeps the shared body READ-FREE, so the client's `.folded` hot paths (drag/keystroke)2458 * still fold — the read is server-side only. Return `false` to deny (→ the default or `opts.message`2459 * forbidden); return `true`/nothing to allow. To reject with a different status/message (a business2460 * rejection, a not-found), throw a {@link RindleApiError} from inside the predicate instead. `principal`2461 * runs before the predicate, so it too may throw `forbidden` for an anonymous caller.2462 */2463export function guardMutator<User, Args>(2464  gen: SharedMutatorWithArgs<Args>,2465  principal: (ctx: MutationContext<User>) => MutatorCtx,2466  predicate: (tx: ServerMutationTx, args: Args, ctx: MutatorCtx) => boolean | void | Promise<boolean | void>,2467  opts?: { message?: string },2468): ApiMutator<User, unknown> {2469  return async (tx, raw, ctx) => {2470    const args = gen.args.parse(raw);2471    const pctx = principal(ctx);2472    if ((await predicate(tx, args, pctx)) === false) {2473      throw new RindleApiError("forbidden", opts?.message ?? "not permitted", 403);2474    }2475    return runSharedMutation(gen, args, pctx, tx);2476  };2477}24782479/** One exemplar invocation for {@link dumpQueryShapes} — the `args`/`user` a query is built with.2480 *  Literal values never matter to the dump (shapes are deduped with literals stripped); what an2481 *  exemplar buys is BRANCH coverage, so supply one per code path a query function can take2482 *  (an optional filter present/absent, each enum axis, …). */2483export interface ShapeExemplar<User = unknown> {2484  args?: unknown;2485  user?: User;2486}24872488/** The query-shapes document `rindle indices suggest` consumes: the app's synced tables (name +2489 *  primary key) and one wire AST per structurally distinct shape a registered query can build. */2490export interface QueryShapesDoc {2491  tables: Array<{ name: string; primaryKey: string[] }>;2492  queries: Array<{ name: string; ast: Ast }>;2493}24942495/**2496 * Dump every registered named query's wire AST — feeder 1 ("exemplar enumeration") of2497 * `rindle indices suggest` (docs/INDEXING.md applied mechanically to the query set).2498 *2499 * Because named queries are FUNCTIONS of `(args, ctx)`, one query can build structurally2500 * different ASTs on different args; each exemplar invocation contributes its shape, and shapes2501 * that differ only in literal values (a limit, a filter string) dedupe to one entry. A query2502 * with no configured exemplars is invoked once with no args. The registry is the app's whole2503 * server-side query surface, so the resulting document is the complete static shape set —2504 * modulo arg-value-dependent branches, which need an exemplar (or runtime shape recording) to2505 * surface.2506 */2507export async function dumpQueryShapes<User>(opts: {2508  schema: Schema;2509  queries: ApiQueries<User>;2510  exemplars?: Partial<Record<string, ReadonlyArray<ShapeExemplar<User>>>>;2511}): Promise<QueryShapesDoc> {2512  const tables = Object.values(opts.schema.tables)2513    // Local-only tables (both `true` and `"session"`) live in the browser's memory source,2514    // never a TableSource — no indexes.2515    .filter((t) => !t.local)2516    .map((t) => ({ name: t.name, primaryKey: [...t.primaryKey] }))2517    .sort((a, b) => a.name.localeCompare(b.name));2518  const queries: QueryShapesDoc["queries"] = [];2519  for (const [name, query] of Object.entries(opts.queries).sort(([a], [b]) => a.localeCompare(b))) {2520    const seen = new Set<string>();2521    for (const ex of opts.exemplars?.[name] ?? [{}]) {2522      const ast = queryResultToAst(await query({ user: ex.user as User }, ex.args));2523      const key = JSON.stringify(normalizeShape(ast));2524      if (seen.has(key)) continue;2525      seen.add(key);2526      queries.push({ name: seen.size > 1 ? `${name}#${seen.size}` : name, ast });2527    }2528  }2529  return { tables, queries };2530}25312532/** The literal-stripped structure of an AST — the dedupe key for {@link dumpQueryShapes}. */2533function normalizeShape(ast: Ast): Record<string, unknown> {2534  return {2535    table: ast.table,2536    where: ast.where && normalizeCondition(ast.where),2537    related: ast.related?.map((r) => ({2538      correlation: r.correlation,2539      subquery: normalizeShape(r.subquery),2540    })),2541    orderBy: ast.orderBy,2542    limit: ast.limit !== undefined,2543    start: ast.start2544      ? { keys: Object.keys(ast.start.row).sort(), exclusive: ast.start.exclusive }2545      : undefined,2546    aggregate: ast.aggregate,2547    groupBy: ast.groupBy,2548    having: ast.having && normalizeCondition(ast.having),2549    one: ast.one,2550  };2551}25522553function normalizeCondition(c: Condition): unknown {2554  switch (c.type) {2555    case "simple":2556      return {2557        type: c.type,2558        op: c.op,2559        left: c.left,2560        right: c.right.type === "literal" ? { type: "literal" } : c.right,2561      };2562    case "and":2563    case "or":2564      return { type: c.type, conditions: c.conditions.map(normalizeCondition) };2565    case "correlatedSubquery":2566      return {2567        type: c.type,2568        op: c.op,2569        related: {2570          correlation: c.related.correlation,2571          subquery: normalizeShape(c.related.subquery),2572        },2573      };2574  }2575}25762577/** Keep outside-SQL driver failures on the infrastructure path even when they happen before the2578 * scoped mutator has opened its mutation transaction. */2579function scopedOutsideSql(sql: ServerSql | undefined): ServerSql {2580  const unavailable = (): BackendError =>2581    new BackendError(new Error("scope.sql is unavailable on this custom MutationBackend"));2582  // An unencodable bind is a deterministic authoring error, not a database failure. Wrapping it in2583  // BackendError would latch `scope.infra` and retry the envelope forever; leaving it a plain throw2584  // lets the scoped harness treat it as a business rejection and advance lmid.2585  const infra = (error: unknown): unknown =>2586    isUnencodableBind(error) ? new Error(errMessage(error)) : error instanceof BackendError ? error : new BackendError(error);2587  return {2588    async execute(text, params = []) {2589      if (!sql) throw unavailable();2590      try {2591        await sql.execute(text, params);2592      } catch (error) {2593        throw infra(error);2594      }2595    },2596    async batch(statements) {2597      if (!sql) throw unavailable();2598      try {2599        await sql.batch(statements);2600      } catch (error) {2601        throw infra(error);2602      }2603    },2604    async query<Row = Record<string, unknown>>(text: string, params: readonly WireValue[] = []): Promise<Row[]> {2605      if (!sql) throw unavailable();2606      try {2607        return await sql.query<Row>(text, params);2608      } catch (error) {2609        throw infra(error);2610      }2611    },2612  };2613}26142615/**2616 * The runtime {@link MutationScope} handed to a {@link ScopedMutator}. It owns the single atomic2617 * transaction (delegating to {@link MutationBackend.runMutation} — the exact machinery a tx-form2618 * mutator uses), but lets the AUTHOR decide when it opens, so server-only work can run outside it.2619 *2620 * It records its outcome so the harness — not the author — enforces the `lmid`-always-advances2621 * invariant: `phase` reports whether the tx committed, business-rejected, or never ran, and `infra`2622 * latches a backend (DB) failure. Because the backend's `runMutation` RETURNS `{accepted:false}` for2623 * a business rejection (having already advanced `lmid` alone) and THROWS only for infra, `transact`2624 * can cleanly re-throw {@link MutationRejected} on the former (for author compensation) and propagate2625 * the raw driver error on the latter.2626 */2627class MutationScopeImpl implements MutationScope {2628  private attempted = false;2629  private readonly backend: MutationBackend;2630  private readonly envelope: MutationEnvelope;2631  private readonly render: RenderIndex;2632  readonly sql: ServerSql;2633  /** Set once `transact` resolved through the backend (accepted OR business-rejected). */2634  outcome?: MutationOutcome;2635  /** The value the backend threw on INFRA (the DB failed) — always propagated, never an `lmid`2636   *  advance. Its presence is tracked by {@link infraLatched}, NOT by testing this for `undefined`:2637   *  a driver may legitimately reject with a falsy value, and misreading that as "no infra" would2638   *  reclassify a lost-connection failure as a business rejection and wrongly advance `lmid`. */2639  infra?: unknown;2640  /** True once an INFRA failure latched, regardless of its (possibly falsy) value. */2641  infraLatched = false;2642  /** The in-flight `transact` promise. `settle` awaits it before sealing, so a transact the author2643   *  FORGOT to await (a floating promise — nothing here lints against it) is still resolved to its2644   *  real outcome first; otherwise the seal would read `untouched`, reply with a phantom no-op, and2645   *  let the real write commit out-of-band after the response was already sent. */2646  pending?: Promise<void>;26472648  constructor(backend: MutationBackend, envelope: MutationEnvelope, render: RenderIndex) {2649    this.backend = backend;2650    this.envelope = envelope;2651    this.render = render;2652    this.sql = scopedOutsideSql(backend.outsideSql);2653  }26542655  transact(2656    first: SharedMutator<any, any> | ((tx: ServerMutationTx) => unknown),2657    args?: unknown,2658    ctx?: MutatorCtx,2659  ): Promise<any> {2660    if (this.attempted) throw new Error("scope.transact may be called at most once per mutation");2661    this.attempted = true;2662    const promise = this.drive(first, args, ctx);2663    // Record the in-flight promise so `settle` can await it even when the author didn't. Errors are2664    // latched onto `this` (outcome / infra), so this tracking copy swallows them — the author's2665    // returned `promise` still rejects for them to await/catch.2666    this.pending = promise.then(2667      () => undefined,2668      () => undefined,2669    );2670    return promise;2671  }26722673  private async drive(2674    first: SharedMutator<any, any> | ((tx: ServerMutationTx) => unknown),2675    args?: unknown,2676    ctx?: MutatorCtx,2677  ): Promise<unknown> {2678    // The callback form's return value, captured INSIDE the transaction and handed back only below,2679    // after the commit is confirmed accepted — so a value can never describe work that rolled back.2680    let value: unknown;2681    // A shared (generator) mutator is driven via the isomorphic seam; a plain callback gets the raw tx.2682    const run = isGeneratorMutator(first)2683      ? async (tx: ServerMutationTx) => {2684          await runSharedMutation(first as SharedMutator<unknown, MutatorCtx>, args, ctx as MutatorCtx, tx);2685        }2686      : async (tx: ServerMutationTx) => {2687          value = await (first as (tx: ServerMutationTx) => unknown)(tx);2688        };2689    let outcome: MutationOutcome;2690    try {2691      outcome = await this.backend.runMutation({ envelope: this.envelope, render: this.render, run });2692    } catch (err) {2693      this.infra = err; // the backend throws ONLY for infra; a business rejection returns {accepted:false}2694      this.infraLatched = true;2695      throw err;2696    }2697    this.outcome = outcome;2698    if (!outcome.accepted) throw new MutationRejected(outcome.reason);2699    return value;2700  }2701}27022703/** Resolve {@link RindleApiServerOptions.rindle} into concrete `daemon` + `database` fields (env2704 *  fallback, explicit fields win), and assert the one hard requirement: SOME control-plane client2705 *  must exist. Returned options carry a non-optional `daemon`, which the rest of construction2706 *  relies on. */2707function withResolvedConnection<User>(2708  opts: RindleApiServerOptions<User>,2709): RindleApiServerOptions<User> & { daemon: RindleDaemonClient } {2710  const connection = opts.rindle;2711  if (connection === undefined) {2712    if (opts.daemon === undefined) {2713      throw new TypeError(2714        "createRindleApiServer needs a connection: pass `rindle: { url, token }` (or `rindle: {}` " +2715          "under `rindle dev`), or an explicit `daemon` client",2716      );2717    }2718    return opts as RindleApiServerOptions<User> & { daemon: RindleDaemonClient };2719  }2720  // Determine which legs actually need deriving BEFORE requiring an endpoint: when every leg is2721  // explicitly configured, `rindle` is inert and must not demand a unified URL — explicit fields2722  // always win. The database leg counts as covered by any explicit SQL machinery (`database`,2723  // `sql`, or `backend`) — an explicit session or backend must not gain a second, owned SQL2724  // client beside it.2725  const needsDaemon = opts.daemon === undefined;2726  const needsDatabase =2727    opts.database === undefined && opts.sql === undefined && opts.backend === undefined;2728  if (!needsDaemon && !needsDatabase) {2729    return opts as RindleApiServerOptions<User> & { daemon: RindleDaemonClient };2730  }2731  // `process` is absent on some serverless runtimes; the env fallback simply does not apply there.2732  const env = typeof process !== "undefined" ? process.env : undefined;2733  const url = connection.url ?? env?.RINDLE_URL;2734  if (url === undefined || url === "") {2735    throw new TypeError(2736      "rindle.url is required: pass it explicitly, or run under `rindle dev`, which exports " +2737        "RINDLE_URL once the rendered topology serves one ingress",2738    );2739  }2740  const token = connection.token ?? env?.RINDLE_DATABASE_TOKEN;2741  if (token === undefined || token === "") {2742    throw new TypeError(2743      "rindle.token is required to derive the unified connection: pass it explicitly, run under " +2744        "`rindle dev` (which exports RINDLE_DATABASE_TOKEN), or configure `database`/`sql`/" +2745        "`backend` and `daemon` yourself",2746    );2747  }2748  const daemon =2749    opts.daemon ??2750    new HttpRindleDaemonClient({2751      baseUrl: url,2752      headers: { authorization: `Bearer ${token}` },2753    });2754  if (!needsDatabase) {2755    return { ...opts, daemon };2756  }2757  return { ...opts, daemon, database: { url, authToken: token } };2758}27592760/** Resolve the public subscription endpoint carried by query leases. This is intentionally derived2761 *  only when the unified `rindle` model is configured; a caller supplying only a custom/split2762 *  daemon retains the legacy wire shape. */2763function resolveRindleWsEndpoint(2764  connection: RindleConnectionOptions | undefined,2765  deriveFromOrigin: boolean,2766): string | undefined {2767  if (connection === undefined) return undefined;2768  if (connection.wsUrl !== undefined && connection.wsUrl !== "") return connection.wsUrl;2769  if (!deriveFromOrigin) return undefined;2770  const env = typeof process !== "undefined" ? process.env : undefined;2771  const origin = connection.url ?? env?.RINDLE_URL;2772  if (origin === undefined || origin === "") return undefined;2773  const url = new URL(origin);2774  if (url.protocol === "http:") url.protocol = "ws:";2775  else if (url.protocol === "https:") url.protocol = "wss:";2776  else {2777    throw new TypeError(`rindle.url must use http: or https: to derive a WebSocket endpoint (got ${url.protocol})`);2778  }2779  const rendered = url.toString();2780  return url.pathname === "/" && url.search === "" && url.hash === "" ? rendered.replace(/\/$/, "") : rendered;2781}27822783export function createRindleApiServer<User = unknown>(options: RindleApiServerOptions<User>): RindleApiServer<User> {2784  const opts = withResolvedConnection(options);2785  // An origin-derived socket belongs only to an origin-derived daemon. If the caller supplied a2786  // split/custom daemon, its leases must not advertise the unrelated SQL origin; only an explicit2787  // wsUrl is sufficiently intentional to accompany that override.2788  const wsEndpoint = resolveRindleWsEndpoint(opts.rindle, options.daemon === undefined);2789  const routes = { ...DEFAULT_RINDLE_API_ROUTES, ...opts.routes };2790  const mode = opts.mode ?? "normalized";2791  // Explicit backend wins; otherwise prefer the versioned Rindle-SQL mutation transport and retain2792  // daemonBackend as the compatibility path for deployments that have not exposed it yet.2793  let ownedSql: SqlClient | undefined;2794  let backend: MutationBackend;2795  if (opts.backend !== undefined) {2796    backend = opts.backend;2797  } else {2798    if (opts.database !== undefined && opts.sql !== undefined) {2799      throw new TypeError("configure either database or sql, not both");2800    }2801    const sql =2802      opts.sql ??2803      (opts.database !== undefined2804        ? // Default FIRST so `database.intMode` can override it; see RindleDatabaseOptions.2805          (ownedSql = createSqlClient({ intMode: "number", ...opts.database }))2806        : undefined);2807    backend = sql === undefined ? daemonBackend(opts.daemon) : sqlBackend(sql);2808  }2809  // Schema-derived render metadata for logical mutator writes; `{}` when no schema is configured (a2810  // logical op then throws loudly — the tx never silently drops a write). Each backend renders in its2811  // own dialect (`backend.dialect`: daemon→sqlite, postgres→postgres).2812  const renderIndex: RenderIndex = opts.schema ? buildRenderIndex(opts.schema) : {};2813  // Names that are ALSO configured pins — a lease for one is forced to a `pinned` policy (the lazy2814  // floor, §4.1) so the first viewer to route to a follower warms it for late joiners.2815  const pinnedNames = new Set((opts.pinnedQueries ?? []).map((p) => p.name));28162817  // Rindle Realtime declaration layer (RINDLE-REALTIME-QUERY-ENABLEMENT §2, slice G-iv-a):2818  // compile the named room profiles and run every "loud at registration" (§2.3) check NOW —2819  // construction is the moment a misconfigured profile or label can still fail the deploy,2820  // not a 3am room boot. The legacy flat `resolveFootprint` stays the anonymous profile and2821  // is deliberately NOT probed or validated (byte-identical legacy behavior).2822  const realtime = opts.realtime;2823  const roomProfiles = compileRoomProfiles<User>({2824    rooms: realtime?.rooms,2825    schema: opts.schema,2826    warn: realtime?.warn,2827  });2828  assertLabeledProfilesExist(opts.queries, roomProfiles);2829  if (realtime !== undefined && roomProfiles.size === 0 && realtime.resolveFootprint === undefined) {2830    throw new Error(2831      "realtime: configure at least one room profile (realtime.rooms) or the legacy resolveFootprint — " +2832        "a realtime host with neither can never boot a room.",2833    );2834  }28352836  // Resolve a named query (+ args) to its AST under a given context — the shared path for both2837  // a per-viewer lease and a system-level pin (which skips per-user authorization).2838  const resolveAst = async (name: string, args: unknown, context: ApiContext<User>): Promise<Ast> => {2839    const query = opts.queries?.[name];2840    if (!query) throw new RindleApiError("not-found", `unknown query: ${name}`, 404);2841    const result = opts.runQuery2842      ? await opts.runQuery({ user: context.user, name, args, query, context })2843      : await query(context, args as never);2844    return queryResultToAst(result);2845  };28462847  // ---------------------------------------------------------------- the room-serve decision2848  //2849  // RINDLE-REALTIME-QUERY-ENABLEMENT §2.1 lease-flow steps 2–5, slice G-iv-b. Everything here is2850  // FAIL-OPEN: any missing wiring, refused proof, or thrown error means the lease is served from2851  // the daemon EXACTLY as today (no `realtime` block, top-level fields untouched) plus a one-time2852  // diagnostic — a coverage/config problem must never block a lease.28532854  const realtimeWarn = realtime?.warn ?? ((message: string) => console.warn(message));2855  // One-time per (queryName, profile): the serve decision runs on EVERY lease, so an uncovered2856  // labeled query would otherwise warn once per viewer per mount.2857  const warnedRoomServe = new Set<string>();2858  const warnRoomServeOnce = (queryName: string, profile: string, reasons: readonly string[]): void => {2859    const key = `${queryName}\u0000${profile}`;2860    if (warnedRoomServe.has(key)) return;2861    warnedRoomServe.add(key);2862    realtimeWarn(2863      `query "${queryName}" is labeled realtime (room profile "${profile}") but is NOT room-served — ` +2864        `${reasons.join("; ")}. It serves from the daemon (correct, just not room-accelerated). ` +2865        `This warning fires once per (query, profile).`,2866    );2867  };28682869  // Room-served aggregates are refused: a room-retargeted query carrying a count()/reduce reads an2870  // `__agg` head only the daemon feed maintains, and the client's room gate DROPS the `__agg` rows2871  // the room publishes — a known-unsupported shape (302 post-impl review). Room serving otherwise2872  // trusts the declaration (302 §5: declared, not derived); this one shape stays a policy refusal2873  // until room-served aggregates are designed.2874  const AGGREGATE_REFUSAL =2875    "the query contains an aggregate/reduce shape — room-served aggregates are not yet supported (the room gate drops `__agg` rows)";28762877  const maybeRoomServe = async (2878    input: QueryLeaseRequest<User>,2879    queryAst: Ast,2880    context: ApiContext<User>,2881    subject: string | undefined,2882  ): Promise<QueryLeaseRealtime | undefined> => {2883    // (a) the label + (b) its profile — the fast bail keeps unlabeled leases byte-identical.2884    const label = queryRealtimeLabel(opts.queries?.[input.name]);2885    if (label === undefined) return undefined;2886    const profile = roomProfiles.get(label.room);2887    if (profile === undefined) return undefined; // unreachable: construction asserted it exists2888    try {2889      // (c) the wiring gates — each absence fail-opens with a one-time, named reason.2890      if (realtime?.locateRoom === undefined) {2891        warnRoomServeOnce(input.name, profile.name, ["realtime.locateRoom is not configured"]);2892        return undefined;2893      }2894      const tokenKey = realtime.roomTokenKey;2895      if (tokenKey === undefined) {2896        warnRoomServeOnce(input.name, profile.name, [2897          "realtime.roomTokenKey is not configured — the room lease token cannot be signed",2898        ]);2899        return undefined;2900      }2901      // The token's subject: the same resolved subject the daemon lease carries, else the2902      // browser's clientId. The shell refuses a subject-less token, so with neither we fail open.2903      const sub = subject ?? input.clientId;2904      if (sub === undefined) {2905        warnRoomServeOnce(input.name, profile.name, [2906          "no token subject — configure `subject` (or have the client send clientId)",2907        ]);2908        return undefined;2909      }29102911      // §2.1: (roomProfile, roomArgs) via the label's args mapping; key + doc minted SERVER-side2912      // (input.args just passed the query's own validation inside resolveAst).2913      const roomArgs = label.args !== undefined ? label.args(input.args) : input.args;2914      const key = profile.key(roomArgs);2915      const doc = mintRoomDoc(profile.name, key);29162917      // The profile footprint for THIS key under the request ctx (works for non-static2918      // profiles), with the §2.3 unwindowed backstop `/room-boot` also applies.2919      const footprintAst = queryResultToAst(await profile.footprint(key, context));2920      assertUnwindowedFootprint(footprintAst, profile.name);29212922      // Trust the declaration (302 §5): a labeled + wired query is room-served, no coverage proof.2923      // The one shape still refused is the aggregate (a policy gate, not a coverage verdict).2924      if (astHasAggregate(queryAst)) {2925        warnRoomServeOnce(input.name, profile.name, [AGGREGATE_REFUSAL]);2926        return undefined;2927      }29282929      // Assemble the realtime block. The room endpoint rides ITS OWN field2930      // (`realtime.wsEndpoint`) — a separate connection from the daemon session's fixed ws host.2931      const { wsEndpoint } = await realtime.locateRoom(doc);2932      const now = Date.now();2933      const ttlMs = realtime.roomTokenTtlMs ?? DEFAULT_ROOM_TOKEN_TTL_MS;2934      const { mintRoomToken, scopeSpecsHash } = await loadRoomTokenModule();2935      // The lease-wire specs, hashed ONCE: the same value is stamped on the token (so the2936      // shell can flag scope skew — a profile edited under a live room, whose gate armed2937      // with the OLD specs at boot) and returned as the client's `tables`.2938      const tables = compileRoomTableSpecs(footprintAst, profile.context);2939      const roomToken = await mintRoomToken({2940        doc,2941        ast: queryAst, // the APPROVED resolved AST — the client carries it, it can't mint/alter it2942        sub,2943        kid: tokenKey.kid,2944        key: tokenKey.secret,2945        ttlMs,2946        now,2947        scopesHash: scopeSpecsHash(tables),2948      });2949      return {2950        // `parse_source_key` (rust/rindle/src/wasm/db.rs): anything but the reserved "daemon" is a room2951        // source; the client-store convention is `room:` + the wire doc.2952        sourceKey: `room:${doc}`,2953        wsEndpoint,2954        roomToken,2955        exp: now + ttlMs,2956        doc,2957        tables,2958      };2959    } catch (e) {2960      // Fail open — a lease is never blocked on room-serve wiring (footprint resolution,2961      // locateRoom, token minting). A failure here just serves the query from the daemon.2962      warnRoomServeOnce(input.name, profile.name, [`room-serve failed: ${errMessage(e)}`]);2963      return undefined;2964    }2965  };29662967  // ---------------------------------------------------------------- the §4 lifecycle mint (I-iii)2968  //2969  // Gated on the OPT-IN `realtime.lifecycle` block: absent, this whole section is dead code and2970  // every lease response is byte-identical to pre-lifecycle. Present, a labeled lease gains the2971  // doorbell system lease and a ROOM-SERVED one the fence bundle (see {@link QueryLeaseLifecycle}).2972  // FAIL-OPEN like the room-serve decision: a mint failure (e.g. a daemon that never ran2973  // `enable_realtime_lifecycle`) warns once per query and the lease ships without the block.29742975  const warnedLifecycle = new Set<string>();2976  const warnLifecycleOnce = (queryName: string, reason: string): void => {2977    if (warnedLifecycle.has(queryName)) return;2978    warnedLifecycle.add(queryName);2979    realtimeWarn(2980      `query "${queryName}" is realtime-labeled with lifecycle configured, but its lifecycle ` +2981        `system leases were not minted — ${reason}. The lease serves without the lifecycle block ` +2982        `(correct, just no §4 upgrade/downgrade plane). This warning fires once per query.`,2983    );2984  };29852986  /** THE SCOPE-KEY DECISION (§4.1): the doorbell scope IS the wire room doc — `"<profile>/<key>"`2987   *  via {@link mintRoomDoc}, the same computation `maybeRoomServe` runs (label args mapping →2988   *  `profile.key`) and the same key `locateRoom`/`/room-boot` address the room by. Occupancy2989   *  (I-iv writes the `_rindle_scope_sessions` rows) must be counted on EXACTLY the key the 1→22990   *  transition provisions, and this is that key. Computed independently of the room-serve2991   *  decision on purpose: the doorbell rides every LABELED lease — an uncovered/unwired labeled2992   *  query still counts toward occupancy (its collaborators still want the upgrade). */2993  const lifecycleScopeDoc = (input: QueryLeaseRequest<User>): string | undefined => {2994    const label = queryRealtimeLabel(opts.queries?.[input.name]);2995    if (label === undefined) return undefined; // unlabeled — no scope to count on2996    const profile = roomProfiles.get(label.room);2997    if (profile === undefined) return undefined; // unreachable: construction asserted it exists2998    const roomArgs = label.args !== undefined ? label.args(input.args) : input.args;2999    return mintRoomDoc(profile.name, profile.key(roomArgs));3000  };30013002  const maybeLifecycle = async (3003    input: QueryLeaseRequest<User>,3004    roomServed: boolean,3005    subject: string | undefined,3006    routingKey: string | undefined,3007  ): Promise<QueryLeaseLifecycle | undefined> => {3008    if (realtime?.lifecycle === undefined) return undefined; // the opt-in gate — mint NOTHING3009    try {3010      const doc = lifecycleScopeDoc(input);3011      if (doc === undefined) return undefined;3012      // Each system lease is an ordinary daemon materialization (the room-boot direct pattern),3013      // carrying the SAME subject/routingKey as the primary lease so a routed deploy co-locates3014      // the system streams on the follower the client's daemon session already lives on. The3015      // daemon dedups by canonical query, so N clients' doorbells over one scope share ONE3016      // materialization (each still minting its own leaseToken); the client-scoped fence ASTs3017      // are per-client by construction.3018      const mint = (ast: Ast) =>3019        opts.daemon.materialize({3020          ast,3021          mode,3022          subject,3023          leaseTtlMs: opts.leaseTtlMs,3024          metadata: routingKey !== undefined ? { routingKey } : undefined,3025          // Lifecycle leases are follower-local exactly like the primary lease. Forward the SAME3026          // opaque placement ticket so every doorbell/fence materialization is minted on the3027          // browser socket's follower instead of independently anycasting across the fleet.3028          ...(input.affinity !== undefined ? { affinity: input.affinity } : {}),3029        });3030      const lease = (table: string, out: MaterializeOutput, id: { scope?: string; doc?: string; clientId?: string }): QueryLeaseLifecycleLease => ({3031        table,3032        leaseToken: out.leaseToken,3033        ...(id.scope !== undefined ? { scope: id.scope } : {}),3034        ...(id.doc !== undefined ? { doc: id.doc } : {}),3035        ...(id.clientId !== undefined ? { clientId: id.clientId } : {}),3036      });3037      const lifecycle: QueryLeaseLifecycle = {3038        doorbell: lease(SCOPE_SESSIONS_TABLE, await mint(scopeSessionsAst(doc)), { scope: doc }),3039      };3040      // The fence bundle only where a room domain exists to fence (room-served leases): the3041      // §4.2 watermark, the §7.1 daemon-carried ledger, and the §3.3 outcome rows.3042      if (roomServed) {3043        const clientId = input.clientId;3044        lifecycle.fence = [3045          lease(ROOM_WATERMARK_TABLE, await mint(roomWatermarkAst(doc)), { doc }),3046          lease(ROOM_CLIENT_MUTATIONS_TABLE, await mint(docClientAst(ROOM_CLIENT_MUTATIONS_TABLE, doc, clientId)), { doc, clientId }),3047          lease(ROOM_MUTATION_OUTCOMES_TABLE, await mint(docClientAst(ROOM_MUTATION_OUTCOMES_TABLE, doc, clientId)), { doc, clientId }),3048        ];3049      }3050      return lifecycle;3051    } catch (e) {3052      warnLifecycleOnce(input.name, errMessage(e)); // fail open — a lease is never blocked3053      return undefined;3054    }3055  };30563057  // ------------------------------------------------------------ the §4.1 occupancy gate (I-iv)3058  //3059  // Runs on EVERY labeled lease under the opt-in `realtime.lifecycle` config (mint AND renewal —3060  // both land on this same route), BEFORE the room-serve decision: (1) sweep + upsert the3061  // caller's session row through the normal write path (the write is the doorbell — I-i's CDC3062  // capture fans the row delta to every solo watcher's doorbell subscription), then (2) read the3063  // occupancy count and return the D6 gate verdict `maybeRoomServe` is conditioned on. The upsert3064  // deliberately precedes the count so the caller's own row is on disk when the verdict is3065  // computed (its own presence rides the `+ 1`, and — more importantly — a concurrent second3066  // client's read sees it). Ordering within the pair is otherwise value-neutral: the count3067  // EXCLUDES the caller's clientId and adds the `+ 1` analytically.3068  //3069  // THE RENEWAL-vs-FRESH DECISION (grounded here because the task forces it): this server is3070  // stateless and the lease request carries no "I am currently room-attached" field, so the gate3071  // CANNOT distinguish a fresh mint from a live room's renewal. Instead of gating on the raw3072  // count (which would suppress a momentarily-solo room's renewal and force the loud client-side3073  // downgrade anomaly), the gate applies the §9.1 hysteresis DIRECTLY FROM THE LINGERING ROWS the3074  // D4 sweep preserves: room-serve iff `liveOthers + self ≥ minSessions` OR some other session3075  // expired within `graceMs`. A renewal is therefore never suppressed until the scope has been3076  // solo SUSTAINED past the grace window — which is exactly Slice I-v's downgrade condition, read3077  // from the same rows; I-v replaces that post-grace loud suppression with the fenced downgrade3078  // dance, refining (not re-deciding) this verdict. A truly fresh solo scope (no other row, live3079  // or lingering) is suppressed immediately — the D6 point.3080  //3081  // Timestamps are `Date.now()` server-side throughout (mint, sweep, count): occupancy tolerates3082  // clock skew between api-server instances up to ~grace — a skewed `now` moves a session between3083  // "live" and "in-grace", both of which hold the gate open; only skew past the grace+slack band3084  // could mis-sweep, and the slack exists to keep that band clear.3085  //3086  // A request with NO `clientId` (a non-shipped client — the shipped one always sends it, see3087  // `postLease`) upserts NO row and contributes NOTHING to occupancy, including to its own gate:3088  // it room-serves only if the OTHER sessions alone reach `minSessions` (there is no session3089  // identity to count it under, D7). It still gets its doorbell (`maybeLifecycle` is independent).3090  //3091  // FAIL-OPEN, like every lifecycle surface: an occupancy failure (e.g. a daemon that never ran3092  // `enable_realtime_lifecycle`) warns once per query and returns `true` — the gate falls away3093  // and the lease serves exactly as pre-I-iv. Suppressing on infrastructure failure would turn3094  // realtime off fleet-wide from one missing table; never block, never suppress, on an error.30953096  const warnedOccupancy = new Set<string>();3097  interface LifecycleOccupancy {3098    /** The D6 room-serve gate verdict — `false` ⇒ suppress the room block (solo/uncovered). */3099    gateOpen: boolean;3100    /** The I-v downgrade guard: a room plausibly hosts this scope (some other-session row lives3101     *  or lingers). Only a `!gateOpen && roomPlausible` reply drains — never a never-shared doc. */3102    roomPlausible: boolean;3103    /** The wire doc the scope maps to (`"<profile>/<key>"`); `undefined` when unlabeled / off. */3104    doc: string | undefined;3105  }3106  const lifecycleOccupancy = async (input: QueryLeaseRequest<User>): Promise<LifecycleOccupancy> => {3107    const lc = realtime?.lifecycle;3108    if (lc === undefined) return { gateOpen: true, roomPlausible: false, doc: undefined }; // lifecycle off — the gate does not exist (inert-until-fed)3109    const doc = lifecycleScopeDoc(input);3110    if (doc === undefined) return { gateOpen: true, roomPlausible: false, doc: undefined }; // unlabeled — no scope to count on, nothing to gate3111    try {3112      const now = Date.now();3113      const minSessions = lc.minSessions ?? DEFAULT_LIFECYCLE_MIN_SESSIONS;3114      const graceMs = lc.graceMs ?? DEFAULT_LIFECYCLE_GRACE_MS;3115      const sessionTtlMs = lc.sessionTtlMs ?? opts.leaseTtlMs ?? DEFAULT_SESSION_TTL_MS;3116      const clientId = input.clientId;3117      // (1) sweep + upsert, ONE write txn (D4: the sweep shares the upsert's transaction — no3118      // separate maintenance pass, and the linger bound holds atomically with the refresh).3119      const statements: SqlStatement[] = [3120        { sql: SESSION_SWEEP_SQL, params: [doc, now - (graceMs + SESSION_SWEEP_SLACK_MS)] },3121      ];3122      if (clientId !== undefined) {3123        statements.push({ sql: SESSION_UPSERT_SQL, params: [doc, clientId, now + sessionTtlMs] });3124      }3125      await opts.daemon.executeSqlTxn({ statements });3126      // (2) the count — read-your-writes via `consistency: "strong"` (see the section note above).3127      const read = await opts.daemon.executeSqlRead({3128        sql: clientId !== undefined ? SESSION_COUNT_OTHERS_SQL : SESSION_COUNT_SQL,3129        params:3130          clientId !== undefined3131            ? [now, now, now - graceMs, doc, clientId]3132            : [now, now, now - graceMs, doc],3133        consistency: "strong",3134      });3135      const cells = read.rows[0] ?? [];3136      const liveOthers = Number(cells[0] ?? 0); // SUM over zero rows is NULL — coerce3137      const graceOthers = Number(cells[1] ?? 0);3138      const totalOthers = Number(cells[2] ?? 0); // ALL other rows (any expiry, pre-sweep)3139      const self = clientId !== undefined ? 1 : 0;3140      return {3141        gateOpen: liveOthers + self >= minSessions || graceOthers > 0,3142        // Room plausibly exists ⇒ this scope was shared (a room was provisioned on the 1→2). A3143        // never-shared solo doc has NO other row and must never drain (no wasted room boot).3144        roomPlausible: totalOthers > 0,3145        doc,3146      };3147    } catch (e) {3148      if (!warnedOccupancy.has(input.name)) {3149        warnedOccupancy.add(input.name);3150        realtimeWarn(3151          `query "${input.name}" is realtime-labeled with lifecycle configured, but the §4.1 ` +3152            `occupancy step failed — ${errMessage(e)}. The occupancy gate fail-opens (the lease ` +3153            `serves exactly as pre-I-iv; no session row was counted). This warning fires once per query.`,3154        );3155      }3156      return { gateOpen: true, roomPlausible: false, doc };3157    }3158  };31593160  // ------------------------------------------------------------ the §4.2 downgrade drain (I-v)3161  //3162  // When the occupancy gate closes for a scope a room plausibly hosted, drain that room to a3163  // COMMITTED flush seq and hand it back as the fence. `drainRoom` (deployment-wired to the room3164  // shell / DO `/drain`) is idempotent (concurrent api-server instances may both call it) and3165  // fails OPEN — a downgrade must never block a lease, so an unconfigured or throwing hook simply3166  // omits the fence (warn-once) and the client falls to its loud legacy downgrade path.3167  const warnedDrain = new Set<string>();3168  const warnDrainOnce = (queryName: string, reason: string): void => {3169    if (warnedDrain.has(queryName)) return;3170    warnedDrain.add(queryName);3171    realtimeWarn(3172      `query "${queryName}" downgraded (occupancy gate closed) but no §4.2 fence was attached — ` +3173        `${reason}. The lease ships without the fence; a room-attached client falls back to its ` +3174        `loud legacy downgrade (correct, just not graceful). This warning fires once per query.`,3175    );3176  };3177  const maybeDrainRoom = async (queryName: string, doc: string): Promise<QueryLeaseRealtimeFence | undefined> => {3178    const drainRoom = realtime?.lifecycle?.drainRoom;3179    if (drainRoom === undefined) {3180      warnDrainOnce(queryName, "realtime.lifecycle.drainRoom is not configured");3181      return undefined;3182    }3183    try {3184      const { finalFlushSeq } = await drainRoom(doc);3185      return { sourceKey: `room:${doc}`, doc, finalFlushSeq };3186    } catch (e) {3187      warnDrainOnce(queryName, `drainRoom threw — ${errMessage(e)}`);3188      return undefined;3189    }3190  };31913192  const createQueryLease = async (input: QueryLeaseRequest<User>): Promise<QueryLeaseResponse> => {3193    const context: ApiContext<User> = { user: input.user, request: input.request };3194    await assertAuthorized(opts.authorizeQuery, {3195      user: input.user,3196      name: input.name,3197      args: input.args,3198      context,3199    });3200    const ast = await resolveAst(input.name, input.args, context);3201    // Lazy pin tier (§4.1): a leased query that is ALSO a configured pin is materialized with a3202    // `pinned` policy regardless of the configured default, so it survives its first viewer's3203    // departure. Otherwise use the configured policy.3204    const policy = pinnedNames.has(input.name)3205      ? ({ kind: "pinned", name: input.name } as MaterializationPolicy)3206      : await resolvePolicy(opts.materializationPolicy, input);3207    const subject = await resolveSubject(opts.subject, input);3208    const routingKey = await resolveRoutingKey(opts.routingKey, input);3209    const out = await opts.daemon.materialize({3210      ast,3211      mode,3212      policy,3213      subject,3214      leaseTtlMs: opts.leaseTtlMs,3215      // The anonymous routing key rides `metadata.routingKey`; the router keys on3216      // `subject ?? metadata.routingKey` (§2.2). Omitted when there is none.3217      metadata: routingKey !== undefined ? { routingKey } : undefined,3218      // Forward the browser's opaque affinity ticket (if any) so the fleet edge places this3219      // materialize to the follower the ws is pinned to (FOLLOWER-AFFINITY-DESIGN.md §4). Opaque —3220      // never verified here. Inert when the reads client is a single daemon (no fleet edge).3221      ...(input.affinity !== undefined ? { affinity: input.affinity } : {}),3222    });3223    const res = queryLeaseResponse(out, wsEndpoint);3224    // I-iv (§4.1): the occupancy step FIRST — session sweep+upsert, then the D6 gate verdict. A3225    // closed gate suppresses the room-serve ONLY (the lease ships without the realtime block,3226    // indistinguishable from a non-room-served query — the daemon path) while the doorbell3227    // below still rides; lifecycle-off ⇒ `gateOpen: true` unconditionally and this line is inert.3228    const occ = await lifecycleOccupancy(input);3229    // G-iv-b: a labeled + wired query ADDITIONALLY gains the realtime block. The daemon lease3230    // above is unconditional (and its fields untouched) — room-serving only ever adds a field,3231    // so a non-room-served/legacy lease stays byte-identical and nothing here can block one.3232    const rt = occ.gateOpen ? await maybeRoomServe(input, ast, context, subject) : undefined;3233    if (rt !== undefined) res.realtime = rt;3234    // I-v (§4.2): the gate CLOSED and a room plausibly hosted this scope — drain it and ride the3235    // fence back so a room-attached client runs the GRACEFUL downgrade instead of the loud legacy3236    // anomaly. A never-shared solo doc (`!roomPlausible`) never drains (no wasted room boot); a3237    // daemon-attached client that receives a stray fence ignores it (its resolver reads only the3238    // daemon fields). `drainRoom` absent/throwing ⇒ no fence (fail-open, warn-once).3239    if (!occ.gateOpen && occ.roomPlausible && occ.doc !== undefined) {3240      const fence = await maybeDrainRoom(input.name, occ.doc);3241      if (fence !== undefined) res.realtimeFence = fence;3242    }3243    // I-iii: under the opt-in `realtime.lifecycle` config a LABELED lease additionally gains the3244    // §4 system-stream block (doorbell always; the fence bundle iff room-served OR downgrade-fenced3245    // — a downgrading client needs the watermark/ledger/outcome streams to run the ghost drop).3246    // Same additive discipline as the realtime block: absent config ⇒ byte-identical response.3247    const lc = await maybeLifecycle(input, rt !== undefined || res.realtimeFence !== undefined, subject, routingKey);3248    if (lc !== undefined) res.lifecycle = lc;3249    return res;3250  };32513252  const readQuery = async (input: QueryReadRequest<User>): Promise<QueryReadResponse> => {3253    const context: ApiContext<User> = { user: input.user, request: input.request };3254    // Same per-viewer gate a live lease passes — a one-shot read returns the same rows a subscribe3255    // would, so it must clear the same authorization.3256    await assertAuthorized(opts.authorizeQuery, {3257      user: input.user,3258      name: input.name,3259      args: input.args,3260      context,3261    });3262    const ast = await resolveAst(input.name, input.args, context);3263    // Scope the one-shot's dedup `QueryKey` by the SAME key the lease path routes on — the routing3264    // key (`subject ?? cookie ?? clientId`) — so the warm pipeline this read leaves behind is the3265    // very one the browser's follow-up `subscribe` reuses, ON THE SAME FOLLOWER HRW will place it3266    // (the warm handoff, SSR-DESIGN.md §3.4 / READ-ROUTER-DESIGN.md §2.4) rather than a second,3267    // viewer-mismatched materialization.3268    const subject = await resolveSubject(opts.subject, input);3269    const routingKey = await resolveRoutingKey(opts.routingKey, input);3270    const visibilityKey = subject ?? routingKey;3271    const out = await opts.daemon.query({3272      ast,3273      visibilityKey,3274      ttlMs: opts.readIdleTtlMs,3275      ...(input.affinity !== undefined ? { affinity: input.affinity } : {}),3276    });3277    return { rows: out.rows, cvMin: out.cvMin, queryKey: out.queryKey };3278  };32793280  const pushMutation = async (input: PushMutationRequest<User>): Promise<PushMutationResponse> => {3281    const context: ApiContext<User> = { user: input.user, request: input.request };3282    const mutator = opts.mutators?.[input.envelope.name];3283    // PRE-FLIGHT rejections — no txn, no data, `lmid` alone (the queue must still drain).3284    if (!mutator) return reject(backend, input.envelope, `unknown mutator: ${input.envelope.name}`);3285    try {3286      await assertAuthorized(opts.authorizeMutation, {3287        user: input.user,3288        envelope: input.envelope,3289        context,3290      });3291    } catch (err) {3292      return reject(backend, input.envelope, errMessage(err));3293    }3294    const mctx: MutationContext<User> = {3295      user: input.user,3296      envelope: input.envelope,3297      daemon: opts.daemon,3298      request: input.request,3299    };33003301    // SCOPED mutator (WORK-OUTSIDE-TX): the author controls the tx boundary via `scope.transact`,3302    // running server-only code before/after it. The `lmid`-always-advances invariant is OURS, not3303    // the author's — we seal the response from the scope's recorded state, so an early return, a3304    // never-called transact, or a swallowed rejection can't wedge the client's pending queue.3305    if (isScoped<User>(mutator)) {3306      const scope = new MutationScopeImpl(backend, input.envelope, renderIndex);3307      // A throw that reaches `settle` AFTER the outcome is already sealed (post-commit effect, or a3308      // compensation handler after a business rejection) can't change the response — but it must not3309      // vanish. Route it to the app's hook, else log so a failed refund is never fully silent.3310      const reportSealed = (err: unknown, phase: "committed" | "rejected"): void => {3311        if (opts.onScopeError) opts.onScopeError(err, { phase, envelope: input.envelope });3312        else console.error(`[rindle api-server] scoped mutator ${input.envelope.name}: post-${phase} code threw (outcome already sealed):`, err);3313      };3314      // Derive the response from the scope's OUTCOME (not the body's return), so control flow in the3315      // author's function can't skip the lmid advance. `caught` distinguishes "the body threw"3316      // (present, even if the thrown value was `undefined`) from "it returned cleanly".3317      const settle = async (caught?: { err: unknown }): Promise<PushMutationResponse> => {3318        // Seal from the REAL outcome even if the author forgot to `await` transact: draining its3319        // in-flight promise here records the outcome/infra before we read it (else a phantom no-op3320        // ships while the real write commits out-of-band). Already-resolved when it WAS awaited.3321        if (scope.pending) await scope.pending;3322        // Infra always wins: the backend threw, the commit state is unknown — never advance lmid.3323        // Keyed on the latched BOOLEAN, so a driver that rejects with a falsy value is still infra.3324        if (scope.infraLatched) throw scope.infra;3325        // transact resolved (committed OR business-rejected): seal from its recorded outcome. A3326        // post-commit / post-reject-compensation throw can't change the sealed outcome (its effects3327        // can't roll the tx back, and lmid already advanced §2.4). Rethrowing the MutationRejected is3328        // the sanctioned "compensated, stay rejected" signal — expected, not surfaced. Any OTHER throw3329        // (a FAILED refund, a post-commit effect) must not vanish — surface it.3330        if (scope.outcome) {3331          if (caught && !(caught.err instanceof MutationRejected)) {3332            reportSealed(caught.err, scope.outcome.accepted ? "committed" : "rejected");3333          }3334          return outcomeToResponse(scope.outcome);3335        }3336        // Never transacted:3337        if (caught) {3338          // A throw before/around transact. A BackendError is the author signaling INFRA (retry);3339          // any other throw is a BUSINESS rejection — advance lmid alone so the prediction snaps back.3340          if (caught.err instanceof BackendError) throw caught.err.driverError;3341          return reject(backend, input.envelope, errMessage(caught.err));3342        }3343        // Clean return with no transact — an accepted no-op that STILL advances lmid (the client3344        // predicted a write; its pending entry must resolve).3345        return outcomeToResponse(3346          await backend.runMutation({ envelope: input.envelope, render: renderIndex, run: async () => {} }),3347        );3348      };3349      try {3350        await (mutator as unknown as ScopedMutator<User, unknown>)(scope, input.envelope.args as never, mctx);3351      } catch (err) {3352        return settle({ err });3353      }3354      return settle();3355    }33563357    // Run the (tx-form) mutator INSIDE the backend's transaction. A throw from the mutator body is a3358    // business rejection (roll data back, advance `lmid`); a BackendError (DB failure) rejects this promise.3359    const outcome = await backend.runMutation({3360      envelope: input.envelope,3361      render: renderIndex,3362      run: async (tx) => {3363        const result = await mutator(tx, input.envelope.args as never, mctx);3364        applyResultToTx(result, tx);3365      },3366    });3367    return outcomeToResponse(outcome);3368  };33693370  const pushMutations = async (input: PushMutationsRequest<User>): Promise<PushMutationResponse[]> => {3371    const out: PushMutationResponse[] = [];3372    for (const envelope of input.envelopes) {3373      out.push(await pushMutation({ user: input.user, envelope, request: input.request }));3374    }3375    return out;3376  };33773378  const assertPins = async (): Promise<void> => {3379    const pins = opts.pinnedQueries;3380    if (!pins?.length) return;3381    const context: ApiContext<User> = { user: opts.pinUser as User, request: undefined };3382    // Resolve every pin's authoritative AST under the system pin user (a transient failure on one3383    // shouldn't strand the rest), then drive the warm-up — fleet fan-out via the router if3384    // configured, else one materialize per pin on the single daemon.3385    const resolved = await Promise.allSettled(3386      pins.map(async (pin) => ({3387        pin,3388        req: {3389          ast: await resolveAst(pin.name, pin.args ?? null, context),3390          mode,3391          policy: { kind: "pinned", name: pin.name } as MaterializationPolicy,3392          leaseTtlMs: opts.leaseTtlMs,3393        } satisfies MaterializeInput,3394      })),3395    );3396    const failures: string[] = [];3397    const pairs: Array<{ pin: PinnedQuery; req: MaterializeInput }> = [];3398    resolved.forEach((r, i) => {3399      if (r.status === "fulfilled") pairs.push(r.value);3400      else failures.push(`${pins[i].name}: ${errMessage(r.reason)}`);3401    });3402    const reqs = pairs.map((p) => p.req);3403    if (reqs.length) {3404      if (opts.pinFanout) {3405        // Push tier (§4.2): fan EACH pin across ALL live followers via the router (fire-and-forget,3406        // router-stateless). A whole-fan-out failure is surfaced so the caller retries.3407        try {3408          await opts.pinFanout.assertPins(reqs);3409        } catch (e) {3410          failures.push(errMessage(e));3411        }3412      } else {3413        // Single-daemon: materialize each pin once (the daemon dedups by canonical query).3414        const results = await Promise.allSettled(reqs.map((req) => opts.daemon.materialize(req)));3415        results.forEach((r, i) => {3416          if (r.status === "rejected") failures.push(`${pairs[i].pin.name}: ${errMessage(r.reason)}`);3417        });3418      }3419    }3420    if (failures.length) {3421      throw new Error(`assertPins: ${failures.length} failed — ${failures.join("; ")}`);3422    }3423  };34243425  // The room write-authority gate (§5.3.1): endpoints are disabled until the app opts in —3426  // the `realtime` block (which also activates `/room-boot`) or the deprecated bare3427  // `authorizeRoom` (trio only). Hosting an authority is never a default.3428  const roomAuthorizer: Authorizer<ApiContext<User>> | undefined = realtime3429    ? (realtime.authorize ?? defaultFlushGate(realtime.shellSecret))3430    : opts.authorizeRoom;3431  const roomGate = async (context: ApiContext<User>): Promise<void> => {3432    if (!roomAuthorizer) {3433      throw new RindleApiError("forbidden", "room authority not configured", 403);3434    }3435    await assertAuthorized(roomAuthorizer, context);3436  };34373438  // §2.1 room-key routing: a "<profile>/<key>" doc resolves through its NAMED profile — with the3439  // boot-time unwindowed backstop (§2.3), which covers footprints that weren't statically3440  // resolvable at construction AND key-dependent branches that window only some docs. Anything3441  // else falls through to the legacy single-profile alias BYTE-IDENTICALLY (the anonymous3442  // profile, bare-key form). A named profile wins over a legacy doc that merely contains "/".3443  // Returns the profile's context set beside the AST (H-iv-b: the scope-spec compilation needs3444  // the §2.2 owned/followed split; the legacy anonymous profile has no declaration — empty set).3445  const resolveRoomFootprint = async (3446    rt: RindleRealtimeOptions<User>,3447    doc: string,3448    context: ApiContext<User>,3449  ): Promise<{ ast: Ast; contextTables: ReadonlySet<string> }> => {3450    const split = splitRoomDoc(doc);3451    if (split !== undefined) {3452      const profile = roomProfiles.get(split.profile);3453      if (profile !== undefined) {3454        const ast = queryResultToAst(await profile.footprint(split.key, context));3455        assertUnwindowedFootprint(ast, profile.name);3456        return { ast, contextTables: profile.context };3457      }3458    }3459    if (rt.resolveFootprint) {3460      return {3461        ast: queryResultToAst(await rt.resolveFootprint(doc, context)),3462        contextTables: new Set<string>(),3463      };3464    }3465    throw new RindleApiError(3466      "not-found",3467      `no room profile matches doc "${doc}" — named profiles are addressed as "<profile>/<key>"`,3468      404,3469    );3470  };34713472  // The store's verdict rides specific statuses + body shapes (fence / conflict /3473  // identity) the room decodes — pass a daemon HTTP error through VERBATIM.3474  const daemonVerdict = (e: unknown): RoomHostResponse => {3475    if (e instanceof DaemonHttpError) {3476      let body: unknown;3477      try {3478        body = JSON.parse(e.body);3479      } catch {3480        body = { error: e.body };3481      }3482      return { status: e.status, body };3483    }3484    throw e;3485  };34863487  // The LM stream plane (LM-STREAM-CHECKPOINT §2). Checkpoints are SYSTEM writes — no clientID, no3488  // mid, no `lmid` advance (that exists to release a CLIENT's optimistic rebase point, and a3489  // server-authored checkpoint has no prediction to release) — so they ride the backend's3490  // outside-transaction SQL surface, whose `batch` is one transaction on every backend. CDC → IVM3491  // fanout happens on any write; it never needed an envelope.3492  if (opts.streams && "tables" in opts.streams.checkpoint) {3493    assertStreamTables(opts.streams.checkpoint.tables, opts.schema);3494  }3495  const streamPlane = opts.streams3496    ? new StreamPlane<User>(3497        opts.streams,3498        backend.outsideSql ? { dialect: backend.dialect, sql: backend.outsideSql } : undefined,3499      )3500    : undefined;3501  const streams = (): StreamPlane<User> => {3502    if (!streamPlane) throw new RindleApiError("forbidden", "streams not configured", 403);3503    return streamPlane;3504  };3505  // The plane's own refusal, translated at the seam (it stays free of the server's error type so3506  // `streams.ts` can be imported without the server — no cycle).3507  const streamForbidden = (e: unknown): never => {3508    if (e instanceof StreamForbidden) throw new RindleApiError("forbidden", "rindle request forbidden", 403);3509    throw e;3510  };35113512  return {3513    routes,3514    close: () => {3515      streamPlane?.closeSync();3516      ownedSql?.close();3517    },3518    createQueryLease,3519    readQuery,3520    assertPins,3521    pushMutation,3522    pushMutations,3523    // `async` throughout so a misconfiguration surfaces as a REJECTION like every other refusal on3524    // this interface, never as a synchronous throw the transport forgot to catch.3525    openStream: async (input) => streams().open(input),3526    subscribeStream: async (input) => streams().subscribe(input).catch(streamForbidden),3527    drainStreams: async () => streamPlane?.drainStreams(),3528    handleStreamJson: async (body, context) => {3529      const msg = parseObject(body, "stream request");3530      const from = msg.from === undefined ? undefined : parseNumber(msg.from, "from");3531      // A non-negative INTEGER, like the GET leg's clamp: a fractional `from` would yield a replay3532      // chunk whose `text.length !== seq - from`, breaking the frame invariant on the client.3533      if (from !== undefined && (!Number.isInteger(from) || from < 0)) {3534        throw new RindleApiError("bad-request", "invalid from", 400);3535      }3536      return streams()3537        .subscribe({3538          user: context.user,3539          streamId: parseString(msg.streamId, "streamId"),3540          ...(from !== undefined ? { from } : {}),3541          request: context.request,3542        })3543        .catch(streamForbidden);3544    },3545    streamResponse: async (request, context) => {3546      try {3547        const { streamId, from } = streamRequestFromHttp(request);3548        const sub = await streams()3549          .subscribe({ user: context.user, streamId, from, request: context.request ?? request })3550          .catch(streamForbidden);3551        const sse = streamFramesToSse(3552          sub,3553          context.keepAliveMs !== undefined ? { keepAliveMs: context.keepAliveMs } : undefined,3554        );3555        return new Response(sse, { headers: STREAM_SSE_HEADERS });3556      } catch (e) {3557        // The route helper OWNS the transport, so refusals become responses here rather than3558        // throws the route forgot to catch. The generic body never reveals stream existence.3559        if (e instanceof RindleApiError) {3560          return new Response(JSON.stringify({ error: e.message }), {3561            status: e.status,3562            headers: { "content-type": "application/json" },3563          });3564        }3565        throw e;3566      }3567    },3568    handleApplyRowChangeTxnJson: async (body, context) => {3569      await roomGate(context);3570      const msg = parseObject(body, "row-change txn");3571      try {3572        const out = await opts.daemon.applyRowChangeTxn(msg as unknown as RowChangeTxn);3573        return { status: 200, body: out };3574      } catch (e) {3575        return daemonVerdict(e);3576      }3577    },3578    handleClaimRoomEpochJson: async (body, context) => {3579      await roomGate(context);3580      const msg = parseObject(body, "claim-room-epoch request");3581      const doc = parseString(msg.doc, "doc");3582      const claim = opts.daemon.claimRoomEpoch?.bind(opts.daemon);3583      if (!claim) {3584        throw new Error("the configured daemon client does not implement claimRoomEpoch");3585      }3586      try {3587        return { status: 200, body: await claim({ doc }) };3588      } catch (e) {3589        return daemonVerdict(e);3590      }3591    },3592    handleRoomLmidsJson: async (body, context) => {3593      await roomGate(context);3594      const msg = parseObject(body, "room-lmids request");3595      const doc = parseString(msg.doc, "doc");3596      if (!Array.isArray(msg.clients) || msg.clients.some((c) => typeof c !== "string")) {3597        throw new RindleApiError("bad-request", "clients must be an array of strings", 400);3598      }3599      const lmids = opts.daemon.roomLmids?.bind(opts.daemon);3600      if (!lmids) {3601        throw new Error("the configured daemon client does not implement roomLmids");3602      }3603      try {3604        return { status: 200, body: await lmids({ doc, clients: msg.clients as string[] }) };3605      } catch (e) {3606        return daemonVerdict(e);3607      }3608    },3609    handleRoomBootJson: async (body, context) => {3610      if (!realtime) {3611        throw new RindleApiError("forbidden", "realtime not configured", 403);3612      }3613      await assertAuthorized(realtime.authorizeBoot ?? defaultBootGate(realtime.shellSecret), context);3614      const msg = parseObject(body, "room-boot request");3615      const doc = parseString(msg.doc, "doc");3616      if (msg.instance !== undefined) parseString(msg.instance, "instance"); // diagnostic identity only3617      const { ast, contextTables } = await resolveRoomFootprint(realtime, doc, context);3618      const claim = opts.daemon.claimRoomEpoch?.bind(opts.daemon);3619      if (!claim) {3620        throw new Error("the configured daemon client does not implement claimRoomEpoch");3621      }3622      try {3623        // Claim FIRST (§2.5): the lease below is minted for THIS placement, so a boot3624        // that loses the epoch race learns it here, before any materialization exists.3625        const { epoch } = await claim({ doc });3626        const lease = await opts.daemon.materialize({3627          ast,3628          // The room's upstream leg IS the normalized protocol (§3) — never the app's3629          // viewer `mode`.3630          mode: "normalized",3631          leaseTtlMs: realtime.upstreamLeaseTtlMs ?? opts.leaseTtlMs,3632        });3633        const headers = realtime.mintFlushHeaders3634          ? await realtime.mintFlushHeaders({ doc, epoch })3635          : {3636              [ROOM_FLUSH_CREDENTIAL_HEADER]: await mintRoomFlushCredential({3637                shellSecret: realtime.shellSecret,3638                doc,3639                epoch,3640              }),3641            };3642        const res: RoomBootResponse = {3643          epoch,3644          upstreamLeaseToken: lease.leaseToken,3645          // H-iv-b: the §3.3 commit-gate scope specs, for named-profile AND legacy docs alike3646          // (the footprint AST is resolved either way; legacy has an empty context set).3647          scopes: compileRoomScopeSpecs(ast, contextTables),3648          flush: {3649            urls: {3650              apply: routes.applyRowChangeTxn,3651              claim: routes.claimRoomEpoch,3652              lmids: routes.roomLmids,3653            },3654            headers,3655          },3656        };3657        if (lease.affinity !== undefined) res.upstreamAffinity = lease.affinity;3658        const upstreamWsEndpoint = realtime.upstreamWsEndpoint;3659        if (upstreamWsEndpoint !== undefined) res.upstreamWsEndpoint = upstreamWsEndpoint;3660        return { status: 200, body: res };3661      } catch (e) {3662        return daemonVerdict(e);3663      }3664    },3665    handleQueryJson: (body, context) => {3666      const msg = parseObject(body, "query request");3667      return createQueryLease({3668        user: context.user,3669        name: parseString(msg.name, "name"),3670        args: msg.args ?? null,3671        request: context.request,3672        clientId: typeof msg.clientId === "string" ? msg.clientId : undefined,3673        affinity: typeof msg.affinity === "string" ? msg.affinity : undefined,3674      });3675    },3676    handleReadJson: (body, context) => {3677      const msg = parseObject(body, "read request");3678      return readQuery({3679        user: context.user,3680        name: parseString(msg.name, "name"),3681        args: msg.args ?? null,3682        request: context.request,3683        clientId: typeof msg.clientId === "string" ? msg.clientId : undefined,3684        affinity: typeof msg.affinity === "string" ? msg.affinity : undefined,3685      });3686    },3687    handleMutateJson: (body, context) => {3688      const msg = parseObject(body, "mutation request");3689      if (Array.isArray(msg.envelopes)) {3690        return pushMutations({3691          user: context.user,3692          envelopes: msg.envelopes.map(parseEnvelope),3693          request: context.request,3694        });3695      }3696      const envelope = parseEnvelope(msg.envelope ?? msg);3697      return pushMutation({ user: context.user, envelope, request: context.request });3698    },3699  };3700}37013702/** Run a SHARED generator mutator (the SAME body the client predicts) against a live server3703 *  transaction (MUTATORS-ISOMORPHIC): bind the tier-agnostic {@link isoTx} factory and drive it —3704 *  each yielded logical op renders + runs against `tx` (dialect SQL, per backend), each `tx.row`3705 *  suspends for read-your-writes, and `tx.all` fans out. A server mutator uses this to delegate its3706 *  write body after parsing untrusted args and applying its server-only authority (principal, policy).3707 *  A mutator-body throw remains a business rejection; a DB failure propagates as infra. */3708export function runSharedMutation<Args, Ctx extends MutatorCtx>(3709  mutator: SharedMutator<Args, Ctx>,3710  args: Args,3711  ctx: Ctx,3712  tx: ServerMutationTx,3713): Promise<void> {3714  return driveMutationAsync(mutator(isoTx, args, ctx), {3715    apply: (op) => applyOpToServerTx(tx, op),3716    read: (table, pk) => tx.row(table, pk),3717    query: async (q) => {3718      // The shared-seam contract is ALWAYS an array of rows: the client's `WriteTxn.query` returns3719      // an array even for a root `.one()` (the root unwrap is the Store's `materialize()`, not the3720      // in-mutator read — rust/rindle/src/wasm/db.rs). The server's `tx.query` returns a scalar (object|null)3721      // for a `.one()` root, so normalize it to `[]`/`[row]` here — one body sees the same shape on3722      // both tiers. (Postgres backend: `tx.query` rejects until POSTGRES-READ-COMPILER-DESIGN.md3723      // Phase B; the daemon/SQLite backend compiles + rides the open session — DAEMON §5.4.)3724      const ast = q.ast();3725      const res = await tx.query(ast);3726      if (ast.one === true) return res == null ? [] : [res as QueryResultRow];3727      return (res ?? []) as QueryResultRow[];3728    },3729  });3730}37313732/** Run one logical {@link MutationOp} (yielded by a shared generator mutator) against the live server3733 *  write surface — the SAME async methods a plain async mutator calls (they render dialect SQL and3734 *  execute/accumulate per backend). */3735function applyOpToServerTx(tx: ServerWriteTx, op: MutationOp): Promise<void> {3736  switch (op.kind) {3737    case "insert":3738      return tx.insert(op.table, op.row);3739    case "upsert":3740      return tx.upsert(op.table, op.row);3741    case "insertIgnore":3742      return tx.insertIgnore(op.table, op.row);3743    case "update":3744      return tx.update(op.table, op.row);3745    case "delete":3746      return tx.delete(op.table, op.pk);3747  }3748}37493750/** Feed a mutator's RETURNED result (the alternative to calling `tx.exec`/logical ops directly) into3751 *  the backend tx: a returned `SqlStatement[]` / `SqlTxn`'s statements are exec'd onto `tx`. A3752 *  mutation's durable retry identity is always its envelope's `(clientID, mid)` — a returned txn's3753 *  own write identity has no meaning here and is ignored. A `void` return is a no-op — the mutator3754 *  already drove the tx. Preserves the return-style contract. */3755function applyResultToTx(result: ApiMutatorResult, tx: ServerMutationTx): void {3756  if (!result) return;3757  const statements = Array.isArray(result) ? result : result.statements;3758  for (const s of statements) tx.exec(s.sql, s.params);3759}37603761/**3762 * The stream table mapping, checked LOUDLY at construction (the room-profile rule — a misconfigured3763 * mapping should fail the deploy, not a 3am generation). Only checkable when a `schema` is3764 * configured; without one the plane's SQL fails at the first checkpoint like any other unschema'd3765 * write. The message table is the APP's, so only the three-to-five columns the plane touches are3766 * asserted — never its shape.3767 */3768function assertStreamTables(tables: StreamTables, schema: Schema | undefined): void {3769  if (!schema) return;3770  const c = resolveStreamColumns(tables.columns);3771  const columnsOf = (table: string): Set<string> | undefined => {3772    const meta = (schema.tables as Record<string, { columns?: Record<string, unknown> }> | undefined)?.[table];3773    return meta?.columns ? new Set(Object.keys(meta.columns)) : undefined;3774  };3775  const check = (table: string, needed: Array<[string, string]>): void => {3776    const cols = columnsOf(table);3777    if (!cols) {3778      throw new TypeError(3779        `streams.checkpoint.tables names "${table}", which is not in the configured schema — add it (and run its ` +3780          `migration; \`streamChunkTableDdl\` generates the chunk table's DDL)`,3781      );3782    }3783    for (const [role, name] of needed) {3784      if (!cols.has(name)) {3785        throw new TypeError(3786          `streams.checkpoint.tables: "${table}" has no column "${name}" (the ${role} column) — ` +3787            `add it, or point \`columns.${role}\` at the one you have. Known columns: ${[...cols].join(", ")}`,3788        );3789      }3790    }3791  };3792  check(tables.message, [3793    ["key", c.key],3794    ["body", c.body],3795    ["status", c.status],3796    ["seq", c.seq],3797    // `cancel`/`error`/`host` are opt-in BY NAMING — present here only when the app asked for them.3798    ...(c.cancel !== undefined ? ([["cancel", c.cancel]] as Array<[string, string]>) : []),3799    ...(c.error !== undefined ? ([["error", c.error]] as Array<[string, string]>) : []),3800    ...(c.host !== undefined ? ([["host", c.host]] as Array<[string, string]>) : []),3801  ]);3802  check(tables.chunks, [3803    ["chunkKey", c.chunkKey],3804    ["chunkStream", c.chunkStream],3805    ["chunkSeq", c.chunkSeq],3806    ["chunkText", c.chunkText],3807  ]);3808}38093810async function assertAuthorized<T>(authorizer: Authorizer<T> | undefined, input: T): Promise<void> {3811  if (!authorizer) return;3812  const result = await authorizer(input);3813  if (result === false) throw new RindleApiError("forbidden", "rindle request forbidden", 403);3814}38153816/** The default flush-trio gate: verify the default epoch-bound credential from the request3817 *  header. The two refusal messages are deliberately distinct — "missing" is a transport wiring3818 *  bug (no `context.request`), "refused" is an invalid credential. */3819function defaultFlushGate<User>(shellSecret: string): Authorizer<ApiContext<User>> {3820  return async (context) => {3821    const credential = requestHeader(context.request, ROOM_FLUSH_CREDENTIAL_HEADER);3822    if (!credential) {3823      throw new RindleApiError(3824        "forbidden",3825        `missing ${ROOM_FLUSH_CREDENTIAL_HEADER} header — is the transport passing its incoming request as context.request?`,3826        403,3827      );3828    }3829    try {3830      await verifyRoomFlushCredential(credential, shellSecret);3831    } catch (e) {3832      throw new RindleApiError("forbidden", `flush credential refused: ${errMessage(e)}`, 403);3833    }3834  };3835}38363837/** The default `/room-boot` gate: `Authorization: Bearer <shell secret>` (README contract),3838 *  compared constant-time. */3839function defaultBootGate<User>(shellSecret: string): Authorizer<ApiContext<User>> {3840  return (context) => {3841    const auth = requestHeader(context.request, "authorization");3842    const bearer = auth && /^bearer\s/i.test(auth) ? auth.replace(/^bearer\s+/i, "") : undefined;3843    if (!bearer || !timingSafeEqualStr(bearer, shellSecret)) {3844      throw new RindleApiError("forbidden", "room-boot: shell secret refused", 403);3845    }3846  };3847}38483849/** Best-effort header extraction from whatever the transport put in `context.request`: a Fetch3850 *  `Request` (`headers.get`), a node `IncomingMessage` (lowercased header map), or a plain3851 *  `{headers: {...}}`. `undefined` when there is no request or no such header. */3852function requestHeader(request: unknown, name: string): string | undefined {3853  if (!request || typeof request !== "object") return undefined;3854  const headers = (request as { headers?: unknown }).headers;3855  if (!headers || typeof headers !== "object") return undefined;3856  if (typeof (headers as { get?: unknown }).get === "function") {3857    return (headers as { get(n: string): string | null }).get(name) ?? undefined;3858  }3859  const map = headers as Record<string, unknown>;3860  const v = map[name.toLowerCase()] ?? map[name];3861  if (typeof v === "string") return v;3862  if (Array.isArray(v) && typeof v[0] === "string") return v[0];3863  return undefined;3864}38653866/** Constant-time string equality (a length mismatch fails fast — length is not the secret). */3867function timingSafeEqualStr(a: string, b: string): boolean {3868  const ab = new TextEncoder().encode(a);3869  const bb = new TextEncoder().encode(b);3870  if (ab.length !== bb.length) return false;3871  let diff = 0;3872  for (let i = 0; i < ab.length; i++) diff |= ab[i] ^ bb[i];3873  return diff === 0;3874}38753876async function resolvePolicy<User>(3877  policy: RindleApiServerOptions<User>["materializationPolicy"],3878  input: QueryLeaseRequest<User>,3879): Promise<MaterializationPolicy> {3880  if (!policy) return { kind: "whileSubscribed" };3881  return typeof policy === "function" ? await policy(input) : policy;3882}38833884async function resolveSubject<User>(3885  subject: RindleApiServerOptions<User>["subject"],3886  input: QueryLeaseRequest<User>,3887): Promise<string | undefined> {3888  if (typeof subject === "function") return subject(input);3889  return subject;3890}38913892/** Resolve the anonymous routing key (READ-ROUTER-DESIGN.md §2.2): an explicit value/resolver if3893 *  configured, otherwise the browser-supplied `clientId`. The router keys on `subject ?? this`. */3894async function resolveRoutingKey<User>(3895  routingKey: RindleApiServerOptions<User>["routingKey"],3896  input: QueryLeaseRequest<User>,3897): Promise<string | undefined> {3898  if (routingKey === undefined) return input.clientId;3899  return typeof routingKey === "function" ? routingKey(input) : routingKey;3900}39013902function queryLeaseResponse(out: MaterializeOutput, wsEndpoint?: string): QueryLeaseResponse {3903  return {3904    leaseToken: out.leaseToken,3905    materializationId: out.materializationId,3906    queryKey: out.queryKey,3907    reused: out.reused,3908    ...(wsEndpoint !== undefined ? { wsEndpoint } : {}),3909    ...(out.affinity !== undefined ? { affinity: out.affinity } : {}),3910  };3911}39123913function errMessage(reason: unknown): string {3914  return String((reason as Error)?.message ?? reason);3915}39163917// --------------------------------------------------------- lifecycle system leases (Slice I-iii)39183919// The four §4 lifecycle system tables, mirrored VERBATIM from the daemon DDL — the source of3920// truth is `rust/rindle-replica/src/mutations.rs` (`realtime_lifecycle_ddl()` + the room-ledger3921// DDL in `enable_client_mutations`); duplicated here like `DEFAULT_ROUTES` is client-side so this3922// package needs no engine import. `Db::enable_realtime_lifecycle` REGISTERS all four, so a3923// hand-built AST over them materializes and resolves `hello` like any base table (the room-boot3924// direct-materialize pattern).3925const SCOPE_SESSIONS_TABLE = "_rindle_scope_sessions";3926const ROOM_WATERMARK_TABLE = "_rindle_room_watermark";3927const ROOM_CLIENT_MUTATIONS_TABLE = "_rindle_room_client_mutations";3928const ROOM_MUTATION_OUTCOMES_TABLE = "_rindle_room_mutation_outcomes";39293930// --------------------------------------------------------- occupancy counting (Slice I-iv, §4.1)3931//3932// The occupancy step rides the NORMAL surfaces end to end: the session upsert + lazy sweep are one3933// `executeSqlTxn` (a plain write txn — CDC-captured since I-i, so the row landing IS the doorbell3934// delta fanning to every subscribed solo client; no clientID/mid — a system write must never3935// advance an lmid — and no producer watermark — a renewal's re-upsert must re-run, that is the3936// refresh), and the count is one `executeSqlRead` with `consistency: "strong"` — the read surface3937// the api-server already has against the daemon (the `RemoteLazyTx` fallback precedent above).3938// "strong" routes the read to the WRITE MASTER in a split deploy, which just serialized our3939// upsert: read-your-writes without a mutation session (the interactive-txn machinery is optional3940// on the daemon interface and far heavier than this two-round-trip pair needs).39413942/** Default {@link RindleRealtimeLifecycleOptions.minSessions} — the §4.1 1→2 trigger. */3943const DEFAULT_LIFECYCLE_MIN_SESSIONS = 2;3944/** Default {@link RindleRealtimeLifecycleOptions.graceMs} — the §9.1 hysteresis window. */3945const DEFAULT_LIFECYCLE_GRACE_MS = 120_000;3946/** Default {@link RindleRealtimeLifecycleOptions.sessionTtlMs} fallback when no `leaseTtlMs` is3947 *  configured either — 5 minutes, the {@link DEFAULT_ROOM_TOKEN_TTL_MS} cadence (see the field doc). */3948const DEFAULT_SESSION_TTL_MS = 5 * 60_000;3949/** Sweep slack past the grace window (D4): rows are deleted only once expired for MORE than3950 *  `graceMs + this` — the linger I-v's downgrade decision reads must comfortably outlive the3951 *  grace comparison itself under clock skew between api-server instances (occupancy tolerates3952 *  skew ≤ grace; the slack keeps the boundary case out of the deletable band). */3953const SESSION_SWEEP_SLACK_MS = 60_000;39543955/** D7 upsert: one row per (scope, clientId) — `(scope, client_id)` is the table's PRIMARY KEY3956 *  (`realtime_lifecycle_ddl()`), so a renewal refreshes `expires_at` in place. */3957const SESSION_UPSERT_SQL =3958  `INSERT INTO ${SCOPE_SESSIONS_TABLE} (scope, client_id, expires_at) VALUES (?, ?, ?) ` +3959  `ON CONFLICT(scope, client_id) DO UPDATE SET expires_at = excluded.expires_at`;3960/** The D4 lazy sweep, in the SAME txn as the upsert: age out THIS scope's long-expired rows.3961 *  Param 2 is `now − (graceMs + SESSION_SWEEP_SLACK_MS)` — never tighter (the linger contract). */3962const SESSION_SWEEP_SQL = `DELETE FROM ${SCOPE_SESSIONS_TABLE} WHERE scope = ? AND expires_at < ?`;3963/** The occupancy read, one SELECT: cell 0 = DISTINCT unexpired sessions (`expires_at > now`;3964 *  distinct by construction — `(scope, client_id)` is the PK), cell 1 = sessions expired WITHIN3965 *  the grace window (`now − graceMs < expires_at ≤ now`) — the upward hysteresis input, cell 2 =3966 *  ALL matching rows regardless of expiry (the I-v "room plausibly exists" signal: a scope with3967 *  ANY other-session row — live OR still lingering pre-sweep — was shared, so a room was3968 *  provisioned; a never-shared solo doc has none and must never drain). Params:3969 *  `[now, now, now − graceMs, scope]`. */3970const SESSION_COUNT_SQL =3971  `SELECT SUM(CASE WHEN expires_at > ? THEN 1 ELSE 0 END), ` +3972  `SUM(CASE WHEN expires_at <= ? AND expires_at > ? THEN 1 ELSE 0 END), ` +3973  `COUNT(*) ` +3974  `FROM ${SCOPE_SESSIONS_TABLE} WHERE scope = ?`;3975/** {@link SESSION_COUNT_SQL} excluding the CALLER's own row (D6 counts *other* sessions; the3976 *  caller contributes itself as the `+ 1`). One extra trailing param: the caller's clientId. */3977const SESSION_COUNT_OTHERS_SQL = `${SESSION_COUNT_SQL} AND client_id <> ?`;39783979/** `col = <string literal>` — the only predicate shape the lifecycle ASTs need. */3980function colEq(name: string, value: string): Condition {3981  return { type: "simple", op: "=", left: { type: "column", name }, right: { type: "literal", value } };3982}39833984/** The doorbell AST (§4.1): every unexpired row under the scope is one live session; the row3985 *  delta arriving through a solo client's daemon subscription IS the 1→2 upgrade signal. The3986 *  expiry filter is deliberately NOT in the predicate — `expires_at > now()` would freeze `now`3987 *  at mint time; liveness is the READER's judgment (I-iv), the stream just carries the rows. */3988function scopeSessionsAst(scope: string): Ast {3989  return { table: SCOPE_SESSIONS_TABLE, where: colEq("scope", scope) };3990}39913992/** The §4.2 fence AST: the doc's monotone `flush_seq` row. */3993function roomWatermarkAst(doc: string): Ast {3994  return { table: ROOM_WATERMARK_TABLE, where: colEq("doc", doc) };3995}39963997/** The §7.1 ledger / §3.3 outcome ASTs share one shape: doc-scoped, and ADDITIONALLY3998 *  client-scoped when the lease request carried the browser's stable `clientId` (the same id the3999 *  mutation envelopes stamp, so it is exactly the ledger/outcome `client_id`). Without it the4000 *  predicate stays doc-only and the client filters to its own rows (defense in depth either4001 *  way — the client always filters). */4002function docClientAst(table: string, doc: string, clientId: string | undefined): Ast {4003  const docCond = colEq("doc", doc);4004  return {4005    table,4006    where: clientId === undefined ? docCond : { type: "and", conditions: [docCond, colEq("client_id", clientId)] },4007  };4008}40094010// --------------------------------------------------------- room-serve helpers (G-iv-b)40114012/** Default room lease token TTL: short (minutes) per RINDLE-REALTIME §4.1 — renewal is a fresh4013 *  lease through the api-server, never an extension of this token. */4014const DEFAULT_ROOM_TOKEN_TTL_MS = 5 * 60_000;401540164017/** Does the AST contain an aggregate/reduce shape ANYWHERE (root, a `related` subquery, or an4018 *  `EXISTS` child)? Room-serving refuses these regardless of coverage: the client's aggregate4019 *  overlay (AGGREGATE-SYNC) is computed against the DAEMON's normalized stream and stays4020 *  daemon-gated until post-G. (`groupBy`/`having` only occur alongside `aggregate`, so testing4021 *  `aggregate` covers them; `having` is still walked for nested EXISTS aggregates.) */4022function astHasAggregate(ast: Ast): boolean {4023  if (ast.aggregate !== undefined) return true;4024  for (const rel of ast.related ?? []) {4025    if (astHasAggregate(rel.subquery)) return true;4026  }4027  return conditionHasAggregate(ast.where) || conditionHasAggregate(ast.having);4028}40294030function conditionHasAggregate(cond: Condition | undefined): boolean {4031  if (cond === undefined) return false;4032  switch (cond.type) {4033    case "simple":4034      return false;4035    case "and":4036    case "or":4037      return cond.conditions.some(conditionHasAggregate);4038    case "correlatedSubquery":4039      return astHasAggregate(cond.related.subquery);4040  }4041}40424043async function reject(4044  backend: MutationBackend,4045  envelope: MutationEnvelope,4046  reason: string,4047): Promise<PushMutationResponse> {4048  const output = await backend.reject({ envelope, reason });4049  return { accepted: false, rejected: true, reason, output };4050}40514052/** The single place a {@link MutationOutcome} becomes the wire {@link PushMutationResponse} — shared4053 *  by the tx-form path and every scoped-mutator seal branch so the accepted/rejected shape can never4054 *  drift between them. */4055function outcomeToResponse(outcome: MutationOutcome): PushMutationResponse {4056  return outcome.accepted4057    ? { accepted: true, rejected: false, output: outcome.output }4058    : { accepted: false, rejected: true, reason: outcome.reason, output: outcome.output };4059}40604061function parseObject(value: unknown, label: string): Record<string, unknown> {4062  if (!value || typeof value !== "object" || Array.isArray(value)) {4063    throw new RindleApiError("bad-request", `invalid ${label}`, 400);4064  }4065  return value as Record<string, unknown>;4066}40674068function parseString(value: unknown, label: string): string {4069  if (typeof value !== "string") throw new RindleApiError("bad-request", `invalid ${label}`, 400);4070  return value;4071}40724073function parseEnvelope(value: unknown): MutationEnvelope {4074  const obj = parseObject(value, "mutation envelope");4075  return {4076    clientID: parseString(obj.clientID, "clientID"),4077    mid: parseNumber(obj.mid, "mid"),4078    name: parseString(obj.name, "name"),4079    args: obj.args ?? null,4080  };4081}40824083function parseNumber(value: unknown, label: string): number {4084  if (typeof value !== "number" || !Number.isFinite(value)) {4085    throw new RindleApiError("bad-request", `invalid ${label}`, 400);4086  }4087  return value;4088}4089