- commit
- 94e79272d1af58cea34ffcb52b0b50534c275438
- parent
- 48a28208ddb67ed1401f119722deb17074d1946a
- author
- Dami <dami.etoile@gmail.com>
| M | docs/ | 7 | ++- |
| M | package/ | 7 | +- |
| A | package/ | 33 | +++++++++++ |
| M | package/ | 60 | +++++++++++++------- |
| M | package/ | 49 | ++++++++++++---- |
| M | package/ | 17 | +++--- |
| M | package/ | 47 | ++++++++++++++-- |
| M | package/ | 31 | ++++++++++- |
| M | package/ | 85 | ++++++++++++++++------------ |
| M | package/ | 93 | ++++++++++++++++++++++-------- |
| M | package/ | 71 | +++++++++++++++++++---- |
| M | package/ | 16 | +++++ |
@@ -137,12 +137,15 @@ export const reserve = s.mutation("reserve", s.insert(app.schema.holds), {
137137 138138 Where the guardrails earn their keep.139139 140-### `s.webhook(url, options?)`140+### `s.webhook(name, url, options?)`141141 142142 POSTs the settled row and its fate to `url`, with the obligation's journal id auto-injected as143143 `Idempotency-Key` so a re-delivery the receiver already saw is dropped receiver-side. The generic144144 integration workhorse and the reference for the at-least-once contract.145145 146+- **`name` is the durable identity.** Keep it stable across releases. URLs and header values are147+ checked only in memory and never become journal keys, so credentials do not leak into persisted148+ effect identifiers. Reusing a name with different configuration fails fast.146149 - **Failure severity follows the response.** A network error, `5xx`, or `429` throws an ordinary147150 error, so the ledger's bounded backoff retries it; any other `4xx` throws `PermanentEffectError`,148151 retiring the obligation rather than retrying a request the receiver will keep refusing.@@ -152,7 +155,7 @@ integration workhorse and the reference for the at-least-once contract.
152155 153156 ```ts154157 export const order = s.mutation("order", s.insert(app.schema.orders), {155- effects: [s.webhook("https://hooks.example.com/orders")],158+ effects: [s.webhook("orders", "https://hooks.example.com/orders")],156159 });157160 ```158161 @@ -1,4 +1,4 @@
1-import type { VNode } from "preact";1+import { Fragment, type VNode } from "preact";22 // Package-owned optional notices surface.33 import { type NoticeEntry, useNotices } from "./use-notices.ts";44 @@ -28,7 +28,7 @@ export type NoticesProps = {
2828 * ```2929 *3030 * @param props Optional region label and a custom per-notice renderer.31- * @returns The live notices region, or `null` when the queue is empty.31+ * @returns The always-mounted live notices region.3232 */3333 export function Notices({ label = "Notifications", children }: NoticesProps): VNode {3434 const { notices, dismiss } = useNotices();@@ -41,11 +41,10 @@ export function Notices({ label = "Notifications", children }: NoticesProps): VN
4141 class="lofi-notices"4242 aria-label={label}4343 aria-live="polite"44- hidden={notices.length === 0}4544 >4645 {notices.map((notice) =>4746 children48- ? <div key={notice.id}>{children(notice, () => dismiss(notice.id))}</div>47+ ? <Fragment key={notice.id}>{children(notice, () => dismiss(notice.id))}</Fragment>4948 : (5049 <div key={notice.id} class="lofi-notice" data-tone={notice.tone} role="status">5150 <p class="lofi-notice-message">{notice.message}</p>@@ -0,0 +1,33 @@
1+import { render } from "npm:preact-render-to-string@6.7.0";2+import { getNoticeQueue } from "../runtime/mod.ts";3+import { Notices } from "./Notices.tsx";4+5+Deno.test("Notices keeps its empty live region exposed for the first announcement", () => {6+ const html = render(<Notices />);7+ if (!html.includes('aria-live="polite"')) {8+ throw new Error("the durable notice surface omitted its live region");9+ }10+ if (html.includes(" hidden")) {11+ throw new Error("an empty live region must remain in the accessibility tree");12+ }13+});14+15+Deno.test("Notices preserves custom renderer markup without a wrapper", async () => {16+ const queue = getNoticeQueue();17+ const id = "notices-custom-renderer-test";18+ await queue.enqueue({ id, message: "Custom", tone: "info", ttlMs: null });19+ try {20+ const html = render(21+ <Notices>{(notice) => <article data-notice={notice.id}>{notice.message}</article>}</Notices>,22+ );23+ if (!html.includes(`<article data-notice="${id}">Custom</article>`)) {24+ throw new Error("the custom notice renderer did not retain its own root markup");25+ }26+ if (html.includes("<div")) {27+ throw new Error("the notices surface inserted a wrapper around custom markup");28+ }29+ } finally {30+ queue.dismiss(id);31+ await queue.flush();32+ }33+});@@ -85,6 +85,7 @@ export class NoticeQueue {
8585 #document: NoticeDocument = { version: 1, entries: [] };8686 #chain: Promise<void> = Promise.resolve();8787 #loaded = false;88+ #loading: Promise<void> | null = null;8889 #listeners = new Set<() => void>();8990 #onCountChange?: (count: number) => void;9091 @@ -105,8 +106,16 @@ export class NoticeQueue {
105106 * re-arm before load resolves — are merged in rather than clobbered: the106107 * persisted set is the base, and any in-memory entry (fresher) wins on id.107108 */108- async load(): Promise<void> {109- if (this.#loaded) return;109+ load(): Promise<void> {110+ if (this.#loaded) return Promise.resolve();111+ this.#loading ??= this.#load().catch((error) => {112+ this.#loading = null;113+ throw error;114+ });115+ return this.#loading;116+ }117+118+ async #load(): Promise<void> {110119 const persisted = parse(await this.#storage.load());111120 const pending = this.#document.entries;112121 const byId = new Map<string, NoticeEntry>();@@ -116,10 +125,10 @@ export class NoticeQueue {
116125 for (const entry of pending) byId.set(entry.id, entry);117126 this.#document = { version: 1, entries: [...byId.values()] };118127 this.#loaded = true;119- this.#sweepExpired();128+ const swept = this.#sweepExpired();120129 // A load that absorbed in-flight entries must re-persist the merged set121130 // and publish it, so the boot-enqueued notice is not silently deferred.122- if (pending.length > 0) this.#persist();131+ if (pending.length > 0 || swept) await this.#persist();123132 this.#emit();124133 }125134 @@ -127,18 +136,23 @@ export class NoticeQueue {
127136 * Enqueues one notice, idempotent by id: an entry whose id is already128137 * present is left untouched, so an at-least-once re-delivery adds nothing.129138 */130- enqueue(input: NoticeEnqueueInput): void {131- if (this.#document.entries.some((entry) => entry.id === input.id)) return;132- const createdAt = this.#now();133- this.#document.entries.push({134- id: input.id,135- message: input.message,136- tone: input.tone,137- createdAt,138- expiresAt: input.ttlMs === null ? null : createdAt + input.ttlMs,139- });140- this.#persist();141- this.#emit();139+ async enqueue(input: NoticeEnqueueInput): Promise<void> {140+ await this.load();141+ const exists = this.#document.entries.some((entry) => entry.id === input.id);142+ if (!exists) {143+ const createdAt = this.#now();144+ this.#document.entries.push({145+ id: input.id,146+ message: input.message,147+ tone: input.tone,148+ createdAt,149+ expiresAt: input.ttlMs === null ? null : createdAt + input.ttlMs,150+ });151+ this.#emit();152+ }153+ // Persist duplicate deliveries too: if the first attempt mutated memory154+ // but its save failed, replay is the retry that makes the entry durable.155+ await this.#persist();142156 }143157 144158 /** Dismisses one entry by id; unknown ids are a no-op. */@@ -146,7 +160,7 @@ export class NoticeQueue {
146160 const next = this.#document.entries.filter((entry) => entry.id !== id);147161 if (next.length === this.#document.entries.length) return;148162 this.#document.entries = next;149- this.#persist();163+ void this.#persist().catch(() => undefined);150164 this.#emit();151165 }152166 @@ -154,7 +168,7 @@ export class NoticeQueue {
154168 dismissAll(): void {155169 if (this.#document.entries.length === 0) return;156170 this.#document.entries = [];157- this.#persist();171+ void this.#persist().catch(() => undefined);158172 this.#emit();159173 }160174 @@ -191,7 +205,7 @@ export class NoticeQueue {
191205 /** Applies TTL retirement, persisting and notifying if anything changed. */192206 sweep(): void {193207 if (this.#sweepExpired()) {194- this.#persist();208+ void this.#persist().catch(() => undefined);195209 this.#emit();196210 }197211 }@@ -205,9 +219,13 @@ export class NoticeQueue {
205219 return this.#document.entries.length !== before;206220 }207221 208- #persist(): void {222+ #persist(): Promise<void> {209223 const snapshot = JSON.stringify(this.#document);210- this.#chain = this.#chain.then(() => this.#storage.save(snapshot)).catch(() => undefined);224+ const attempt = this.#chain.then(() => this.#storage.save(snapshot));225+ // Keep the serialized queue usable after a failed attempt while returning226+ // that failure to the effect handler that requires durable delivery.227+ this.#chain = attempt.catch(() => undefined);228+ return attempt;211229 }212230 213231 #emit(): void {@@ -12,30 +12,30 @@ function queue(now: () => number = () => 1000): {
1212 return { q: new NoticeQueue(storage, now, (count) => counts.push(count)), storage, counts };1313 }1414 15-Deno.test("enqueue is idempotent by id, so an at-least-once re-delivery adds one entry", () => {15+Deno.test("enqueue is idempotent by id, so an at-least-once re-delivery adds one entry", async () => {1616 const { q, counts } = queue();17- q.enqueue({ id: "w1:notice#1", message: "Saved.", tone: "success", ttlMs: null });18- q.enqueue({ id: "w1:notice#1", message: "Saved.", tone: "success", ttlMs: null });17+ await q.enqueue({ id: "w1:notice#1", message: "Saved.", tone: "success", ttlMs: null });18+ await q.enqueue({ id: "w1:notice#1", message: "Saved.", tone: "success", ttlMs: null });1919 assertCount(q.list().length, 1, "the second enqueue with one id must be a no-op");2020 assertCount(counts.at(-1) ?? -1, 1, "the active-notice count must reflect one entry");2121 });2222 23-Deno.test("dismiss removes one entry and notifies subscribers", () => {23+Deno.test("dismiss removes one entry and notifies subscribers", async () => {2424 const { q } = queue();2525 let notified = 0;2626 q.subscribe(() => notified += 1);27- q.enqueue({ id: "a", message: "one", tone: "info", ttlMs: null });28- q.enqueue({ id: "b", message: "two", tone: "info", ttlMs: null });27+ await q.enqueue({ id: "a", message: "one", tone: "info", ttlMs: null });28+ await q.enqueue({ id: "b", message: "two", tone: "info", ttlMs: null });2929 q.dismiss("a");3030 assertCount(q.list().length, 1, "dismiss must drop exactly the named entry");3131 assert(q.list()[0].id === "b", "the surviving entry must be the one not dismissed");3232 assert(notified >= 3, "each mutation must notify subscribers");3333 });3434 35-Deno.test("a TTL entry retires once its window closes", () => {35+Deno.test("a TTL entry retires once its window closes", async () => {3636 let clock = 1000;3737 const { q } = queue(() => clock);38- q.enqueue({ id: "ttl", message: "temporary", tone: "warning", ttlMs: 5000 });38+ await q.enqueue({ id: "ttl", message: "temporary", tone: "warning", ttlMs: 5000 });3939 assertCount(q.list().length, 1, "the entry is live inside its window");4040 clock = 6001;4141 assertCount(q.list().length, 0, "list must not surface an entry past its TTL");@@ -50,8 +50,8 @@ Deno.test("an entry enqueued during the load window is merged, not clobbered", a
5050 );5151 const q = new NoticeQueue(storage, () => 1000);5252 // An effect fires at a boot re-arm and enqueues before load() resolves.53- q.enqueue({ id: "boot", message: "fresh", tone: "success", ttlMs: null });54- await q.load();53+ const enqueued = q.enqueue({ id: "boot", message: "fresh", tone: "success", ttlMs: null });54+ await Promise.all([q.load(), enqueued]);5555 const ids = q.list().map((entry) => entry.id).sort();5656 assertCount(ids.length, 2, "the boot entry and the persisted entry both survive");5757 assert(ids.includes("boot") && ids.includes("persisted"), "neither entry is clobbered");@@ -61,8 +61,8 @@ Deno.test("a persisted queue reloads its entries and drops expired ones", async
6161 let clock = 1000;6262 const storage = createMemoryJournalStorage();6363 const first = new NoticeQueue(storage, () => clock);64- first.enqueue({ id: "keep", message: "durable", tone: "info", ttlMs: null });65- first.enqueue({ id: "drop", message: "fleeting", tone: "info", ttlMs: 100 });64+ await first.enqueue({ id: "keep", message: "durable", tone: "info", ttlMs: null });65+ await first.enqueue({ id: "drop", message: "fleeting", tone: "info", ttlMs: 100 });6666 await first.flush();6767 6868 clock = 5000;@@ -72,3 +72,28 @@ Deno.test("a persisted queue reloads its entries and drops expired ones", async
7272 assert(ids.includes("keep"), "a persisted entry must survive a reload");7373 assert(!ids.includes("drop"), "an entry whose TTL passed while away must not reappear");7474 });75+76+Deno.test("enqueue rejects a failed save and replay retries persistence", async () => {77+ const saves: string[] = [];78+ let attempts = 0;79+ const q = new NoticeQueue({80+ load: () => Promise.resolve(null),81+ save(text) {82+ attempts += 1;83+ if (attempts === 1) return Promise.reject(new Error("quota"));84+ saves.push(text);85+ return Promise.resolve();86+ },87+ });88+ const input = { id: "retry", message: "durable", tone: "error" as const, ttlMs: null };89+ let rejected = false;90+ try {91+ await q.enqueue(input);92+ } catch {93+ rejected = true;94+ }95+ assert(rejected, "the effect handler must observe a failed durable save");96+ await q.enqueue(input);97+ assert(attempts === 2, "re-delivery must retry persistence for an in-memory duplicate");98+ assert(saves[0]?.includes('"id":"retry"') === true, "the replayed entry must become durable");99+});@@ -54,6 +54,8 @@ export type JournalWriteStage = "saved" | "synced" | "rejected";
5454 export type JournalWriteRecord = {5555 /** The stable package-owned write id. */5656 writeId: string;57+ /** Parent obligation retaining this child until it commits, or `null`. */58+ retainedBy?: string | null;5759 /** The declaring verb's name, or `null` for writes without a verb. */5860 verb: string | null;5961 /** The written table's name. */@@ -250,11 +252,7 @@ function localStorageStorage(key: string): JournalStorage | null {
250252 }251253 },252254 save(text: string) {253- try {254- localStorage.setItem(key, text);255- } catch {256- // A private-mode quota failure degrades to session-only pending state.257- }255+ localStorage.setItem(key, text);258256 return Promise.resolve();259257 },260258 };@@ -280,6 +278,7 @@ export class WriteJournal {
280278 readonly #storage: JournalStorage;281279 #document: JournalDocument = emptyJournal();282280 #chain: Promise<void> = Promise.resolve();281+ #lastAttempt: Promise<void> = Promise.resolve();283282 #loaded = false;284283 285284 /** Creates a journal over one storage location. */@@ -305,13 +304,13 @@ export class WriteJournal {
305304 update(mutate: (document: JournalDocument) => void): void {306305 mutate(this.#document);307306 const snapshot = JSON.stringify(this.#document);308- this.#chain = this.#chain309- .then(() => this.#storage.save(snapshot))310- .catch(() => undefined);307+ const attempt = this.#chain.then(() => this.#storage.save(snapshot));308+ this.#lastAttempt = attempt;309+ this.#chain = attempt.catch(() => undefined);311310 }312311 313312 /** Resolves once every scheduled persistence has settled. */314313 flush(): Promise<void> {315- return this.#chain;314+ return this.#lastAttempt;316315 }317316 }@@ -51,6 +51,7 @@ import { createWriteHandle, type WriteHandle, type WriteHandleController } from
5151 import {5252 createDefaultJournalStorage,5353 hashJournalValue,54+ type JournalDocument,5455 type JournalEffectState,5556 journalIdFor,5657 type JournalStorage,@@ -110,6 +111,10 @@ export type LedgerWriteOptions = {
110111 units?: readonly EffectUnit<{ id: string }>[];111112 /** The intent's lifespan in milliseconds, or `null` for none. */112113 expiresAfterMs?: number | null;114+ /** Internal deterministic identity for a chain child. */115+ writeId?: string;116+ /** Parent obligation retaining this chain child until parent completion. */117+ retainedBy?: string | null;113118 };114119 115120 type VendorWrite = {@@ -279,7 +284,7 @@ export class WriteLedger {
279284 */280285 perform<T>(request: LedgerWriteRequest, options: LedgerWriteOptions = {}): WriteHandle<T> {281286 if (this.#disposed) throw new Error("write ledger has been disposed");282- const { handle, controller } = createWriteHandle<T>(crypto.randomUUID());287+ const { handle, controller } = createWriteHandle<T>(options.writeId ?? crypto.randomUUID());283288 const db = this.#db;284289 const guard = this.#environment.guardWrite;285290 if (db && !guard) {@@ -328,12 +333,17 @@ export class WriteLedger {
328333 }329334 330335 /** Performs one declared verb call; the verb dispatcher's entry point. */331- performVerb(descriptor: MutationDescriptor, args: readonly unknown[]): WriteHandle<unknown> {336+ performVerb(337+ descriptor: MutationDescriptor,338+ args: readonly unknown[],339+ internal: Pick<LedgerWriteOptions, "writeId" | "retainedBy"> = {},340+ ): WriteHandle<unknown> {332341 const table = descriptor.op.table as TableProxy<unknown, unknown>;333342 const options: LedgerWriteOptions = {334343 verb: descriptor.verbName,335344 units: descriptor.units,336345 expiresAfterMs: descriptor.expiresAfterMs,346+ ...internal,337347 };338348 if (descriptor.op.kind === "insert") {339349 return this.perform({ kind: "insert", table, values: args[0] }, options);@@ -349,6 +359,22 @@ export class WriteLedger {
349359 return this.perform({ kind: "remove", table, id: args[0] as string }, options);350360 }351361 362+ /** Performs a chain child once and retains its record through parent commit. */363+ async performChainedVerb(364+ descriptor: MutationDescriptor,365+ args: readonly unknown[],366+ parentJournalId: string,367+ ): Promise<void> {368+ const writeId = `chain:${parentJournalId}`;369+ if (this.#journal.document.writes[writeId]) {370+ await this.#journal.flush();371+ return;372+ }373+ const handle = this.performVerb(descriptor, args, { writeId, retainedBy: parentJournalId });374+ await handle.saved;375+ await this.#journal.flush();376+ }377+352378 /** Re-attempts outstanding obligations of one effect name after late registration. */353379 retryObligationsFor(effectName: string): void {354380 for (const entry of Object.values(this.#journal.document.writes)) {@@ -564,6 +590,7 @@ export class WriteLedger {
564590 const createdAt = this.#environment.now();565591 const entry: JournalWriteRecord = {566592 writeId: handle.writeId,593+ retainedBy: options.retainedBy ?? null,567594 verb: options.verb ?? null,568595 table: (request.table as { _table?: string })._table ?? "",569596 op: request.kind,@@ -730,8 +757,9 @@ export class WriteLedger {
730757 const fate = entry.stage === "synced" ? "synced" as const : "rejected" as const;731758 const handler = fate === "synced" ? unit.handlers.onSynced : unit.handlers.onRejected;732759 if (!handler) {733- this.#journal.update(() => {760+ this.#journal.update((document) => {734761 state.status = "done";762+ this.#releaseRetainedChildren(document, journalId);735763 });736764 this.#afterChange();737765 return;@@ -760,9 +788,10 @@ export class WriteLedger {
760788 };761789 try {762790 await handler(row as EffectRow<{ id: string }>, context);763- this.#journal.update(() => {791+ this.#journal.update((document) => {764792 state.status = "done";765793 state.lastError = null;794+ this.#releaseRetainedChildren(document, journalId);766795 });767796 } catch (error) {768797 const maxAttempts = unit.maxAttempts ?? defaultMaxAttempts;@@ -819,6 +848,12 @@ export class WriteLedger {
819848 return { id: entry.rowId };820849 }821850 851+ #releaseRetainedChildren(document: JournalDocument, parentJournalId: string): void {852+ for (const child of Object.values(document.writes)) {853+ if (child.retainedBy === parentJournalId) child.retainedBy = null;854+ }855+ }856+822857 #afterChange(): void {823858 const document = this.#journal.document;824859 for (const entry of Object.values(document.writes)) {@@ -829,7 +864,7 @@ export class WriteLedger {
829864 }830865 const settled = Object.values(entry.effects)831866 .every((state) => isRetired(state.status));832- if (settled) {867+ if (settled && !entry.retainedBy) {833868 // Fully settled entries leave the journal; failed-but-not-quarantined834869 // obligations keep theirs so the next boot can re-arm the handler.835870 this.#journal.update(() => {@@ -967,6 +1002,8 @@ export function dismissAllNotices(): void {
9671002 // package installs dispatch, so verbs work wherever islands import hooks.9681003 setMutationRuntime({9691004 dispatch: (descriptor, args) => getWriteLedger().performVerb(descriptor, args),1005+ dispatchChained: (descriptor, args, parentJournalId) =>1006+ getWriteLedger().performChainedVerb(descriptor, args, parentJournalId),9701007 recordLog: (label, context) =>9711008 updateRuntimeDiagnostics((diagnostics) =>9721009 recordEffectLogEntry(diagnostics, {@@ -1,6 +1,11 @@
11 import type { Db, MutationErrorEvent, TableProxy } from "jazz-tools";22 import { createDiagnostics } from "./diagnostics.ts";3-import { type EffectContext, type EffectUnit, PermanentEffectError } from "../schema/effects.ts";3+import {4+ type EffectContext,5+ type EffectUnit,6+ type MutationDescriptor,7+ PermanentEffectError,8+} from "../schema/effects.ts";49 import { assert, assertCount } from "./test-assert.ts";510 import {611 createMemoryJournalStorage,@@ -818,3 +823,27 @@ Deno.test("quarantine retires a permanently failing handler after its attempt bo
818823 assertCount(calls.length, 0, "a quarantined handler never reports success");819824 second.ledger.dispose();820825 });826+827+Deno.test("a chained child uses one retained deterministic write across replay", async () => {828+ const fixture = harness();829+ await fixture.ledger.arm();830+ const descriptor: MutationDescriptor = {831+ verbName: "createChild",832+ op: { kind: "insert", table: table as TableProxy<unknown, unknown> },833+ units: [],834+ expiresAfterMs: null,835+ };836+ const parentJournalId = "parent-write:chain#0";837+ await fixture.ledger.performChainedVerb(descriptor, [{ title: "child" }], parentJournalId);838+ assertCount(fixture.db.nextBatch, 1, "the first delivery must issue one child mutation");839+ const writeId = `chain:${parentJournalId}`;840+ const first = JSON.parse(fixture.storage.text() ?? "{}") as JournalDocument;841+ assert(842+ first.writes[writeId]?.retainedBy === parentJournalId,843+ "the child record must remain retained until its parent obligation commits",844+ );845+846+ await fixture.ledger.performChainedVerb(descriptor, [{ title: "child" }], parentJournalId);847+ assertCount(fixture.db.nextBatch, 1, "replay must observe the retained child, not issue another");848+ fixture.ledger.dispose();849+});@@ -28,6 +28,7 @@ import type { WriteHandle } from "../runtime/write-handle.ts";
2828 import {2929 anonymousUnit,3030 cachedBuiltin,31+ dispatchChainedMutation,3132 type EffectContext,3233 type EffectHandlers,3334 type EffectRow,@@ -139,10 +140,10 @@ export function notice<Row extends { id: string } = { id: string }>(
139140 resolver: NoticeResolver<Row> | undefined,140141 tone: NoticeInput["tone"],141142 ) =>142- (row: EffectRow<{ id: string }>, context: EffectContext) => {143+ async (row: EffectRow<{ id: string }>, context: EffectContext) => {143144 if (resolver === undefined) return;144145 const runtime = requireMutationRuntime();145- runtime.enqueueNotice?.(146+ await runtime.enqueueNotice?.(146147 { message: resolveNotice(resolver, row as EffectRow<Row>), tone, ttlMs: ttlMs ?? null },147148 context,148149 );@@ -225,9 +226,10 @@ export function mark<T extends { id: string }, Init>(
225226 * is an ordinary verb, so its own effects and rejection handling apply. Fires226227 * only on `synced`; a rejected write starts no chain.227228 *228- * The follow-up is itself journaled and idempotent by its own write id. This229- * unit's obligation is marked delivered once the next verb is durably saved,230- * so a crash between links re-issues from the last settled row.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.231233 *232234 * 233235 * ```ts@@ -236,15 +238,17 @@ export function mark<T extends { id: string }, Init>(
236238 * });237239 * ```238240 */239-export function chain<Row extends { id: string }, NextInput>(240- next: (input: NextInput) => WriteHandle<unknown>,241+export function chain<Row extends { id: string }, NextInput, NextResult>(242+ next: (input: NextInput) => WriteHandle<NextResult>,241243 toInput: (row: EffectRow<Row>) => NextInput,242244 ): EffectUnit<Row> {243245 return anonymousUnit<{ id: string }>("chain", {244- onSynced: async (row) => {245- // Await local durability of the follow-up: only then is this link safe246- // to mark delivered, so a crash re-issues rather than dropping it.247- await next(toInput(row as EffectRow<Row>)).saved;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);248252 },249253 }) as unknown as EffectUnit<Row>;250254 }@@ -271,7 +275,8 @@ export type WebhookOptions = {
271275 const defaultWebhookExpiry = 24 * 60 * 60 * 1000;272276 273277 /**274- * External unit: POSTs the settled row and its fate to `url`, with the278+ * External unit: POSTs the settled row and its fate to `url`, using `name` as279+ * its safe durable identity and with the275280 * obligation's journal id auto-injected as `Idempotency-Key` so a re-delivery276281 * the receiver already saw is dropped receiver-side. The generic277282 * outside-world workhorse and the reference for the at-least-once contract.@@ -286,25 +291,41 @@ const defaultWebhookExpiry = 24 * 60 * 60 * 1000;
286291 * 287292 * ```ts288293 * export const order = s.mutation("order", s.insert(app.orders), {289- * effects: [s.webhook("https://hooks.example.com/orders")],294+ * effects: [s.webhook("orders", "https://hooks.example.com/orders")],290295 * });291296 * ```292297 */293-export function webhook(url: string, options: WebhookOptions = {}): EffectUnit<{ id: string }> {298+export 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");294304 if (!url.trim()) throw new Error("webhook url must not be empty");295305 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+ });296318 const build = (): EffectHandlers<{ id: string }> => {297319 const post = async (row: EffectRow<{ id: string }>, context: EffectContext) => {298320 if (!fates.has(context.fate)) return;299321 let response: Response;300322 try {323+ const headers = new Headers(configuredHeaders);324+ headers.set("content-type", "application/json");325+ headers.set("idempotency-key", context.journalId);301326 response = await fetch(url, {302327 method: "POST",303- headers: {304- "content-type": "application/json",305- "idempotency-key": context.journalId,306- ...options.headers,307- },328+ headers,308329 body: JSON.stringify({309330 journalId: context.journalId,310331 verb: context.verb,@@ -316,33 +337,25 @@ export function webhook(url: string, options: WebhookOptions = {}): EffectUnit<{
316337 row,317338 }),318339 });319- } catch (error) {340+ } catch {320341 // A network-level failure is transient by nature: rethrow so the321342 // ledger retries with backoff inside the delivery window.322- throw new Error(`webhook ${url} unreachable: ${(error as Error).message}`);343+ throw new Error(`webhook "${name}" unreachable`);323344 }324345 if (response.ok) return;325346 if (response.status >= 400 && response.status < 500 && response.status !== 429) {326- throw new PermanentEffectError(`webhook ${url} refused with ${response.status}`);347+ throw new PermanentEffectError(`webhook "${name}" refused with ${response.status}`);327348 }328- throw new Error(`webhook ${url} transient failure ${response.status}`);349+ throw new Error(`webhook "${name}" transient failure ${response.status}`);329350 };330351 return { onSynced: post, onRejected: post };331352 };332- // Keyed by url + options so two verbs posting the same configured webhook333- // share one unit (no name collision), while a different config is distinct.334- const name = `webhook:${url}:${335- JSON.stringify({336- on: options.on ?? null,337- expiresAfterMs: options.expiresAfterMs ?? null,338- maxAttempts: options.maxAttempts ?? null,339- headers: options.headers ?? null,340- })341- }`;342- return cachedBuiltin(name, build, {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, {343356 expiresAfterMs: options.expiresAfterMs === undefined344357 ? defaultWebhookExpiry345- : options.expiresAfterMs ?? undefined,358+ : options.expiresAfterMs,346359 ...(options.maxAttempts !== undefined ? { maxAttempts: options.maxAttempts } : {}),347- });360+ }, signature);348361 }@@ -23,7 +23,11 @@ type Recorder = {
2323 traces: Array<{ label: string | null; context: EffectContext }>;2424 notices: Array<{ input: NoticeInput; context: EffectContext }>;2525 marks: Array<{ rowId: string; patch: Record<string, unknown> }>;26- dispatched: Array<{ descriptor: unknown; args: readonly unknown[] }>;26+ dispatched: Array<{27+ descriptor: unknown;28+ args: readonly unknown[];29+ parentJournalId?: string;30+ }>;2731 };2832 2933 function install(): Recorder {@@ -33,12 +37,17 @@ function install(): Recorder {
3337 recorder.dispatched.push({ descriptor, args });3438 return { saved: Promise.resolve() } as unknown as WriteHandle<unknown>;3539 },40+ dispatchChained(descriptor, args, parentJournalId) {41+ recorder.dispatched.push({ descriptor, args, parentJournalId });42+ return Promise.resolve();43+ },3644 recordLog() {},3745 recordTrace(label, context) {3846 recorder.traces.push({ label, context });3947 },4048 enqueueNotice(input, context) {4149 recorder.notices.push({ input, context });50+ return Promise.resolve();4251 },4352 applyMark(_table, rowId, patch) {4453 recorder.marks.push({ rowId, patch });@@ -103,29 +112,36 @@ Deno.test("trace records a span with the saved-to-fate latency on either fate",
103112 104113 // The content-named built-ins share one unit per identity instead of throwing105114 // a duplicate-name error when reused across verbs.106-Deno.test("trace and webhook share one unit per content across verbs", () => {115+Deno.test("trace and webhook share one unit per durable identity", () => {107116 clearEffectDeclarations();108117 install();109118 assert(trace("checkout") === trace("checkout"), "one label shares one trace unit");110119 assert(trace() === trace(), "the unlabeled trace shares one unit");111120 const url = "https://hooks.example.com/x";112- assert(webhook(url) === webhook(url), "one url+config shares one webhook unit");121+ assert(webhook("orders", url) === webhook("orders", url), "one name+config shares one unit");122+ let mismatched = false;123+ try {124+ webhook("orders", url, { maxAttempts: 9 });125+ } catch (error) {126+ mismatched = !(error as Error).message.includes(url);127+ }128+ assert(mismatched, "reusing a name with different config must fail without leaking config");113129 assert(114- webhook(url) !== webhook(url, { maxAttempts: 9 }),115- "a different config is a distinct unit",130+ webhook("orders-v2", url, { maxAttempts: 9 }) !== webhook("orders", url),131+ "a distinct author name creates a distinct durable unit",116132 );117133 clearEffectDeclarations();118134 });119135 120-Deno.test("notice enqueues a message keyed by the obligation journal id", () => {136+Deno.test("notice enqueues a message keyed by the obligation journal id", async () => {121137 clearEffectDeclarations();122138 const recorder = install();123139 const unit = notice<Claim>({124140 synced: "Submitted.",125141 rejected: (claim) => `Could not submit "${claim.title ?? "claim"}".`,126142 });127- unit.handlers.onSynced?.({ id: "row-1" }, context({ journalId: "w1:notice#1" }));128- unit.handlers.onRejected?.(143+ await unit.handlers.onSynced?.({ id: "row-1" }, context({ journalId: "w1:notice#1" }));144+ await unit.handlers.onRejected?.(129145 { id: "row-1", title: "roof" } as Claim & { id: string },130146 context({ fate: "rejected", journalId: "w2:notice#1" }),131147 );@@ -143,13 +159,13 @@ Deno.test("notice enqueues a message keyed by the obligation journal id", () =>
143159 clearEffectDeclarations();144160 });145161 146-Deno.test("notice fires only the configured fate", () => {162+Deno.test("notice fires only the configured fate", async () => {147163 clearEffectDeclarations();148164 const recorder = install();149165 const unit = notice({ rejected: "Failed." });150- unit.handlers.onSynced?.({ id: "row-1" }, context());166+ await unit.handlers.onSynced?.({ id: "row-1" }, context());151167 assertCount(recorder.notices.length, 0, "an unconfigured synced fate enqueues nothing");152- unit.handlers.onRejected?.({ id: "row-1" }, context({ fate: "rejected" }));168+ await unit.handlers.onRejected?.({ id: "row-1" }, context({ fate: "rejected" }));153169 assertCount(recorder.notices.length, 1, "the configured rejected fate enqueues");154170 clearEffectDeclarations();155171 });@@ -180,18 +196,24 @@ Deno.test("mark skips a rejected insert, which left no row to patch", async () =
180196 181197 Deno.test("chain issues the follow-up verb only on synced", async () => {182198 clearEffectDeclarations();183- install();184- const calls: Array<{ holdId: string }> = [];185- const charge = (input: { holdId: string }): WriteHandle<unknown> => {186- calls.push(input);187- return { saved: Promise.resolve() } as unknown as WriteHandle<unknown>;188- };189- const unit = chain(charge, (row: { id: string }) => ({ holdId: row.id }));199+ const recorder = install();200+ const charge = mutation("chargeClaim", insert(app.claims));201+ const unit = chain(charge, (row: { id: string }) => ({202+ title: row.id,203+ status: "charging",204+ }));190205 await unit.handlers.onRejected?.({ id: "row-1" }, context({ fate: "rejected" }));191- assertCount(calls.length, 0, "a rejected write starts no chain");206+ assertCount(recorder.dispatched.length, 0, "a rejected write starts no chain");192207 await unit.handlers.onSynced?.({ id: "row-1" }, context());193- assertCount(calls.length, 1, "a synced write issues the follow-up verb once");194- assert(calls[0].holdId === "row-1", "the mapper must receive the settled row");208+ assertCount(recorder.dispatched.length, 1, "a synced write issues the follow-up verb once");209+ assert(210+ (recorder.dispatched[0].args[0] as { title: string }).title === "row-1",211+ "the mapper must receive the settled row",212+ );213+ assert(214+ recorder.dispatched[0].parentJournalId === "w1:unit#1",215+ "the child must be tied to its parent obligation identity",216+ );195217 clearEffectDeclarations();196218 });197219 @@ -205,7 +227,11 @@ Deno.test("webhook injects the journal id as the idempotency key and posts the f
205227 return Promise.resolve(new Response(null, { status: 200 }));206228 }) as typeof fetch;207229 try {208- const unit = webhook("https://hooks.example.com/claims");230+ const unit = webhook("claims", "https://hooks.example.com/claims", {231+ headers: { authorization: "secret", "idempotency-key": "override" },232+ });233+ assert(unit.effectName === "webhook:claims", "durable identity must contain only the name");234+ assert(!unit.effectName.includes("secret"), "header secrets must not enter durable identity");209235 await unit.handlers.onSynced?.({ id: "row-1" }, context({ journalId: "w1:webhook#1" }));210236 assertCount(requests.length, 1, "a synced write posts once");211237 const key = new Headers(requests[0].init.headers as HeadersInit).get("idempotency-key") ?? "";@@ -226,7 +252,7 @@ Deno.test("webhook treats a 4xx as permanent and a 5xx as retryable", async () =
226252 globalThis.fetch = (() => Promise.resolve(new Response(null, { status }))) as typeof fetch;227253 };228254 try {229- const unit = webhook("https://hooks.example.com/claims");255+ const unit = webhook("claims", "https://hooks.example.com/claims");230256 respond(400);231257 let permanent = false;232258 try {@@ -277,12 +303,29 @@ Deno.test("a custom unit on the public contract is idempotent by journal id", as
277303 Deno.test("webhook defaults to a finite 24-hour delivery window", () => {278304 clearEffectDeclarations();279305 install();280- const unit = webhook("https://hooks.example.com/claims");306+ const unit = webhook("claims", "https://hooks.example.com/claims");281307 assert(unit.expiresAfterMs === 24 * 60 * 60 * 1000, "external delivery must default to 24h");282- const forever = webhook("https://hooks.example.com/claims2", { expiresAfterMs: null });308+ const forever = webhook("claims-forever", "https://hooks.example.com/claims2", {309+ expiresAfterMs: null,310+ });283311 assert(284312 forever.expiresAfterMs === null || forever.expiresAfterMs === undefined,285313 "null must opt into no expiry",286314 );287315 clearEffectDeclarations();288316 });317+318+Deno.test("webhook default expiry and explicit no-expiry do not alias", () => {319+ clearEffectDeclarations();320+ install();321+ const url = "https://hooks.example.com/expiry";322+ webhook("expiry", url);323+ let thrown = false;324+ try {325+ webhook("expiry", url, { expiresAfterMs: null });326+ } catch {327+ thrown = true;328+ }329+ assert(thrown, "explicit null must not reuse the default 24-hour configuration");330+ clearEffectDeclarations();331+});@@ -276,6 +276,12 @@ export type NoticeInput = {
276276 export type MutationRuntime = {277277 /** Performs one declared verb call and returns its write handle. */278278 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>;279285 /** Records one structured {@link log} entry. */280286 recordLog(label: string, context: EffectContext): void;281287 /** Records one {@link trace} span from the write's journaling to its fate. */@@ -283,7 +289,7 @@ export type MutationRuntime = {
283289 /** Records one {@link debug} timeline event; a no-op in production builds. */284290 recordDebug?(event: string, context: EffectContext): void;285291 /** Enqueues one durable {@link notice} entry. */286- enqueueNotice?(input: NoticeInput, context: EffectContext): void;292+ enqueueNotice?(input: NoticeInput, context: EffectContext): Promise<void>;287293 /**288294 * Patches a row on behalf of a {@link mark} unit: a bare update carrying no289295 * further units, so write fate becomes replicated row data without@@ -303,8 +309,13 @@ type EffectsSlot = {
303309 effects: Map<string, EffectUnit<{ id: string }>>;304310 verbs: Map<string, MutationDescriptor>;305311 logs: Map<string, EffectUnit<{ id: string }>>;306- /** Cache for content-named built-ins (trace, debug, webhook) shared by name. */307- cache: Map<string, EffectUnit<{ id: string }>>;312+ /** Cache for shared built-ins (trace, debug, webhook) keyed by durable name. */313+ cache: Map<string, CachedBuiltinEntry>;314+};315+316+type CachedBuiltinEntry = {317+ unit: EffectUnit<{ id: string }>;318+ signature: string | null;308319 };309320 310321 const slotName = "__LOFI_EFFECT_DECLARATIONS__";@@ -486,26 +497,40 @@ export function anonymousUnit<Row extends { id: string }>(
486497 487498 /**488499 * Returns the shared built-in unit registered under `name`, building and489- * registering it on first use. For content-named observation and external490- * units (`s.trace`, `s.debug`, `s.webhook`) whose whole identity is their491- * name, so reusing one across verbs shares a single unit instead of colliding492- * — the same discipline {@link log} follows.500+ * 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.493506 */494507 export function cachedBuiltin<Row extends { id: string }>(495508 name: string,496509 build: () => EffectHandlers<Row>,497- options: EffectUnitOptions = {},510+ options: Omit<EffectUnitOptions, "expiresAfterMs"> & { expiresAfterMs?: number | null } = {},511+ signature?: string,498512 ): EffectUnit<Row> {499513 const state = slot();500514 const existing = state.cache.get(name);501- if (existing) return existing as unknown as EffectUnit<Row>;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);502524 const unit: EffectUnit<Row> = {503525 effectName: name,504526 handlers: build(),505527 expiresAfterMs: options.expiresAfterMs ?? null,506528 ...(options.maxAttempts !== undefined ? { maxAttempts: options.maxAttempts } : {}),507529 };508- state.cache.set(name, unit as unknown as EffectUnit<{ id: string }>);530+ state.cache.set(name, {531+ unit: unit as unknown as EffectUnit<{ id: string }>,532+ signature: signature ?? null,533+ });509534 registerUnit(unit as unknown as EffectUnit<{ id: string }>);510535 return unit;511536 }@@ -554,9 +579,15 @@ export function mutation<T extends { id: string }, Init, Kind extends MutationOp
554579 }555580 const units: EffectUnit<{ id: string }>[] = [];556581 const seen = new Set<string>();582+ const seenDeclarations = new Set<EffectUnit<T>>();557583 const declared = options.effects ?? [];558584 for (let index = 0; index < declared.length; index += 1) {559- let unit = declared[index] as EffectUnit<{ id: string }>;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 }>;560591 if (unit.anonymousPrefix !== undefined) {561592 // An anonymous built-in (s.notice/s.mark/s.chain) is named and562593 // registered here, from the verb it is attached to and its position in@@ -592,5 +623,23 @@ export function mutation<T extends { id: string }, Init, Kind extends MutationOp
592623 };593624 state.verbs.set(name, descriptor);594625 const verb = (...args: readonly unknown[]) => requireRuntime().dispatch(descriptor, args);626+ Object.defineProperty(verb, mutationDescriptor, { value: descriptor });595627 return verb as MutationVerb<MutationOp<T, Init, Kind>>;596628 }629+630+const mutationDescriptor = Symbol.for("@nzip/lofi/mutationDescriptor");631+632+/** Dispatches a declared mutation as a deterministic child of an effect. */633+export 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+}@@ -19,6 +19,9 @@ function installRecorder(): { dispatched: Dispatched[]; logs: string[] } {
1919 dispatched.push({ descriptor, args });2020 return { stage: "saving" } as unknown as WriteHandle<unknown>;2121 },22+ dispatchChained() {23+ return Promise.resolve();24+ },2225 recordLog(label) {2326 logs.push(label);2427 },@@ -99,6 +102,19 @@ Deno.test("verb names are unique per app and collide fast", () => {
99102 clearEffectDeclarations();100103 });101104 105+Deno.test("one anonymous unit object cannot occupy two mutation positions", () => {106+ clearEffectDeclarations();107+ const unit = s.notice({ synced: "Saved." });108+ let thrown = false;109+ try {110+ s.mutation("duplicateNotice", s.insert(app.orders), { effects: [unit, unit] });111+ } catch {112+ thrown = true;113+ }114+ assert(thrown, "reusing one anonymous declaration must fail before it gains two identities");115+ clearEffectDeclarations();116+});117+102118 Deno.test("s.log reuses one unit per label and records through the runtime", () => {103119 clearEffectDeclarations();104120 const recorder = installRecorder();