/**
 * 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 setInterval.
 * (node-cron 4.2.1 silently skipped scheduled executions in the long-running server process;
 * setInterval fires reliably in the same process — replaced to fix the silent skip.)
 *
 * Source: poller.ts pattern (runPoll/startBrokerPoller)
 */
import { z } from 'zod';
import { and, eq, lte } from 'drizzle-orm';
import { db } from '../db/client.js';
import { calendarEvents, calendarOutbox, calendars, 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 { buildVeventString, extractRruleString, RRULE_PRESETS } from './vevent.js';
import type { FastmailClient } from './client.js';
import { dispatchEventChange } from '../lib/eventChangeDispatcher.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;

// ── Outbox payload re-validation (IN-03) ─────────────────────────────────────

/**
 * IN-03: re-validate the JSON payload read back out of calendar_outbox before building
 * a VEVENT from it. The payload was zod-validated at enqueue (routes/events.ts
 * eventFieldsSchema), but a manually-inserted row or enqueue→drain schema drift could
 * feed undefined/wrong-typed fields into buildVeventString, producing SUMMARY:undefined
 * or an Invalid Date. Such a row can NEVER succeed, so on validation failure the caller
 * hard-fails the row (no retry) instead of burning the backoff budget.
 *
 * Mirrors eventFieldsSchema in routes/events.ts. `_preservedRrule` (added by the edit-as-
 * move route, CR-01) is allowed via .passthrough() so the move payload still validates.
 */
const outboxPayloadSchema = z
  .object({
    title: z.string().min(1).max(255),
    allDay: z.boolean(),
    start: z.string().min(1).max(64),
    end: z.string().min(1).max(64),
    location: z.string().max(2000).optional(),
    description: z.string().max(2000).optional(),
    recurrence: z.enum(['none', 'daily', 'weekly', 'monthly', 'yearly']).optional(),
    calendarUrl: z.string().url().max(1024).optional(),
    _preservedRrule: z.string().max(1024).optional(),
    // D-06: recurrence bounding (RRULE UNTIL / COUNT)
    // CR-01: defense-in-depth — validate the exact 'YYYY-MM-DD' shape here too (the route
    // schema validates on ingress, but the outbox payload is re-parsed from stored JSON).
    // Guarantees .replace(/-/g,'') in assembleRruleString emits digits-only, closing the
    // RRULE-part injection vector. int().min(1) prevents zero/negative counts.
    recurrenceUntil: z
      .string()
      .regex(/^\d{4}-\d{2}-\d{2}$/)
      .optional(), // 'YYYY-MM-DD' → RRULE UNTIL
    recurrenceCount: z.number().int().min(1).optional(), // integer ≥ 1 → RRULE COUNT
  })
  .passthrough();

type OutboxPayloadFields = z.infer<typeof outboxPayloadSchema>;

// ── D-06: RRULE bound assembly ───────────────────────────────────────────────

/**
 * Assembles a complete RRULE string by appending a COUNT or UNTIL bound to a
 * base preset string (e.g. 'FREQ=WEEKLY').
 *
 * Rules (RFC 5545 §3.3.10):
 *   - COUNT takes precedence over UNTIL when both are supplied (mutual exclusion).
 *   - For all-day events (allDay=true), UNTIL serializes as DATE form: `YYYYMMDD`.
 *   - For timed events (allDay=false), UNTIL serializes as DATETIME UTC: `YYYYMMDDTHHMMSSZ`.
 *     Using T235959Z (end of UTC day) is a safe universal choice per RESEARCH.md Pitfall 2.
 *   - When neither bound is supplied the base preset is returned unchanged.
 *
 * T-06-02 security note: `until.replace(/-/g,'')` emits only digits stripped from a
 * length-bounded string (max 10 via Zod). The fixed `;UNTIL=` / `;COUNT=` templates
 * prevent injection of extra `;`-delimited RRULE parts. The assembled string is later
 * passed through `ICAL.Recur.fromString` which rejects malformed RRULE values.
 *
 * @param basePreset - Base RRULE string, e.g. 'FREQ=DAILY' (from RRULE_PRESETS or extracted)
 * @param until - Optional 'YYYY-MM-DD' end date
 * @param count - Optional occurrence count (integer ≥ 1)
 * @param allDay - Whether the event is all-day (controls UNTIL value-type)
 * @returns Assembled RRULE string
 */
export function assembleRruleString(
  basePreset: string,
  until?: string,
  count?: number,
  allDay?: boolean,
): string {
  let s = basePreset;
  if (count !== undefined) {
    // COUNT wins over UNTIL (mutual exclusion)
    s += `;COUNT=${count}`;
  } else if (until) {
    const dateDigits = until.replace(/-/g, '');
    if (allDay) {
      // DATE form for all-day events: YYYYMMDD (RFC 5545 §3.3.10)
      s += `;UNTIL=${dateDigits}`;
    } else {
      // DATETIME UTC form for timed events: YYYYMMDDTHHMMSSZ (end of UTC day)
      s += `;UNTIL=${dateDigits}T235959Z`;
    }
  }
  return s;
}

// ── Drain concurrency guard (CR-05) ──────────────────────────────────────────

/**
 * Module-level drain guard — prevents overlapping 15s drain cycles from
 * double-dispatching the same still-pending outbox row.
 *
 * SINGLE-PROCESS LIMITATION: This guard is valid ONLY for the single-process
 * Unraid deployment of this two-user app where all drain cycles share the same
 * Node.js module instance. A multi-process or multi-replica deployment (e.g.
 * running multiple API containers behind a load balancer) would require a
 * durable DB row-claim instead:
 *   UPDATE calendar_outbox SET status='processing'
 *   WHERE id=? AND status='pending'
 * and only the process that wins the affected-rows check would dispatch the row.
 * Do not remove this comment if deploying to multi-process infrastructure.
 */
let isDraining = false;

/**
 * WR-06: max time to wait on the post-write targeted re-sync before marking the
 * outbox row 'done'. A stalled Fastmail connection cannot wedge the single-process
 * drain loop beyond this cap; the PWA's next sync-status poll reconciles any cache
 * that the timed-out re-sync did not refresh.
 */
const RESYNC_TIMEOUT_MS = 10_000;

// ── 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.
 *
 * IN-01: accepts an optional per-drain-cycle client cache. Without it, every settled or
 * conflicted row independently reloaded + AES-GCM-decrypted the member credential,
 * widening the window the decrypted app password lives in memory (T-03-13). When a cache
 * is supplied, the decrypted client is built at most once per userId per drain cycle.
 */
async function triggerTargetedResync(
  calendarUrl: string,
  userId: number,
  clientCache?: Map<number, FastmailClient>,
): Promise<void> {
  try {
    // loadClientForUser may throw in test environments — caught below
    let client = clientCache?.get(userId);
    if (!client) {
      client = await loadClientForUser(userId);
      clientCache?.set(userId, client);
    }
    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;
    }

    // NOTIF-03: pass onChanges so this-member writes push to the other member.
    // actor = userId (the member who wrote via the outbox — D-03).
    await syncCalendar(client, davCal, userId, (changes) => {
      for (const change of changes) {
        dispatchEventChange(change, userId).catch((err: unknown) => {
          console.error(
            '[outboxWorker] dispatchEventChange error:',
            err instanceof Error ? err.message : String(err),
          );
        });
      }
    });
  } 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> {
  // CR-03: fail closed on credential errors — let loadClientForUser throw.
  // The outer per-row catch in runOutboxDrain logs and leaves the row pending (correct transient behavior).
  // Do NOT add an empty-credential fallback — that would silently PUT with no authentication.
  const client = await loadClientForUser(row.userId);

  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',
      };
    }
    // CR-02: parse the stored form JSON and build a real VCALENDAR string
    let rawFields: Record<string, unknown>;
    try {
      rawFields = JSON.parse(row.payload) as Record<string, unknown>;
    } catch {
      return {
        success: false,
        conflict: false,
        hardFail: true,
        transient: false,
        error: 'payload parse failed',
      };
    }
    // IN-03: re-validate the parsed payload. A schema-invalid row can never succeed —
    // hard-fail it (no retry) rather than feeding undefined/Invalid Date into the VEVENT.
    const parsedFields = outboxPayloadSchema.safeParse(rawFields);
    if (!parsedFields.success) {
      return {
        success: false,
        conflict: false,
        hardFail: true,
        transient: false,
        error: `payload validation failed: ${parsedFields.error.message}`,
      };
    }
    const fields: OutboxPayloadFields = parsedFields.data;
    // WR-01: recurrence preservation. The PWA omits `recurrence` from an edit payload
    // (it cannot read the existing RRULE — not in the occurrence contract, D-03), so on
    // update we must NOT rebuild the VEVENT with no RRULE — that would silently convert a
    // recurring series into a single event. When the payload carries no explicit
    // recurrence, fall back to the RRULE already stored in calendarEvents.rawVevent.
    // An explicit recurrence value (including 'none') still overrides — that is a
    // deliberate user change. Read rawVevent in the same scoped query as the fresh etag.
    let preservedRrule: string | undefined;
    const hasExplicitRecurrence = Object.prototype.hasOwnProperty.call(fields, 'recurrence');
    const rruleFromPayload =
      fields.recurrence && fields.recurrence !== 'none'
        ? RRULE_PRESETS[fields.recurrence as string]
        : undefined;
    // WR-02: re-read the freshest etag from calendarEvents just before PUT.
    // Rapid successive edits to the same uid enqueue multiple update rows, each
    // carrying the etag at enqueue time. If a prior edit succeeded and triggered
    // a re-sync, calendarEvents.etag was updated but the next update row still
    // carries the old enqueue-time etag — guaranteed 412 on the second edit.
    // Using the freshest cached etag here prevents the spurious conflict toast
    // while still preserving genuine conflict detection (D-08): a real external
    // change updates calendarEvents.etag differently from any pending row's etag.
    // CR-02: calendar_events is keyed (calendarId, uid), and a shared Fastmail
    // account (D-16) caches the same uid once per member's calendar. A uid-only
    // re-read returns multiple rows and an arbitrary [0] — potentially the OTHER
    // member's etag, which would spuriously 412 (false conflict → edit dropped,
    // D-08) or coincidentally match and overwrite. Scope the re-read to THIS row's
    // own calendar by joining through calendars on the outbox row's userId +
    // calendarUrl so the freshest etag belongs to the writing member.
    let etagForPut: string | null = row.etag ?? null;
    const freshEtagRows = (await db
      .select({ etag: calendarEvents.etag, rawVevent: calendarEvents.rawVevent })
      .from(calendarEvents)
      .innerJoin(calendars, eq(calendarEvents.calendarId, calendars.id))
      .where(
        and(
          eq(calendarEvents.uid, row.uid),
          eq(calendars.userId, row.userId),
          eq(calendars.url, row.calendarUrl),
        ),
      )
      .limit(1)) as Array<{ etag: string | null; rawVevent: string | null }>;
    if (freshEtagRows.length > 0 && freshEtagRows[0].etag != null) {
      etagForPut = freshEtagRows[0].etag;
    }

    // WR-01: when the edit payload carries no explicit recurrence, preserve the RRULE
    // already on the stored event so an edit does not strip a recurring series.
    if (!hasExplicitRecurrence && freshEtagRows.length > 0 && freshEtagRows[0].rawVevent) {
      preservedRrule = extractRruleString(freshEtagRows[0].rawVevent);
    }

    // D-06: assemble the final RRULE string, combining the preset or preserved RRULE
    // with an optional UNTIL/COUNT bound from the payload.
    // Pitfall 3: on series edit with bound change only (no new preset), parse the preserved
    // RRULE, STRIP any existing UNTIL/COUNT, then re-apply the new bound — never naive-
    // concatenate onto `FREQ=WEEKLY;BYDAY=...` which would produce double-UNTIL.
    // WR-01 note: preservedRrule is only set when !hasExplicitRecurrence (see above),
    // so the hasExplicitRecurrence branch always takes precedence over preserved RRULE.
    let finalRruleString: string | undefined;
    if (hasExplicitRecurrence) {
      // Explicit recurrence wins — recurrence:'none' yields undefined (no RRULE emitted)
      finalRruleString = rruleFromPayload
        ? assembleRruleString(
            rruleFromPayload,
            fields.recurrenceUntil,
            fields.recurrenceCount,
            fields.allDay,
          )
        : undefined;
    } else if (preservedRrule) {
      if (fields.recurrenceUntil || fields.recurrenceCount !== undefined) {
        // Series edit with bound change only: strip existing UNTIL/COUNT, then re-apply
        const strippedPreset = preservedRrule.replace(/;(UNTIL|COUNT)=[^;]*/g, '');
        finalRruleString = assembleRruleString(
          strippedPreset,
          fields.recurrenceUntil,
          fields.recurrenceCount,
          fields.allDay,
        );
      } else {
        finalRruleString = preservedRrule;
      }
    } else {
      finalRruleString = rruleFromPayload;
    }

    const { icsString } = buildVeventString({
      uid: row.uid,
      summary: fields.title,
      allDay: fields.allDay,
      dtstart: fields.allDay ? fields.start : new Date(fields.start),
      dtend: fields.allDay ? fields.end : new Date(fields.end),
      location: fields.location,
      description: fields.description,
      rruleString: finalRruleString,
    });

    response = await updateCalendarEvent(client, row.calendarObjectUrl, icsString, etagForPut);
  } else {
    // create
    if (!row.payload) {
      return {
        success: false,
        conflict: false,
        hardFail: true,
        transient: false,
        error: 'create operation missing payload',
      };
    }
    // CR-02: parse the stored form JSON and build a real VCALENDAR string
    let rawFields: Record<string, unknown>;
    try {
      rawFields = JSON.parse(row.payload) as Record<string, unknown>;
    } catch {
      return {
        success: false,
        conflict: false,
        hardFail: true,
        transient: false,
        error: 'payload parse failed',
      };
    }
    // IN-03: re-validate the parsed payload — hard-fail a schema-invalid create row.
    const parsedFields = outboxPayloadSchema.safeParse(rawFields);
    if (!parsedFields.success) {
      return {
        success: false,
        conflict: false,
        hardFail: true,
        transient: false,
        error: `payload validation failed: ${parsedFields.error.message}`,
      };
    }
    const fields: OutboxPayloadFields = parsedFields.data;
    // CR-01: edit-as-move RRULE preservation. The same-calendar `update` branch
    // preserves a recurring series' RRULE by reading rawVevent; the `create` branch
    // (used for the create half of an edit-as-move, D-04) has no source for the
    // original RRULE because it writes under a brand-new uid. The edit route extracts
    // the source event's RRULE and stashes it on the payload as `_preservedRrule` so
    // the worker can re-apply it here. An explicit `recurrence` on the payload still
    // wins (deliberate user change); the preserved RRULE only fills the gap when the
    // edit omitted recurrence — matching the update-branch semantics and the WR-01 fix.
    const hasExplicitRecurrence = Object.prototype.hasOwnProperty.call(fields, 'recurrence');
    const rruleFromPayload =
      fields.recurrence && fields.recurrence !== 'none'
        ? RRULE_PRESETS[fields.recurrence as string]
        : undefined;
    const preservedRrule =
      typeof fields._preservedRrule === 'string' && fields._preservedRrule.length > 0
        ? fields._preservedRrule
        : undefined;

    // D-06: assemble the final RRULE string with optional UNTIL/COUNT bound.
    // CR-01: an explicit recurrence preset wins over _preservedRrule (deliberate user choice).
    // recurrence:'none' explicitly clears any RRULE — including when _preservedRrule is present.
    // If no explicit recurrence, fall back to _preservedRrule (edit-as-move RRULE carry-through).
    let finalRruleString: string | undefined;
    if (hasExplicitRecurrence) {
      // Explicit recurrence wins — recurrence:'none' yields undefined (no RRULE emitted)
      finalRruleString = rruleFromPayload
        ? assembleRruleString(
            rruleFromPayload,
            fields.recurrenceUntil,
            fields.recurrenceCount,
            fields.allDay,
          )
        : undefined;
    } else if (preservedRrule) {
      if (fields.recurrenceUntil || fields.recurrenceCount !== undefined) {
        // Bound change on preserved RRULE: strip existing UNTIL/COUNT first (Pitfall 3)
        const strippedPreset = preservedRrule.replace(/;(UNTIL|COUNT)=[^;]*/g, '');
        finalRruleString = assembleRruleString(
          strippedPreset,
          fields.recurrenceUntil,
          fields.recurrenceCount,
          fields.allDay,
        );
      } else {
        finalRruleString = preservedRrule;
      }
    } else {
      finalRruleString = rruleFromPayload;
    }

    const { icsString } = buildVeventString({
      uid: row.uid,
      summary: fields.title,
      allDay: fields.allDay,
      dtstart: fields.allDay ? fields.start : new Date(fields.start),
      dtend: fields.allDay ? fields.end : new Date(fields.end),
      location: fields.location,
      description: fields.description,
      rruleString: finalRruleString,
    });
    // 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, icsString);
  }

  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 };
  }

  // IN-02: an unmapped 4xx (e.g. 405, 409, 422) is a permanent client error — retrying it
  // for the full backoff window just delays settling and burns the attempt budget before
  // dead-lettering. The transient-eligible 4xx codes (408 request timeout, 429 too many
  // requests) are already in TRANSIENT_STATUSES and handled above, so any remaining 4xx
  // here is a hard fail. 5xx, network failures, and truly unknown statuses still fall
  // through to transient so genuinely recoverable conditions keep their retries.
  if (status >= 400 && status < 500) {
    return {
      success: false,
      conflict: false,
      hardFail: true,
      transient: false,
      error: `Hard fail: HTTP ${status} for uid=${row.uid}`,
    };
  }

  // Unknown / 5xx 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).
 *
 * Concurrency guard (CR-05): the module-level `isDraining` flag ensures overlapping
 * 15s scheduler invocations are no-ops for the single-process deployment.
 *
 * Durable create-before-delete gate (CR-04): for delete rows with a groupId, the
 * worker queries the DB for the sibling create row's status. It does NOT rely on
 * both rows co-occurring in the same in-memory batch.
 *
 * Per-row errors are caught and logged so one bad row cannot crash the loop.
 */
