familysync/.planning/phases/09-faster-write-back/09-PATTERNS.md

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):

import { EventEmitter } from 'node:events';

Module-level singleton pattern (lines 22–23):

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):

export function publishListEvent(listId: number, event: ListEvent): void {
  emitter.emit(`list:${listId}`, event);
}

For outboxTrigger, adapt as:

export function signalOutboxDrain(): void {
  emitter.emit('drain');
}

Subscribe + unsubscribe pattern (lines 47–54):

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:

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):

let isDraining = false;

New drainRequested flag goes immediately after, at the same module scope:

let drainRequested = false;

Existing runOutboxDrain guard + finally (lines 602–605, 783–785):

export async function runOutboxDrain(): Promise<void> {
  // 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:

/**
 * 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:

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:

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):

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:

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:

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):

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):

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):

import { startOutboxWorker } from './broker/outboxWorker.js';

Update to:

import { startOutboxWorker, initOutboxTrigger } from './broker/outboxWorker.js';

Existing startup block (lines 112, 136–141):

if (isMainModule()) {
  // ... VAPID setup ...
  startBrokerPoller();
  startOutboxWorker();
  startReminderScheduler();
  serve({ fetch: app.fetch, port: 3000 }, ...);
}

Add initOutboxTrigger() immediately after startOutboxWorker():

  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):

import { runOutboxDrain, assembleRruleString } from '../../src/broker/outboxWorker.js';

Extend to include new exports:

import { runOutboxDrain, assembleRruleString, scheduleOutboxDrain } from '../../src/broker/outboxWorker.js';

Also import the signal function for trigger wiring tests:

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):

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):

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):

// Flush microtask queue so the promise chain from scheduleOutboxDrain completes
await new Promise<void>(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:

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

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)

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