# Phase 9: Faster Write-Back — Pattern Map **Mapped:** 2026-06-12 **Files analyzed:** 5 (1 new, 4 modified) **Analogs found:** 5 / 5 --- ## File Classification | New/Modified File | Role | Data Flow | Closest Analog | Match Quality | |---|---|---|---|---| | `apps/api/src/lib/outboxTrigger.ts` | utility / singleton | event-driven | `apps/api/src/lib/listEmitter.ts` | exact | | `apps/api/src/broker/outboxWorker.ts` | service / broker | event-driven + batch | `apps/api/src/broker/outboxWorker.ts` (self — extend) | exact | | `apps/api/src/routes/events.ts` | route / controller | request-response | `apps/api/src/routes/events.ts` (self — extend) | exact | | `apps/api/src/index.ts` | config / startup | request-response | `apps/api/src/index.ts` (self — extend) | exact | | `apps/api/tests/broker/outboxWorker.test.ts` | test | batch | `apps/api/tests/broker/outboxWorker.test.ts` (self — extend) | exact | --- ## Pattern Assignments ### `apps/api/src/lib/outboxTrigger.ts` (NEW — utility, event-driven) **Analog:** `apps/api/src/lib/listEmitter.ts` **Imports pattern** (lines 18): ```typescript import { EventEmitter } from 'node:events'; ``` **Module-level singleton pattern** (lines 22–23): ```typescript const emitter = new EventEmitter(); emitter.setMaxListeners(200); // listEmitter sets 200 for fan-out; outboxTrigger needs only 1 // NOTE: for outboxTrigger, setMaxListeners is NOT needed — single subscriber, default 10 is fine ``` **Publish function pattern** (lines 39–41): ```typescript export function publishListEvent(listId: number, event: ListEvent): void { emitter.emit(`list:${listId}`, event); } ``` For outboxTrigger, adapt as: ```typescript export function signalOutboxDrain(): void { emitter.emit('drain'); } ``` **Subscribe + unsubscribe pattern** (lines 47–54): ```typescript export function subscribeListEvents( listId: number, handler: (event: ListEvent) => void, ): () => void { const channel = `list:${listId}`; emitter.on(channel, handler); return () => emitter.off(channel, handler); } ``` For outboxTrigger, adapt as: ```typescript export function onOutboxDrain(handler: () => void): () => void { emitter.on('drain', handler); return () => emitter.off('drain', handler); } ``` **No imports from broker/, routes/, db/, or index.ts** — `listEmitter.ts` also imports nothing from the project. `outboxTrigger.ts` must follow the same zero-internal-dependency rule to avoid circular imports. --- ### `apps/api/src/broker/outboxWorker.ts` (MODIFIED — service/broker, event-driven + batch) **Analog:** self (extend existing file) **Existing `isDraining` flag declaration** (line 159): ```typescript let isDraining = false; ``` New `drainRequested` flag goes immediately after, at the same module scope: ```typescript let drainRequested = false; ``` **Existing `runOutboxDrain` guard + finally** (lines 602–605, 783–785): ```typescript export async function runOutboxDrain(): Promise { // CR-05: single-process concurrency guard if (isDraining) return; isDraining = true; // ... (do not touch the body) } finally { isDraining = false; } } ``` `runOutboxDrain` is NOT modified — `scheduleOutboxDrain` wraps it externally. **New `scheduleOutboxDrain` wrapper** — add between `isDraining` declaration and `runOutboxDrain`: ```typescript /** * Scheduler entrypoint called by both the EventEmitter listener (initOutboxTrigger) * and the 15s setInterval. Implements the drain-again-if-requested loop (D-05) so * the last edit in a burst is drained promptly without double-draining (D-02 / D-07). * Fire-and-forget — callers do not await this. */ export function scheduleOutboxDrain(): void { if (isDraining) { drainRequested = true; return; } runOutboxDrain() .catch((err: unknown) => { console.error('[outboxWorker] Unhandled runOutboxDrain error:', err); }) .finally(() => { if (drainRequested) { drainRequested = false; // reset FIRST, then recurse (Pitfall 3) scheduleOutboxDrain(); // exactly one trailing re-drain (D-05) } }); } ``` **`initOutboxTrigger` export** — add in the Scheduler section alongside `startOutboxWorker`: ```typescript import { onOutboxDrain } from '../lib/outboxTrigger.js'; export function initOutboxTrigger(): void { onOutboxDrain(() => scheduleOutboxDrain()); } ``` **Existing `startOutboxWorker`** (lines 796–802) — update to call `scheduleOutboxDrain` instead of `runOutboxDrain` directly: ```typescript export function startOutboxWorker(): void { setInterval(() => { scheduleOutboxDrain(); }, 15 * 1000); } ``` (Existing `.catch` wrapper is absorbed into `scheduleOutboxDrain` — remove the per-call `.catch` in the interval body.) **Export rationale comment** (line 19 in file header): ``` * runOutboxDrain is exported for unit testing. * startOutboxWorker wraps it in a 15-second setInterval. ``` Add: `scheduleOutboxDrain is exported for unit testing.` --- ### `apps/api/src/routes/events.ts` (MODIFIED — route/controller, request-response) **Analog:** self (extend existing file) **Import to add** (after existing imports, lines 24–35): ```typescript import { signalOutboxDrain } from '../lib/outboxTrigger.js'; ``` **Site 1 — POST /create, single insert** (lines 300–309). Insert is a bare `await db.insert(...)` (no transaction). Signal fires after the awaited insert, before the `return`: ```typescript await db.insert(calendarOutbox).values({ userId: currentUserId, operation: 'create', status: 'pending', uid, calendarUrl: targetCalendarUrl, payload: JSON.stringify(payload), }); signalOutboxDrain(); // fire-and-forget (D-04) return c.json({ uid }, 202); ``` **Site 2 — PATCH /:uid/edit, edit-as-move transaction** (lines 408–432). Signal fires AFTER `await db.transaction(...)` resolves — NOT inside the callback: ```typescript await db.transaction(async (tx) => { await tx.insert(calendarOutbox).values({ operation: 'delete', ... }); await tx.insert(calendarOutbox).values({ operation: 'create', ... }); }); signalOutboxDrain(); // after transaction commits — both rows durable (D-03) return c.json({ uid: newUid }, 202); ``` **Site 3 — PATCH /:uid/edit, same-calendar update** (lines 436–447): ```typescript await db.insert(calendarOutbox).values({ operation: 'update', ... }); signalOutboxDrain(); // fire-and-forget (D-04) return c.json({ uid }, 202); ``` **Site 4 — DELETE /:uid** (lines 511–521): ```typescript await db.insert(calendarOutbox).values({ operation: 'delete', ... }); signalOutboxDrain(); // fire-and-forget (D-04) return c.json({ uid }, 202); ``` All four sites are inside existing `try/catch` blocks — the signal call slots between the `await` and the `return`, matching the same pattern at each site. --- ### `apps/api/src/index.ts` (MODIFIED — config/startup, request-response) **Analog:** self (extend existing file) **Existing import** (line 16): ```typescript import { startOutboxWorker } from './broker/outboxWorker.js'; ``` Update to: ```typescript import { startOutboxWorker, initOutboxTrigger } from './broker/outboxWorker.js'; ``` **Existing startup block** (lines 112, 136–141): ```typescript if (isMainModule()) { // ... VAPID setup ... startBrokerPoller(); startOutboxWorker(); startReminderScheduler(); serve({ fetch: app.fetch, port: 3000 }, ...); } ``` Add `initOutboxTrigger()` immediately after `startOutboxWorker()`: ```typescript startOutboxWorker(); initOutboxTrigger(); // subscribe drain signal listener (D-01) startReminderScheduler(); ``` **`isMainModule()` function** (lines 100–107) — do not modify. The trigger listener must remain gated by this guard (same as all other background workers) to prevent listener registration when tests import `app`. --- ### `apps/api/tests/broker/outboxWorker.test.ts` (MODIFIED — test, batch) **Analog:** self (extend existing file) **Existing import line** (line 21): ```typescript import { runOutboxDrain, assembleRruleString } from '../../src/broker/outboxWorker.js'; ``` Extend to include new exports: ```typescript import { runOutboxDrain, assembleRruleString, scheduleOutboxDrain } from '../../src/broker/outboxWorker.js'; ``` Also import the signal function for trigger wiring tests: ```typescript import { signalOutboxDrain } from '../../src/lib/outboxTrigger.js'; ``` **`vi.hoisted` + `vi.mock` scaffold** (lines 31–101) — copy verbatim into new describe block. The `mockPendingRows`, `makeRow`, `makeResponse`, and `wireMockChain` helpers are already at module scope and are reusable from the new tests without re-declaration. **`beforeEach` pattern** (lines 170–174): ```typescript beforeEach(() => { vi.resetAllMocks(); mockPendingRows = []; wireMockChain(); }); ``` New describe block uses the same `beforeEach`. Note: `vi.resetAllMocks()` clears all mock state between tests — adequate for trigger wiring tests since `isDraining` and `drainRequested` are module-level state that resets between test runs only if the module is re-imported. Use `vi.isolateModules` or reset state explicitly if module-level flags need isolation between tests. **`makeResponse` helper** (line 131–132): ```typescript const makeResponse = (status: number): Response => ({ status, ok: status >= 200 && status < 300, headers: new Headers() }) as unknown as Response; ``` Reuse in new tests — already in scope. **Async flush pattern for signal→drain tests** (no fake timers needed): ```typescript // Flush microtask queue so the promise chain from scheduleOutboxDrain completes await new Promise(resolve => setImmediate(resolve)); ``` **`vi.useFakeTimers()` pattern** — needed only for tests that advance the 15s interval. Not needed for pure signal→drain tests. See existing `startOutboxWorker` interval tests for the pattern if needed. **New describe block placement** — add after the final existing `describe` block, before EOF: ```typescript describe('scheduleOutboxDrain — trigger wiring (D-09)', () => { beforeEach(() => { vi.resetAllMocks(); mockPendingRows = []; wireMockChain(); }); // Test A: signal → drain without timer wait (SC-1) // Test B: signal mid-drain → exactly one trailing re-drain (D-05) // Test C: concurrent scheduleOutboxDrain calls → each row dispatched once (D-07) }); ``` --- ## Shared Patterns ### `isMainModule()` startup gate **Source:** `apps/api/src/index.ts` lines 100–107 + 112 **Apply to:** `initOutboxTrigger()` call in index.ts ```typescript function isMainModule(): boolean { if (!process.argv[1]) return false; try { return fileURLToPath(import.meta.url) === realpathSync(process.argv[1]); } catch { return false; } } if (isMainModule()) { // all background worker starts here } ``` All listener registrations and `setInterval` calls must be inside this guard. `initOutboxTrigger` follows the same gating as `startBrokerPoller`, `startOutboxWorker`, `startReminderScheduler`. ### Error swallowing in setInterval workers **Source:** `apps/api/src/broker/outboxWorker.ts` lines 797–801 **Apply to:** `scheduleOutboxDrain` wrapper (the `.catch` on the `runOutboxDrain()` call) ```typescript runOutboxDrain().catch((err: unknown) => { console.error('[outboxWorker] Unhandled runOutboxDrain error:', err); }); ``` `scheduleOutboxDrain` absorbs this pattern into its own `.catch` — the interval body becomes a bare `scheduleOutboxDrain()` call with no inline error handling. ### Module-level singleton + typed publish/subscribe API **Source:** `apps/api/src/lib/listEmitter.ts` lines 18–54 **Apply to:** `apps/api/src/lib/outboxTrigger.ts` Pattern: `const emitter = new EventEmitter()` at module scope; named export functions wrap all emitter access; no direct EventEmitter reference leaks to callers. `outboxTrigger.ts` follows this exactly. --- ## No Analog Found None. All five files have direct analogs (three are self-extensions of existing files). --- ## Metadata **Analog search scope:** `apps/api/src/lib/`, `apps/api/src/broker/`, `apps/api/src/routes/`, `apps/api/src/`, `apps/api/tests/broker/` **Files read:** `listEmitter.ts`, `outboxWorker.ts`, `events.ts`, `index.ts`, `tests/broker/outboxWorker.test.ts` **Pattern extraction date:** 2026-06-12