1/// <reference path="../runtime/env.d.ts" />2/**3 * Author-declared verbs and effect units.4 *5 * Tables are the nouns of an application: they declare what exists, and6 * policies attach to them because a policy must be safe to evaluate on every7 * replica. Mutations are the verbs: they declare what happens, and8 * consequences attach to them because a consequence needs an owner — the9 * device that performed the intent. This module is the declarative half of10 * that grammar: {@link effect} pairs an action with its compensation under a11 * durable name, {@link mutation} binds effect units to a named verb once, at12 * declaration, so call sites carry nothing. The runtime half journals each13 * unit per write and fires handlers when the write's fate settles.14 *15 * This module stays authoring-only: it declares names, tables, and handlers,16 * and delegates every write to the runtime installed through17 * {@link setMutationRuntime}.18 *19 * @module20 */2122import type { TableProxy } from "jazz-tools";23import type { WriteHandle } from "../runtime/write-handle.ts";2425/**26 * The row an effect handler receives. In the session that performed the write27 * it is the write's snapshot: the full row for inserts, the changed columns28 * for updates, only the id for removes. After a reload the journal holds no29 * column values: a `synced` handler receives the row fetched live from the30 * store — the final merged state — and a `rejected` handler receives the id31 * alone, because the engine rolled the row back and identity plus32 * {@link EffectContext.cause} is all that remains. Treat every column except33 * `id` as optional.34 */35export type EffectRow<Row> = Partial<Row> & { id: string };3637/**38 * Delivery metadata passed to every effect handler. Delivery is39 * at-least-once: a crash between handler start and journal completion re-runs40 * the handler at the next boot, so handlers calling external services should41 * pass {@link EffectContext.journalId} as an idempotency key.42 */43export type EffectContext = {44 /** The `(write id, effect name)` idempotency key for this obligation. */45 journalId: string;46 /** The journaled write's stable id. */47 writeId: string;48 /** The declaring verb's name, or `null` for writes without a verb. */49 verb: string | null;50 /** The written table's name. */51 table: string;52 /** Which operation the write performed. */53 op: "insert" | "update" | "remove";54 /** The written row's id. */55 rowId: string;56 /** Epoch milliseconds when the write was journaled, for latency spans. */57 writeCreatedAt: number;58 /** Which fate settled the write. */59 fate: "synced" | "rejected";60 /**61 * Why a `rejected` write settled: `denied` for a store verdict; `null` on62 * the `synced` fate. `expired` is reserved for store-side expiry63 * enforcement — at the pinned runtime an overdue intent is surfaced, never64 * retired ({@link MutationOptions.expiresAfterMs}), so rejections today65 * carry66 * `denied`.67 */68 cause: "denied" | "expired" | null;69 /** The adjudicated rejection code, on the `rejected` fate. */70 code: string | null;71 /** The adjudicated rejection reason, on the `rejected` fate. */72 reason: string | null;73};7475/**76 * Thrown from an effect handler to retire the obligation immediately instead77 * of re-arming it. Delivery is at-least-once by default: an ordinary thrown78 * error is *retryable* — the handler re-runs at the next boot until it79 * succeeds or {@link EffectUnitOptions.maxAttempts} quarantines it. Some80 * failures are known to be *permanent* — a webhook receiver answered `400`, a81 * row a handler needed is gone for good — and retrying only burns attempts and82 * delays the quarantine diagnostic. Throwing this retires the obligation now,83 * counted as a permanent handler failure. The message reaches diagnostics.84 *85 * @example86 * ```ts87 * s.effect("charge", app.orders, {88 * onSynced: async (order, { journalId }) => {89 * const res = await fetch(url, { headers: { "Idempotency-Key": journalId } });90 * if (res.status >= 400 && res.status < 500) {91 * throw new PermanentEffectError(`charge refused: ${res.status}`);92 * }93 * if (!res.ok) throw new Error(`charge transient failure: ${res.status}`);94 * },95 * });96 * ```97 */98export class PermanentEffectError extends Error {99 /** Stable class name for diagnostics and the ledger's severity check. */100 override readonly name = "PermanentEffectError";101 /** Creates the permanent-failure signal with a diagnostics message. */102 constructor(message: string) {103 super(message);104 }105}106107/** True when a handler failure asked to retire rather than retry. */108export function isPermanentEffectError(error: unknown): error is PermanentEffectError {109 return error instanceof PermanentEffectError;110}111112/** The action and compensation handlers one effect unit pairs. */113export type EffectHandlers<Row> = {114 /** Runs on the originating device once the store confirms the write. */115 onSynced?: (row: EffectRow<Row>, context: EffectContext) => void | Promise<void>;116 /**117 * Runs on the originating device when the store adjudicates the write and118 * denies it permanently. The engine has already rolled the row out of local119 * query results, so compensate what the user was told — notices, external120 * calls, follow-up writes — not the row data. After a reload the payload is121 * identity-only ({@link EffectRow}). A synchronous local refusal throws122 * from the verb call instead and never fires this handler.123 */124 onRejected?: (row: EffectRow<Row>, context: EffectContext) => void | Promise<void>;125};126127/**128 * Retention options for one effect unit: how long delivery may lag the write,129 * and how many failing attempts are made before quarantine.130 */131export type EffectUnitOptions = {132 /**133 * The delivery window in milliseconds, measured from the write. When the134 * write synced but this obligation could not be delivered inside the135 * window, the obligation retires as expired and no handler fires — the136 * write happened, so compensation would be wrong. External-tier handlers137 * whose receivers keep finite idempotency windows should set this.138 */139 expiresAfterMs?: number;140 /**141 * Failing handler attempts before quarantine: past this count the142 * obligation retires as failed-permanent instead of re-arming at every143 * boot. The retirement is counted in runtime diagnostics and the journal144 * entry becomes prunable.145 *146 * @default 5147 */148 maxAttempts?: number;149};150151/**152 * A named, reusable pairing of action and compensation. The name is the153 * durable identity the journal uses to re-arm handlers after a reload; a154 * mutation declares its units once, at the verb declaration.155 */156export type EffectUnit<Row = { id: string }> = {157 /** The app-unique durable name of this unit. */158 readonly effectName: string;159 /** The unit's handlers. */160 readonly handlers: EffectHandlers<Row>;161 /** The delivery window in milliseconds, or `null` for none. */162 readonly expiresAfterMs?: number | null;163 /** Failing attempts before quarantine retires the obligation. */164 readonly maxAttempts?: number;165 /**166 * Set on a built-in unit the author did not name (`s.notice`, `s.mark`,167 * `s.chain`). Such a unit is registered lazily, when a {@link mutation}168 * includes it: its durable name becomes `<verb>#<position>` — stable across169 * reloads because it derives from the author-chosen verb name and the unit's170 * fixed position in that verb's `effects`, not from module-evaluation order.171 * Absent on named units, which register at declaration.172 */173 readonly anonymousPrefix?: string;174};175176/** Which operation a {@link MutationOp} performs. */177export type MutationOpKind = "insert" | "update" | "remove";178179/** A table operation a verb is declared over; see {@link insert}, {@link update}, {@link remove}. */180export type MutationOp<T, Init, Kind extends MutationOpKind = MutationOpKind> = {181 /** The operation kind. */182 readonly kind: Kind;183 /** The declared table the verb writes. */184 readonly table: TableProxy<T, Init>;185};186187/**188 * Declares that a verb inserts rows into `table`. A {@link mutation} declared189 * over this op is called as `(values) => WriteHandle<Row>`; `await` resolves190 * at `saved` with the created row.191 */192export function insert<T, Init>(table: TableProxy<T, Init>): MutationOp<T, Init, "insert"> {193 return { kind: "insert", table };194}195196/**197 * Declares that a verb updates rows of `table`. A {@link mutation} declared198 * over this op is called as `(id, patch) => WriteHandle<void>`; `await`199 * resolves at `saved`.200 */201export function update<T, Init>(table: TableProxy<T, Init>): MutationOp<T, Init, "update"> {202 return { kind: "update", table };203}204205/**206 * Declares that a verb removes rows from `table`. A {@link mutation} declared207 * over this op is called as `(id) => WriteHandle<void>`; `await` resolves at208 * `saved`.209 */210export function remove<T, Init>(table: TableProxy<T, Init>): MutationOp<T, Init, "remove"> {211 return { kind: "remove", table };212}213214/** A callable verb declared over an insert operation. */215export type InsertVerb<T, Init> = (values: Init) => WriteHandle<T>;216/** A callable verb declared over an update operation. */217export type UpdateVerb<Init> = (id: string, patch: Partial<Init>) => WriteHandle<void>;218/** A callable verb declared over a remove operation. */219export type RemoveVerb = (id: string) => WriteHandle<void>;220221/** The callable shape of a declared verb, derived from its operation. */222export type MutationVerb<Op> = Op extends MutationOp<infer T, infer Init, "insert">223 ? InsertVerb<T, Init>224 : Op extends MutationOp<infer _T, infer Init, "update"> ? UpdateVerb<Init>225 : Op extends MutationOp<infer _T, infer _Init, "remove"> ? RemoveVerb226 : never;227228/** Options accepted by {@link mutation}: declared units plus inline sugar. */229export type MutationOptions<Row> = {230 /** The effect units this verb carries; declared once, reused by every call site. */231 effects?: readonly EffectUnit<Row>[];232 /** Inline sugar for an implicit single unit named after the verb. */233 onSynced?: EffectHandlers<Row>["onSynced"];234 /** Inline sugar for an implicit single unit named after the verb. */235 onRejected?: EffectHandlers<Row>["onRejected"];236 /**237 * The intent's lifespan in milliseconds. A write of this verb still pending238 * past its lifespan is surfaced through pending-writes state and239 * diagnostics. The engine offers no withdrawal for a locally accepted write240 * at the pinned runtime, so an overdue intent is reported, never retired:241 * retiring it as rejected could fire compensation for a write that later242 * syncs anyway.243 */244 expiresAfterMs?: number;245};246247/** The registered declaration the runtime dispatches for one verb. */248export type MutationDescriptor = {249 /** The verb's app-unique name. */250 readonly verbName: string;251 /** The operation the verb performs. */252 readonly op: MutationOp<unknown, unknown>;253 /** The verb's effect units, implicit unit included. */254 readonly units: readonly EffectUnit<{ id: string }>[];255 /** The intent's lifespan in milliseconds, or `null` for none. */256 readonly expiresAfterMs: number | null;257};258259/**260 * One durable notice a {@link notice} unit enqueues. The queue is persistent261 * and UI-agnostic: entries may be created at a boot re-arm with nothing262 * mounted, and a component renders them later. `tone` classifies the message263 * for the render; `ttlMs` bounds its life when the author sets no explicit264 * dismissal.265 */266export type NoticeInput = {267 /** The user-facing message. */268 message: string;269 /** How to classify the message for rendering. */270 tone: "info" | "success" | "warning" | "error";271 /** Lifespan in milliseconds before the queue retires the entry, or `null`. */272 ttlMs: number | null;273};274275/** The runtime half installed by the package runtime before verbs are called. */276export type MutationRuntime = {277 /** Performs one declared verb call and returns its write handle. */278 dispatch(descriptor: MutationDescriptor, args: readonly unknown[]): WriteHandle<unknown>;279 /** Performs a chain child under a deterministic id retained by its parent. */280 dispatchChained(281 descriptor: MutationDescriptor,282 args: readonly unknown[],283 parentJournalId: string,284 ): Promise<void>;285 /** Records one structured {@link log} entry. */286 recordLog(label: string, context: EffectContext): void;287 /** Records one {@link trace} span from the write's journaling to its fate. */288 recordTrace?(label: string | null, context: EffectContext): void;289 /** Records one {@link debug} timeline event; a no-op in production builds. */290 recordDebug?(event: string, context: EffectContext): void;291 /** Enqueues one durable {@link notice} entry. */292 enqueueNotice?(input: NoticeInput, context: EffectContext): Promise<void>;293 /**294 * Patches a row on behalf of a {@link mark} unit: a bare update carrying no295 * further units, so write fate becomes replicated row data without296 * recursion. Resolves at local durability; rejects if the row is gone.297 */298 applyMark?(299 table: TableProxy<unknown, unknown>,300 rowId: string,301 patch: Record<string, unknown>,302 ): Promise<void>;303 /** Re-attempts outstanding journal obligations after a late unit registration. */304 unitRegistered?(name: string): void;305};306307type EffectsSlot = {308 runtime: MutationRuntime | null;309 effects: Map<string, EffectUnit<{ id: string }>>;310 verbs: Map<string, MutationDescriptor>;311 logs: Map<string, EffectUnit<{ id: string }>>;312 /** Cache for shared built-ins (trace, debug, webhook) keyed by durable name. */313 cache: Map<string, CachedBuiltinEntry>;314};315316type CachedBuiltinEntry = {317 unit: EffectUnit<{ id: string }>;318 signature: string | null;319};320321const slotName = "__LOFI_EFFECT_DECLARATIONS__";322const effectsGlobal = globalThis as typeof globalThis & { [slotName]?: EffectsSlot };323324function slot(): EffectsSlot {325 effectsGlobal[slotName] ??= {326 runtime: null,327 effects: new Map(),328 verbs: new Map(),329 logs: new Map(),330 cache: new Map(),331 };332 return effectsGlobal[slotName];333}334335/**336 * Installs the runtime half verbs dispatch through. The package runtime337 * installs it at boot; application code never calls this. Tests install a338 * deterministic fake to unit-test verbs without a booted runtime.339 */340export function setMutationRuntime(runtime: MutationRuntime): void {341 slot().runtime = runtime;342}343344/** Resolves a registered effect unit by its durable name, for the runtime. */345export function resolveEffectUnit(name: string): EffectUnit<{ id: string }> | null {346 return slot().effects.get(name) ?? null;347}348349/** Clears every declaration; deterministic-test seam. */350export function clearEffectDeclarations(): void {351 const state = slot();352 state.effects.clear();353 state.verbs.clear();354 state.logs.clear();355 state.cache.clear();356}357358// During dev hot replacement author modules re-evaluate routinely; the newest359// declaration replaces its predecessor under the same durable name. Outside360// dev, a duplicate name is a real collision and fails fast.361const hotReplacement = Boolean(import.meta.hot);362363function registerUnit(unit: EffectUnit<{ id: string }>): void {364 const state = slot();365 const existing = state.effects.get(unit.effectName);366 if (existing && existing !== unit && !hotReplacement) {367 throw new Error(368 `effect name "${unit.effectName}" is already declared; effect names are the journal's ` +369 `durable identities and must be unique per app`,370 );371 }372 state.effects.set(unit.effectName, unit);373 // A unit declared after boot may satisfy obligations journaled before the374 // reload; the runtime re-attempts them on registration.375 state.runtime?.unitRegistered?.(unit.effectName);376}377378/**379 * Declares a named, reusable, typed effect unit over one table.380 *381 * The name is durable: the journal re-arms this unit's handlers by name after382 * a reload, so renaming a unit orphans obligations journaled under the old383 * name. Names must be unique per app; a duplicate declaration throws.384 *385 * Handlers run once, on the originating device, with at-least-once delivery:386 * a crash between handler start and journal completion re-runs the handler at387 * the next boot, so handlers that call external services should pass388 * {@link EffectContext.journalId} as the idempotency key.389 *390 * @example391 * ```ts392 * const chargeCard = s.effect("chargeCard", app.schema.orders, {393 * onSynced: async (order, { journalId }) => {394 * // The world accepted the order. journalId is the idempotency key.395 * },396 * onRejected: async (order) => {397 * // The order never happened: compensate what the user saw.398 * },399 * });400 * ```401 *402 * @throws Error when `name` is empty or already declared by a different unit;403 * in dev hot replacement the newest declaration replaces its predecessor404 * instead of throwing.405 */406export function effect<T extends { id: string }, Init>(407 name: string,408 table: TableProxy<T, Init>,409 handlers: EffectHandlers<T>,410 options: EffectUnitOptions = {},411): EffectUnit<T> {412 if (!name.trim()) throw new Error("effect name must not be empty");413 void table;414 const unit: EffectUnit<T> = {415 effectName: name,416 handlers,417 expiresAfterMs: options.expiresAfterMs ?? null,418 ...(options.maxAttempts !== undefined ? { maxAttempts: options.maxAttempts } : {}),419 };420 registerUnit(unit as EffectUnit<{ id: string }>);421 return unit;422}423424/**425 * A built-in effect unit that records a structured entry in runtime426 * diagnostics on either fate. Repeated calls with one label return the same427 * unit, so a label can be shared by several verbs.428 *429 * @example430 * ```ts431 * export const addTask = s.mutation("addTask", s.insert(app.schema.tasks), {432 * effects: [s.log("task-writes")],433 * });434 * // Each addTask write records a "task-writes" diagnostics entry when it435 * // syncs or is rejected.436 * ```437 */438export function log(label: string): EffectUnit<{ id: string }> {439 if (!label.trim()) throw new Error("log label must not be empty");440 const state = slot();441 const name = `log:${label}`;442 let unit = state.logs.get(name);443 if (!unit) {444 const record = (_row: EffectRow<{ id: string }>, context: EffectContext) => {445 requireRuntime().recordLog(label, context);446 };447 unit = { effectName: name, handlers: { onSynced: record, onRejected: record } };448 state.logs.set(name, unit);449 registerUnit(unit);450 }451 return unit;452}453454function requireRuntime(): MutationRuntime {455 const runtime = slot().runtime;456 if (!runtime) {457 throw new Error(458 "the lofi runtime is not installed; import @nzip/lofi (or @nzip/lofi/preact) in the " +459 "module that calls this verb",460 );461 }462 return runtime;463}464465/**466 * The installed runtime, or a thrown explanation when none is. The built-in467 * effect library ({@link EffectUnit} factories in `effect-library.ts`) reaches468 * the runtime through this so it stays on the public authoring surface — the469 * same requirement custom units must meet.470 */471export function requireMutationRuntime(): MutationRuntime {472 return requireRuntime();473}474475/**476 * Builds an anonymous built-in unit (a notice, mark, or chain the author did477 * not name) without registering it. Registration is deferred to478 * {@link mutation}, which names it `<verb>#<position>` from the verb it is479 * attached to and its position in that verb's `effects`. That identity is480 * stable across reloads regardless of which module evaluates first, so a481 * journaled obligation always re-arms against the same logical unit; the only482 * way to orphan one is to reorder the effects within its own verb.483 */484export function anonymousUnit<Row extends { id: string }>(485 prefix: string,486 handlers: EffectHandlers<Row>,487 options: EffectUnitOptions = {},488): EffectUnit<Row> {489 return {490 effectName: "",491 handlers,492 anonymousPrefix: prefix,493 expiresAfterMs: options.expiresAfterMs ?? null,494 ...(options.maxAttempts !== undefined ? { maxAttempts: options.maxAttempts } : {}),495 };496}497498/**499 * Returns the shared built-in unit registered under `name`, building and500 * registering it on first use. For observation and external units (`s.trace`,501 * `s.debug`, `s.webhook`) whose whole identity is their durable name, so502 * reusing one across verbs shares a single unit instead of colliding — the503 * same discipline {@link log} follows. A supplied in-memory signature makes504 * reuse under different configuration fail without putting that configuration505 * into the durable name.506 */507export function cachedBuiltin<Row extends { id: string }>(508 name: string,509 build: () => EffectHandlers<Row>,510 options: Omit<EffectUnitOptions, "expiresAfterMs"> & { expiresAfterMs?: number | null } = {},511 signature?: string,512): EffectUnit<Row> {513 const state = slot();514 const existing = state.cache.get(name);515 if (existing && "unit" in existing) {516 if (existing.signature !== (signature ?? null)) {517 throw new Error(`built-in effect "${name}" is already declared with different options`);518 }519 return existing.unit as unknown as EffectUnit<Row>;520 }521 // A dev hot-reload may retain the pre-signature cache shape. Rebuild that522 // entry so future declarations receive the same configuration checks.523 if (existing) state.cache.delete(name);524 const unit: EffectUnit<Row> = {525 effectName: name,526 handlers: build(),527 expiresAfterMs: options.expiresAfterMs ?? null,528 ...(options.maxAttempts !== undefined ? { maxAttempts: options.maxAttempts } : {}),529 };530 state.cache.set(name, {531 unit: unit as unknown as EffectUnit<{ id: string }>,532 signature: signature ?? null,533 });534 registerUnit(unit as unknown as EffectUnit<{ id: string }>);535 return unit;536}537538/**539 * Declares a typed, callable verb: a named mutation over one table operation,540 * carrying its effect units. Call sites invoke the verb like a function and541 * receive a {@link WriteHandle}; `await` resolves at `saved`.542 *543 * Verb names are app-unique and durable — the journal attributes re-armed544 * obligations through them — and read as imperative verb phrases545 * (`placeOrder`, not `orderInsert`). Inline `onSynced`/`onRejected` are sugar546 * for an implicit single unit named after the verb.547 *548 * @example549 * ```ts550 * export const placeOrder = s.mutation("placeOrder", s.insert(app.schema.orders), {551 * effects: [chargeCard, s.log("order-placed")],552 * });553 *554 * const write = placeOrder({ item, qty }); // WriteHandle<Order>555 * await write; // saved: durable on this device556 * await write.synced; // confirmed by the store557 * ```558 *559 * @throws Error at declaration when `name` is empty, already declared560 * (outside dev hot replacement, where the newest declaration replaces its561 * predecessor under the same name), or when `options` list one effect unit562 * twice.563 * @throws Error from a call to the returned verb when the local policy564 * refuses the write: the refusal is synchronous, creates no stage and no565 * journal entry, and fires no effect — see {@link EffectHandlers.onRejected}566 * for the adjudicated case. Calls also throw when the lofi runtime is not567 * installed.568 */569export function mutation<T extends { id: string }, Init, Kind extends MutationOpKind>(570 name: string,571 op: MutationOp<T, Init, Kind>,572 options: MutationOptions<T> = {},573): MutationVerb<MutationOp<T, Init, Kind>> {574 if (!name.trim()) throw new Error("mutation name must not be empty");575 const state = slot();576 const existing = state.verbs.get(name);577 if (existing && !hotReplacement) {578 throw new Error(`mutation name "${name}" is already declared; verb names are unique per app`);579 }580 const units: EffectUnit<{ id: string }>[] = [];581 const seen = new Set<string>();582 const seenDeclarations = new Set<EffectUnit<T>>();583 const declared = options.effects ?? [];584 for (let index = 0; index < declared.length; index += 1) {585 const declaration = declared[index];586 if (seenDeclarations.has(declaration)) {587 throw new Error(`mutation "${name}" declares the same effect unit twice`);588 }589 seenDeclarations.add(declaration);590 let unit = declaration as EffectUnit<{ id: string }>;591 if (unit.anonymousPrefix !== undefined) {592 // An anonymous built-in (s.notice/s.mark/s.chain) is named and593 // registered here, from the verb it is attached to and its position in594 // this verb's effects — a durable identity independent of module load595 // order. Register a finalized copy so the ledger resolves it at re-arm.596 unit = { ...unit, effectName: `${name}#${index}`, anonymousPrefix: undefined };597 registerUnit(unit);598 }599 if (seen.has(unit.effectName)) {600 throw new Error(`mutation "${name}" declares effect "${unit.effectName}" twice`);601 }602 seen.add(unit.effectName);603 units.push(unit);604 }605 if (options.onSynced || options.onRejected) {606 if (seen.has(name)) {607 throw new Error(608 `mutation "${name}" cannot combine inline handlers with an effect unit named "${name}"`,609 );610 }611 const implicit: EffectUnit<T> = {612 effectName: name,613 handlers: { onSynced: options.onSynced, onRejected: options.onRejected },614 };615 registerUnit(implicit as EffectUnit<{ id: string }>);616 units.push(implicit as EffectUnit<{ id: string }>);617 }618 const descriptor: MutationDescriptor = {619 verbName: name,620 op: op as MutationOp<unknown, unknown>,621 units,622 expiresAfterMs: options.expiresAfterMs ?? null,623 };624 state.verbs.set(name, descriptor);625 const verb = (...args: readonly unknown[]) => requireRuntime().dispatch(descriptor, args);626 Object.defineProperty(verb, mutationDescriptor, { value: descriptor });627 return verb as MutationVerb<MutationOp<T, Init, Kind>>;628}629630const mutationDescriptor = Symbol.for("@nzip/lofi/mutationDescriptor");631632/** Dispatches a declared mutation as a deterministic child of an effect. */633export function dispatchChainedMutation<Input, Result>(634 verb: (input: Input) => WriteHandle<Result>,635 args: readonly [Input],636 parentJournalId: string,637): Promise<void> {638 const descriptor = (verb as unknown as { [mutationDescriptor]?: MutationDescriptor })[639 mutationDescriptor640 ];641 if (!descriptor) {642 throw new Error("chain target must be a mutation declared with s.mutation");643 }644 return requireRuntime().dispatchChained(descriptor, args, parentJournalId);645}