1/// <reference path="../runtime/env.d.ts" />2/**3 * The built-in effect library: reusable {@link EffectUnit}s tiered by risk,4 * each idempotent by construction so it doubles as a reference implementation5 * of the at-least-once discipline a custom unit author must follow.6 *7 * The tiers are a learning ramp, one new concept each:8 *9 * - **Observation** — {@link trace}, {@link debug}. Cannot change anything;10 * they only record. (The first observation unit, `s.log`, lives in11 * `effects.ts` beside the core because the runtime records it directly.)12 * - **Data-internal** — {@link notice}, {@link mark}, {@link chain}. Write13 * back into the app: a durable message, a status column, a follow-up verb.14 * Fate stops being a callback and becomes replicated data.15 * - **External** — {@link webhook}. Calls the outside world, where the16 * idempotency key and a bounded, backed-off retry earn their keep.17 *18 * Every unit here is built on the public authoring surface from `effects.ts`19 * ({@link effect}, {@link EffectContext}, {@link PermanentEffectError}, and the20 * optional runtime capabilities the package runtime installs) — nothing21 * private. The authoring guide points custom-unit authors at these as models.22 *23 * @module24 */2526import type { TableProxy } from "jazz-tools";27import type { WriteHandle } from "../runtime/write-handle.ts";28import {29 anonymousUnit,30 cachedBuiltin,31 dispatchChainedMutation,32 type EffectContext,33 type EffectHandlers,34 type EffectRow,35 type EffectUnit,36 type NoticeInput,37 PermanentEffectError,38 requireMutationRuntime,39} from "./effects.ts";4041/**42 * Observation unit: a span from the write's journaling to its settled fate,43 * recorded in runtime diagnostics as an OpenTelemetry-shaped event with the44 * saved→synced/rejected latency. Pure instrumentation — it changes no state45 * and cannot fail a write. The optional label groups related spans; without46 * one the verb name labels the span. Repeated calls with one label share a47 * unit, so a label can be reused across verbs.48 *49 * @example50 * ```ts51 * export const placeOrder = s.mutation("placeOrder", s.insert(app.orders), {52 * effects: [s.trace("checkout")],53 * });54 * ```55 */56export function trace(label?: string): EffectUnit<{ id: string }> {57 if (label !== undefined && !label.trim()) throw new Error("trace label must not be empty");58 const name = label ? `trace:${label}` : "trace";59 return cachedBuiltin(name, () => {60 const record = (_row: EffectRow<{ id: string }>, context: EffectContext) => {61 requireMutationRuntime().recordTrace?.(label ?? null, context);62 };63 return { onSynced: record, onRejected: record };64 });65}6667/**68 * Observation unit: a development-only timeline entry for each fate an69 * obligation settles, for eyeballing effect delivery in the inspector.70 * Stripped from production builds — in a `PROD` bundle the handlers record71 * nothing, so it costs a closure and no more.72 *73 * @example74 * ```ts75 * export const editNote = s.mutation("editNote", s.update(app.notes), {76 * effects: [s.debug()],77 * });78 * ```79 */80export function debug(): EffectUnit<{ id: string }> {81 return cachedBuiltin("debug", () => {82 const production = !import.meta.env.DEV;83 const record = (_row: EffectRow<{ id: string }>, context: EffectContext) => {84 if (production) return;85 requireMutationRuntime().recordDebug?.(context.fate, context);86 };87 return { onSynced: record, onRejected: record };88 });89}9091/** A notice message resolved from the settled row, or a static string. */92export type NoticeResolver<Row> = string | ((row: EffectRow<Row>) => string);9394/** Per-fate notice configuration for {@link notice}. */95export type NoticeConfig<Row> = {96 /** The message (or resolver) enqueued when the write syncs. */97 synced?: NoticeResolver<Row>;98 /** The message (or resolver) enqueued when the write is rejected. */99 rejected?: NoticeResolver<Row>;100 /**101 * Lifespan of an enqueued entry before the durable queue retires it. Omit102 * for the queue default; pass `null` to keep the entry until dismissed.103 */104 ttlMs?: number | null;105};106107function resolveNotice<Row>(resolver: NoticeResolver<Row>, row: EffectRow<Row>): string {108 return typeof resolver === "function" ? resolver(row) : resolver;109}110111/**112 * Data-internal unit: enqueues a durable, user-visible message when the write113 * settles — the fix for the "a rejected write still flashed success" failure114 * mode. The queue is durable and UI-agnostic: an entry created at a boot115 * re-arm survives with nothing mounted, and a component (the built-in notices116 * surface, or an author's) renders it. Toasts are a userland wrapper over the117 * queue, never an imperative call from here.118 *119 * Idempotent by the obligation's journal id: a re-delivered handler enqueues120 * the same keyed entry once, so a crash-and-replay shows one message.121 *122 * @example123 * ```ts124 * export const publish = s.mutation("publish", s.update(app.posts), {125 * effects: [s.notice<Post>({126 * synced: "Published.",127 * rejected: (post) => `Could not publish "${post.title}".`,128 * })],129 * });130 * ```131 */132export function notice<Row extends { id: string } = { id: string }>(133 config: NoticeConfig<Row>,134): EffectUnit<Row> {135 if (config.synced === undefined && config.rejected === undefined) {136 throw new Error("notice must configure at least one of synced or rejected");137 }138 const ttlMs = config.ttlMs;139 const enqueue = (140 resolver: NoticeResolver<Row> | undefined,141 tone: NoticeInput["tone"],142 ) =>143 async (row: EffectRow<{ id: string }>, context: EffectContext) => {144 if (resolver === undefined) return;145 const runtime = requireMutationRuntime();146 await runtime.enqueueNotice?.(147 { message: resolveNotice(resolver, row as EffectRow<Row>), tone, ttlMs: ttlMs ?? null },148 context,149 );150 };151 return anonymousUnit<{ id: string }>("notice", {152 onSynced: enqueue(config.synced, "success"),153 onRejected: enqueue(config.rejected, "error"),154 }) as unknown as EffectUnit<Row>;155}156157/** Per-fate row patches for {@link mark}. */158export type MarkConfig<Init> = {159 /** The patch applied to the row when the write syncs. */160 synced?: Partial<Init>;161 /** The patch applied to the row when the write is rejected. */162 rejected?: Partial<Init>;163};164165/**166 * Data-internal unit: patches the written row when its fate resolves, so write167 * fate becomes replicated data every device and query sees — the hand-rolled168 * `status` column made declarative. The patch is absolute (a static169 * set-column-to-value object), so it is convergent under re-delivery: applying170 * it twice lands the same row.171 *172 * A rejected *insert* has no row to mark — the engine rolled it out — so the173 * rejected patch is skipped for inserts; on updates and removes the row174 * survives the rollback and the patch records the failure.175 *176 * The mark is best-effort: its patch is an ordinary update that the store may177 * itself deny (a policy that governs the status column), or that finds no row178 * (an update whose row was concurrently removed elsewhere). In those cases the179 * mark simply does not land — it is a convenience over the manual status180 * column, not a delivery guarantee. For a fate signal that must survive, pair181 * it with {@link notice} (durable queue) or a `webhook`.182 *183 * @example184 * ```ts185 * export const submit = s.mutation("submit", s.update(app.claims), {186 * effects: [s.mark(app.claims, {187 * synced: { status: "confirmed" },188 * rejected: { status: "failed" },189 * })],190 * });191 * ```192 */193export function mark<T extends { id: string }, Init>(194 table: TableProxy<T, Init>,195 config: MarkConfig<Init>,196): EffectUnit<T> {197 if (config.synced === undefined && config.rejected === undefined) {198 throw new Error("mark must configure at least one of synced or rejected");199 }200 const apply = async (201 patch: Partial<Init> | undefined,202 row: EffectRow<T>,203 skipForInsertOp: boolean,204 context: EffectContext,205 ) => {206 if (patch === undefined) return;207 // A rejected insert left no row behind; patching a vanished id would only208 // fail the obligation into a pointless retry.209 if (skipForInsertOp && context.op === "insert") return;210 await requireMutationRuntime().applyMark?.(211 table as TableProxy<unknown, unknown>,212 row.id,213 patch as Record<string, unknown>,214 );215 };216 return anonymousUnit<T>("mark", {217 onSynced: (row, context) => apply(config.synced, row, false, context),218 onRejected: (row, context) => apply(config.rejected, row, true, context),219 });220}221222/**223 * Data-internal unit: issues a follow-up verb once the write syncs, mapping224 * the settled row to the next verb's input. Reifies a chain of writes225 * (reserve → charge → fulfill) declaratively, without a saga API — each link226 * is an ordinary verb, so its own effects and rejection handling apply. Fires227 * only on `synced`; a rejected write starts no chain.228 *229 * The follow-up is journaled under a deterministic child write id derived230 * from this obligation. Its record remains retained until the parent commits231 * delivery, so a crash between those steps reuses the child instead of232 * issuing the mutation again.233 *234 * @example235 * ```ts236 * export const reserve = s.mutation("reserve", s.insert(app.holds), {237 * effects: [s.chain(charge, (hold) => ({ holdId: hold.id, amount: hold.total }))],238 * });239 * ```240 */241export function chain<Row extends { id: string }, NextInput, NextResult>(242 next: (input: NextInput) => WriteHandle<NextResult>,243 toInput: (row: EffectRow<Row>) => NextInput,244): EffectUnit<Row> {245 return anonymousUnit<{ id: string }>("chain", {246 onSynced: async (row, context) => {247 // The ledger derives the child write id from this obligation and keeps248 // its record until the parent is committed done. A replay therefore249 // observes the existing child instead of issuing the mutation twice.250 const input = toInput(row as EffectRow<Row>);251 await dispatchChainedMutation(next, [input], context.journalId);252 },253 }) as unknown as EffectUnit<Row>;254}255256/** Options for the {@link webhook} unit. */257export type WebhookOptions = {258 /** Which fates POST; defaults to both. */259 on?: ReadonlyArray<"synced" | "rejected">;260 /**261 * Delivery window in milliseconds. Receiver idempotency windows are finite262 * (Stripe forgets keys after ~24h), so external delivery defaults to a263 * conservative 24 hours rather than infinity; pass `null` to opt into no264 * expiry.265 *266 * @default 86_400_000267 */268 expiresAfterMs?: number | null;269 /** Failing attempts before the obligation is quarantined. */270 maxAttempts?: number;271 /** Extra headers merged onto the POST (the idempotency key is always set). */272 headers?: Record<string, string>;273};274275const defaultWebhookExpiry = 24 * 60 * 60 * 1000;276277/**278 * External unit: POSTs the settled row and its fate to `url`, using `name` as279 * its safe durable identity and with the280 * obligation's journal id auto-injected as `Idempotency-Key` so a re-delivery281 * the receiver already saw is dropped receiver-side. The generic282 * outside-world workhorse and the reference for the at-least-once contract.283 *284 * Failure severity follows the response: a transient failure (network error,285 * `5xx`, `429`) throws an ordinary error, so the ledger's bounded backoff286 * retries it; a `4xx` (other than `429`) throws {@link PermanentEffectError},287 * retiring the obligation without burning the retry budget on a request the288 * receiver will keep refusing. Because receiver keys expire, delivery defaults289 * to a 24-hour window ({@link WebhookOptions.expiresAfterMs}).290 *291 * @example292 * ```ts293 * export const order = s.mutation("order", s.insert(app.orders), {294 * effects: [s.webhook("orders", "https://hooks.example.com/orders")],295 * });296 * ```297 */298export function webhook(299 name: string,300 url: string,301 options: WebhookOptions = {},302): EffectUnit<{ id: string }> {303 if (!name.trim()) throw new Error("webhook name must not be empty");304 if (!url.trim()) throw new Error("webhook url must not be empty");305 const fates = new Set(options.on ?? ["synced", "rejected"]);306 const configuredHeaders = new Headers(options.headers);307 configuredHeaders.delete("content-type");308 configuredHeaders.delete("idempotency-key");309 const signature = JSON.stringify({310 url,311 on: [...fates].sort(),312 expiresAfterMs: options.expiresAfterMs === undefined313 ? defaultWebhookExpiry314 : options.expiresAfterMs,315 maxAttempts: options.maxAttempts ?? null,316 headers: [...configuredHeaders.entries()].sort(([left], [right]) => left.localeCompare(right)),317 });318 const build = (): EffectHandlers<{ id: string }> => {319 const post = async (row: EffectRow<{ id: string }>, context: EffectContext) => {320 if (!fates.has(context.fate)) return;321 let response: Response;322 try {323 const headers = new Headers(configuredHeaders);324 headers.set("content-type", "application/json");325 headers.set("idempotency-key", context.journalId);326 response = await fetch(url, {327 method: "POST",328 headers,329 body: JSON.stringify({330 journalId: context.journalId,331 verb: context.verb,332 table: context.table,333 op: context.op,334 fate: context.fate,335 code: context.code,336 reason: context.reason,337 row,338 }),339 });340 } catch {341 // A network-level failure is transient by nature: rethrow so the342 // ledger retries with backoff inside the delivery window.343 throw new Error(`webhook "${name}" unreachable`);344 }345 if (response.ok) return;346 if (response.status >= 400 && response.status < 500 && response.status !== 429) {347 throw new PermanentEffectError(`webhook "${name}" refused with ${response.status}`);348 }349 throw new Error(`webhook "${name}" transient failure ${response.status}`);350 };351 return { onSynced: post, onRejected: post };352 };353 // Only the author-chosen name is durable. Configuration (including headers)354 // is compared in memory so secrets never enter journal identifiers.355 return cachedBuiltin(`webhook:${name}`, build, {356 expiresAfterMs: options.expiresAfterMs === undefined357 ? defaultWebhookExpiry358 : options.expiresAfterMs,359 ...(options.maxAttempts !== undefined ? { maxAttempts: options.maxAttempts } : {}),360 }, signature);361}