From eb8db7f27de0ba05705e57be85ef3a4cbdcebb98 Mon Sep 17 00:00:00 2001 From: Linus Rath <139418639+rathlinus@users.noreply.github.com> Date: Tue, 3 Nov 2026 05:30:11 +0100 Subject: [PATCH] feat: keep offline changes, sends and conflicts in the local store's outbox --- lib/store/__tests__/outbox.test.ts | 591 +++++++++++++++++++++++++++++ lib/store/areas.ts | 3 +- lib/store/outbox/calls.ts | 287 ++++++++++++++ lib/store/outbox/dedupe.ts | 149 ++++++++ lib/store/outbox/email-set.ts | 213 +++++++++++ lib/store/outbox/flush-patches.ts | 143 +++++++ lib/store/outbox/index.ts | 13 + lib/store/outbox/items.ts | 105 +++++ lib/store/outbox/patches.ts | 155 ++++++++ lib/store/outbox/schema.ts | 66 ++++ 10 files changed, 1724 insertions(+), 1 deletion(-) create mode 100644 lib/store/__tests__/outbox.test.ts create mode 100644 lib/store/outbox/calls.ts create mode 100644 lib/store/outbox/dedupe.ts create mode 100644 lib/store/outbox/email-set.ts create mode 100644 lib/store/outbox/flush-patches.ts create mode 100644 lib/store/outbox/index.ts create mode 100644 lib/store/outbox/items.ts create mode 100644 lib/store/outbox/patches.ts create mode 100644 lib/store/outbox/schema.ts diff --git a/lib/store/__tests__/outbox.test.ts b/lib/store/__tests__/outbox.test.ts new file mode 100644 index 000000000..cbdbcdf46 --- /dev/null +++ b/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; + +beforeEach(async () => { + t = setup(); + await t.engine.requestSync(); +}); + +const set = (args: Record) => t.engine.answer([['Email/set', args, 's']]); +const update = (patches: Record>) => set({ update: patches }); +const pending = () => t.engine.areaCall('outbox.pendingPatches', {}) as Promise; +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 => ({ + 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 = {}) => 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); + }); +}); diff --git a/lib/store/areas.ts b/lib/store/areas.ts index 6bff665ca..ef64b3775 100644 --- a/lib/store/areas.ts +++ b/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]; diff --git a/lib/store/outbox/calls.ts b/lib/store/outbox/calls.ts new file mode 100644 index 000000000..2e7cb06d9 --- /dev/null +++ b/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('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('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('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( + 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('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 }>; + 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> = { + '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, +}; diff --git a/lib/store/outbox/dedupe.ts b/lib/store/outbox/dedupe.ts new file mode 100644 index 000000000..ca50ca900 --- /dev/null +++ b/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; +} + +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; + keywords?: Record; +} + +function bare(id: string): string { + return id.trim().replace(/^<|>$/g, ''); +} + +async function findCopies(io: LookupIo, args: LookupArgs, folders: { drafts?: string; sent?: string }): Promise { + 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> { + 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 { + 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' }; +} diff --git a/lib/store/outbox/email-set.ts b/lib/store/outbox/email-set.ts new file mode 100644 index 000000000..5451e1c21 --- /dev/null +++ b/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('SELECT * FROM outbox_patch WHERE jmap = ? AND type = ? AND record_id = ?', [jmap, EMAIL, id]); +} + +function savePatch(db: StorageAdapter, row: Omit, 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): Record { + 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 = { 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); + } + if (args.update) { + const updated: Record = {}; + const notUpdated: Record = {}; + for (const [id, patch] of Object.entries(args.update as Record)) { + 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 { + 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(); + 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; + } +}; diff --git a/lib/store/outbox/flush-patches.ts b/lib/store/outbox/flush-patches.ts new file mode 100644 index 000000000..361cf817d --- /dev/null +++ b/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; + /** 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 { + 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 = {}; + const destroy: string[] = []; + const sent = new Map(); + 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 = { ...(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, +): Promise { + const byAccount = new Map(); + 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; +} diff --git a/lib/store/outbox/index.ts b/lib/store/outbox/index.ts new file mode 100644 index 000000000..4d65885cb --- /dev/null +++ b/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, +}; diff --git a/lib/store/outbox/items.ts b/lib/store/outbox/items.ts new file mode 100644 index 000000000..3e726ebbd --- /dev/null +++ b/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; +} diff --git a/lib/store/outbox/patches.ts b/lib/store/outbox/patches.ts new file mode 100644 index 000000000..5666623e6 --- /dev/null +++ b/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; + +export interface RecordState { + keywords?: Record; + mailboxIds?: Record; +} + +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(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) }; + continue; + } + const map = next[parsed.map] as Record; + 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) }; + 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 | undefined, b: Record | 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 | 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; + 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 ?? {} }; +} diff --git a/lib/store/outbox/schema.ts b/lib/store/outbox/schema.ts new file mode 100644 index 000000000..2fb40b308 --- /dev/null +++ b/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; + `, + }, +];