export async function runOutboxDrain(): Promise<void> {
  // CR-05: single-process concurrency guard (see isDraining declaration for limitations)
  if (isDraining) return;
  isDraining = true;

  try {
    // 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).
    // This is a fast path; the authoritative gate is the durable DB sibling-status check below.
    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 within this batch (fast path for same-batch pairs).
    // Cross-batch ordering is enforced durably by the DB sibling-status check inside the loop.
    const failedCreateGroups = new Set<string>();

    // IN-01: per-drain-cycle client cache so triggerTargetedResync decrypts each member's
    // credential at most once per cycle. Discarded when the drain returns — never persisted.
    const clientCache = new Map<number, FastmailClient>();

    for (const row of sorted) {
      // D-04 fast path: if the create for this group already failed in this batch, skip the 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 this batch (D-04)`,
        );
        continue;
      }

      // CR-04: Durable create-before-delete gate — query DB for sibling create status.
      // This prevents the delete from running when the create pair straddles drain batches.
      if (row.operation === 'delete' && row.groupId) {
        const siblingRows = (await db
          .select({ status: calendarOutbox.status })
          .from(calendarOutbox)
          .where(
            and(eq(calendarOutbox.groupId, row.groupId), eq(calendarOutbox.operation, 'create')),
          )) as Array<{ status: string }>;

        const siblingStatus = siblingRows[0]?.status;

        if (siblingStatus !== 'done') {
          if (siblingStatus === 'failed' || siblingStatus === 'dead') {
            // Sibling create failed permanently — skip this delete forever (D-04: original preserved)
            console.warn(
              `[outboxWorker] Paired create for groupId=${row.groupId} is ${siblingStatus} — marking delete row.id=${row.id} failed (original event preserved, D-04)`,
            );
            await db
              .update(calendarOutbox)
              .set({
                status: 'failed',
                lastError: 'paired create did not succeed — original preserved',
              })
              .where(eq(calendarOutbox.id, row.id));
          } else {
            // Sibling create is still pending/processing — defer this delete to a later cycle
            console.warn(
              `[outboxWorker] Deferring delete row.id=${row.id} — sibling create (groupId=${row.groupId}) is not yet done (status=${siblingStatus ?? 'not found'})`,
            );
            // Leave the delete row pending; do NOT update its status
          }
          continue;
        }
      }

      try {
        const result = await dispatchRow(row);

        if (result.conflict) {
          // WR-06: distinguish an edit-as-move create-412 from a same-calendar conflict.
          // For a move (D-04) the create runs first; on 412 the paired delete is later
          // marked failed and the original event survives — so this is NOT a "the event
          // changed elsewhere" conflict, it is "the move could not be applied". The PWA
          // set lastSyncedUid to the NEW (move) uid, whose only outbox row is this failed
          // create, so without a distinct message the user sees the wrong conflict copy
          // and has no cue to retry. Emit a move-specific lastError that does NOT contain
          // '412' so the toast routes it to the dedicated move-failed copy instead of the
          // generic etag-conflict copy.
          const isMoveCreate = !!row.groupId && row.operation === 'create';
          const conflictError = isMoveCreate
            ? 'move-failed: the event could not be moved — re-open it and save again'
            : (result.error ?? '412 conflict');
          // 412 — mark failed (no retry), re-sync calendar so UI sees authoritative state (D-08)
          await db
            .update(calendarOutbox)
            .set({ status: 'failed', lastError: conflictError })
            .where(eq(calendarOutbox.id, row.id));
          await triggerTargetedResync(row.calendarUrl, row.userId, clientCache);

          if (row.groupId && row.operation === 'create') {
            failedCreateGroups.add(row.groupId);
          }
        } else if (result.success) {
          // Success — refresh the local cache BEFORE marking done. The PWA's
          // SyncStateToast polls sync-status and invalidates ['events'] the
          // instant it sees status='done'; if we marked done first, that refetch
          // raced the re-sync and returned stale cache (deleted event still
          // present, edit not yet applied) — forcing a manual refresh. Re-syncing
          // first means 'done' guarantees the cache already reflects the write.
          //
          // WR-06: bound the re-sync with a timeout. triggerTargetedResync does
          // unbounded network I/O against Fastmail (fetchCalendars + syncCalendar);
          // a hang would leave this row 'pending' from the DB's view for the full
          // duration, the 15s isDraining guard would no-op the next cycle, and the
          // PWA would poll 'pending' indefinitely — wedging the single-process
          // drain loop. On timeout we proceed to mark 'done' and let the PWA's next
          // poll/refetch reconcile (the same documented refetch path the eager
          // re-sync was optimizing). triggerTargetedResync already swallows its own
          // errors, so the race only needs to cap the wait.
          await Promise.race([
            triggerTargetedResync(row.calendarUrl, row.userId, clientCache),
            new Promise<void>((resolve) => setTimeout(resolve, RESYNC_TIMEOUT_MS)),
          ]);
          await db
            .update(calendarOutbox)
            .set({ status: 'done' })
            .where(eq(calendarOutbox.id, row.id));
        } 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 {
            // WR-01: use row.attemptCount (the attempt that just failed, 0-based) as the backoff index.
            // This makes the first retry wait BACKOFF_SECONDS[0]=15s, not BACKOFF_SECONDS[1]=60s.
            const backoffMs = (BACKOFF_SECONDS[row.attemptCount] ?? 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),
        );
      }
    }
  } finally {
    isDraining = false;
  }
}

// ── Scheduler ────────────────────────────────────────────────────────────────

/**
 * Starts the 15-second background outbox drain schedule.
 * Call once at API startup (wired in index.ts beside startBrokerPoller).
 * Uses setInterval instead of node-cron: node-cron 4.2.1 silently skipped executions
 * in the long-running server process; setInterval fires reliably.
 */
export function startOutboxWorker(): void {
  setInterval(() => {
    runOutboxDrain().catch((err: unknown) => {
      console.error('[outboxWorker] Unhandled runOutboxDrain error:', err);
    });
  }, 15 * 1000);
}