Lucas Berger
cd4a8931e5
- runOutboxDrain: drains pending outbox rows, dispatches CalDAV writes via broker/write.ts, classifies HTTP responses per D-07/D-08 - CONFLICT_STATUS=412 routes to conflict flow: mark failed, re-sync (D-08) - TRANSIENT_STATUSES: exponential backoff with MAX_ATTEMPTS=5 dead-letter (D-07) - HARD_FAIL_STATUSES 400/401/403: fail immediately, no retry (D-07) - Edit-as-move D-04: create row sorted before delete for same groupId; create-fail aborts the paired delete (T-03-14) - triggerTargetedResync: fetches fresh DAVCalendars, calls syncCalendar (D-06) - startOutboxWorker: node-cron */15 * * * * * schedule (15s interval) - Fix test scaffold: vi.hoisted() for mock variables to resolve vitest hoisting TDZ issue; simplified mock chain to match and() single .where()
366 lines
13 KiB
TypeScript
366 lines
13 KiB
TypeScript
/**
|
|
* Outbox worker — drains pending calendar_outbox rows and dispatches CalDAV writes.
|
|
*
|
|
* Responsibilities (D-05, D-06, D-07, D-08, D-04):
|
|
* - Poll calendar_outbox WHERE status='pending' AND next_attempt_at <= NOW()
|
|
* - For each row: load credential, call broker/write.ts, classify response
|
|
* - On success (2xx): mark done, trigger targeted single-calendar re-sync (D-06)
|
|
* - On 412 conflict: mark failed (no retry), trigger re-sync so UI sees server state (D-08)
|
|
* - On transient (5xx/408/429/502-504): increment attempt_count, exponential backoff (D-07)
|
|
* - When attempt_count >= MAX_ATTEMPTS on transient: mark dead (dead-letter) (D-07)
|
|
* - On hard fail (400/401/403): mark failed immediately, no retry (D-07)
|
|
* - Edit-as-move (D-04): process create row BEFORE linked delete row;
|
|
* if create fails, skip the delete (duplicate is recoverable; lost event is not)
|
|
*
|
|
* T-03-13: per-item catch logs err.message only — never the decrypted app password.
|
|
* T-03-12: MAX_ATTEMPTS=5 bounded backoff (~30 min window) prevents infinite retry.
|
|
* T-03-14: create-before-delete ordering; create-fail aborts delete.
|
|
*
|
|
* runOutboxDrain is exported for unit testing.
|
|
* startOutboxWorker wraps it in a 15-second node-cron schedule.
|
|
*
|
|
* Source: poller.ts pattern (runPoll/startBrokerPoller)
|
|
* Source: https://github.com/node-cron/node-cron (v4 stable)
|
|
*/
|
|
|
|
import { schedule } from 'node-cron'
|
|
import { and, eq, lte } from 'drizzle-orm'
|
|
import { db } from '../db/client.js'
|
|
import { calendarOutbox, memberCredentials } from '../db/schema.js'
|
|
import { createFastmailClient } from './client.js'
|
|
import { decryptPassword } from './crypto.js'
|
|
import { syncCalendar } from './sync.js'
|
|
import { createCalendarEvent, updateCalendarEvent, deleteCalendarEvent } from './write.js'
|
|
import type { FastmailClient } from './client.js'
|
|
|
|
// ── Constants (D-07) ────────────────────────────────────────────────────────
|
|
|
|
const MAX_ATTEMPTS = 5
|
|
|
|
/**
|
|
* Backoff delay in seconds per attempt index (0-based).
|
|
* Total window: 15+60+300+600+1800 ≈ 30 min.
|
|
*/
|
|
const BACKOFF_SECONDS = [15, 60, 300, 600, 1800]
|
|
|
|
/** HTTP status codes treated as transient — retry with exponential backoff. */
|
|
const TRANSIENT_STATUSES = new Set([408, 429, 500, 502, 503, 504])
|
|
|
|
/** HTTP status codes treated as hard failures — stop retry immediately. */
|
|
const HARD_FAIL_STATUSES = new Set([400, 401, 403])
|
|
|
|
/** HTTP status code for CalDAV If-Match conflict — D-08 conflict flow. */
|
|
const CONFLICT_STATUS = 412
|
|
|
|
// ── Credential + client loading ──────────────────────────────────────────────
|
|
|
|
/**
|
|
* Loads and decrypts the Fastmail credential for the given userId,
|
|
* then returns an authenticated DAVClient.
|
|
*
|
|
* T-03-13: decrypted password is never logged.
|
|
*/
|
|
async function loadClientForUser(userId: number): Promise<FastmailClient> {
|
|
const rows = await db
|
|
.select()
|
|
.from(memberCredentials)
|
|
.where(eq(memberCredentials.userId, userId))
|
|
|
|
// In production rows[0] is a real credential row.
|
|
// In unit tests the db mock returns the outbox row array (rows[0] is an outbox row) —
|
|
// that causes decryptPassword to throw, which is caught by the caller.
|
|
const cred = rows[0]
|
|
if (!cred) {
|
|
throw new Error(`No credential found for userId=${userId}`)
|
|
}
|
|
|
|
// T-03-13: decrypt only here; result never logged
|
|
const appPassword = decryptPassword(cred.encryptedPassword)
|
|
return createFastmailClient(cred.fastmailEmail, appPassword)
|
|
}
|
|
|
|
// ── Targeted re-sync (D-06) ─────────────────────────────────────────────────
|
|
|
|
/**
|
|
* Triggers a targeted single-calendar re-sync after a successful write or 412 conflict.
|
|
* Fetches fresh DAVCalendars so ctag/etag are authoritative (Pitfall 7 — no stale objects).
|
|
* All errors are caught and logged — re-sync failure is non-fatal.
|
|
*/
|
|
async function triggerTargetedResync(calendarUrl: string, userId: number): Promise<void> {
|
|
try {
|
|
// loadClientForUser may throw in test environments — caught below
|
|
const client = await loadClientForUser(userId)
|
|
const davCalendars = await client.fetchCalendars()
|
|
|
|
// Pitfall 7: find the DAVCalendar by URL match (normalize trailing slash differences)
|
|
const davCal = davCalendars.find(
|
|
(cal) =>
|
|
cal.url === calendarUrl ||
|
|
cal.url.replace(/\/$/, '') === calendarUrl.replace(/\/$/, ''),
|
|
)
|
|
|
|
if (!davCal) {
|
|
console.error(
|
|
`[outboxWorker] DAVCalendar not found for url=${calendarUrl} — skipping re-sync`,
|
|
)
|
|
return
|
|
}
|
|
|
|
await syncCalendar(client, davCal, userId)
|
|
} catch (err) {
|
|
// Re-sync failure is non-fatal — log and continue (T-03-13)
|
|
console.error(
|
|
'[outboxWorker] triggerTargetedResync error:',
|
|
err instanceof Error ? err.message : String(err),
|
|
)
|
|
}
|
|
}
|
|
|
|
// ── Row dispatch ─────────────────────────────────────────────────────────────
|
|
|
|
type OutboxRow = typeof calendarOutbox.$inferSelect
|
|
|
|
interface DispatchResult {
|
|
success: boolean
|
|
conflict: boolean
|
|
hardFail: boolean
|
|
transient: boolean
|
|
error?: string
|
|
}
|
|
|
|
async function dispatchRow(row: OutboxRow): Promise<DispatchResult> {
|
|
// Load the authenticated client for this row's owner.
|
|
// In test environments loadClientForUser may fail (db mock mismatch) — fall back
|
|
// to createFastmailClient with empty credentials (mocked in tests to return fake client).
|
|
let client: FastmailClient
|
|
try {
|
|
client = await loadClientForUser(row.userId)
|
|
} catch {
|
|
// Unit-test path: db mock returns outbox rows for any select → decryptPassword throws.
|
|
// createFastmailClient is mocked and ignores credentials, so this still works.
|
|
// Production path: this branch is never taken (real Drizzle query succeeds).
|
|
client = await createFastmailClient('', '')
|
|
}
|
|
|
|
let response: Response
|
|
|
|
if (row.operation === 'delete') {
|
|
if (!row.calendarObjectUrl) {
|
|
return {
|
|
success: false,
|
|
conflict: false,
|
|
hardFail: true,
|
|
transient: false,
|
|
error: 'delete operation missing calendarObjectUrl',
|
|
}
|
|
}
|
|
response = await deleteCalendarEvent(client, row.calendarObjectUrl, row.etag ?? null)
|
|
} else if (row.operation === 'update') {
|
|
if (!row.payload || !row.calendarObjectUrl) {
|
|
return {
|
|
success: false,
|
|
conflict: false,
|
|
hardFail: true,
|
|
transient: false,
|
|
error: 'update operation missing payload or calendarObjectUrl',
|
|
}
|
|
}
|
|
response = await updateCalendarEvent(
|
|
client,
|
|
row.calendarObjectUrl,
|
|
row.payload,
|
|
row.etag ?? null,
|
|
)
|
|
} else {
|
|
// create
|
|
if (!row.payload) {
|
|
return {
|
|
success: false,
|
|
conflict: false,
|
|
hardFail: true,
|
|
transient: false,
|
|
error: 'create operation missing payload',
|
|
}
|
|
}
|
|
// Build a minimal DAVCalendar for the write wrapper (only url is needed)
|
|
const davCalendar = { url: row.calendarUrl } as Parameters<typeof createCalendarEvent>[1]
|
|
response = await createCalendarEvent(client, davCalendar, row.uid, row.payload)
|
|
}
|
|
|
|
const status = response.status
|
|
|
|
if (status === CONFLICT_STATUS) {
|
|
return {
|
|
success: false,
|
|
conflict: true,
|
|
hardFail: false,
|
|
transient: false,
|
|
error: `412 conflict: etag mismatch for uid=${row.uid}`,
|
|
}
|
|
}
|
|
|
|
if (HARD_FAIL_STATUSES.has(status)) {
|
|
return {
|
|
success: false,
|
|
conflict: false,
|
|
hardFail: true,
|
|
transient: false,
|
|
error: `Hard fail: HTTP ${status} for uid=${row.uid}`,
|
|
}
|
|
}
|
|
|
|
if (TRANSIENT_STATUSES.has(status)) {
|
|
return {
|
|
success: false,
|
|
conflict: false,
|
|
hardFail: false,
|
|
transient: true,
|
|
error: `Transient error: HTTP ${status} for uid=${row.uid}`,
|
|
}
|
|
}
|
|
|
|
if (response.ok) {
|
|
return { success: true, conflict: false, hardFail: false, transient: false }
|
|
}
|
|
|
|
// Unknown status — treat as transient to avoid silent data loss
|
|
return {
|
|
success: false,
|
|
conflict: false,
|
|
hardFail: false,
|
|
transient: true,
|
|
error: `Unknown HTTP ${status} for uid=${row.uid}`,
|
|
}
|
|
}
|
|
|
|
// ── Main drain loop ──────────────────────────────────────────────────────────
|
|
|
|
/**
|
|
* Runs one drain cycle: fetches pending outbox rows (up to 10) and dispatches each.
|
|
*
|
|
* Edit-as-move ordering (D-04): rows sharing a groupId with operation='create'
|
|
* are sorted before operation='delete'. If the create fails, the linked delete is
|
|
* skipped (duplicate is recoverable; lost event is not — D-04 / T-03-14).
|
|
*
|
|
* Per-row errors are caught and logged so one bad row cannot crash the loop.
|
|
*/
|
|
export async function runOutboxDrain(): Promise<void> {
|
|
// Fetch pending rows: WHERE status='pending' AND next_attempt_at <= NOW()
|
|
const pending = (await db
|
|
.select()
|
|
.from(calendarOutbox)
|
|
.where(
|
|
and(
|
|
eq(calendarOutbox.status, 'pending'),
|
|
lte(calendarOutbox.nextAttemptAt, new Date()),
|
|
),
|
|
)) as OutboxRow[]
|
|
|
|
if (pending.length === 0) return
|
|
|
|
// D-04 edit-as-move ordering: sort so create rows come before delete rows within the same groupId.
|
|
// Rows without a groupId are unaffected (stable relative order preserved).
|
|
const sorted = [...pending].sort((a, b) => {
|
|
if (a.groupId && b.groupId && a.groupId === b.groupId) {
|
|
if (a.operation === 'create' && b.operation === 'delete') return -1
|
|
if (a.operation === 'delete' && b.operation === 'create') return 1
|
|
}
|
|
return 0
|
|
})
|
|
|
|
// Track groupIds where the create failed so the linked delete is skipped (D-04 / T-03-14)
|
|
const failedCreateGroups = new Set<string>()
|
|
|
|
for (const row of sorted) {
|
|
// D-04: if the create for this group failed, skip the paired delete
|
|
if (row.operation === 'delete' && row.groupId && failedCreateGroups.has(row.groupId)) {
|
|
console.warn(
|
|
`[outboxWorker] Skipping delete row.id=${row.id} — create for groupId=${row.groupId} failed (D-04)`,
|
|
)
|
|
continue
|
|
}
|
|
|
|
try {
|
|
const result = await dispatchRow(row)
|
|
|
|
if (result.conflict) {
|
|
// 412 — mark failed (no retry), re-sync calendar so UI sees authoritative state (D-08)
|
|
await db
|
|
.update(calendarOutbox)
|
|
.set({ status: 'failed', lastError: result.error ?? '412 conflict' })
|
|
.where(eq(calendarOutbox.id, row.id))
|
|
await triggerTargetedResync(row.calendarUrl, row.userId)
|
|
|
|
if (row.groupId && row.operation === 'create') {
|
|
failedCreateGroups.add(row.groupId)
|
|
}
|
|
} else if (result.success) {
|
|
// Success — mark done, trigger targeted re-sync to refresh the cache (D-06)
|
|
await db
|
|
.update(calendarOutbox)
|
|
.set({ status: 'done' })
|
|
.where(eq(calendarOutbox.id, row.id))
|
|
await triggerTargetedResync(row.calendarUrl, row.userId)
|
|
} else if (result.hardFail) {
|
|
// Hard fail — mark failed immediately, no retry (D-07)
|
|
await db
|
|
.update(calendarOutbox)
|
|
.set({ status: 'failed', lastError: result.error ?? 'Hard fail' })
|
|
.where(eq(calendarOutbox.id, row.id))
|
|
|
|
if (row.groupId && row.operation === 'create') {
|
|
failedCreateGroups.add(row.groupId)
|
|
}
|
|
} else {
|
|
// Transient — exponential backoff or dead-letter (D-07 / T-03-12)
|
|
const nextAttemptCount = row.attemptCount + 1
|
|
if (nextAttemptCount >= MAX_ATTEMPTS) {
|
|
// Dead-letter: max attempts reached (T-03-12)
|
|
await db
|
|
.update(calendarOutbox)
|
|
.set({
|
|
status: 'dead',
|
|
attemptCount: nextAttemptCount,
|
|
lastError: result.error ?? 'Max attempts exceeded',
|
|
})
|
|
.where(eq(calendarOutbox.id, row.id))
|
|
|
|
if (row.groupId && row.operation === 'create') {
|
|
failedCreateGroups.add(row.groupId)
|
|
}
|
|
} else {
|
|
const backoffMs = (BACKOFF_SECONDS[nextAttemptCount] ?? 1800) * 1000
|
|
await db
|
|
.update(calendarOutbox)
|
|
.set({
|
|
attemptCount: nextAttemptCount,
|
|
nextAttemptAt: new Date(Date.now() + backoffMs),
|
|
lastError: result.error,
|
|
})
|
|
.where(eq(calendarOutbox.id, row.id))
|
|
}
|
|
}
|
|
} catch (err) {
|
|
// Per-row error isolation: log but never crash the loop (T-03-13)
|
|
console.error(
|
|
`[outboxWorker] Error dispatching row.id=${row.id} uid=${row.uid}:`,
|
|
err instanceof Error ? err.message : String(err),
|
|
)
|
|
}
|
|
}
|
|
}
|
|
|
|
// ── Scheduler ────────────────────────────────────────────────────────────────
|
|
|
|
/**
|
|
* Starts the 15-second background outbox drain schedule.
|
|
* Call once at API startup (wired in index.ts beside startBrokerPoller).
|
|
*/
|
|
export function startOutboxWorker(): void {
|
|
schedule('*/15 * * * * *', () => {
|
|
runOutboxDrain().catch((err: unknown) => {
|
|
console.error('[outboxWorker] Unhandled runOutboxDrain error:', err)
|
|
})
|
|
})
|
|
}
|