feat: keep offline changes, sends and conflicts in the local store's outbox

lib/store/__tests__/outbox.test.ts

@@ -0,0 +1,591 @@
+// @vitest-environment node
+import { beforeEach, describe, expect, it } from 'vitest';
+import { DEFAULT_SCOPE } from '../area';
+import { AccountEngine } from '../engine/account-engine';
+import { TransportError, type MethodCall } from '../engine/transport';
+import { mailArea } from '../mail';
+import { outboxArea } from '../outbox';
+import { CLOCK_SKEW_MARGIN_MS, decideRetry, LOOKUP_TRUSTED_MS, type LookupArgs } from '../outbox/dedupe';
+import { flushPatches, type PatchOutcome, type PendingPatch } from '../outbox/flush-patches';
+import type { StoredConflict, StoredOutboxEntry } from '../outbox/items';
+import { mergePatches, patchConflicts } from '../outbox/patches';
+import { FAKE_SESSION, JMAP_ACCOUNT, type FakeEmail } from './helpers/fake-jmap';
+import { FakeJmapMail } from './helpers/fake-jmap-mail';
+import { openMemoryStore } from './helpers/node-adapter';
+
+const NOW = Date.UTC(2026, 10, 3);
+const BOXES = [
+  { id: 'inbox', name: 'Inbox', role: 'inbox', totalEmails: 2, unreadEmails: 1 },
+  { id: 'trash', name: 'Trash', role: 'trash', totalEmails: 1, unreadEmails: 0 },
+  { id: 'drafts', name: 'Drafts', role: 'drafts', totalEmails: 0, unreadEmails: 0 },
+  { id: 'sent', name: 'Sent', role: 'sent', totalEmails: 0, unreadEmails: 0 },
+];
+
+function setup() {
+  const server = new FakeJmapMail({ mailboxes: BOXES.map((box) => ({ ...box })) });
+  server.addEmail({ id: 'a', threadId: 't1', subject: 'Aye', keywords: {}, mailboxIds: { inbox: true } });
+  server.addEmail({ id: 'b', threadId: 't1', subject: 'Bee', keywords: { $seen: true }, mailboxIds: { inbox: true } });
+  server.addEmail({ id: 'c', threadId: 't2', subject: 'Sea', keywords: { $seen: true }, mailboxIds: { trash: true } });
+  const db = openMemoryStore();
+  const engine = new AccountEngine({
+    account: 'a@fake.test',
+    db,
+    session: FAKE_SESSION,
+    transport: server,
+    areas: [mailArea, outboxArea],
+    scope: { ...DEFAULT_SCOPE, days: null },
+    now: () => NOW,
+    sleep: async () => {},
+  });
+  return { server, db, engine };
+}
+
+let t: ReturnType<typeof setup>;
+
+beforeEach(async () => {
+  t = setup();
+  await t.engine.requestSync();
+});
+
+const set = (args: Record<string, unknown>) => t.engine.answer([['Email/set', args, 's']]);
+const update = (patches: Record<string, Record<string, unknown>>) => set({ update: patches });
+const pending = () => t.engine.areaCall('outbox.pendingPatches', {}) as Promise<PendingPatch[]>;
+const read = () => t.engine.areaCall('outbox.read', {}) as Promise<{ patches: number; entries: StoredOutboxEntry[]; conflicts: StoredConflict[] }>;
+
+async function local(id: string) {
+  const answer = await t.engine.answer([['Email/get', { ids: [id], properties: ['keywords', 'mailboxIds', 'subject'] }, 'g']]);
+  return answer?.methodResponses[0][1].list[0];
+}
+
+async function boxes() {
+  const answer = await t.engine.answer([['Mailbox/get', {}, 'm']]);
+  return Object.fromEntries((answer?.methodResponses[0][1].list as Array<{ id: string; totalEmails: number; unreadEmails: number }>).map((box) => [box.id, [box.totalEmails, box.unreadEmails]]));
+}
+
+const io = (server: FakeJmapMail = t.server, isRefusal: (error: unknown) => boolean = () => false) => ({
+  request: (calls: MethodCall[], using: string[]) => server.request(calls, using),
+  isRefusal,
+});
+
+async function flush(ioOverride = io()) {
+  const outcomes: PatchOutcome[] = [];
+  const finished = await flushPatches(ioOverride, await pending(), async (batch) => {
+    outcomes.push(...batch);
+    await t.engine.areaCall('outbox.patchResults', { outcomes: batch });
+  });
+  return { outcomes, finished };
+}
+
+describe('merging patches per record', () => {
+  it('keeps one row per record, with later paths on top of earlier ones', async () => {
+    await update({ a: { 'keywords/$seen': true } });
+    await update({ a: { 'keywords/$flagged': true } });
+    await update({ a: { 'keywords/$seen': null } });
+    const rows = await pending();
+    expect(rows).toHaveLength(1);
+    expect(rows[0].patch).toEqual({ 'keywords/$seen': null, 'keywords/$flagged': true });
+    expect((await local('a')).keywords).toEqual({ $flagged: true });
+  });
+
+  it('lets a whole map replace earlier paths into it, and takes later paths into the map', () => {
+    expect(mergePatches({ 'keywords/$seen': true, 'mailboxIds/x': true }, { keywords: { $flagged: true } }))
+      .toEqual({ keywords: { $flagged: true }, 'mailboxIds/x': true });
+    expect(mergePatches({ keywords: { $flagged: true } }, { 'keywords/$seen': true }))
+      .toEqual({ keywords: { $flagged: true, $seen: true } });
+    expect(mergePatches({ keywords: { $flagged: true } }, { 'keywords/$flagged': null })).toEqual({ keywords: {} });
+    expect(mergePatches({ mailboxIds: { a: true } }, { mailboxIds: { b: true } })).toEqual({ mailboxIds: { b: true } });
+  });
+
+  it('reads escaped keyword names', () => {
+    expect(mergePatches({ keywords: {} }, { 'keywords/a~1b': true })).toEqual({ keywords: { 'a/b': true } });
+  });
+
+  it('queues a change to a message it doesn\'t keep, and answers it as done', async () => {
+    const answer = await update({ unknown: { 'keywords/$seen': true } });
+    expect(answer?.methodResponses[0][1].updated).toEqual({ unknown: null });
+    expect((await pending()).map((row) => [row.recordId, row.base])).toEqual([['unknown', null]]);
+  });
+
+  it('refuses per record what can\'t wait: other properties, a new folder, no folder at all', async () => {
+    const answer = await update({
+      a: { subject: 'x' },
+      b: { 'mailboxIds/#new': true },
+      c: { mailboxIds: {} },
+    });
+    const result = answer?.methodResponses[0][1];
+    expect(Object.keys(result.notUpdated).sort()).toEqual(['a', 'b', 'c']);
+    expect(result.notUpdated.a.type).toBe('forbidden');
+    expect(result.notUpdated.c.type).toBe('invalidProperties');
+    expect(await pending()).toEqual([]);
+    const emptied = await update({ c: { 'mailboxIds/trash': null } });
+    expect(emptied?.methodResponses[0][1].notUpdated.c.type).toBe('invalidProperties');
+    expect(await pending()).toEqual([]);
+    expect((await local('c')).mailboxIds).toEqual({ trash: true });
+  });
+
+  it('answers a create per message, so the composer\'s "not saved" path runs (seam for the drafts agent)', async () => {
+    const answer = await set({ create: { draft: { subject: 'x' } } });
+    expect(answer?.methodResponses[0][0]).toBe('Email/set');
+    expect(answer?.methodResponses[0][1].notCreated.draft.type).toBe('serverUnavailable');
+  });
+});
+
+describe('showing changes at once', () => {
+  it('moves a message and keeps the folder counts right', async () => {
+    expect(await boxes()).toMatchObject({ inbox: [2, 1], trash: [1, 0] });
+    await update({ a: { 'mailboxIds/inbox': null, 'mailboxIds/trash': true } });
+    expect(await boxes()).toMatchObject({ inbox: [1, 0], trash: [2, 1] });
+    const answer = await t.engine.answer([['Email/query', { filter: { inMailbox: 'trash' }, sort: [{ property: 'receivedAt', isAscending: false }] }, 'q']]);
+    expect(answer?.methodResponses[0][1].ids).toEqual(['c', 'a']);
+  });
+
+  it('counts a read change once, also after later changes to the same record', async () => {
+    await update({ a: { 'keywords/$seen': true } });
+    expect((await boxes()).inbox).toEqual([2, 0]);
+    await update({ a: { 'keywords/$flagged': true } });
+    expect((await boxes()).inbox).toEqual([2, 0]);
+    await update({ a: { 'keywords/$seen': null } });
+    expect((await boxes()).inbox).toEqual([2, 1]);
+  });
+
+  it('whole-map patches count the same', async () => {
+    await update({ a: { keywords: { $seen: true } } });
+    expect((await boxes()).inbox).toEqual([2, 0]);
+    await update({ b: { mailboxIds: { trash: true } } });
+    expect(await boxes()).toMatchObject({ inbox: [1, 0], trash: [2, 0] });
+  });
+
+  it('deletes for good right away, from lists, threads and counts', async () => {
+    const answer = await set({ destroy: ['a'] });
+    expect(answer?.methodResponses[0][1].destroyed).toEqual(['a']);
+    expect(await boxes()).toMatchObject({ inbox: [1, 0] });
+    const check = await t.engine.answer([
+      ['Email/get', { ids: ['a'], properties: ['id'] }, 'g'],
+      ['Thread/get', { ids: ['t1'] }, 't'],
+    ]);
+    expect(check?.methodResponses[0][1].notFound).toEqual(['a']);
+    expect(check?.methodResponses[1][1].list).toEqual([{ id: 't1', emailIds: ['b'] }]);
+    expect((await pending()).map((row) => row.op)).toEqual(['destroy']);
+  });
+
+  it('empties a folder through the back-reference the client sends', async () => {
+    const answer = await t.engine.answer([
+      ['Email/query', { filter: { inMailbox: 'trash' }, limit: 500 }, '0'],
+      ['Email/set', { '#destroy': { resultOf: '0', name: 'Email/query', path: '/ids' } }, '1'],
+    ]);
+    expect(answer?.methodResponses[1][1].destroyed).toEqual(['c']);
+    expect((await boxes()).trash).toEqual([0, 0]);
+    const again = await t.engine.answer([['Email/query', { filter: { inMailbox: 'trash' } }, '0']]);
+    expect(again?.methodResponses[0][1].ids).toEqual([]);
+    await flush();
+    expect(t.server.emails.has('c')).toBe(false);
+    expect(await pending()).toEqual([]);
+  });
+
+  it('a change to a message waiting to be deleted is not found', async () => {
+    await set({ destroy: ['a'] });
+    const answer = await update({ a: { 'keywords/$seen': true } });
+    expect(answer?.methodResponses[0][1].notUpdated.a.type).toBe('notFound');
+    expect((await pending())[0].op).toBe('destroy');
+  });
+});
+
+describe('sync and waiting changes', () => {
+  it('reapplies a waiting change over what the server sends', async () => {
+    await update({ a: { 'keywords/$flagged': true } });
+    t.server.updateEmail('a', { subject: 'Renamed', keywords: { $seen: true } });
+    await t.engine.requestSync();
+    expect(await local('a')).toMatchObject({ subject: 'Renamed', keywords: { $seen: true, $flagged: true } });
+    expect(await pending()).toHaveLength(1);
+  });
+
+  it('survives a full resync, and keeps a pending delete away', async () => {
+    await update({ a: { 'keywords/$flagged': true } });
+    await set({ destroy: ['b'] });
+    t.server.truncateLog('Email');
+    t.server.addEmail({ id: 'new' });
+    await t.engine.requestSync();
+    expect((await local('a')).keywords).toEqual({ $flagged: true });
+    expect(await local('b')).toBeUndefined();
+    expect(await local('new')).toBeTruthy();
+  });
+
+  it('doesn\'t bring back a message waiting to be deleted when the server changes it', async () => {
+    await set({ destroy: ['a'] });
+    t.server.updateEmail('a', { keywords: { $flagged: true } });
+    await t.engine.requestSync();
+    expect(await local('a')).toBeUndefined();
+  });
+
+  it('follows the server\'s new counts without counting a waiting change twice', async () => {
+    await update({ a: { 'keywords/$seen': true } });
+    expect((await boxes()).inbox).toEqual([2, 0]);
+    t.server.mailboxes.get('inbox')!.totalEmails = 3;
+    t.server.mailboxes.get('inbox')!.unreadEmails = 2;
+    t.server.coll.Mailbox.push('updated', 'inbox');
+    t.server.addEmail({ id: 'new', keywords: {} });
+    await t.engine.requestSync();
+    expect((await boxes()).inbox).toEqual([3, 1]);
+  });
+
+  it('drops the patch once the server took it, and the next run brings the truth', async () => {
+    await update({ a: { 'keywords/$seen': true } });
+    const { outcomes, finished } = await flush();
+    expect(finished).toBe(true);
+    expect(outcomes).toEqual([expect.objectContaining({ recordId: 'a', outcome: 'done' })]);
+    expect(await pending()).toEqual([]);
+    expect((t.server.emails.get('a') as FakeEmail).keywords).toEqual({ $seen: true });
+    t.server.updateEmail('a', { keywords: { $seen: true, $answered: true } });
+    await t.engine.requestSync();
+    expect((await local('a')).keywords).toEqual({ $seen: true, $answered: true });
+  });
+});
+
+describe('conflicts', () => {
+  it('holds back a change to what the server changed since, and shows the server\'s state', async () => {
+    await update({ a: { 'keywords/$flagged': true } });
+    t.server.updateEmail('a', { keywords: { $flagged: true, $seen: true } });
+    // The same outcome as wanted: not a conflict.
+    let run = await flush();
+    expect(run.outcomes[0]).toMatchObject({ outcome: 'done' });
+
+    await update({ b: { 'keywords/$seen': null } });
+    t.server.updateEmail('b', { keywords: { $seen: true, $flagged: true } });
+    // Another keyword moved: this one still stands.
+    run = await flush();
+    expect(run.outcomes[0]).toMatchObject({ outcome: 'done' });
+
+    await update({ c: { 'mailboxIds/trash': null, 'mailboxIds/inbox': true } });
+    t.server.updateEmail('c', { mailboxIds: { sent: true } });
+    run = await flush();
+    expect(run.outcomes[0]).toMatchObject({ outcome: 'conflict', reason: 'changed' });
+    const { conflicts, patches } = await read();
+    expect(patches).toBe(0);
+    expect(conflicts).toEqual([expect.objectContaining({ recordId: 'c', jmap: JMAP_ACCOUNT, summary: { subject: 'Sea', op: 'update', reason: 'changed' } })]);
+    expect((t.server.emails.get('c') as FakeEmail).mailboxIds).toEqual({ sent: true });
+    expect((await local('c')).mailboxIds).toEqual({ sent: true });
+  });
+
+  it('never sends a conflicting change, and sends the rest of the batch', async () => {
+    await update({ a: { 'mailboxIds/inbox': null, 'mailboxIds/trash': true }, b: { 'keywords/$flagged': true } });
+    t.server.updateEmail('a', { mailboxIds: { sent: true } });
+    await flush();
+    expect((t.server.emails.get('a') as FakeEmail).mailboxIds).toEqual({ sent: true });
+    expect((t.server.emails.get('b') as FakeEmail).keywords).toEqual({ $seen: true, $flagged: true });
+  });
+
+  it('shows what the server refused', async () => {
+    await update({ a: { 'keywords/$flagged': true } });
+    t.server.refuse.set('a', { type: 'forbidden', description: 'Read only' });
+    await flush();
+    const { conflicts } = await read();
+    expect(conflicts[0].summary).toEqual({ subject: 'Aye', op: 'update', reason: 'refused', detail: 'Read only' });
+    expect((await local('a')).keywords).toEqual({});
+    await t.engine.areaCall('outbox.dismissConflict', { id: conflicts[0].id });
+    expect((await read()).conflicts).toEqual([]);
+  });
+
+  it('shows a change to a message gone from the server, and removes the local copy', async () => {
+    await update({ a: { 'keywords/$flagged': true } });
+    t.server.destroyEmail('a');
+    await flush();
+    expect((await read()).conflicts[0].summary).toMatchObject({ reason: 'gone', subject: 'Aye' });
+    expect(await local('a')).toBeUndefined();
+  });
+
+  it('a delete of what\'s already gone is simply done', async () => {
+    await set({ destroy: ['a'] });
+    t.server.destroyEmail('a');
+    const { outcomes } = await flush();
+    expect(outcomes[0]).toMatchObject({ outcome: 'done' });
+    expect((await read()).conflicts).toEqual([]);
+  });
+
+  it('holds back a delete of what changed on the server, and the message comes back', async () => {
+    await set({ destroy: ['a'] });
+    t.server.updateEmail('a', { keywords: { $flagged: true } });
+    await t.engine.requestSync();
+    await flush();
+    expect(t.server.emails.has('a')).toBe(true);
+    expect((await read()).conflicts[0].summary).toMatchObject({ op: 'destroy', reason: 'changed' });
+    expect((await local('a')).keywords).toEqual({ $flagged: true });
+  });
+
+  it('shows a refused delete', async () => {
+    await set({ destroy: ['a'] });
+    t.server.refuse.set('a', { type: 'forbidden' });
+    await flush();
+    expect((await read()).conflicts[0].summary).toMatchObject({ op: 'destroy', reason: 'refused', detail: 'forbidden' });
+    expect((await local('a'))?.subject).toBe('Aye');
+  });
+
+  it('compares per property, ignoring case in keyword names', () => {
+    const base = { keywords: { '$Label:A': true }, mailboxIds: { x: true } };
+    expect(patchConflicts({ 'keywords/$label:a': null }, base, { keywords: { '$label:a': true }, mailboxIds: { x: true } })).toBe(false);
+    expect(patchConflicts({ 'keywords/$seen': true }, base, { keywords: {}, mailboxIds: { x: true } })).toBe(false);
+    expect(patchConflicts({ keywords: { $seen: true } }, base, { keywords: {}, mailboxIds: { x: true } })).toBe(true);
+    expect(patchConflicts({ keywords: { $seen: true } }, base, { keywords: { $seen: true }, mailboxIds: { x: true } })).toBe(false);
+    expect(patchConflicts({ keywords: { $seen: true } }, null, { keywords: {}, mailboxIds: {} })).toBe(false);
+  });
+});
+
+describe('when the server can\'t be reached during a flush', () => {
+  it('stops and keeps every change, without calling anything a conflict', async () => {
+    await update({ a: { 'keywords/$seen': true } });
+    t.server.offline = true;
+    const { outcomes, finished } = await flush();
+    expect(finished).toBe(false);
+    expect(outcomes).toEqual([]);
+    expect(await pending()).toHaveLength(1);
+    expect((await read()).conflicts).toEqual([]);
+    expect((await local('a')).keywords).toEqual({ $seen: true });
+  });
+
+  it('stops on "not now" from the server too', async () => {
+    await update({ a: { 'keywords/$seen': true } });
+    t.server.failNext.set('Email/set', 'serverUnavailable');
+    const { finished } = await flush();
+    expect(finished).toBe(false);
+    expect(await pending()).toHaveLength(1);
+  });
+
+  it('turns a request the server refuses for good into a conflict, not a stuck queue', async () => {
+    await update({ a: { 'keywords/$seen': true } });
+    const refusing = {
+      request: async () => { throw new Error('Request failed: 400 - bad'); },
+      isRefusal: (error: unknown) => /^Request failed: 400/.test((error as Error).message),
+    };
+    const { outcomes, finished } = await flush(refusing);
+    expect(finished).toBe(true);
+    expect(outcomes[0]).toMatchObject({ outcome: 'conflict', reason: 'refused' });
+    expect(await pending()).toEqual([]);
+  });
+
+  it('keeps a change made while the flush ran, and doesn\'t compare it with the state the flush caused', async () => {
+    await update({ a: { 'keywords/$seen': true } });
+    const rows = await pending();
+    await update({ a: { 'keywords/$flagged': true } });
+    await flushPatches(io(), rows, async (batch) => { await t.engine.areaCall('outbox.patchResults', { outcomes: batch }); });
+    const left = await pending();
+    expect(left).toHaveLength(1);
+    expect(left[0].base).toBeNull();
+    await flush();
+    expect((t.server.emails.get('a') as FakeEmail).keywords).toEqual({ $seen: true, $flagged: true });
+    expect((await read()).conflicts).toEqual([]);
+  });
+});
+
+describe('sends', () => {
+  const summary = { subject: 'S', to: ['b@example.org'] };
+  const item = { kind: 'email', args: [['b@example.org'], 'S', 'x'] };
+
+  it('takes an entry for sending once: a second claim loses', async () => {
+    const id = await t.engine.areaCall('outbox.queueSend', { summary, item, messageId: 'fixed@example.org' }) as number;
+    expect(await t.engine.areaCall('outbox.claim', { id })).toBe(true);
+    expect(await t.engine.areaCall('outbox.claim', { id })).toBe(false);
+    const entry = (await read()).entries[0];
+    expect(entry).toMatchObject({ status: 'sending', sendingSince: NOW, firstAttemptAt: NOW, messageId: 'fixed@example.org', item });
+  });
+
+  it('marks what a closed tab left behind uncertain', async () => {
+    const id = await t.engine.areaCall('outbox.queueSend', { summary, item }) as number;
+    await t.engine.areaCall('outbox.claim', { id });
+    await t.engine.areaCall('outbox.settleStale', { before: NOW });
+    expect((await read()).entries[0].status).toBe('sending');
+    await t.engine.areaCall('outbox.settleStale', { before: NOW + 1 });
+    expect((await read()).entries[0].status).toBe('uncertain');
+  });
+
+  it('keeps the first attempt time across retries, so the lookup covers the first copy', async () => {
+    const id = await t.engine.areaCall('outbox.queueSend', { summary, item }) as number;
+    await t.engine.areaCall('outbox.claim', { id });
+    await t.engine.areaCall('outbox.update', { id, status: 'waiting' });
+    await t.engine.areaCall('outbox.claim', { id });
+    expect((await read()).entries[0].firstAttemptAt).toBe(NOW);
+  });
+
+  it('keeps order across changes and sends, and a later merge moves a change later', async () => {
+    await update({ a: { 'keywords/$seen': true } });
+    await t.engine.areaCall('outbox.queueSend', { summary, item });
+    await update({ b: { 'keywords/$flagged': true } });
+    const { entries } = await read();
+    const before = await t.engine.areaCall('outbox.pendingPatches', { upToSeq: entries[0].seq }) as PendingPatch[];
+    expect(before.map((row) => row.recordId)).toEqual(['a']);
+    await update({ a: { 'keywords/$flagged': true } });
+    expect(await t.engine.areaCall('outbox.pendingPatches', { upToSeq: entries[0].seq })).toEqual([]);
+  });
+
+  it('removes the files of a removed entry, and keeps an update from bringing a deleted one back', async () => {
+    const data = new TextEncoder().encode('hello');
+    const held = await t.engine.areaCall('outbox.holdFile', { name: 'note.txt', type: 'text/plain', data }) as { blobId: string; size: number };
+    expect(held).toMatchObject({ size: 5, type: 'text/plain' });
+    expect(held.blobId).toMatch(/^bulwark-local-/);
+    const file = await t.engine.areaCall('outbox.heldFile', { blobId: held.blobId }) as { name: string; data: Uint8Array };
+    expect(file.name).toBe('note.txt');
+    expect(new TextDecoder().decode(file.data)).toBe('hello');
+    const id = await t.engine.areaCall('outbox.queueSend', { summary, item: { kind: 'raw', blobId: held.blobId, identityId: 'i' } }) as number;
+    await t.engine.areaCall('outbox.remove', { id });
+    expect(await t.engine.areaCall('outbox.heldFile', { blobId: held.blobId })).toBeNull();
+    await t.engine.areaCall('outbox.update', { id, status: 'failed' });
+    expect((await read()).entries).toEqual([]);
+  });
+
+  it('drops old files nothing needs', async () => {
+    const data = new Uint8Array([1]);
+    const orphan = await t.engine.areaCall('outbox.holdFile', { name: 'a', type: '', data }) as { blobId: string };
+    const used = await t.engine.areaCall('outbox.holdFile', { name: 'b', type: '', data }) as { blobId: string };
+    await t.engine.areaCall('outbox.queueSend', { summary, item: { kind: 'email', args: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, [{ blobId: used.blobId }]] } });
+    expect(await t.engine.areaCall('outbox.dropOrphans', { olderThan: NOW })).toBe(0);
+    expect(await t.engine.areaCall('outbox.dropOrphans', { olderThan: NOW + 1 })).toBe(1);
+    expect(await t.engine.areaCall('outbox.heldFile', { blobId: orphan.blobId })).toBeNull();
+    expect(await t.engine.areaCall('outbox.heldFile', { blobId: used.blobId })).not.toBeNull();
+  });
+});
+
+describe('looking a send up before it goes again', () => {
+  const MESSAGE_ID = 'k3x.11111111@example.org';
+  const args = (over: Partial<LookupArgs> = {}): LookupArgs => ({
+    jmap: JMAP_ACCOUNT,
+    submissionJmap: JMAP_ACCOUNT,
+    messageId: MESSAGE_ID,
+    attemptAt: NOW - 60_000,
+    now: NOW,
+    copyKept: true,
+    ...over,
+  });
+
+  function server() {
+    return new FakeJmapMail({ mailboxes: BOXES.map((box) => ({ ...box })) });
+  }
+
+  const inDrafts = (s: FakeJmapMail, over: Partial<FakeEmail> = {}) => s.addEmail({
+    id: 'copy',
+    messageId: [MESSAGE_ID],
+    mailboxIds: { drafts: true },
+    keywords: { $seen: true, $draft: true },
+    receivedAt: new Date(NOW - 30_000).toISOString(),
+    ...over,
+  });
+
+  // The five cases of decision 4.
+  it('request never arrived: send again', async () => {
+    const s = server();
+    // A message from before, and another with its own id.
+    s.addEmail({ id: 'other', messageId: ['other@example.org'], mailboxIds: { sent: true }, receivedAt: new Date(NOW - 10_000).toISOString() });
+    expect(await decideRetry(io(s), args())).toEqual({ action: 'send' });
+  });
+
+  it('draft arrived, submission didn\'t: submit that copy, don\'t create another', async () => {
+    const s = server();
+    inDrafts(s);
+    expect(await decideRetry(io(s), args())).toEqual({ action: 'submit', emailId: 'copy' });
+  });
+
+  it('both arrived: done', async () => {
+    const s = server();
+    inDrafts(s, { mailboxIds: { sent: true }, keywords: { $seen: true } });
+    s.addSubmission('copy', 'final');
+    expect(await decideRetry(io(s), args())).toEqual({ action: 'done', emailId: 'copy', why: 'sent' });
+    // A record Stalwart already removed doesn't change that.
+    s.submissions.clear();
+    expect(await decideRetry(io(s), args())).toEqual({ action: 'done', emailId: 'copy', why: 'sent' });
+  });
+
+  it('scheduled and waiting: done, leave it', async () => {
+    const s = server();
+    inDrafts(s);
+    s.addSubmission('copy', 'pending');
+    expect(await decideRetry(io(s), args())).toEqual({ action: 'done', emailId: 'copy', why: 'scheduled' });
+  });
+
+  it('canceled: someone decided against it, so it doesn\'t go', async () => {
+    const s = server();
+    inDrafts(s);
+    s.addSubmission('copy', 'canceled');
+    expect(await decideRetry(io(s), args())).toEqual({ action: 'done', emailId: 'copy', why: 'canceled' });
+  });
+
+  it('never uses a header or text filter, which find nothing on either version', async () => {
+    const s = server();
+    inDrafts(s, { mailboxIds: { sent: true }, keywords: {} });
+    await decideRetry(io(s), args());
+    const filters = JSON.stringify(s.calls.filter(([name]) => name === 'Email/query').map(([, a]) => a.filter));
+    expect(filters).not.toMatch(/header|text/);
+    expect(filters).toMatch(/inMailbox/);
+  });
+
+  it('finds a copy whose server time is behind this device\'s clock, within the margin', async () => {
+    const s = server();
+    const skew = CLOCK_SKEW_MARGIN_MS - 60_000;
+    inDrafts(s, { receivedAt: new Date(NOW - 60_000 - skew).toISOString() });
+    expect((await decideRetry(io(s), args())).action).toBe('submit');
+  });
+
+  it('matches with or without angle brackets', async () => {
+    const s = server();
+    inDrafts(s);
+    expect((await decideRetry(io(s), args({ messageId: `<${MESSAGE_ID}>` }))).action).toBe('submit');
+  });
+
+  it('can\'t tell without a Message-ID (a send from before they were fixed)', async () => {
+    expect(await decideRetry(io(server()), args({ messageId: undefined }))).toEqual({ action: 'undecided', why: 'no-message-id' });
+  });
+
+  it('can\'t tell without the folders', async () => {
+    const s = new FakeJmapMail();
+    expect(await decideRetry(io(s), args())).toEqual({ action: 'undecided', why: 'no-folders' });
+  });
+
+  it('can\'t tell when nothing stays after a good send, and nothing is found', async () => {
+    expect(await decideRetry(io(server()), args({ copyKept: false }))).toEqual({ action: 'undecided', why: 'no-copy-kept' });
+  });
+
+  it('can\'t tell once the server may have forgotten the submission', async () => {
+    const old = args({ attemptAt: NOW - LOOKUP_TRUSTED_MS - 1000 });
+    expect(await decideRetry(io(server()), old)).toEqual({ action: 'undecided', why: 'too-old' });
+    const s = server();
+    inDrafts(s, { receivedAt: new Date(old.attemptAt).toISOString() });
+    expect(await decideRetry(io(s), old)).toEqual({ action: 'undecided', why: 'too-old' });
+  });
+
+  it('can\'t tell when more mail arrived than it reads, unless the copy is among it', async () => {
+    const s = server();
+    for (let i = 0; i < 2500; i += 1) s.addEmail({ mailboxIds: { sent: true }, receivedAt: new Date(NOW - 1000).toISOString() });
+    expect(await decideRetry(io(s), args())).toEqual({ action: 'undecided', why: 'too-many' });
+  }, 30_000);
+
+  it('lets a failed lookup fail, so the caller leaves the entry alone', async () => {
+    const s = server();
+    s.offline = true;
+    await expect(decideRetry(io(s), args())).rejects.toBeInstanceOf(TransportError);
+  });
+});
+
+describe('moving the old outbox in', () => {
+  const legacy = {
+    entries: [
+      { summary: { subject: 'Old', to: [] }, item: { kind: 'email', args: [] }, status: 'uncertain', createdAt: 5 },
+      { summary: { subject: 'Older file', to: [] }, item: { kind: 'raw', blobId: 'bulwark-local-x', identityId: 'i' }, status: 'waiting', createdAt: 6 },
+    ],
+    patches: [{ jmap: JMAP_ACCOUNT, update: { a: { 'keywords/$seen': true } } }],
+    files: [{ blobId: 'bulwark-local-x', name: 'x.eml', type: 'message/rfc822', data: new Uint8Array([1, 2]) }],
+  };
+
+  it('takes all of it once, keeping stand-in ids and order', async () => {
+    expect(await t.engine.areaCall('outbox.importLegacy', legacy)).toBe(true);
+    expect(await t.engine.areaCall('outbox.importLegacy', legacy)).toBe(false);
+    const { entries, patches } = await read();
+    expect(entries.map((e) => [e.summary.subject, e.status, e.createdAt])).toEqual([['Old', 'uncertain', 5], ['Older file', 'waiting', 6]]);
+    expect(patches).toBe(1);
+    expect(await t.engine.areaCall('outbox.heldFile', { blobId: 'bulwark-local-x' })).toMatchObject({ name: 'x.eml' });
+    expect((await local('a')).keywords).toEqual({ $seen: true });
+  });
+
+  it('takes nothing when a part of it fails, and can run again', async () => {
+    const broken = { ...legacy, files: [{ blobId: 'x', name: null, type: null, data: null }] };
+    await expect(t.engine.areaCall('outbox.importLegacy', broken)).rejects.toThrow();
+    expect((await read()).entries).toEqual([]);
+    expect(await pending()).toEqual([]);
+    expect(await t.engine.areaCall('outbox.importLegacy', legacy)).toBe(true);
+  });
+});

lib/store/areas.ts

@@ -2,5 +2,6 @@
 
 import type { StoreArea } from './area';
 import { mailArea } from './mail';
+import { outboxArea } from './outbox';
 
-export const STORE_AREAS: readonly StoreArea[] = [mailArea];
+export const STORE_AREAS: readonly StoreArea[] = [mailArea, outboxArea];

lib/store/outbox/calls.ts

@@ -0,0 +1,287 @@
+/** What the page asks the outbox. Each call is one SQLite transaction, so two tabs can't interleave. */
+
+import type { AreaCall, StoreContext } from '../area';
+import { metaGet, metaSet } from '../engine/core-schema';
+import { deleteEmail, upsertEmail, type EmailObject } from '../mail/records';
+import { nextSeq, queueUpdate, type PatchRow } from './email-set';
+import type { PatchOutcome, PendingPatch } from './flush-patches';
+import {
+  heldBlobIds,
+  LOCAL_BLOB_PREFIX,
+  type ConflictSummary,
+  type OutboxError,
+  type OutboxStatus,
+  type OutboxSummary,
+  type PolicyFingerprint,
+  type SendItem,
+  type StoredConflict,
+  type StoredOutboxEntry,
+} from './items';
+import type { Patch, RecordState } from './patches';
+
+interface OutboxRow {
+  id: number;
+  seq: number;
+  kind: string;
+  created_at: number;
+  status: string;
+  sending_since: number | null;
+  first_attempt_at: number | null;
+  payload: string;
+  summary: string;
+  message_id: string | null;
+  copy_id: string | null;
+  skip_lookup: number;
+  error: string | null;
+}
+
+function toEntry(row: OutboxRow): StoredOutboxEntry {
+  const payload = JSON.parse(row.payload) as { item: SendItem; fingerprint?: PolicyFingerprint };
+  return {
+    id: Number(row.id),
+    seq: Number(row.seq),
+    createdAt: Number(row.created_at),
+    status: row.status as OutboxStatus,
+    summary: JSON.parse(row.summary) as OutboxSummary,
+    item: payload.item,
+    ...(payload.fingerprint ? { fingerprint: payload.fingerprint } : {}),
+    ...(row.sending_since ? { sendingSince: Number(row.sending_since) } : {}),
+    ...(row.first_attempt_at ? { firstAttemptAt: Number(row.first_attempt_at) } : {}),
+    ...(row.error ? { error: JSON.parse(row.error) as OutboxError } : {}),
+    ...(row.message_id ? { messageId: row.message_id } : {}),
+    ...(row.copy_id ? { copyId: row.copy_id } : {}),
+    ...(row.skip_lookup ? { skipLookup: true } : {}),
+  };
+}
+
+interface QueueParams {
+  summary: OutboxSummary;
+  item: SendItem;
+  fingerprint?: PolicyFingerprint;
+  messageId?: string;
+  status?: OutboxStatus;
+  createdAt?: number;
+  error?: OutboxError;
+}
+
+function insertEntry(ctx: StoreContext, params: QueueParams): number {
+  return ctx.db.run(
+    `INSERT INTO outbox (seq, kind, created_at, status, payload, summary, message_id, error) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
+    [
+      nextSeq(ctx.db),
+      params.item.kind,
+      params.createdAt ?? ctx.now(),
+      params.status ?? 'waiting',
+      JSON.stringify({ item: params.item, fingerprint: params.fingerprint }),
+      JSON.stringify(params.summary),
+      params.messageId ?? null,
+      params.error ? JSON.stringify(params.error) : null,
+    ],
+  ).lastInsertRowid;
+}
+
+function queueSend(ctx: StoreContext, params: QueueParams): number {
+  return ctx.db.transaction(() => insertEntry(ctx, params));
+}
+
+function read(ctx: StoreContext): { patches: number; entries: StoredOutboxEntry[]; conflicts: StoredConflict[] } {
+  const patches = Number(ctx.db.get<{ n: number }>('SELECT COUNT(*) AS n FROM outbox_patch')?.n ?? 0);
+  const entries = ctx.db.all<OutboxRow>('SELECT * FROM outbox ORDER BY seq ASC').map(toEntry);
+  const conflicts = ctx.db.all<{ id: number; jmap: string; record_id: string; summary: string; created_at: number }>(
+    'SELECT id, jmap, record_id, summary, created_at FROM conflict ORDER BY id ASC',
+  ).map((row) => ({
+    id: Number(row.id),
+    jmap: row.jmap,
+    recordId: row.record_id,
+    createdAt: Number(row.created_at),
+    summary: JSON.parse(row.summary) as ConflictSummary,
+  }));
+  return { patches, entries, conflicts };
+}
+
+/** Takes a waiting entry for sending: false when it isn't waiting any more (another tab took it, the user deleted it). */
+function claim(ctx: StoreContext, params: { id: number }): boolean {
+  const now = ctx.now();
+  return ctx.db.run(
+    `UPDATE outbox SET status = 'sending', sending_since = ?, first_attempt_at = COALESCE(first_attempt_at, ?) WHERE id = ? AND status = 'waiting'`,
+    [now, now, params.id],
+  ).changes === 1;
+}
+
+/** Deleted meanwhile stays deleted: there's no row to update. */
+function update(ctx: StoreContext, params: { id: number; status: OutboxStatus; error?: OutboxError; copyId?: string | null; skipLookup?: boolean }): void {
+  ctx.db.run(
+    `UPDATE outbox SET status = ?, error = ?, sending_since = NULL,
+       copy_id = CASE WHEN ? THEN ? ELSE copy_id END,
+       skip_lookup = COALESCE(?, skip_lookup)
+     WHERE id = ?`,
+    [
+      params.status,
+      params.error ? JSON.stringify(params.error) : null,
+      params.copyId === undefined ? 0 : 1,
+      params.copyId ?? null,
+      params.skipLookup === undefined ? null : (params.skipLookup ? 1 : 0),
+      params.id,
+    ],
+  );
+}
+
+/** A send a closed tab left behind may have gone out. */
+function settleStale(ctx: StoreContext, params: { before: number }): number {
+  return ctx.db.run(
+    `UPDATE outbox SET status = 'uncertain', sending_since = NULL WHERE status = 'sending' AND COALESCE(sending_since, 0) < ?`,
+    [params.before],
+  ).changes;
+}
+
+/** Removes an entry and the files it held. */
+function remove(ctx: StoreContext, params: { id: number }): void {
+  ctx.db.transaction(() => {
+    const row = ctx.db.get<OutboxRow>('SELECT * FROM outbox WHERE id = ?', [params.id]);
+    if (!row) return;
+    for (const blobId of heldBlobIds(toEntry(row).item)) ctx.db.run('DELETE FROM upload WHERE local_id = ?', [blobId]);
+    ctx.db.run('DELETE FROM outbox WHERE id = ?', [params.id]);
+  });
+}
+
+// ---- Files attached offline ----
+
+function holdFile(ctx: StoreContext, params: { jmap?: string; name: string; type: string; data: Uint8Array; localId?: string }): { blobId: string; size: number; type: string } {
+  const blobId = params.localId ?? `${LOCAL_BLOB_PREFIX}${crypto.randomUUID()}`;
+  ctx.db.run(
+    'INSERT OR REPLACE INTO upload (local_id, jmap, type, name, size, data, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)',
+    [blobId, params.jmap ?? null, params.type, params.name, params.data.byteLength, params.data, ctx.now()],
+  );
+  return { blobId, size: params.data.byteLength, type: params.type || 'application/octet-stream' };
+}
+
+function heldFile(ctx: StoreContext, params: { blobId: string }): { name: string; type: string; data: Uint8Array } | null {
+  const row = ctx.db.get<{ name: string | null; type: string | null; data: Uint8Array }>('SELECT name, type, data FROM upload WHERE local_id = ?', [params.blobId]);
+  return row ? { name: row.name ?? 'attachment', type: row.type ?? 'application/octet-stream', data: row.data } : null;
+}
+
+/** Files held for no entry (attached offline to a message that was then discarded). SEAM: the drafts agent keeps its drafts' files here too. */
+function dropOrphans(ctx: StoreContext, params: { olderThan: number }): number {
+  return ctx.db.transaction(() => {
+    const needed = new Set(ctx.db.all<OutboxRow>('SELECT * FROM outbox').flatMap((row) => heldBlobIds(toEntry(row).item)));
+    let dropped = 0;
+    for (const row of ctx.db.all<{ local_id: string }>('SELECT local_id FROM upload WHERE created_at < ?', [params.olderThan])) {
+      if (needed.has(String(row.local_id))) continue;
+      ctx.db.run('DELETE FROM upload WHERE local_id = ?', [row.local_id]);
+      dropped += 1;
+    }
+    return dropped;
+  });
+}
+
+// ---- Changes ----
+
+function pendingPatches(ctx: StoreContext, params: { upToSeq?: number | null }): PendingPatch[] {
+  const rows = ctx.db.all<PatchRow>(
+    params.upToSeq === undefined || params.upToSeq === null
+      ? 'SELECT * FROM outbox_patch ORDER BY seq ASC'
+      : 'SELECT * FROM outbox_patch WHERE seq < ? ORDER BY seq ASC',
+    params.upToSeq === undefined || params.upToSeq === null ? [] : [params.upToSeq],
+  );
+  return rows.map((row) => ({
+    jmap: row.jmap,
+    type: row.type,
+    recordId: row.record_id,
+    op: row.op,
+    patch: JSON.parse(row.patch) as Patch,
+    base: row.base ? JSON.parse(row.base) as RecordState : null,
+    subject: row.subject,
+    seq: Number(row.seq),
+  }));
+}
+
+function subjectFor(row: PatchRow, server: EmailObject | null | undefined): string {
+  if (typeof server?.subject === 'string') return server.subject;
+  return row.subject ?? '';
+}
+
+/**
+ * What came of a flush. A change made while it ran stays and goes next time,
+ * without comparing against a state this flush itself just moved. After a
+ * conflict the local copy is the server's again: nothing else will correct it
+ * if sync already passed that point.
+ */
+function patchResults(ctx: StoreContext, params: { outcomes: PatchOutcome[] }): number {
+  let conflicts = 0;
+  ctx.db.transaction(() => {
+    for (const outcome of params.outcomes) {
+      const key = [outcome.jmap, outcome.type, outcome.recordId];
+      const row = ctx.db.get<PatchRow>('SELECT * FROM outbox_patch WHERE jmap = ? AND type = ? AND record_id = ?', key);
+      if (!row) continue;
+      if (outcome.outcome === 'done' && Number(row.seq) !== outcome.seq) {
+        ctx.db.run('UPDATE outbox_patch SET base = NULL WHERE jmap = ? AND type = ? AND record_id = ?', key);
+        continue;
+      }
+      ctx.db.run('DELETE FROM outbox_patch WHERE jmap = ? AND type = ? AND record_id = ?', key);
+      if (outcome.outcome === 'done') continue;
+      conflicts += 1;
+      const summary: ConflictSummary = {
+        subject: subjectFor(row, outcome.server),
+        op: row.op,
+        reason: outcome.reason,
+        ...(outcome.detail ? { detail: outcome.detail } : {}),
+      };
+      ctx.db.run(
+        'INSERT INTO conflict (jmap, type, record_id, summary, local, created_at) VALUES (?, ?, ?, ?, ?, ?)',
+        [outcome.jmap, outcome.type, outcome.recordId, JSON.stringify(summary), JSON.stringify({ op: row.op, patch: JSON.parse(row.patch) }), ctx.now()],
+      );
+      if (outcome.server === null) {
+        deleteEmail(ctx.db, outcome.jmap, outcome.recordId);
+      } else if (outcome.server) {
+        const gen = Number(ctx.db.get<{ gen: number | null }>('SELECT MAX(gen) AS gen FROM email WHERE jmap = ?', [outcome.jmap])?.gen ?? 0);
+        upsertEmail(ctx.db, outcome.jmap, outcome.server, gen);
+      }
+    }
+  });
+  if (conflicts > 0) ctx.emit({ type: 'changed', account: ctx.account, collections: ['Email', 'Mailbox', 'Thread'] });
+  return conflicts;
+}
+
+function dismissConflict(ctx: StoreContext, params: { id: number }): void {
+  ctx.db.run('DELETE FROM conflict WHERE id = ?', [params.id]);
+}
+
+// ---- What the old offline database held ----
+
+export interface LegacyImport {
+  entries: QueueParams[];
+  patches: Array<{ jmap: string; update: Record<string, Patch> }>;
+  files: Array<{ blobId: string; name: string; type: string; data: Uint8Array }>;
+}
+
+const LEGACY_DONE = 'legacy_outbox_moved';
+
+/** One transaction: all of it or none. A second run after a partial failure elsewhere adds nothing. */
+function importLegacy(ctx: StoreContext, params: LegacyImport): boolean {
+  return ctx.db.transaction(() => {
+    if (metaGet(ctx.db, LEGACY_DONE)) return false;
+    for (const file of params.files) holdFile(ctx, { name: file.name, type: file.type, data: file.data, localId: file.blobId });
+    for (const entry of params.entries) insertEntry(ctx, entry);
+    for (const { jmap, update: changes } of params.patches) {
+      for (const [id, patch] of Object.entries(changes)) queueUpdate(ctx.db, jmap, id, patch, ctx.now());
+    }
+    metaSet(ctx.db, LEGACY_DONE, '1');
+    return true;
+  });
+}
+
+export const OUTBOX_CALLS: Readonly<Record<string, AreaCall>> = {
+  'outbox.queueSend': queueSend,
+  'outbox.read': read,
+  'outbox.claim': claim,
+  'outbox.update': update,
+  'outbox.settleStale': settleStale,
+  'outbox.remove': remove,
+  'outbox.holdFile': holdFile,
+  'outbox.heldFile': heldFile,
+  'outbox.dropOrphans': dropOrphans,
+  'outbox.pendingPatches': pendingPatches,
+  'outbox.patchResults': patchResults,
+  'outbox.dismissConflict': dismissConflict,
+  'outbox.importLegacy': importLegacy,
+};

lib/store/outbox/dedupe.ts

@@ -0,0 +1,149 @@
+/**
+ * Before a send whose response got lost goes out again: did it reach the
+ * server? JMAP has no idempotency key and a second create or submission is
+ * simply accepted, so this looks. The `header` and `text` filters find no
+ * Message-ID on either Stalwart, so it reads the recent Drafts and Sent and
+ * matches on the client (stalwart-changes-prf.md section 1, decision 4).
+ *
+ * | Lookup                            | Decision                             |
+ * | no match                          | send it again                        |
+ * | in Drafts, no submission          | submit the copy that's there         |
+ * | in Sent                           | done                                 |
+ * | in Drafts, a submission waiting   | done, it's scheduled                 |
+ * | in Drafts, a canceled submission  | done, someone decided against it     |
+ *
+ * It answers "undecided" when it can't tell, and the entry stays with the user.
+ */
+
+import { responseFor, type JmapResponse, type MethodCall } from '../engine/transport';
+import { MAIL_USING } from '../mail/records';
+
+/**
+ * `after` compares the server's receivedAt with this device's clock, so it
+ * looks a day further back than the attempt. Only an exact Message-ID matches,
+ * so looking further costs nothing but a page or two.
+ */
+export const CLOCK_SKEW_MARGIN_MS = 24 * 60 * 60 * 1000;
+/**
+ * Stalwart drops submissions after 3 days (`expungeSubmissionsAfter`). Past
+ * this, "no submission" and "not found" no longer say the send never happened.
+ */
+export const LOOKUP_TRUSTED_MS = 2 * 24 * 60 * 60 * 1000;
+const PAGE = 500;
+const MAX_PAGES = 5;
+const SUBMISSION_USING = [...MAIL_USING, 'urn:ietf:params:jmap:submission'];
+
+export type RetryDecision =
+  | { action: 'send' }
+  | { action: 'submit'; emailId: string }
+  | { action: 'done'; emailId: string; why: 'sent' | 'scheduled' | 'canceled' }
+  | { action: 'undecided'; why: 'no-message-id' | 'no-folders' | 'too-many' | 'too-old' | 'no-copy-kept' | 'unexpected' };
+
+export interface LookupIo {
+  request(calls: MethodCall[], using: string[]): Promise<JmapResponse>;
+}
+
+export interface LookupArgs {
+  /** The JMAP account the send creates its copy in. */
+  jmap: string;
+  submissionJmap: string;
+  messageId: string | undefined;
+  /** When the first attempt was handed to the server. */
+  attemptAt: number;
+  now: number;
+  /**
+   * False when a successful send leaves no copy (no copy in Sent, a submit-only
+   * raw send): then finding nothing doesn't mean it never arrived.
+   */
+  copyKept: boolean;
+}
+
+interface FoundEmail {
+  id: string;
+  messageId?: string[] | null;
+  mailboxIds?: Record<string, boolean>;
+  keywords?: Record<string, boolean>;
+}
+
+function bare(id: string): string {
+  return id.trim().replace(/^<|>$/g, '');
+}
+
+async function findCopies(io: LookupIo, args: LookupArgs, folders: { drafts?: string; sent?: string }): Promise<FoundEmail[] | 'too-many'> {
+  const boxes = [folders.drafts, folders.sent].filter((box): box is string => Boolean(box));
+  const wanted = bare(args.messageId as string);
+  const filter = {
+    operator: 'AND',
+    conditions: [
+      { operator: 'OR', conditions: boxes.map((inMailbox) => ({ inMailbox })) },
+      { after: new Date(args.attemptAt - CLOCK_SKEW_MARGIN_MS).toISOString() },
+    ],
+  };
+  const found: FoundEmail[] = [];
+  for (let page = 0; page < MAX_PAGES; page += 1) {
+    const response = await io.request([
+      ['Email/query', {
+        accountId: args.jmap,
+        filter,
+        sort: [{ property: 'receivedAt', isAscending: false }],
+        position: page * PAGE,
+        limit: PAGE,
+      }, 'q'],
+      ['Email/get', {
+        accountId: args.jmap,
+        '#ids': { resultOf: 'q', name: 'Email/query', path: '/ids' },
+        properties: ['id', 'messageId', 'mailboxIds', 'keywords'],
+      }, 'g'],
+    ], MAIL_USING);
+    const ids = responseFor(response, 'q')[1].ids as string[];
+    const list = responseFor(response, 'g')[1].list as FoundEmail[];
+    found.push(...list.filter((email) => (email.messageId ?? []).some((id) => bare(id) === wanted)));
+    if (ids.length < PAGE) return found;
+  }
+  // More than this since the attempt, and nothing matched so far: it may be further on.
+  return found.length > 0 ? found : 'too-many';
+}
+
+async function submissionsOf(io: LookupIo, jmap: string, emailId: string): Promise<Array<{ undoStatus?: string }>> {
+  const response = await io.request([
+    ['EmailSubmission/query', { accountId: jmap, filter: { emailIds: [emailId] } }, 'q'],
+    ['EmailSubmission/get', {
+      accountId: jmap,
+      '#ids': { resultOf: 'q', name: 'EmailSubmission/query', path: '/ids' },
+      properties: ['id', 'undoStatus'],
+    }, 'g'],
+  ], SUBMISSION_USING);
+  return responseFor(response, 'g')[1].list as Array<{ undoStatus?: string }>;
+}
+
+/** What to do with a send that may have reached the server. Throws when the lookup itself fails. */
+export async function decideRetry(io: LookupIo, args: LookupArgs): Promise<RetryDecision> {
+  if (!args.messageId) return { action: 'undecided', why: 'no-message-id' };
+  const boxesResponse = await io.request([['Mailbox/get', { accountId: args.jmap, properties: ['id', 'role'] }, 'm']], MAIL_USING);
+  const boxes = responseFor(boxesResponse, 'm')[1].list as Array<{ id: string; role?: string | null }>;
+  const folders = { drafts: boxes.find((box) => box.role === 'drafts')?.id, sent: boxes.find((box) => box.role === 'sent')?.id };
+  if (!folders.drafts && !folders.sent) return { action: 'undecided', why: 'no-folders' };
+
+  const copies = await findCopies(io, args, folders);
+  if (copies === 'too-many') return { action: 'undecided', why: 'too-many' };
+  const trusted = args.now - args.attemptAt <= LOOKUP_TRUSTED_MS;
+
+  const inSent = copies.find((email) => folders.sent && email.mailboxIds?.[folders.sent]);
+  if (inSent) return { action: 'done', emailId: inSent.id, why: 'sent' };
+
+  const inDrafts = copies.find((email) => folders.drafts && email.mailboxIds?.[folders.drafts]);
+  if (inDrafts) {
+    const submissions = await submissionsOf(io, args.submissionJmap, inDrafts.id);
+    if (submissions.length > 0) {
+      const live = submissions.some((submission) => submission.undoStatus !== 'canceled');
+      return { action: 'done', emailId: inDrafts.id, why: live ? 'scheduled' : 'canceled' };
+    }
+    if (!trusted) return { action: 'undecided', why: 'too-old' };
+    if (!inDrafts.keywords?.$draft) return { action: 'undecided', why: 'unexpected' };
+    return { action: 'submit', emailId: inDrafts.id };
+  }
+
+  if (!args.copyKept) return { action: 'undecided', why: 'no-copy-kept' };
+  if (!trusted) return { action: 'undecided', why: 'too-old' };
+  return { action: 'send' };
+}

lib/store/outbox/email-set.ts

@@ -0,0 +1,213 @@
+/**
+ * Email/set answered from the store: changes wait in outbox_patch, merged per
+ * record, and show at once. Deleting for good and emptying a folder wait the
+ * same way. Answers carry no state: a made-up one would reach the server later.
+ */
+
+import { MethodError, type MethodContext, type MethodHandler, type RecordOverlay } from '../area';
+import { metaGet, metaSet } from '../engine/core-schema';
+import { deleteEmail, upsertEmail, type EmailObject } from '../mail/records';
+import type { StorageAdapter } from '../storage';
+import { applyPatch, mergePatches, normalizePatch, patchProblem, stateOf, type Patch, type RecordState } from './patches';
+
+export interface PatchRow {
+  jmap: string;
+  type: string;
+  record_id: string;
+  op: 'update' | 'destroy';
+  patch: string;
+  base: string | null;
+  server: string | null;
+  subject: string | null;
+  seq: number;
+  queued_at: number;
+}
+
+const EMAIL = 'Email';
+
+/** The next place in the order changes and sends were made. */
+export function nextSeq(db: StorageAdapter): number {
+  const next = Number(metaGet(db, 'outbox_seq') ?? 0) + 1;
+  metaSet(db, 'outbox_seq', String(next));
+  return next;
+}
+
+function pendingRow(db: StorageAdapter, jmap: string, id: string): PatchRow | undefined {
+  return db.get<PatchRow>('SELECT * FROM outbox_patch WHERE jmap = ? AND type = ? AND record_id = ?', [jmap, EMAIL, id]);
+}
+
+function savePatch(db: StorageAdapter, row: Omit<PatchRow, 'seq' | 'queued_at' | 'type'>, now: number): void {
+  db.run(
+    `INSERT INTO outbox_patch (jmap, type, record_id, op, patch, base, server, subject, seq, queued_at)
+     VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+     ON CONFLICT (jmap, type, record_id) DO UPDATE SET op = excluded.op, patch = excluded.patch, seq = excluded.seq`,
+    [row.jmap, EMAIL, row.record_id, row.op, row.patch, row.base, row.server, row.subject, nextSeq(db), now],
+  );
+}
+
+function localEmail(db: StorageAdapter, jmap: string, id: string): { email: EmailObject; gen: number } | null {
+  const row = db.get<{ json: string; gen: number }>('SELECT json, gen FROM email WHERE jmap = ? AND id = ?', [jmap, id]);
+  return row ? { email: JSON.parse(row.json) as EmailObject, gen: Number(row.gen) } : null;
+}
+
+function subjectOf(email: EmailObject | undefined): string | null {
+  return typeof email?.subject === 'string' ? email.subject : null;
+}
+
+/** Queues one change and shows it. The error for `notUpdated`, or null. */
+export function queueUpdate(db: StorageAdapter, jmap: string, id: string, rawPatch: Patch, now: number): { type: string; description?: string } | null {
+  const problem = patchProblem(rawPatch);
+  if (problem) return problem;
+  const patch = normalizePatch(rawPatch);
+  const existing = pendingRow(db, jmap, id);
+  if (existing?.op === 'destroy') return { type: 'notFound' };
+  const local = localEmail(db, jmap, id);
+  const after = applyPatch(local?.email ?? ({} as EmailObject), patch);
+  // JMAP has no message outside every folder. Deleting is a destroy.
+  if ((local || 'mailboxIds' in patch) && Object.keys(after.mailboxIds ?? {}).length === 0) {
+    return { type: 'invalidProperties', description: 'A message has to stay in a folder' };
+  }
+  const state = local ? JSON.stringify(stateOf(local.email)) : null;
+  savePatch(db, {
+    jmap,
+    record_id: id,
+    op: 'update',
+    patch: JSON.stringify(existing ? mergePatches(JSON.parse(existing.patch) as Patch, patch) : patch),
+    base: state,
+    server: state,
+    subject: subjectOf(local?.email),
+  }, now);
+  if (local) upsertEmail(db, jmap, after as EmailObject, local.gen);
+  return null;
+}
+
+/** Queues a delete and shows it. */
+export function queueDestroy(db: StorageAdapter, jmap: string, id: string, now: number): void {
+  const local = localEmail(db, jmap, id);
+  const state = local ? JSON.stringify(stateOf(local.email)) : null;
+  savePatch(db, { jmap, record_id: id, op: 'destroy', patch: '{}', base: state, server: state, subject: subjectOf(local?.email) }, now);
+  // savePatch keeps base and server of a row already there; only the op changes.
+  if (local) deleteEmail(db, jmap, id);
+}
+
+/**
+ * SEAM for the drafts agent: Email/set create is theirs. Until it lands every
+ * create is refused per message, so createDraft throws and the composer shows
+ * its "draft not saved" state, as it does today.
+ */
+function createEmails(_ctx: MethodContext, _jmap: string, create: Record<string, unknown>): Record<string, { type: string; description: string }> {
+  return Object.fromEntries(Object.keys(create).map((id) => [id, { type: 'serverUnavailable', description: 'Saving a draft needs the server' }]));
+}
+
+export const emailSet: MethodHandler = (ctx, args) => {
+  const jmap = (args.accountId as string | undefined) ?? ctx.defaultJmap;
+  if (args.destroy !== undefined && args.destroy !== null && !Array.isArray(args.destroy)) {
+    throw new MethodError('invalidArguments', 'destroy has to be a list');
+  }
+  const result: Record<string, unknown> = { accountId: jmap };
+  ctx.db.transaction(() => {
+    if (args.create && Object.keys(args.create as object).length > 0) {
+      result.created = {};
+      result.notCreated = createEmails(ctx, jmap, args.create as Record<string, unknown>);
+    }
+    if (args.update) {
+      const updated: Record<string, null> = {};
+      const notUpdated: Record<string, unknown> = {};
+      for (const [id, patch] of Object.entries(args.update as Record<string, Patch>)) {
+        const problem = queueUpdate(ctx.db, jmap, id, patch, ctx.now());
+        if (problem) notUpdated[id] = problem;
+        else updated[id] = null;
+      }
+      result.updated = updated;
+      if (Object.keys(notUpdated).length > 0) result.notUpdated = notUpdated;
+    }
+    if (Array.isArray(args.destroy)) {
+      for (const id of args.destroy as string[]) queueDestroy(ctx.db, jmap, id, ctx.now());
+      result.destroyed = args.destroy;
+    }
+  });
+  ctx.emit({ type: 'changed', account: ctx.account, collections: ['Email', 'Mailbox', 'Thread'] });
+  return result;
+};
+
+/** What waiting changes do to the folder counts: from the server's copy to what the change makes of it. */
+function countDeltas(db: StorageAdapter, jmap: string): Map<string, { total: number; unread: number }> {
+  const rows = db.all<{ op: string; patch: string; server: string | null }>('SELECT op, patch, server FROM outbox_patch WHERE jmap = ? AND type = ?', [jmap, EMAIL]);
+  const deltas = new Map<string, { total: number; unread: number }>();
+  const bump = (box: string, total: number, unread: number) => {
+    const delta = deltas.get(box) ?? { total: 0, unread: 0 };
+    delta.total += total;
+    delta.unread += unread;
+    deltas.set(box, delta);
+  };
+  for (const row of rows) {
+    if (!row.server) continue;
+    const before = JSON.parse(row.server) as RecordState;
+    const wasUnread = before.keywords?.$seen ? 0 : 1;
+    const boxesBefore = Object.keys(before.mailboxIds ?? {}).filter((box) => before.mailboxIds?.[box]);
+    if (row.op === 'destroy') {
+      for (const box of boxesBefore) bump(box, -1, -wasUnread);
+      continue;
+    }
+    const after = applyPatch(before, JSON.parse(row.patch) as Patch);
+    const isUnread = after.keywords?.$seen ? 0 : 1;
+    const boxesAfter = Object.keys(after.mailboxIds ?? {});
+    for (const box of boxesBefore) if (!after.mailboxIds?.[box]) bump(box, -1, -wasUnread);
+    for (const box of boxesAfter) {
+      if (!before.mailboxIds?.[box]) bump(box, 1, isUnread);
+      else if (wasUnread !== isUnread) bump(box, 0, isUnread - wasUnread);
+    }
+  }
+  return deltas;
+}
+
+/**
+ * Sync must not lose a change that waits: what the server sends goes under it.
+ * A record waiting to be deleted stays away. The folder counts stay the
+ * server's, with what waits added when they're read, so a sync can't count a
+ * change twice or forget it.
+ */
+export const outboxOverlay: RecordOverlay = (ctx, jmap, type, records) => {
+  if (!ctx.db.get('SELECT 1 AS x FROM outbox_patch WHERE jmap = ? LIMIT 1', [jmap])) return records;
+  switch (type) {
+    case EMAIL: {
+      const out: Array<{ id: string }> = [];
+      for (const record of records) {
+        const row = pendingRow(ctx.db, jmap, record.id);
+        if (!row) {
+          out.push(record);
+          continue;
+        }
+        const email = record as EmailObject;
+        ctx.db.run('UPDATE outbox_patch SET server = ?, subject = COALESCE(subject, ?) WHERE jmap = ? AND type = ? AND record_id = ?', [
+          JSON.stringify(stateOf(email)), subjectOf(email), jmap, EMAIL, record.id,
+        ]);
+        if (row.op === 'update') out.push(applyPatch(email, JSON.parse(row.patch) as Patch) as EmailObject);
+      }
+      return out;
+    }
+    case 'Thread': {
+      const gone = new Set(ctx.db.all<{ record_id: string }>(
+        "SELECT record_id FROM outbox_patch WHERE jmap = ? AND type = ? AND op = 'destroy'",
+        [jmap, EMAIL],
+      ).map((row) => String(row.record_id)));
+      return (records as Array<{ id: string; emailIds: string[] }>).map((thread) => ({
+        ...thread,
+        emailIds: thread.emailIds.filter((id) => !gone.has(id)),
+      }));
+    }
+    case 'Mailbox': {
+      const deltas = countDeltas(ctx.db, jmap);
+      if (deltas.size === 0) return records;
+      return (records as Array<{ id: string; totalEmails?: unknown; unreadEmails?: unknown }>).map((box) => {
+        const delta = deltas.get(box.id);
+        if (!delta) return box;
+        const count = (value: unknown, by: number) => (typeof value === 'number' ? Math.max(0, value + by) : value);
+        // Messages only: how the conversations change can't be told from here.
+        return { ...box, totalEmails: count(box.totalEmails, delta.total), unreadEmails: count(box.unreadEmails, delta.unread) };
+      });
+    }
+    default:
+      return records;
+  }
+};

lib/store/outbox/flush-patches.ts

@@ -0,0 +1,143 @@
+/**
+ * Sends the changes that waited. Written against a small `request`, so the page
+ * can run it through the JMAP client (its auth, its retries) while the logic
+ * stays testable next to the store.
+ *
+ * Per batch: read what the server has now, keep back what it changed since the
+ * change was queued, send the rest, and report per record. A change the server
+ * took is dropped and the next /changes run brings the truth. A change it
+ * refused, or that met a record changed since, goes to the user as a conflict.
+ * Anything that looks like an outage stops the run: the changes wait.
+ */
+
+import { JmapMethodError, responseFor, type JmapResponse, type MethodCall } from '../engine/transport';
+import { EMAIL_LIST_PROPERTIES, MAIL_USING, type EmailObject } from '../mail/records';
+import type { ConflictReason } from './items';
+import { destroyConflicts, patchConflicts, type Patch, type RecordState } from './patches';
+
+export interface PendingPatch {
+  jmap: string;
+  type: string;
+  recordId: string;
+  op: 'update' | 'destroy';
+  patch: Patch;
+  base: RecordState | null;
+  subject: string | null;
+  seq: number;
+}
+
+export type PatchOutcome =
+  | { jmap: string; type: string; recordId: string; seq: number; outcome: 'done' }
+  | {
+    jmap: string;
+    type: string;
+    recordId: string;
+    seq: number;
+    outcome: 'conflict';
+    reason: ConflictReason;
+    detail?: string;
+    /** The record as the server has it, or null when it's gone there. Undefined when unknown. */
+    server?: EmailObject | null;
+  };
+
+export interface PatchIo {
+  request(calls: MethodCall[], using: string[]): Promise<JmapResponse>;
+  /** The server refused the request itself, for good: not an outage, and no use retrying. */
+  isRefusal(error: unknown): boolean;
+}
+
+const BATCH = 200;
+/** Method errors that mean "not now" rather than "no". */
+const RETRY_LATER = new Set(['serverUnavailable', 'serverFail', 'serverPartialFail', 'rateLimit', 'requestTooLarge']);
+
+function isLater(error: unknown, io: PatchIo): boolean {
+  if (error instanceof JmapMethodError) return RETRY_LATER.has(error.type);
+  return !io.isRefusal(error);
+}
+
+function conflict(row: PendingPatch, reason: ConflictReason, server?: EmailObject | null, detail?: string): PatchOutcome {
+  return { jmap: row.jmap, type: row.type, recordId: row.recordId, seq: row.seq, outcome: 'conflict', reason, detail, server };
+}
+
+function done(row: PendingPatch): PatchOutcome {
+  return { jmap: row.jmap, type: row.type, recordId: row.recordId, seq: row.seq, outcome: 'done' };
+}
+
+function errorText(error: unknown): string {
+  return error instanceof Error ? error.message : String(error);
+}
+
+async function flushBatch(io: PatchIo, jmap: string, rows: PendingPatch[]): Promise<PatchOutcome[]> {
+  const read = await io.request([['Email/get', {
+    accountId: jmap,
+    ids: rows.map((row) => row.recordId),
+    properties: EMAIL_LIST_PROPERTIES,
+  }, 'g']], MAIL_USING);
+  const [, got] = responseFor(read, 'g');
+  const current = new Map((got.list as EmailObject[]).map((email) => [email.id, email]));
+
+  const outcomes: PatchOutcome[] = [];
+  const update: Record<string, Patch> = {};
+  const destroy: string[] = [];
+  const sent = new Map<string, PendingPatch>();
+  for (const row of rows) {
+    const server = current.get(row.recordId);
+    if (!server) {
+      // Asked to delete what's already gone: that's the same result.
+      outcomes.push(row.op === 'destroy' ? done(row) : conflict(row, 'gone', null));
+      continue;
+    }
+    const changed = row.op === 'destroy' ? destroyConflicts(row.base, server) : patchConflicts(row.patch, row.base, server);
+    if (changed) {
+      outcomes.push(conflict(row, 'changed', server));
+      continue;
+    }
+    sent.set(row.recordId, row);
+    if (row.op === 'destroy') destroy.push(row.recordId);
+    else update[row.recordId] = row.patch;
+  }
+  if (sent.size === 0) return outcomes;
+
+  const response = await io.request([['Email/set', {
+    accountId: jmap,
+    ...(Object.keys(update).length > 0 ? { update } : {}),
+    ...(destroy.length > 0 ? { destroy } : {}),
+  }, 's']], MAIL_USING);
+  const [, result] = responseFor(response, 's');
+  const refused: Record<string, { type?: string; description?: string }> = { ...(result.notUpdated ?? {}), ...(result.notDestroyed ?? {}) };
+  for (const [id, row] of sent) {
+    const refusal = refused[id];
+    if (!refusal) {
+      outcomes.push(done(row));
+    } else if (refusal.type === 'notFound') {
+      outcomes.push(row.op === 'destroy' ? done(row) : conflict(row, 'gone', null));
+    } else {
+      outcomes.push(conflict(row, 'refused', current.get(id), refusal.description ?? refusal.type));
+    }
+  }
+  return outcomes;
+}
+
+/** True when all went through to an answer, false when the run stopped on an outage. */
+export async function flushPatches(
+  io: PatchIo,
+  pending: readonly PendingPatch[],
+  report: (outcomes: PatchOutcome[]) => Promise<void>,
+): Promise<boolean> {
+  const byAccount = new Map<string, PendingPatch[]>();
+  for (const row of pending) byAccount.set(row.jmap, [...(byAccount.get(row.jmap) ?? []), row]);
+  for (const [jmap, rows] of byAccount) {
+    for (let i = 0; i < rows.length; i += BATCH) {
+      const batch = rows.slice(i, i + BATCH);
+      let outcomes: PatchOutcome[];
+      try {
+        outcomes = await flushBatch(io, jmap, batch);
+      } catch (error) {
+        if (isLater(error, io)) return false;
+        outcomes = batch.map((row) => conflict(row, 'refused', undefined, errorText(error)));
+      }
+      await report(outcomes);
+    }
+  }
+  return true;
+}

lib/store/outbox/index.ts

@@ -0,0 +1,13 @@
+import type { StoreArea } from '../area';
+import { OUTBOX_CALLS } from './calls';
+import { emailSet, outboxOverlay } from './email-set';
+import { OUTBOX_MIGRATIONS } from './schema';
+
+export const outboxArea: StoreArea = {
+  name: 'outbox',
+  migrations: OUTBOX_MIGRATIONS,
+  methods: { 'Email/set': emailSet },
+  syncers: [],
+  calls: OUTBOX_CALLS,
+  overlay: outboxOverlay,
+};

lib/store/outbox/items.ts

@@ -0,0 +1,105 @@
+/**
+ * What a finished send looks like when it waits. An outbox entry is a finished
+ * message: everything a send does up to the transport ran when the user clicked
+ * Send (the organisation's checks, the label, signatures, plugin send checks,
+ * S/MIME or PGP with the user present). What is kept is the arguments of the
+ * final send, or the finished raw message of a plugin's send. Sending it later
+ * can't skip any of that.
+ */
+
+/** The arguments of useEmailStore.sendEmail after the client (positional, as the client takes them). */
+export interface EmailSendItem {
+  kind: 'email';
+  args: unknown[];
+}
+
+export interface RawSendItem {
+  kind: 'raw';
+  /** The finished message, kept as an upload until it is sent. */
+  blobId: string;
+  identityId: string;
+  delayedUntil?: string;
+  envelopeRecipients?: string[];
+  /** submitRaw: no copy in Sent. */
+  submitOnly?: boolean;
+}
+
+export type SendItem = EmailSendItem | RawSendItem;
+
+export interface OutboxSummary {
+  subject: string;
+  to: string[];
+}
+
+/**
+ * - waiting: goes out by itself when the server answers;
+ * - sending: on its way (a tab took it);
+ * - failed: the server refused it; the error says why;
+ * - uncertain: the connection broke after it may have reached the server.
+ *   Looked up when the server is back; stays this way only when the lookup
+ *   can't tell, and then it's the user's to send again or delete;
+ * - held: the organisation's rules got stricter while it waited.
+ */
+export type OutboxStatus = 'waiting' | 'sending' | 'failed' | 'uncertain' | 'held';
+
+/** Why an entry failed: one of ours (a code the UI translates) or the server's own words. */
+export interface OutboxError {
+  code?: 'file_missing' | 'no_sent_mailbox' | 'no_drafts_mailbox';
+  text?: string;
+}
+
+/** What the organisation required when the message was finished. */
+export interface PolicyFingerprint {
+  labelRequired: boolean;
+  hasLabel: boolean;
+  guards: string[];
+}
+
+export interface StoredOutboxEntry {
+  id: number;
+  seq: number;
+  createdAt: number;
+  summary: OutboxSummary;
+  status: OutboxStatus;
+  sendingSince?: number;
+  firstAttemptAt?: number;
+  error?: OutboxError;
+  item: SendItem;
+  fingerprint?: PolicyFingerprint;
+  messageId?: string;
+  copyId?: string;
+  skipLookup?: boolean;
+}
+
+/** Stand-in blob ids of files kept on this device start with this. */
+export const LOCAL_BLOB_PREFIX = 'bulwark-local-';
+
+export function isLocalBlobId(blobId: unknown): blobId is string {
+  return typeof blobId === 'string' && blobId.startsWith(LOCAL_BLOB_PREFIX);
+}
+
+/** The stand-in blob ids an outbox entry needs uploaded first. */
+export function heldBlobIds(item: SendItem): string[] {
+  if (item.kind === 'raw') return isLocalBlobId(item.blobId) ? [item.blobId] : [];
+  const attachments = item.args[10];
+  if (!Array.isArray(attachments)) return [];
+  return attachments.map((a) => (a as { blobId?: unknown }).blobId).filter(isLocalBlobId);
+}
+
+export type ConflictReason = 'changed' | 'gone' | 'refused';
+
+export interface ConflictSummary {
+  subject: string;
+  op: 'update' | 'destroy';
+  reason: ConflictReason;
+  /** The server's word when it refused. */
+  detail?: string;
+}
+
+export interface StoredConflict {
+  id: number;
+  jmap: string;
+  recordId: string;
+  createdAt: number;
+  summary: ConflictSummary;
+}

lib/store/outbox/patches.ts

@@ -0,0 +1,155 @@
+/**
+ * JMAP PatchObjects for the two properties that can change offline, keywords
+ * and mailboxIds, in both forms the client sends: `keywords/$seen` paths and
+ * the whole map.
+ */
+
+export type Patch = Record<string, unknown>;
+
+export interface RecordState {
+  keywords?: Record<string, boolean>;
+  mailboxIds?: Record<string, boolean>;
+}
+
+const MAPS = ['keywords', 'mailboxIds'] as const;
+type MapName = (typeof MAPS)[number];
+
+function unescapeToken(token: string): string {
+  return token.replace(/~1/g, '/').replace(/~0/g, '~');
+}
+
+/** The map and the key a path changes, or null when it isn't one of ours. */
+function parsePath(path: string): { map: MapName; key: string | null } | null {
+  const slash = path.indexOf('/');
+  const name = slash < 0 ? path : path.slice(0, slash);
+  if (!(MAPS as readonly string[]).includes(name)) return null;
+  return { map: name as MapName, key: slash < 0 ? null : unescapeToken(path.slice(slash + 1)) };
+}
+
+/** Why this patch can't wait offline, or null when it can. */
+export function patchProblem(patch: Patch): { type: string; description: string } | null {
+  const paths = Object.keys(patch);
+  if (paths.length === 0) return { type: 'invalidPatch', description: 'An empty change' };
+  for (const path of paths) {
+    const parsed = parsePath(path);
+    if (!parsed) return { type: 'forbidden', description: `Changing ${path} needs the server` };
+    // A folder made in the same request: that part needs the server, so this does too.
+    if (parsed.map === 'mailboxIds') {
+      const keys = parsed.key !== null ? [parsed.key] : Object.keys((patch[path] ?? {}) as object);
+      if (keys.some((key) => key.startsWith('#'))) return { type: 'forbidden', description: 'A new folder needs the server' };
+    }
+  }
+  return null;
+}
+
+/** `false` means removal to the client's code; on the wire only null does. */
+export function normalizePatch(patch: Patch): Patch {
+  const out: Patch = {};
+  for (const [path, value] of Object.entries(patch)) out[path] = path.includes('/') && !value ? null : value;
+  return out;
+}
+
+export function applyPatch<T extends object>(record: T, patch: Patch): T & RecordState {
+  const state = record as RecordState;
+  const next: T & RecordState = {
+    ...record,
+    keywords: { ...(state.keywords ?? {}) },
+    mailboxIds: { ...(state.mailboxIds ?? {}) },
+  };
+  for (const [path, value] of Object.entries(patch)) {
+    const parsed = parsePath(path);
+    if (!parsed) continue;
+    if (parsed.key === null) {
+      next[parsed.map] = { ...((value ?? {}) as Record<string, boolean>) };
+      continue;
+    }
+    const map = next[parsed.map] as Record<string, boolean>;
+    if (value) map[parsed.key] = true;
+    else delete map[parsed.key];
+  }
+  return next;
+}
+
+/**
+ * A later change to the same record on top of an earlier one. A whole map
+ * replaces every earlier change to that map; a path into a map the earlier
+ * change replaced goes into that map, since the whole map is what gets sent.
+ */
+export function mergePatches(earlier: Patch, later: Patch): Patch {
+  const out: Patch = { ...earlier };
+  for (const [path, value] of Object.entries(later)) {
+    const parsed = parsePath(path);
+    if (!parsed) {
+      out[path] = value;
+      continue;
+    }
+    if (parsed.key === null) {
+      for (const existing of Object.keys(out)) {
+        if (existing === parsed.map || existing.startsWith(`${parsed.map}/`)) delete out[existing];
+      }
+      out[path] = value;
+      continue;
+    }
+    if (parsed.map in out) {
+      const map = { ...(out[parsed.map] as Record<string, boolean>) };
+      if (value) map[parsed.key] = true;
+      else delete map[parsed.key];
+      out[parsed.map] = map;
+      continue;
+    }
+    out[path] = value;
+  }
+  return out;
+}
+
+function sameMap(a: Record<string, boolean> | undefined, b: Record<string, boolean> | undefined): boolean {
+  const left = Object.keys(a ?? {}).filter((key) => a?.[key]).map((key) => key.toLowerCase()).sort();
+  const right = Object.keys(b ?? {}).filter((key) => b?.[key]).map((key) => key.toLowerCase()).sort();
+  return left.length === right.length && left.every((key, i) => key === right[i]);
+}
+
+function has(map: Record<string, boolean> | undefined, key: string): boolean {
+  const wanted = key.toLowerCase();
+  return Object.keys(map ?? {}).some((name) => name.toLowerCase() === wanted && map?.[name]);
+}
+
+/**
+ * Whether the server changed what this patch touches since it was queued. A
+ * change elsewhere on the record isn't a conflict, and neither is the server
+ * already having what the patch wants. Folders count as a whole: a move here
+ * after a move on another device would leave the message in both places.
+ * Keywords stand alone.
+ */
+export function patchConflicts(patch: Patch, base: RecordState | null, serverRecord: object): boolean {
+  if (!base) return false;
+  const server = serverRecord as RecordState;
+  const paths = Object.keys(patch).map((path) => parsePath(path));
+  if (paths.some((parsed) => parsed?.map === 'mailboxIds')
+    && !sameMap(base.mailboxIds, server.mailboxIds)
+    && !sameMap(applyPatch(base, patch).mailboxIds, server.mailboxIds)) return true;
+  for (const [path, value] of Object.entries(patch)) {
+    const parsed = parsePath(path);
+    if (!parsed || parsed.map === 'mailboxIds') continue;
+    if (parsed.key === null) {
+      const wanted = value as Record<string, boolean>;
+      if (!sameMap(base[parsed.map], server[parsed.map]) && !sameMap(wanted, server[parsed.map])) return true;
+      continue;
+    }
+    const before = has(base[parsed.map], parsed.key);
+    const now = has(server[parsed.map], parsed.key);
+    if (before !== now && now !== Boolean(value)) return true;
+  }
+  return false;
+}
+
+/** Deleting a record the server changed since: the user should see that. */
+export function destroyConflicts(base: RecordState | null, serverRecord: object): boolean {
+  if (!base) return false;
+  const server = serverRecord as RecordState;
+  return !sameMap(base.keywords, server.keywords) || !sameMap(base.mailboxIds, server.mailboxIds);
+}
+
+export function stateOf(record: object): RecordState {
+  const state = record as RecordState;
+  return { keywords: state.keywords ?? {}, mailboxIds: state.mailboxIds ?? {} };
+}

lib/store/outbox/schema.ts

@@ -0,0 +1,66 @@
+import type { Migration } from '../storage';
+
+export const OUTBOX_MIGRATIONS: readonly Migration[] = [
+  {
+    id: 200,
+    name: 'outbox',
+    sql: `
+      -- Finished messages waiting for the server. seq is shared with
+      -- outbox_patch, so changes and sends go out in the order they were made.
+      CREATE TABLE outbox (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        seq INTEGER NOT NULL,
+        kind TEXT NOT NULL,
+        created_at INTEGER NOT NULL,
+        status TEXT NOT NULL,
+        sending_since INTEGER,
+        -- When it was first handed to the server: where a retry's lookup starts.
+        first_attempt_at INTEGER,
+        payload TEXT NOT NULL,
+        summary TEXT NOT NULL,
+        message_id TEXT,
+        -- The copy already on the server that only needs submitting.
+        copy_id TEXT,
+        -- The user said send again: no lookup first.
+        skip_lookup INTEGER NOT NULL DEFAULT 0,
+        error TEXT
+      );
+      -- One row per record, later changes merged in. base is the record's state
+      -- when first queued, to tell whether the server changed it since; server
+      -- is the newest state the server sent, for the folder counts.
+      CREATE TABLE outbox_patch (
+        jmap TEXT NOT NULL,
+        type TEXT NOT NULL,
+        record_id TEXT NOT NULL,
+        op TEXT NOT NULL,
+        patch TEXT NOT NULL,
+        base TEXT,
+        server TEXT,
+        subject TEXT,
+        seq INTEGER NOT NULL,
+        queued_at INTEGER NOT NULL,
+        PRIMARY KEY (jmap, type, record_id)
+      ) WITHOUT ROWID;
+      CREATE INDEX outbox_patch_seq ON outbox_patch (seq);
+      CREATE TABLE conflict (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        jmap TEXT NOT NULL,
+        type TEXT NOT NULL,
+        record_id TEXT NOT NULL,
+        summary TEXT NOT NULL,
+        local TEXT NOT NULL,
+        created_at INTEGER NOT NULL
+      );
+      -- Files attached offline, under a stand-in blob id.
+      CREATE TABLE upload (
+        local_id TEXT PRIMARY KEY,
+        jmap TEXT,
+        type TEXT,
+        name TEXT,
+        size INTEGER NOT NULL,
+        data BLOB NOT NULL,
+        created_at INTEGER NOT NULL
+      ) WITHOUT ROWID;
+    `,
+  },
+];