feat: add an encrypted local mail store that syncs in a worker

This commit is contained in:
Linus Rath
2026-11-03 04:57:20 +01:00
parent 1dfe09d321
commit 6da9a6a6f5
36 changed files with 18299 additions and 3 deletions
+5
View File
@@ -12,11 +12,16 @@ scripts/
!scripts/lite/
!scripts/build-plugin-sandbox.mjs
!scripts/build-pst-worker.mjs
!scripts/build-store-worker.mjs
!scripts/fetch-sqlite3mc.mjs
!scripts/healthcheck.mjs
!scripts/sourcemaps.mjs
!scripts/standalone-guard.mjs
# Built in the image by scripts/build-plugin-sandbox.mjs
public/plugin-sandbox-assets/
# Built in the image by scripts/build-store-worker.mjs
public/store-worker.js
public/sqlite3mc/
TODO.md
*.md
!README.md
+3
View File
@@ -23,6 +23,9 @@
/public/plugin-sandbox-assets/
# scripts/build-pst-worker.mjs
/public/pst-worker.js
# scripts/build-store-worker.mjs
/public/store-worker.js
/public/sqlite3mc/
# production
/build
+1
View File
@@ -35,6 +35,7 @@ RUN rm -rf "app/api/dev-jmap" && \
\( -name __tests__ -o -name "*.test.ts" -o -name "*.test.tsx" \) \
-prune -exec rm -rf {} \; && \
node scripts/build-pst-worker.mjs && \
node scripts/build-store-worker.mjs && \
node scripts/build-plugin-sandbox.mjs && \
npx next build --webpack && \
node scripts/standalone-guard.mjs
+285
View File
@@ -0,0 +1,285 @@
/**
* A small JMAP server for engine tests, modelled on how Stalwart really
* behaves (local-data/desktop-offline-facts-parts/stalwart-changes-prf.md):
* a 5000 cap on /changes pages, fixed log ranges, the three kinds of state
* refusal, 500 ids per Email/get, 16 calls per request, `ids: null` giving the
* first 500.
*/
import { TransportError, type JmapResponse, type JmapTransport, type MethodCall, type MethodResponse, type StoreSession } from '../../engine/transport';
export type FakeEmail = Record<string, unknown> & { id: string };
export const JMAP_ACCOUNT = 'acc1';
export const FAKE_SESSION: StoreSession = {
apiUrl: 'https://fake.test/jmap/',
downloadUrl: 'https://fake.test/d/{accountId}/{blobId}/{name}?type={type}',
uploadUrl: 'https://fake.test/u/{accountId}',
primaryAccounts: { 'urn:ietf:params:jmap:mail': JMAP_ACCOUNT },
accounts: { [JMAP_ACCOUNT]: { name: 'a@example.org', isPersonal: true, accountCapabilities: { 'urn:ietf:params:jmap:mail': {} } } },
capabilities: { 'urn:ietf:params:jmap:mail': {} },
};
type ChangeKind = 'created' | 'updated' | 'destroyed';
interface Log { kind: ChangeKind; id: string }
class Collection {
log: Log[] = [];
/** Below this log position, states are refused as truncated. */
truncatedBelow = 0;
state(): string {
return `s${this.log.length}`;
}
push(kind: ChangeKind, id: string): void {
this.log.push({ kind, id });
}
changes(since: string, maxChanges: number): { created: string[]; updated: string[]; destroyed: string[]; newState: string; hasMoreChanges: boolean } {
const from = Number(since.slice(1));
const slice = this.log.slice(from, from + maxChanges);
const created = new Set<string>();
const updated = new Set<string>();
const destroyed = new Set<string>();
for (const { kind, id } of slice) {
if (kind === 'created') created.add(id);
else if (kind === 'updated') {
if (!created.has(id)) updated.add(id);
} else if (created.delete(id)) {
// Created and destroyed in the window: nowhere.
} else {
updated.delete(id);
destroyed.add(id);
}
}
const end = from + slice.length;
return {
created: [...created],
updated: [...updated],
destroyed: [...destroyed],
newState: `s${end}`,
hasMoreChanges: end < this.log.length,
};
}
}
export interface FakeOptions {
/** Ten mailboxes by default, a single Inbox here unless given. */
mailboxes?: Array<Record<string, unknown> & { id: string }>;
}
export class FakeJmap implements JmapTransport {
readonly emails = new Map<string, FakeEmail>();
readonly mailboxes = new Map<string, Record<string, unknown> & { id: string }>();
readonly threads = new Map<string, string[]>();
readonly coll = { Email: new Collection(), Mailbox: new Collection(), Thread: new Collection() };
readonly calls: MethodCall[] = [];
requests = 0;
/** Set to make every request fail like a network drop. */
offline = false;
/** Fails the next N requests as busy. */
busyNext = 0;
/** Called before each request; can change data while a sync is running. */
beforeRequest: ((n: number) => void) | null = null;
nextId = 1;
constructor(options: FakeOptions = {}) {
for (const box of options.mailboxes ?? [{ id: 'inbox', name: 'Inbox', role: 'inbox', totalEmails: 0, unreadEmails: 0 }]) {
this.mailboxes.set(box.id, box);
this.coll.Mailbox.push('created', box.id);
}
}
addEmail(partial: Partial<FakeEmail> = {}): FakeEmail {
const n = this.nextId++;
const id = partial.id ?? `e${n}`;
const threadId = (partial.threadId as string) ?? `t${n}`;
const email: FakeEmail = {
id,
threadId,
mailboxIds: { inbox: true },
keywords: {},
size: 100,
receivedAt: new Date(Date.UTC(2026, 9, 1) + n * 1000).toISOString(),
sentAt: new Date(Date.UTC(2026, 9, 1) + n * 1000).toISOString(),
from: [{ name: `Sender ${n}`, email: `s${n}@example.org` }],
to: [{ name: 'A', email: 'a@example.org' }],
cc: [],
subject: `Subject ${n}`,
preview: `Preview ${n}`,
hasAttachment: false,
blobId: `b${n}`,
textBody: [{ partId: '1', type: 'text/plain' }],
htmlBody: [{ partId: '1', type: 'text/plain' }],
bodyValues: { 1: { value: `Body of message ${n}`, isTruncated: false, isEncodingProblem: false } },
attachments: [],
messageId: [`m${n}@example.org`],
headers: [],
bodyStructure: { partId: '1', type: 'text/plain' },
...partial,
};
this.emails.set(id, email);
this.coll.Email.push('created', id);
const members = this.threads.get(threadId);
if (members) {
members.push(id);
this.coll.Thread.push('updated', threadId);
} else {
this.threads.set(threadId, [id]);
this.coll.Thread.push('created', threadId);
}
return email;
}
updateEmail(id: string, patch: Partial<FakeEmail>): void {
Object.assign(this.emails.get(id) as FakeEmail, patch);
this.coll.Email.push('updated', id);
}
destroyEmail(id: string): void {
const email = this.emails.get(id);
if (!email) return;
this.emails.delete(id);
this.coll.Email.push('destroyed', id);
const threadId = email.threadId as string;
const members = (this.threads.get(threadId) ?? []).filter((m) => m !== id);
if (members.length > 0) {
this.threads.set(threadId, members);
this.coll.Thread.push('updated', threadId);
} else {
this.threads.delete(threadId);
this.coll.Thread.push('destroyed', threadId);
}
}
/** The next /changes from before this point gets cannotCalculateChanges. */
truncateLog(type: 'Email' | 'Mailbox' | 'Thread'): void {
this.coll[type].truncatedBelow = this.coll[type].log.length;
}
private queryIds(args: Record<string, unknown>): string[] {
const filter = args.filter as { after?: string; inMailbox?: string } | undefined;
let list = [...this.emails.values()];
if (filter?.after) list = list.filter((e) => Date.parse(String(e.receivedAt)) >= Date.parse(filter.after as string));
if (filter?.inMailbox) list = list.filter((e) => (e.mailboxIds as Record<string, boolean>)[filter.inMailbox as string]);
list.sort((a, b) => Date.parse(String(b.receivedAt)) - Date.parse(String(a.receivedAt)));
return list.map((e) => e.id);
}
private handle(name: string, args: Record<string, unknown>): Record<string, unknown> {
switch (name) {
case 'Mailbox/get': {
const ids = args.ids as string[] | null;
const all = [...this.mailboxes.values()];
const list = ids ? all.filter((b) => ids.includes(b.id)) : all;
return { accountId: args.accountId, state: this.coll.Mailbox.state(), list, notFound: [] };
}
case 'Email/query': {
const ids = this.queryIds(args);
const position = Number(args.position ?? 0);
const limit = Number(args.limit ?? 5000) || 5000;
return {
accountId: args.accountId,
queryState: this.coll.Email.state(),
canCalculateChanges: true,
position,
ids: ids.slice(position, position + Math.min(limit, 5000)),
...(args.calculateTotal ? { total: ids.length } : {}),
};
}
case 'Email/get': {
const ids = args.ids as string[] | null;
if (ids && ids.length > 500) throw Object.assign(new Error('requestTooLarge'), { jmapError: 'requestTooLarge' });
const chosen = ids ?? [...this.emails.keys()].slice(0, 500);
const properties = args.properties as string[] | undefined;
const list: Record<string, unknown>[] = [];
const notFound: string[] = [];
for (const id of chosen) {
const email = this.emails.get(id);
if (!email) {
notFound.push(id);
continue;
}
if (!properties) {
list.push({ ...email });
continue;
}
const out: Record<string, unknown> = { id };
for (const p of properties) if (p in email) out[p] = email[p];
list.push(out);
}
return { accountId: args.accountId, state: this.coll.Email.state(), list, notFound };
}
case 'Thread/get': {
const ids = (args.ids ?? []) as string[];
return {
accountId: args.accountId,
state: this.coll.Thread.state(),
list: ids.filter((id) => this.threads.has(id)).map((id) => ({ id, emailIds: this.threads.get(id) })),
notFound: ids.filter((id) => !this.threads.has(id)),
};
}
case 'Email/changes':
case 'Mailbox/changes':
case 'Thread/changes': {
const type = name.split('/')[0] as 'Email' | 'Mailbox' | 'Thread';
const coll = this.coll[type];
const since = String(args.sinceState);
if (!/^s\d+$/.test(since)) throw Object.assign(new Error('bad state'), { jmapError: 'invalidArguments' });
const from = Number(since.slice(1));
if (from < coll.truncatedBelow || from > coll.log.length) {
throw Object.assign(new Error('log truncated'), { jmapError: 'cannotCalculateChanges' });
}
const maxChanges = Math.min(Number(args.maxChanges ?? 5000) || 5000, 5000);
return { accountId: args.accountId, oldState: since, ...coll.changes(since, maxChanges) };
}
default:
throw Object.assign(new Error(name), { jmapError: 'unknownMethod' });
}
}
async request(calls: MethodCall[]): Promise<JmapResponse> {
this.requests += 1;
this.beforeRequest?.(this.requests);
if (this.offline) throw new TransportError('network', 'Failed to fetch');
if (this.busyNext > 0) {
this.busyNext -= 1;
throw new TransportError('rate', 'Server busy', 429, undefined, 1);
}
if (calls.length > 16) throw new TransportError('http', 'too many calls', 400, 'urn:ietf:params:jmap:error:limit');
const done: MethodResponse[] = [];
for (const call of calls) {
this.calls.push(call);
const [name, raw, id] = call;
const args: Record<string, unknown> = {};
for (const [key, value] of Object.entries(raw)) {
if (key.startsWith('#')) {
const ref = value as { resultOf: string; path: string };
const source = done.find(([, , callId]) => callId === ref.resultOf);
args[key.slice(1)] = source?.[1][ref.path.slice(1)];
} else {
args[key] = value;
}
}
try {
done.push([name, this.handle(name, args), id]);
} catch (error) {
const type = (error as { jmapError?: string }).jmapError;
if (!type) throw error;
done.push(['error', { type }, id]);
}
}
return { methodResponses: done };
}
async download(): Promise<Uint8Array> {
return new Uint8Array();
}
async upload(): Promise<{ blobId: string; type: string; size: number }> {
return { blobId: 'uploaded', type: 'application/octet-stream', size: 0 };
}
}
@@ -0,0 +1,36 @@
/** The storage adapter for tests: real SQLite with FTS5 through node:sqlite. */
import { DatabaseSync, type SQLInputValue } from 'node:sqlite';
import { runTransaction, type SqlRow, type SqlValue, type StorageAdapter } from '../../storage';
export function openMemoryStore(): StorageAdapter {
return adapterFor(new DatabaseSync(':memory:'));
}
export function openFileStore(file: string): StorageAdapter {
return adapterFor(new DatabaseSync(file));
}
function adapterFor(db: DatabaseSync): StorageAdapter {
const tx = { depth: 0 };
const prepared = new Map<string, ReturnType<DatabaseSync['prepare']>>();
const statement = (sql: string) => {
let stmt = prepared.get(sql);
if (!stmt) {
stmt = db.prepare(sql);
prepared.set(sql, stmt);
}
return stmt;
};
return {
exec: (sql) => db.exec(sql),
run: (sql, params = []) => {
const result = statement(sql).run(...(params as SQLInputValue[]));
return { changes: Number(result.changes), lastInsertRowid: Number(result.lastInsertRowid) };
},
all: <T extends object = SqlRow>(sql: string, params: readonly SqlValue[] = []) => statement(sql).all(...(params as SQLInputValue[])) as T[],
get: <T extends object = SqlRow>(sql: string, params: readonly SqlValue[] = []) => statement(sql).get(...(params as SQLInputValue[])) as T | undefined,
transaction: (fn) => runTransaction(tx, (sql) => db.exec(sql), fn),
close: () => db.close(),
};
}
+390
View File
@@ -0,0 +1,390 @@
// @vitest-environment node
import { describe, expect, it } from 'vitest';
import { mkdtempSync, rmSync } from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import { DEFAULT_SCOPE, type StoreEvent } from '../area';
import { AccountEngine } from '../engine/account-engine';
import { runChangesLoop } from '../engine/changes';
import { TransportError, type MethodCall } from '../engine/transport';
import { mailArea } from '../mail';
import { FAKE_SESSION, FakeJmap, JMAP_ACCOUNT } from './helpers/fake-jmap';
import { openFileStore, openMemoryStore } from './helpers/node-adapter';
const EVERYTHING = { ...DEFAULT_SCOPE, days: null };
const INBOX = { inMailbox: 'inbox' };
function engineFor(server: FakeJmap, db = openMemoryStore(), emit?: (e: StoreEvent) => void) {
return new AccountEngine({
account: 'a@fake.test',
db,
session: FAKE_SESSION,
transport: server,
areas: [mailArea],
scope: EVERYTHING,
emit,
now: () => Date.UTC(2026, 10, 3),
sleep: async () => {},
});
}
function count(db: ReturnType<typeof openMemoryStore>, table: string): number {
return Number(db.get<{ n: number }>(`SELECT COUNT(*) AS n FROM ${table}`)?.n);
}
function callsNamed(server: FakeJmap, name: string): MethodCall[] {
return server.calls.filter(([n]) => n === name);
}
describe('full sync', () => {
it('stores every message, thread and body across pages', async () => {
const server = new FakeJmap();
for (let i = 0; i < 1234; i += 1) server.addEmail();
const db = openMemoryStore();
await engineFor(server, db).requestSync();
expect(count(db, 'email')).toBe(1234);
expect(count(db, 'email_body')).toBe(1234);
expect(count(db, 'thread')).toBe(1234);
expect(count(db, 'email_mailbox')).toBe(1234);
});
it('captures the state before paging, so a message landing mid-sync is not lost', async () => {
const server = new FakeJmap();
for (let i = 0; i < 600; i += 1) server.addEmail();
let landed = false;
server.beforeRequest = (n) => {
if (n === 4 && !landed) {
landed = true;
server.addEmail({ id: 'late' });
}
};
const db = openMemoryStore();
await engineFor(server, db).requestSync();
expect(landed).toBe(true);
expect(db.get('SELECT id FROM email WHERE id = ?', ['late'])).toBeTruthy();
});
it('keeps only the chosen window', async () => {
const server = new FakeJmap();
server.addEmail({ id: 'old', receivedAt: '2020-01-01T00:00:00Z' });
server.addEmail({ id: 'new', receivedAt: '2026-10-30T00:00:00Z' });
const db = openMemoryStore();
const engine = new AccountEngine({
account: 'a@fake.test', db, session: FAKE_SESSION, transport: server, areas: [mailArea],
scope: { ...DEFAULT_SCOPE, days: 30 }, now: () => Date.UTC(2026, 10, 3), sleep: async () => {},
});
await engine.requestSync();
expect(db.all('SELECT id FROM email').map((r) => r.id)).toEqual(['new']);
});
});
describe('persisted state', () => {
it('survives a reload and continues from the saved state with a delta', async () => {
const dir = mkdtempSync(path.join(os.tmpdir(), 'store-'));
const file = path.join(dir, 's.db');
try {
const server = new FakeJmap();
for (let i = 0; i < 50; i += 1) server.addEmail();
const first = openFileStore(file);
await engineFor(server, first).requestSync();
first.close();
server.addEmail({ id: 'after-reload' });
server.calls.length = 0;
const second = openFileStore(file);
await engineFor(server, second).requestSync();
expect(second.get('SELECT id FROM email WHERE id = ?', ['after-reload'])).toBeTruthy();
expect(count(second, 'email')).toBe(51);
// A delta, not another full listing.
expect(callsNamed(server, 'Email/query')).toHaveLength(0);
expect(callsNamed(server, 'Email/changes').length).toBeGreaterThan(0);
second.close();
} finally {
rmSync(dir, { recursive: true, force: true });
}
});
it('resumes an interrupted full sync from its cursor', async () => {
const dir = mkdtempSync(path.join(os.tmpdir(), 'store-'));
const file = path.join(dir, 's.db');
try {
const server = new FakeJmap();
for (let i = 0; i < 4500; i += 1) server.addEmail();
const first = openFileStore(file);
// Goes down after the first page of the list.
server.beforeRequest = (n) => {
if (n === 4) server.offline = true;
};
await engineFor(server, first).requestSync();
const listed = count(first, 'email');
expect(listed).toBeGreaterThan(0);
expect(listed).toBeLessThan(4500);
first.close();
server.offline = false;
server.beforeRequest = null;
server.calls.length = 0;
const second = openFileStore(file);
await engineFor(server, second).requestSync();
expect(count(second, 'email')).toBe(4500);
const firstQuery = callsNamed(server, 'Email/query')[0];
expect(firstQuery[1].position).toBe(listed);
second.close();
} finally {
rmSync(dir, { recursive: true, force: true });
}
});
});
describe('deltas', () => {
it('applies created, updated and destroyed, and leaves bodies alone on an update', async () => {
const server = new FakeJmap();
const a = server.addEmail();
const b = server.addEmail();
const db = openMemoryStore();
const engine = engineFor(server, db);
await engine.requestSync();
server.updateEmail(a.id, { keywords: { $seen: true } });
server.destroyEmail(b.id);
server.addEmail({ id: 'c' });
await engine.requestSync();
expect(db.all('SELECT id FROM email ORDER BY id').map((r) => r.id)).toEqual(['c', a.id]);
expect(db.get('SELECT 1 AS x FROM email_keyword WHERE keyword = ?', ['$seen'])).toBeTruthy();
const kept = db.get<{ body_state: number }>('SELECT body_state FROM email WHERE id = ?', [a.id]);
expect(kept?.body_state).toBe(1);
// The destroyed message left nothing behind in the search index.
expect(db.all("SELECT rowid FROM email_fts_word WHERE email_fts_word MATCH ?", [`"Subject" `]).length).toBe(2);
});
it('pages through more changes than one call gives', async () => {
const server = new FakeJmap();
server.addEmail();
const db = openMemoryStore();
const engine = engineFor(server, db);
await engine.requestSync();
server.calls.length = 0;
for (let i = 0; i < 1500; i += 1) server.addEmail();
await engine.requestSync();
expect(count(db, 'email')).toBe(1501);
expect(callsNamed(server, 'Email/changes').length).toBeGreaterThanOrEqual(3);
const page = callsNamed(server, 'Email/changes')[0][1];
expect(page.maxChanges).toBe(500);
});
it('sends sinceState last, because a malformed one otherwise fails the whole request on Stalwart', async () => {
const server = new FakeJmap();
server.addEmail();
const engine = engineFor(server);
await engine.requestSync();
server.addEmail();
server.calls.length = 0;
await engine.requestSync();
const keys = Object.keys(callsNamed(server, 'Email/changes')[0][1]);
expect(keys[keys.length - 1]).toBe('sinceState');
});
it('keeps going while behind the pushed state, since hasMoreChanges false does not mean caught up', async () => {
const server = new FakeJmap();
server.addEmail();
const applied: string[] = [];
let wrote = false;
const outcome = await runChangesLoop({
transport: server,
type: 'Email',
jmap: JMAP_ACCOUNT,
using: [],
since: 's0',
target: 's2',
apply: async (page) => {
applied.push(page.newState);
if (!wrote) {
wrote = true;
server.addEmail();
}
},
});
expect(outcome).toMatchObject({ outcome: 'caughtUp', state: 's2' });
expect(applied).toEqual(['s1', 's2']);
});
it('stops when the state stands still even though the push named a newer one', async () => {
const server = new FakeJmap();
server.addEmail();
const outcome = await runChangesLoop({
transport: server, type: 'Email', jmap: JMAP_ACCOUNT, using: [], since: 's1', target: 's9',
apply: async () => {},
});
expect(outcome.outcome).toBe('caughtUp');
expect(server.requests).toBe(1);
});
});
describe('resync', () => {
it('starts Email over on cannotCalculateChanges, keeps bodies, and drops what is gone', async () => {
const server = new FakeJmap();
const keep = server.addEmail({ id: 'keep' });
server.addEmail({ id: 'gone' });
const db = openMemoryStore();
const engine = engineFor(server, db);
await engine.requestSync();
server.destroyEmail('gone');
server.updateEmail(keep.id, { keywords: { $flagged: true } });
server.truncateLog('Email');
server.calls.length = 0;
await engine.requestSync();
expect(callsNamed(server, 'Email/query').length).toBeGreaterThan(0);
expect(db.all('SELECT id FROM email').map((r) => r.id)).toEqual(['keep']);
expect(db.get('SELECT 1 AS x FROM email_keyword WHERE keyword = ?', ['$flagged'])).toBeTruthy();
// The body wasn't fetched a second time.
const bodyFetches = callsNamed(server, 'Email/get').filter(([, a]) => a.fetchTextBodyValues);
expect(bodyFetches).toHaveLength(0);
expect(count(db, 'email_body')).toBe(1);
});
it.each([
['invalidArguments', { jmapError: 'invalidArguments' }],
['cannotCalculateChanges', { jmapError: 'cannotCalculateChanges' }],
])('treats %s as a reason to start over', async (_name, thrown) => {
const server = new FakeJmap();
const original = server.request.bind(server);
let refused = false;
server.request = async (calls) => {
if (!refused && calls[0][0] === 'Email/changes') {
refused = true;
return { methodResponses: [['error', { type: thrown.jmapError }, 'c']] };
}
return original(calls);
};
const outcome = await runChangesLoop({ transport: server, type: 'Email', jmap: JMAP_ACCOUNT, using: [], since: 's0', apply: async () => {} });
expect(outcome.outcome).toBe('resync');
});
it('treats an HTTP 400 notRequest as a reason to start over, not as an outage', async () => {
const server = new FakeJmap();
server.request = async () => {
throw new TransportError('http', 'bad', 400, 'urn:ietf:params:jmap:error:notRequest');
};
const outcome = await runChangesLoop({ transport: server, type: 'Email', jmap: JMAP_ACCOUNT, using: [], since: 's0', apply: async () => {} });
expect(outcome.outcome).toBe('resync');
});
it('never sends a stored state a server cannot have sent', async () => {
const server = new FakeJmap();
const outcome = await runChangesLoop({ transport: server, type: 'Email', jmap: JMAP_ACCOUNT, using: [], since: 'bad state\n', apply: async () => {} });
expect(outcome.outcome).toBe('resync');
expect(server.requests).toBe(0);
});
});
describe('when the server is busy or gone', () => {
it('backs off and carries on after a 429', async () => {
const server = new FakeJmap();
server.addEmail();
server.busyNext = 2;
const db = openMemoryStore();
await engineFor(server, db).requestSync();
expect(count(db, 'email')).toBe(1);
});
it('ends the run quietly when the network is gone, and picks up when back', async () => {
const server = new FakeJmap();
server.addEmail();
server.offline = true;
const events: StoreEvent[] = [];
const db = openMemoryStore();
const engine = engineFor(server, db, (e) => events.push(e));
await engine.requestSync();
expect(count(db, 'email')).toBe(0);
expect(events.some((e) => e.type === 'error')).toBe(false);
server.offline = false;
await engine.requestSync();
expect(count(db, 'email')).toBe(1);
});
it('tells the page about new mail only for messages it had not seen', async () => {
const server = new FakeJmap();
server.addEmail();
const events: StoreEvent[] = [];
const engine = engineFor(server, openMemoryStore(), (e) => events.push(e));
await engine.requestSync();
expect(events.filter((e) => e.type === 'created')).toHaveLength(0);
server.addEmail({ id: 'fresh' });
await engine.requestSync();
expect(events.filter((e) => e.type === 'created')).toEqual([
{ type: 'created', account: 'a@fake.test', jmap: JMAP_ACCOUNT, collection: 'Email', ids: ['fresh'] },
]);
});
});
describe('answers from the store', () => {
async function seeded() {
const server = new FakeJmap();
server.addEmail({ id: 'one', subject: 'Zahlungserinnerung Rechnungsbetrag', keywords: { $seen: true } });
server.addEmail({ id: 'two', subject: 'Hello', mailboxIds: { inbox: true }, bodyValues: { 1: { value: 'See you Tuesday' } } });
const db = openMemoryStore();
const engine = engineFor(server, db);
await engine.requestSync();
return { engine, db, server };
}
it('gives every call its own answer, in order, with a per-call error for what it cannot answer', async () => {
const { engine } = await seeded();
const answer = await engine.answer([
['Email/query', { filter: INBOX, sort: [{ property: 'receivedAt', isAscending: false }], calculateTotal: true }, 'a'],
['SieveScript/get', {}, 'b'],
['Email/get', { '#ids': { resultOf: 'a', name: 'Email/query', path: '/ids' }, properties: ['id', 'subject'] }, 'c'],
]);
expect(answer?.methodResponses.map(([name, , id]) => [name, id])).toEqual([
['Email/query', 'a'], ['error', 'b'], ['Email/get', 'c'],
]);
expect(answer?.methodResponses[0][1]).toMatchObject({ ids: ['two', 'one'], total: 2, position: 0 });
expect(answer?.methodResponses[1][1].type).toBe('serverUnavailable');
expect(answer?.methodResponses[2][1].list).toEqual([
{ id: 'two', subject: 'Hello' },
{ id: 'one', subject: 'Zahlungserinnerung Rechnungsbetrag' },
]);
});
it('answers null when nothing in the request could be answered', async () => {
const { engine } = await seeded();
expect(await engine.answer([['SieveScript/get', {}, 'a']])).toBeNull();
});
it('finds a German compound from inside the word, and a word start', async () => {
const { engine } = await seeded();
const search = async (text: string) => (await engine.answer([['Email/query', { filter: { text } }, 'q']]))?.methodResponses[0][1].ids;
expect(await search('rechnung')).toEqual(['one']);
expect(await search('erinnerung')).toEqual(['one']);
expect(await search('zahl*')).toEqual(['one']);
expect(await search('tuesday')).toEqual(['two']);
expect(await search('nichtda')).toEqual([]);
});
it('reads keyword sort the way the client sends it for that server', async () => {
const { engine } = await seeded();
const sorted = async () => (await engine.answer([['Email/query', {
filter: INBOX,
sort: [{ property: 'hasKeyword', keyword: '$seen', isAscending: false }],
}, 'q']]))?.methodResponses[0][1].ids;
expect(await sorted()).toEqual(['one', 'two']);
engine.setInvertedKeywordSort(true);
expect(await sorted()).toEqual(['two', 'one']);
});
it('does not make up a message with empty text when the body is not kept', async () => {
const { engine, db } = await seeded();
db.run('DELETE FROM email_body');
const answer = await engine.answer([['Email/get', { ids: ['one'], fetchTextBodyValues: true }, 'g']]);
expect(answer?.methodResponses[0][1]).toMatchObject({ list: [], notFound: ['one'] });
});
it('answers with the real server state, never a made-up one', async () => {
const { engine, server } = await seeded();
const answer = await engine.answer([['Email/get', { ids: ['one'], properties: ['id'] }, 'g']]);
expect(answer?.methodResponses[0][1].state).toBe(server.coll.Email.state());
});
});
+197
View File
@@ -0,0 +1,197 @@
/**
* SQLite3MultipleCiphers on OPFS, for the owner tab's worker.
*
* Rules that come from bugs reproduced in 2.6.1, see
* local-data/desktop-offline-facts-parts/sqlite-wasm.md section 5:
* - PRAGMA key is the very first thing on a connection. Reading first deletes
* a hot journal (#292).
* - The page size must be 8192. On 4096, rollback doesn't work (#293).
* - No PRAGMA rekey (#291, #294). The key is random and only ever re-wrapped.
* - Always the multipleciphers VFS. On plain opfs-sahpool, PRAGMA key throws
* and the connection still writes in the clear if that's ignored.
*/
import { runTransaction, type SqlRow, type SqlValue, type StorageAdapter } from '../storage';
const VFS = 'multipleciphers-opfs-sahpool';
const INIT_TIMEOUT_MS = 30_000;
const REQUIRED_PAGE_SIZE = 8192;
const POOL_DIRECTORY = '.bulwark-store';
const INITIAL_CAPACITY = 16;
/** A database and its journal, plus room for one more. */
const SLOTS_PER_DATABASE = 3;
/* eslint-disable @typescript-eslint/no-explicit-any */
interface Pool {
getFileNames(): string[];
getCapacity(): number;
getFileCount(): number;
addCapacity(n: number): Promise<number>;
unlink(name: string): boolean;
wipeFiles(): Promise<void>;
pauseVfs(): void;
unpauseVfs(): Promise<void>;
isPaused(): boolean;
}
export class StoreKeyError extends Error {
constructor() {
super('The local store could not be opened with this key');
this.name = 'StoreKeyError';
}
}
function withTimeout<T>(promise: Promise<T>, ms: number, what: string): Promise<T> {
return new Promise<T>((resolve, reject) => {
const timer = setTimeout(() => reject(new Error(`${what} timed out`)), ms);
promise.then((value) => {
clearTimeout(timer);
resolve(value);
}, (error) => {
clearTimeout(timer);
reject(error);
});
});
}
class WasmAdapter implements StorageAdapter {
private readonly tx = { depth: 0 };
private readonly statements = new Map<string, any>();
constructor(private readonly sqlite3: any, private readonly db: any) {}
exec(sql: string): void {
this.db.exec(sql);
}
private statement(sql: string): any {
let stmt = this.statements.get(sql);
if (!stmt) {
stmt = this.db.prepare(sql);
this.statements.set(sql, stmt);
}
return stmt;
}
private bindAndRun<T>(sql: string, params: readonly SqlValue[] | undefined, read: (stmt: any) => T): T {
const stmt = this.statement(sql);
try {
if (params && params.length > 0) stmt.bind(params as SqlValue[]);
return read(stmt);
} finally {
stmt.reset(true);
}
}
run(sql: string, params?: readonly SqlValue[]): { changes: number; lastInsertRowid: number } {
this.bindAndRun(sql, params, (stmt) => {
while (stmt.step()) { /* no rows wanted */ }
});
return {
changes: Number(this.db.changes()),
lastInsertRowid: Number(this.sqlite3.capi.sqlite3_last_insert_rowid(this.db.pointer)),
};
}
all<T extends object = SqlRow>(sql: string, params?: readonly SqlValue[]): T[] {
return this.bindAndRun(sql, params, (stmt) => {
const rows: T[] = [];
while (stmt.step()) rows.push(stmt.get({}) as T);
return rows;
});
}
get<T extends object = SqlRow>(sql: string, params?: readonly SqlValue[]): T | undefined {
return this.bindAndRun(sql, params, (stmt) => (stmt.step() ? stmt.get({}) as T : undefined));
}
transaction<T>(fn: () => T): T {
return runTransaction(this.tx, (sql) => this.db.exec(sql), fn);
}
close(): void {
for (const stmt of this.statements.values()) stmt.finalize();
this.statements.clear();
this.db.close();
}
}
/** The module and the pool, once. Only the owner gets here. */
export class WasmStorage {
private constructor(private readonly sqlite3: any, private readonly pool: Pool) {}
/**
* Call only while holding the owner lock. The holder before may still be
* letting go, so a refusal is retried for a while.
*/
static async install(moduleUrl: string): Promise<WasmStorage> {
const { default: init } = await import(/* webpackIgnore: true */ moduleUrl) as { default: (o: unknown) => Promise<any> };
// A failed wasm load never settles the promise from the module's init.
const sqlite3 = await withTimeout(init({ print() {}, printErr() {} }), INIT_TIMEOUT_MS, 'Loading the store engine');
let lastError: unknown;
for (let attempt = 0; attempt < 20; attempt += 1) {
try {
const pool = await sqlite3.installOpfsSAHPoolVfs({
name: 'opfs-sahpool',
directory: POOL_DIRECTORY,
initialCapacity: INITIAL_CAPACITY,
preserveOnInitFailure: true,
forceReinitIfPreviouslyFailed: true,
}) as Pool;
return new WasmStorage(sqlite3, pool);
} catch (error) {
lastError = error;
await new Promise((resolve) => setTimeout(resolve, 100 + attempt * 50));
}
}
throw lastError;
}
/** Opens one account's database. `keyHex` is 32 random bytes in hex. */
async open(file: string, keyHex: string): Promise<StorageAdapter> {
if (!/^\/a[0-9a-f]+\.db$/.test(file)) throw new Error('Store file names carry no account data');
if (!/^[0-9a-f]{64}$/.test(keyHex)) throw new Error('A store key is 32 bytes of hex');
const needed = this.pool.getFileCount() + SLOTS_PER_DATABASE;
if (this.pool.getCapacity() < needed) await this.pool.addCapacity(needed - this.pool.getCapacity());
const db = new this.sqlite3.oo1.DB({ filename: file, vfs: VFS, flags: 'c' });
try {
db.exec(`PRAGMA key = "raw:${keyHex}"`);
try {
db.selectValue('SELECT count(*) FROM sqlite_master');
} catch {
throw new StoreKeyError();
}
const pageSize = Number(db.selectValue('PRAGMA page_size'));
if (pageSize !== REQUIRED_PAGE_SIZE) throw new Error(`Refusing an unsafe page size: ${pageSize}`);
db.exec('PRAGMA temp_store = MEMORY');
db.exec('PRAGMA foreign_keys = OFF');
db.exec('PRAGMA synchronous = NORMAL');
} catch (error) {
db.close();
throw error;
}
return new WasmAdapter(this.sqlite3, db);
}
remove(file: string): boolean {
let removed = false;
for (const name of this.pool.getFileNames()) {
if (name === file || name === `${file}-journal`) removed = this.pool.unlink(name) || removed;
}
return removed;
}
/** Lets go of the OPFS handles. Every database must be closed first. */
pause(): void {
this.pool.pauseVfs();
}
async unpause(): Promise<void> {
await this.pool.unpauseVfs();
}
files(): string[] {
return this.pool.getFileNames();
}
}
+101
View File
@@ -0,0 +1,101 @@
/**
* What one part of the local store (mail, calendar, contacts, files, ...)
* plugs into the engine: its tables, the JMAP methods it answers while the
* server can't be reached, and how it syncs.
*
* Migration id ranges: core 1-99, mail 100-199, outbox 200-299, search
* 300-399, drafts 400-499, calendar 500-599, contacts 600-699, files 700-799,
* settings and reminders 800-899, wipe and lock 900-999.
*/
import type { JmapTransport, MethodCall, StoreSession } from './engine/transport';
import type { Migration, StorageAdapter } from './storage';
export interface StoreScope {
/** How far back mail is kept, in days. Null keeps everything. */
days: number | null;
/** Whether attachments are kept. */
attachments: boolean;
/** The largest attachment kept, and the total for all of them. */
maxAttachmentBytes: number;
maxTotalBytes: number;
}
export const DEFAULT_SCOPE: StoreScope = {
days: 90,
attachments: true,
maxAttachmentBytes: 25 * 1024 * 1024,
maxTotalBytes: 2 * 1024 * 1024 * 1024,
};
export interface SyncStateRow {
state: string | null;
/** The newest state the server told us about, to know when a run is caught up. */
target: string | null;
/** The area's own phase name; null when in normal delta mode. */
phase: string | null;
/** The area's own resume point. */
cursor: Record<string, unknown> | null;
}
export type StoreEvent =
| { type: 'progress'; account: string; area: string; phase: string; done: number; total: number | null }
| { type: 'synced'; account: string; at: number }
| { type: 'changed'; account: string; collections: string[] }
| { type: 'created'; account: string; jmap: string; collection: string; ids: string[] }
| { type: 'error'; account: string; area: string; message: string };
export interface StoreContext {
account: string;
db: StorageAdapter;
session: StoreSession;
scope: StoreScope;
now(): number;
emit(event: StoreEvent): void;
}
export interface SyncContext extends StoreContext {
transport: JmapTransport;
/** True once the engine is closing or lost the network: stop at the next batch. */
stopped(): boolean;
syncState(jmap: string, type: string): SyncStateRow;
/** Call inside the transaction that writes the data, so the two can't disagree. */
setSyncState(jmap: string, type: string, row: Partial<SyncStateRow>): void;
}
export interface Syncer {
/** A short name for progress and errors. */
name: string;
/** The JMAP capability an account must advertise for this to run for it. */
capability: string;
/** Runs one pass for one JMAP account. Must be safe to stop between batches and start again. */
run(ctx: SyncContext, jmap: string): Promise<void>;
}
export type MethodResult = Record<string, unknown>;
/** Thrown by a method handler: the client gets a per-call JMAP error. */
export class MethodError extends Error {
constructor(readonly type: string, readonly description?: string) {
super(description ?? type);
this.name = 'MethodError';
}
}
export interface MethodContext extends StoreContext {
/** The account's own default JMAP account for calls that don't say. */
defaultJmap: string;
/** Whether keyword sort comparators arrive inverted from this client. */
invertedKeywordSort: boolean;
/** Earlier responses in this request, for creation-id references. */
createdIds: Map<string, string>;
}
export type MethodHandler = (ctx: MethodContext, args: Record<string, unknown>, call: MethodCall) => MethodResult | Promise<MethodResult>;
export interface StoreArea {
name: string;
migrations: readonly Migration[];
methods: Readonly<Record<string, MethodHandler>>;
syncers: readonly Syncer[];
}
+6
View File
@@ -0,0 +1,6 @@
/** Every part of the local store. New areas go here. */
import type { StoreArea } from './area';
import { mailArea } from './mail';
export const STORE_AREAS: readonly StoreArea[] = [mailArea];
+173
View File
@@ -0,0 +1,173 @@
/**
* The per-account database keys, as the page sees them.
*
* Each account's database has a random 32-byte key. It is only ever stored
* wrapped: AES-GCM with a non-extractable key that lives in IndexedDB, then
* any extra layers on top (the desktop app's OS keychain, app lock). Forgetting
* the record makes the file unreadable even if deleting it failed.
*/
const DB_NAME = 'bulwark-store';
const DB_VERSION = 1;
const KEYS = 'keys';
const ACCOUNTS = 'accounts';
const WRAP_ID = 'wrap';
export interface WrapLayer {
/** A name for the layer: 'safeStorage', 'pin', 'prf'. */
name: string;
wrap(data: Uint8Array): Promise<Uint8Array>;
unwrap(data: Uint8Array): Promise<Uint8Array>;
}
export interface AccountRecord {
account: string;
/** `/a<hex>.db`. */
file: string;
iv: Uint8Array;
wrapped: Uint8Array;
/** Layers applied after the WebCrypto wrap, innermost first. */
layers: string[];
createdAt: number;
}
function promisify<T>(request: IDBRequest<T>): Promise<T> {
return new Promise((resolve, reject) => {
request.onsuccess = () => resolve(request.result);
request.onerror = () => reject(request.error);
});
}
function openDb(): Promise<IDBDatabase> {
const request = indexedDB.open(DB_NAME, DB_VERSION);
request.onupgradeneeded = () => {
const db = request.result;
if (!db.objectStoreNames.contains(KEYS)) db.createObjectStore(KEYS, { keyPath: 'id' });
if (!db.objectStoreNames.contains(ACCOUNTS)) db.createObjectStore(ACCOUNTS, { keyPath: 'account' });
};
return promisify(request);
}
async function withDb<T>(fn: (db: IDBDatabase) => Promise<T>): Promise<T> {
const db = await openDb();
try {
return await fn(db);
} finally {
db.close();
}
}
async function wrapKey(db: IDBDatabase): Promise<CryptoKey> {
const read = db.transaction(KEYS, 'readonly').objectStore(KEYS).get(WRAP_ID);
const existing = await promisify(read) as { key: CryptoKey } | undefined;
if (existing) return existing.key;
const key = await crypto.subtle.generateKey({ name: 'AES-GCM', length: 256 }, false, ['encrypt', 'decrypt']);
const tx = db.transaction(KEYS, 'readwrite');
const store = tx.objectStore(KEYS);
// Another tab may have got there first: keep theirs.
const raced = await promisify(store.get(WRAP_ID)) as { key: CryptoKey } | undefined;
if (raced) return raced.key;
await promisify(store.put({ id: WRAP_ID, key }));
return key;
}
function toHex(bytes: Uint8Array): string {
return Array.from(bytes, (b) => b.toString(16).padStart(2, '0')).join('');
}
function asBuffer(bytes: Uint8Array): ArrayBuffer {
return bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength) as ArrayBuffer;
}
export function newStoreFileName(): string {
return `/a${toHex(crypto.getRandomValues(new Uint8Array(8)))}.db`;
}
export async function getAccountRecord(account: string): Promise<AccountRecord | null> {
return withDb(async (db) => {
const found = await promisify(db.transaction(ACCOUNTS, 'readonly').objectStore(ACCOUNTS).get(account)) as AccountRecord | undefined;
return found ?? null;
});
}
export async function listAccountRecords(): Promise<AccountRecord[]> {
return withDb((db) => promisify(db.transaction(ACCOUNTS, 'readonly').objectStore(ACCOUNTS).getAll()) as Promise<AccountRecord[]>);
}
/** The account's key in hex, created on first use. Returns the record too, for the file name. */
export async function openAccountKey(account: string, layers: readonly WrapLayer[] = []): Promise<{ keyHex: string; record: AccountRecord }> {
return withDb(async (db) => {
const wrapping = await wrapKey(db);
const found = await promisify(db.transaction(ACCOUNTS, 'readonly').objectStore(ACCOUNTS).get(account)) as AccountRecord | undefined;
if (found) {
let data = found.wrapped;
for (const name of [...found.layers].reverse()) {
const layer = layers.find((candidate) => candidate.name === name);
if (!layer) throw new StoreKeyLayerMissingError(name);
data = await layer.unwrap(data);
}
const raw = new Uint8Array(await crypto.subtle.decrypt({ name: 'AES-GCM', iv: asBuffer(found.iv) }, wrapping, asBuffer(data)));
const keyHex = toHex(raw);
raw.fill(0);
return { keyHex, record: found };
}
const raw = crypto.getRandomValues(new Uint8Array(32));
const iv = crypto.getRandomValues(new Uint8Array(12));
let wrapped = new Uint8Array(await crypto.subtle.encrypt({ name: 'AES-GCM', iv }, wrapping, raw));
for (const layer of layers) wrapped = new Uint8Array(await layer.wrap(wrapped));
const record: AccountRecord = {
account,
file: newStoreFileName(),
iv,
wrapped,
layers: layers.map((layer) => layer.name),
createdAt: Date.now(),
};
await promisify(db.transaction(ACCOUNTS, 'readwrite').objectStore(ACCOUNTS).put(record));
const keyHex = toHex(raw);
raw.fill(0);
return { keyHex, record };
});
}
/** Adds or removes wrap layers without touching the database key. */
export async function rewrapAccountKey(
account: string,
from: readonly WrapLayer[],
to: readonly WrapLayer[],
): Promise<void> {
await withDb(async (db) => {
const found = await promisify(db.transaction(ACCOUNTS, 'readonly').objectStore(ACCOUNTS).get(account)) as AccountRecord | undefined;
if (!found) return;
let data = found.wrapped;
for (const name of [...found.layers].reverse()) {
const layer = from.find((candidate) => candidate.name === name);
if (!layer) throw new StoreKeyLayerMissingError(name);
data = await layer.unwrap(data);
}
for (const layer of to) data = new Uint8Array(await layer.wrap(data));
const record: AccountRecord = { ...found, wrapped: data, layers: to.map((layer) => layer.name) };
await promisify(db.transaction(ACCOUNTS, 'readwrite').objectStore(ACCOUNTS).put(record));
});
}
/** Forgets the account's key. The database file can't be read after this. */
export async function forgetAccountKey(account: string): Promise<void> {
await withDb((db) => promisify(db.transaction(ACCOUNTS, 'readwrite').objectStore(ACCOUNTS).delete(account)).then(() => undefined));
}
export class StoreKeyLayerMissingError extends Error {
constructor(readonly layer: string) {
super(`The key needs the ${layer} layer`);
this.name = 'StoreKeyLayerMissingError';
}
}
/** Deletes every key at sign-out. Works with no worker running, on any page. */
export function deleteKeyDatabase(): void {
try {
indexedDB.deleteDatabase(DB_NAME);
} catch {
// Private mode or similar: nothing was stored either.
}
}
+261
View File
@@ -0,0 +1,261 @@
/**
* The page's side of the local store. Each tab starts its own worker, which
* waits for the owner lock. Calls go over a BroadcastChannel to whichever
* worker owns the store, including this tab's own, so a tab doesn't need to
* know whether it's the owner. A key never goes over the channel.
*/
import type { StoreEvent, StoreScope } from '../area';
import type { JmapResponse, MethodCall, StoreSession } from '../engine/transport';
import { SQLITE3MC_PUBLIC_DIR } from '../sqlite-version';
import {
RPC_CHANNEL,
type ChannelMessage,
type OpenAccountParams,
type PageToWorker,
type RpcReply,
type StoreRpc,
type StoreStatus,
type WorkerToPage,
} from '../worker/protocol';
const CALL_TIMEOUT_MS = 60_000;
const OWNER_WAIT_MS = 15_000;
export class StoreOwnerGoneError extends Error {
constructor() {
super('The local store is not available in this browser window');
this.name = 'StoreOwnerGoneError';
}
}
interface Pending {
resolve(value: unknown): void;
reject(error: Error): void;
timer: ReturnType<typeof setTimeout>;
}
export interface StoreClientOptions {
/** NEXT_PUBLIC_BASE_PATH plus the build id, passed in so this works in tests. */
basePath?: string;
buildId?: string;
onEvent?(event: StoreEvent): void;
/** The tab became the owner: open its accounts. */
onOwner?(): void;
/** The worker asks for a fresh Authorization header. */
onNeedAuth?(account: string): void;
}
export class StoreClient {
readonly clientId = crypto.randomUUID();
private worker: Worker | null = null;
private channel: BroadcastChannel | null = null;
private owner: string | null = null;
private isOwner = false;
private readonly pending = new Map<string, Pending>();
private readonly ownerWaiters: Array<() => void> = [];
constructor(private readonly options: StoreClientOptions = {}) {}
static supported(): boolean {
return typeof Worker === 'function'
&& typeof BroadcastChannel === 'function'
&& typeof navigator !== 'undefined'
&& Boolean(navigator.locks)
&& typeof navigator.storage?.getDirectory === 'function'
&& typeof FileSystemFileHandle !== 'undefined'
&& 'createSyncAccessHandle' in FileSystemFileHandle.prototype;
}
start(): void {
if (this.worker) return;
const base = this.options.basePath ?? '';
const query = this.options.buildId ? `?v=${encodeURIComponent(this.options.buildId)}` : '';
this.channel = new BroadcastChannel(RPC_CHANNEL);
this.channel.onmessage = (event: MessageEvent<ChannelMessage>) => this.onChannel(event.data);
this.worker = new Worker(`${base}/store-worker.js${query}`, { type: 'module' });
this.worker.onmessage = (event: MessageEvent<WorkerToPage>) => this.onWorker(event.data);
this.post({
kind: 'init',
clientId: this.clientId,
moduleUrl: new URL(`${base}/${SQLITE3MC_PUBLIC_DIR}/sqlite3.mjs`, location.origin).href,
});
this.bindLifecycle();
}
private bindLifecycle(): void {
// A clean handover when the tab goes. If this doesn't get to run, the lock
// and the handles die with the worker and the next owner retries.
addEventListener('pagehide', (event) => {
if (!event.persisted) void this.release().catch(() => {});
});
}
private post(message: PageToWorker): void {
this.worker?.postMessage(message);
}
private onWorker(message: WorkerToPage): void {
switch (message.kind) {
case 'owner':
this.isOwner = true;
this.owner = this.clientId;
this.flushOwnerWaiters();
this.options.onOwner?.();
break;
case 'result': {
const pending = this.pending.get(message.id);
if (!pending) return;
this.pending.delete(message.id);
clearTimeout(pending.timer);
if (message.error) {
const error = new Error(message.error);
if (message.errorName) error.name = message.errorName;
pending.reject(error);
} else {
pending.resolve(message.result);
}
break;
}
case 'need-auth':
this.options.onNeedAuth?.(message.account);
break;
case 'event':
// The owner's events also reach every other tab over the channel.
this.options.onEvent?.(message.event);
break;
}
}
private onChannel(message: ChannelMessage): void {
switch (message.kind) {
case 'reply':
if ((message as RpcReply).to === this.clientId) this.settle(message);
break;
case 'event':
if (!this.isOwner) this.options.onEvent?.(message.event);
break;
case 'owner-changed':
this.owner = message.owner;
if (message.owner) this.flushOwnerWaiters();
break;
default:
break;
}
}
private settle(message: RpcReply): void {
const pending = this.pending.get(message.id);
if (!pending) return;
this.pending.delete(message.id);
clearTimeout(pending.timer);
if (message.error) pending.reject(new Error(message.error));
else pending.resolve(message.result);
}
private flushOwnerWaiters(): void {
for (const waiter of this.ownerWaiters.splice(0)) waiter();
}
private track<T>(id: string, timeoutMs: number): Promise<T> {
return new Promise<T>((resolve, reject) => {
const timer = setTimeout(() => {
this.pending.delete(id);
reject(new StoreOwnerGoneError());
}, timeoutMs);
this.pending.set(id, { resolve: resolve as (value: unknown) => void, reject, timer });
});
}
private async waitForOwner(): Promise<void> {
if (this.owner) return;
await new Promise<void>((resolve, reject) => {
const timer = setTimeout(() => reject(new StoreOwnerGoneError()), OWNER_WAIT_MS);
this.ownerWaiters.push(() => {
clearTimeout(timer);
resolve();
});
});
}
private async rpc<K extends keyof StoreRpc>(method: K, params: Parameters<StoreRpc[K]>[0]): Promise<ReturnType<StoreRpc[K]>> {
await this.waitForOwner();
const id = crypto.randomUUID();
const result = this.track<ReturnType<StoreRpc[K]>>(id, CALL_TIMEOUT_MS);
this.channel?.postMessage({ kind: 'rpc', id, from: this.clientId, method, params } as ChannelMessage);
return result;
}
private direct<T>(message: PageToWorker & { id: string }): Promise<T> {
const result = this.track<T>(message.id, CALL_TIMEOUT_MS);
this.post(message);
return result;
}
// ---- Owner only: these carry keys or manage files ----
get ownsStore(): boolean {
return this.isOwner;
}
openAccount(params: OpenAccountParams): Promise<void> {
return this.direct({ kind: 'open', id: crypto.randomUUID(), params });
}
closeAccount(account: string): Promise<void> {
return this.direct({ kind: 'close-account', id: crypto.randomUUID(), account });
}
removeAccount(account: string, file: string): Promise<boolean> {
return this.direct({ kind: 'remove-account', id: crypto.randomUUID(), account, file });
}
setAuth(account: string, auth: string | null): void {
this.post({ kind: 'set-auth', account, auth });
}
setSession(account: string, session: StoreSession): void {
this.post({ kind: 'set-session', account, session });
}
setScope(account: string, scope: StoreScope): void {
this.post({ kind: 'set-scope', account, scope });
}
/** Hands the store over: the worker closes everything, then lets go of the lock. */
async release(): Promise<void> {
if (!this.worker) return;
const worker = this.worker;
if (this.isOwner) {
await this.direct({ kind: 'release', id: crypto.randomUUID() });
this.isOwner = false;
this.owner = null;
}
worker.terminate();
this.worker = null;
this.channel?.close();
this.channel = null;
}
// ---- Any tab ----
answer(account: string, calls: MethodCall[]): Promise<JmapResponse | null> {
return this.rpc('answer', { account, calls });
}
sync(account: string): Promise<void> {
return this.rpc('sync', { account });
}
noteStateChange(account: string, changed: Record<string, Record<string, string>>): Promise<void> {
return this.rpc('noteStateChange', { account, changed });
}
setOnline(online: boolean): Promise<void> {
return this.rpc('setOnline', { online });
}
status(account: string): Promise<StoreStatus | null> {
return this.rpc('status', { account });
}
}
+223
View File
@@ -0,0 +1,223 @@
/**
* One local account in the store: its database, its sync runs, its answers.
* It acts as a JMAP client toward the real server and as a JMAP server toward
* the UI. Nothing here touches the DOM, so tests run it on node:sqlite.
*/
import { MethodError, type MethodHandler, type StoreArea, type StoreEvent, type StoreScope, type SyncContext, type SyncStateRow, type Syncer } from '../area';
import { migrate, type StorageAdapter } from '../storage';
import { answerRequest, buildMethodTable } from './responder';
import { TransportError, type JmapResponse, type JmapTransport, type MethodCall, type StoreSession } from './transport';
import { CORE_MIGRATIONS, metaGet, metaSet } from './core-schema';
export interface AccountEngineOptions {
account: string;
db: StorageAdapter;
session: StoreSession;
transport: JmapTransport;
areas: readonly StoreArea[];
scope: StoreScope;
invertedKeywordSort?: boolean;
emit?: (event: StoreEvent) => void;
now?: () => number;
/** Waits before a retry after the server said it's busy. Tests pass a stub. */
sleep?: (ms: number) => Promise<void>;
}
const RATE_RETRIES = 5;
const RATE_BACKOFF_MS = 2_000;
export class AccountEngine {
readonly account: string;
private readonly db: StorageAdapter;
private session: StoreSession;
private scope: StoreScope;
private readonly transport: JmapTransport;
private readonly syncers: readonly Syncer[];
private readonly methods: ReadonlyMap<string, MethodHandler>;
private readonly emitter: (event: StoreEvent) => void;
private readonly now: () => number;
private readonly sleep: (ms: number) => Promise<void>;
private invertedKeywordSort: boolean;
private closed = false;
private online = true;
private running: Promise<void> | null = null;
private again = false;
constructor(options: AccountEngineOptions) {
this.account = options.account;
this.db = options.db;
this.session = options.session;
this.scope = options.scope;
this.transport = options.transport;
this.syncers = options.areas.flatMap((area) => area.syncers);
this.methods = buildMethodTable(options.areas);
this.emitter = options.emit ?? (() => {});
this.now = options.now ?? Date.now;
this.sleep = options.sleep ?? ((ms) => new Promise((resolve) => setTimeout(resolve, ms)));
this.invertedKeywordSort = options.invertedKeywordSort ?? false;
migrate(this.db, [...CORE_MIGRATIONS, ...options.areas.flatMap((area) => area.migrations)]);
metaSet(this.db, 'session', JSON.stringify(options.session));
}
/** The session kept from the last time, for a start without a network. */
static storedSession(db: StorageAdapter): StoreSession | null {
migrate(db, CORE_MIGRATIONS);
const raw = metaGet(db, 'session');
return raw ? JSON.parse(raw) as StoreSession : null;
}
setSession(session: StoreSession): void {
this.session = session;
metaSet(this.db, 'session', JSON.stringify(session));
}
setScope(scope: StoreScope): void {
this.scope = scope;
metaSet(this.db, 'scope', JSON.stringify(scope));
}
setInvertedKeywordSort(inverted: boolean): void {
this.invertedKeywordSort = inverted;
}
setOnline(online: boolean): void {
this.online = online;
if (online) this.requestSync();
}
private syncState(jmap: string, type: string): SyncStateRow {
const row = this.db.get<{ state: string | null; target: string | null; phase: string | null; cursor: string | null }>(
'SELECT state, target, phase, cursor FROM sync_state WHERE jmap = ? AND type = ?',
[jmap, type],
);
return {
state: row?.state ?? null,
target: row?.target ?? null,
phase: row?.phase ?? null,
cursor: row?.cursor ? JSON.parse(row.cursor) as Record<string, unknown> : null,
};
}
private setSyncState(jmap: string, type: string, patch: Partial<SyncStateRow>): void {
const row = { ...this.syncState(jmap, type), ...patch };
this.db.run(
`INSERT INTO sync_state (jmap, type, state, target, phase, cursor, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)
ON CONFLICT (jmap, type) DO UPDATE SET state = excluded.state, target = excluded.target, phase = excluded.phase,
cursor = excluded.cursor, updated_at = excluded.updated_at`,
[jmap, type, row.state, row.target, row.phase, row.cursor ? JSON.stringify(row.cursor) : null, this.now()],
);
}
/** What the server pushed: jmap account id to { type: state }. */
noteStateChange(changed: Record<string, Record<string, string>>): void {
this.db.transaction(() => {
for (const [jmap, types] of Object.entries(changed)) {
for (const [type, state] of Object.entries(types)) this.setSyncState(jmap, type, { target: state });
}
});
this.requestSync();
}
/** Starts a run, or asks the running one to go again when it's done. */
requestSync(): Promise<void> {
if (this.closed) return Promise.resolve();
if (this.running) {
this.again = true;
return this.running;
}
this.running = this.runLoop().finally(() => {
this.running = null;
});
return this.running;
}
private async runLoop(): Promise<void> {
do {
this.again = false;
await this.runOnce();
} while (this.again && !this.closed && this.online);
}
private async runOnce(): Promise<void> {
if (!this.online) return;
const ctx = this.syncContext();
const jmapAccounts = Object.keys(this.session.accounts);
for (const syncer of this.syncers) {
for (const jmap of jmapAccounts) {
const capabilities = this.session.accounts[jmap].accountCapabilities ?? this.session.capabilities;
if (!(syncer.capability in capabilities)) continue;
if (this.closed || !this.online) return;
if (!(await this.runSyncer(ctx, syncer, jmap))) return;
}
}
if (!this.closed && this.online) this.emitter({ type: 'synced', account: this.account, at: this.now() });
}
/** False when the run should end: no network, no credentials. */
private async runSyncer(ctx: SyncContext, syncer: Syncer, jmap: string): Promise<boolean> {
for (let attempt = 0; ; attempt += 1) {
try {
await syncer.run(ctx, jmap);
return true;
} catch (error) {
if (error instanceof TransportError) {
if (error.kind === 'network' || error.kind === 'auth') return false;
if (error.kind === 'rate' && attempt < RATE_RETRIES) {
await this.sleep(error.retryAfterMs ?? RATE_BACKOFF_MS * 2 ** attempt);
continue;
}
}
// One part failing doesn't hold up the others. It runs again next time.
this.emitter({
type: 'error',
account: this.account,
area: syncer.name,
message: error instanceof Error ? error.message : String(error),
});
return true;
}
}
}
private syncContext(): SyncContext {
return {
account: this.account,
db: this.db,
session: this.session,
scope: this.scope,
transport: this.transport,
now: this.now,
emit: this.emitter,
stopped: () => this.closed || !this.online,
syncState: (jmap, type) => this.syncState(jmap, type),
setSyncState: (jmap, type, row) => this.setSyncState(jmap, type, row),
};
}
/** An answer from the device. Null when nothing in the request could be answered. */
async answer(calls: MethodCall[]): Promise<JmapResponse | null> {
if (this.closed) throw new MethodError('serverUnavailable', 'The local store is closed');
const primary = this.session.primaryAccounts['urn:ietf:params:jmap:mail'] ?? Object.keys(this.session.accounts)[0];
return answerRequest(
{
account: this.account,
db: this.db,
session: this.session,
scope: this.scope,
now: this.now,
emit: this.emitter,
defaultJmap: primary,
invertedKeywordSort: this.invertedKeywordSort,
},
this.methods,
calls,
);
}
/** Stops syncing and waits for the current batch. The database stays open for the caller to close. */
async close(): Promise<void> {
this.closed = true;
await this.running?.catch(() => {});
}
}
+101
View File
@@ -0,0 +1,101 @@
/**
* The /changes loop, written for how servers really answer (see
* local-data/desktop-offline-facts-parts/stalwart-changes-prf.md):
*
* - sinceState goes last: on Stalwart a malformed state followed by more
* arguments is an HTTP 400 that fails the whole request.
* - hasMoreChanges false doesn't mean caught up. A page walks a fixed range of
* the log, so the loop also runs while newState differs from the state the
* server last pushed.
* - An unusable state comes as cannotCalculateChanges, invalidArguments or that
* HTTP 400. All three mean: start this type over.
*/
import { JmapMethodError, responseFor, TransportError, type JmapTransport } from './transport';
export interface ChangesPage {
created: string[];
updated: string[];
destroyed: string[];
newState: string;
/** Set when only these properties changed (Mailbox counts). */
updatedProperties: string[] | null;
}
export interface ChangesLoopOptions {
transport: JmapTransport;
type: string;
jmap: string;
using: string[];
since: string;
/** The newest state the server pushed, if any. */
target?: string | null;
maxChanges?: number;
stopped?: () => boolean;
/** Stores one page and its newState. Runs to completion before the next call. */
apply(page: ChangesPage): Promise<void>;
}
export type ChangesOutcome =
| { outcome: 'caughtUp'; state: string; pages: number }
| { outcome: 'stopped'; state: string; pages: number }
| { outcome: 'resync'; state: string; pages: number };
const MAX_PAGES_PAST_TARGET = 3;
/** A stored state a server can't have sent: don't send it, it can fail the whole request. */
export function isUsableState(state: unknown): state is string {
return typeof state === 'string' && state.length > 0 && state.length < 1024 && /^[\w.~+=:/-]+$/.test(state);
}
function isStateRefusal(error: unknown): boolean {
if (error instanceof JmapMethodError) {
return error.type === 'cannotCalculateChanges' || error.type === 'invalidArguments';
}
if (error instanceof TransportError) {
return error.kind === 'http' && error.status === 400 && /notRequest/.test(error.problem ?? '');
}
return false;
}
export async function runChangesLoop(options: ChangesLoopOptions): Promise<ChangesOutcome> {
const { transport, type, jmap, using, target } = options;
const maxChanges = options.maxChanges ?? 500;
let state = options.since;
let pages = 0;
let pastTarget = 0;
if (!isUsableState(state)) return { outcome: 'resync', state, pages };
for (;;) {
if (options.stopped?.()) return { outcome: 'stopped', state, pages };
let page: ChangesPage & { hasMoreChanges: boolean };
try {
const response = await transport.request(
[[`${type}/changes`, { accountId: jmap, maxChanges, sinceState: state }, 'c']],
using,
);
const [, result] = responseFor(response, 'c');
page = {
created: result.created ?? [],
updated: result.updated ?? [],
destroyed: result.destroyed ?? [],
newState: String(result.newState),
updatedProperties: Array.isArray(result.updatedProperties) ? result.updatedProperties : null,
hasMoreChanges: Boolean(result.hasMoreChanges),
};
} catch (error) {
if (isStateRefusal(error)) return { outcome: 'resync', state, pages };
throw error;
}
const before = state;
await options.apply(page);
state = page.newState;
pages += 1;
if (page.hasMoreChanges) continue;
if (!target || state === target) return { outcome: 'caughtUp', state, pages };
// Behind the pushed state: the log moved on while we paged. A state that
// stands still means the server has nothing more for this type.
if (state === before) return { outcome: 'caughtUp', state, pages };
pastTarget += 1;
if (pastTarget >= MAX_PAGES_PAST_TARGET) return { outcome: 'caughtUp', state, pages };
}
}
+57
View File
@@ -0,0 +1,57 @@
import type { Migration, StorageAdapter } from '../storage';
export const CORE_MIGRATIONS: readonly Migration[] = [
{
id: 1,
name: 'core',
sql: `
CREATE TABLE meta (k TEXT PRIMARY KEY, v TEXT) WITHOUT ROWID;
CREATE TABLE sync_state (
jmap TEXT NOT NULL,
type TEXT NOT NULL,
state TEXT,
target TEXT,
phase TEXT,
cursor TEXT,
updated_at INTEGER NOT NULL,
PRIMARY KEY (jmap, type)
) WITHOUT ROWID;
CREATE TABLE object (
jmap TEXT NOT NULL,
type TEXT NOT NULL,
id TEXT NOT NULL,
json TEXT NOT NULL,
PRIMARY KEY (jmap, type, id)
) WITHOUT ROWID;
`,
},
];
export function metaGet(db: StorageAdapter, key: string): string | null {
return db.get<{ v: string | null }>('SELECT v FROM meta WHERE k = ?', [key])?.v ?? null;
}
export function metaSet(db: StorageAdapter, key: string, value: string | null): void {
db.run('INSERT INTO meta (k, v) VALUES (?, ?) ON CONFLICT (k) DO UPDATE SET v = excluded.v', [key, value]);
}
export function putObject(db: StorageAdapter, jmap: string, type: string, id: string, value: unknown): void {
db.run(
'INSERT INTO object (jmap, type, id, json) VALUES (?, ?, ?, ?) ON CONFLICT (jmap, type, id) DO UPDATE SET json = excluded.json',
[jmap, type, id, JSON.stringify(value)],
);
}
export function deleteObject(db: StorageAdapter, jmap: string, type: string, id: string): void {
db.run('DELETE FROM object WHERE jmap = ? AND type = ? AND id = ?', [jmap, type, id]);
}
export function getObject<T = Record<string, unknown>>(db: StorageAdapter, jmap: string, type: string, id: string): T | null {
const row = db.get<{ json: string }>('SELECT json FROM object WHERE jmap = ? AND type = ? AND id = ?', [jmap, type, id]);
return row ? JSON.parse(row.json) as T : null;
}
export function listObjects<T = Record<string, unknown>>(db: StorageAdapter, jmap: string, type: string): T[] {
return db.all<{ json: string }>('SELECT json FROM object WHERE jmap = ? AND type = ?', [jmap, type])
.map((row) => JSON.parse(row.json) as T);
}
+89
View File
@@ -0,0 +1,89 @@
/**
* Answers a JMAP request from the local store: the store acting as a server
* toward the UI. Every call gets its own answer, in order, as a server would:
* one the store can't answer gets a per-call error and the rest still work.
*/
import { MethodError, type MethodContext, type MethodHandler, type MethodResult, type StoreArea } from '../area';
import type { JmapResponse, MethodCall, MethodResponse } from './transport';
class BadReference extends Error {}
/** RFC 8620 §3.7 result references, with the `*` array step. */
export function resolvePointer(value: unknown, path: string): unknown {
const steps = path.split('/').slice(1).map((s) => s.replace(/~1/g, '/').replace(/~0/g, '~'));
const walk = (node: unknown, i: number): unknown => {
if (i === steps.length) return node;
const step = steps[i];
if (step === '*') {
if (!Array.isArray(node)) throw new BadReference();
return node.flatMap((item) => {
const out = walk(item, i + 1);
return Array.isArray(out) ? out : [out];
});
}
if (node === null || typeof node !== 'object') throw new BadReference();
return walk((node as Record<string, unknown>)[step], i + 1);
};
return walk(value, 0);
}
function resolveArgs(args: Record<string, unknown>, done: readonly MethodResponse[]): Record<string, unknown> {
const out: Record<string, unknown> = {};
for (const [key, value] of Object.entries(args)) {
if (!key.startsWith('#')) {
out[key] = value;
continue;
}
const ref = value as { resultOf?: string; name?: string; path?: string };
const source = done.find(([name, , id]) => id === ref.resultOf && name === ref.name);
if (!source || typeof ref.path !== 'string') throw new BadReference();
out[key.slice(1)] = resolvePointer(source[1], ref.path);
}
return out;
}
export function buildMethodTable(areas: readonly StoreArea[]): Map<string, MethodHandler> {
const table = new Map<string, MethodHandler>();
for (const area of areas) {
for (const [name, handler] of Object.entries(area.methods)) {
if (table.has(name)) throw new Error(`Two store areas answer ${name}`);
table.set(name, handler);
}
}
return table;
}
/** Answers are copied by structured clone on the way to the page, so callers can't mutate stored objects. */
export async function answerRequest(
ctx: Omit<MethodContext, 'createdIds'>,
methods: ReadonlyMap<string, MethodHandler>,
calls: readonly MethodCall[],
): Promise<JmapResponse | null> {
const done: MethodResponse[] = [];
const full: MethodContext = { ...ctx, createdIds: new Map() };
let answered = 0;
for (const call of calls) {
const [name, rawArgs, callId] = call;
const handler = methods.get(name);
if (!handler) {
done.push(['error', { type: 'serverUnavailable', description: `${name} needs the server` }, callId]);
continue;
}
try {
const args = resolveArgs(rawArgs, done);
const result: MethodResult = await handler(full, args, call);
done.push([name, result, callId]);
answered += 1;
} catch (error) {
if (error instanceof BadReference) {
done.push(['error', { type: 'invalidResultReference' }, callId]);
} else if (error instanceof MethodError) {
done.push(['error', { type: error.type, description: error.description }, callId]);
} else {
throw error;
}
}
}
return answered === 0 ? null : { methodResponses: done };
}
+142
View File
@@ -0,0 +1,142 @@
/**
* The engine's way to the real JMAP server. Bulk sync runs in the worker with
* its own fetch, so large JSON isn't parsed on the main thread. Auth stays
* with the page: it hands over the header and refreshes it on request.
*/
export type MethodCall = [string, Record<string, unknown>, string];
// Shaped like the server's answer (any per method, as the callers read it).
// eslint-disable-next-line @typescript-eslint/no-explicit-any
export type MethodResponse = [string, any, string];
export interface JmapResponse {
methodResponses: MethodResponse[];
sessionState?: string;
}
/** The parts of a JMAP session the engine uses. */
export interface StoreSession {
apiUrl: string;
downloadUrl: string;
uploadUrl: string;
primaryAccounts: Record<string, string>;
accounts: Record<string, { name?: string; isPersonal?: boolean; accountCapabilities?: Record<string, unknown> }>;
capabilities: Record<string, unknown>;
}
export type TransportFailure = 'network' | 'auth' | 'rate' | 'http';
export class TransportError extends Error {
constructor(
readonly kind: TransportFailure,
message: string,
readonly status?: number,
/** The `type` of an RFC 7807 problem body, when there was one. */
readonly problem?: string,
readonly retryAfterMs?: number,
) {
super(message);
this.name = 'TransportError';
}
}
export interface JmapTransport {
request(calls: MethodCall[], using: string[]): Promise<JmapResponse>;
download(jmap: string, blobId: string, type: string, name: string): Promise<Uint8Array>;
upload(jmap: string, data: Uint8Array, type: string): Promise<{ blobId: string; type: string; size: number }>;
}
export interface FetchTransportOptions {
session(): StoreSession;
/** The current Authorization header value, or null when the page has none yet. */
auth(): string | null;
/** Asks the page for a fresh header; resolves null when it can't get one. */
refreshAuth(): Promise<string | null>;
fetch?: typeof fetch;
}
const REQUEST_TIMEOUT_MS = 120_000;
function fillTemplate(template: string, values: Record<string, string>): string {
return template.replace(/\{(\w+)\}/g, (whole, key: string) => (key in values ? encodeURIComponent(values[key]) : whole));
}
export class FetchTransport implements JmapTransport {
constructor(private readonly options: FetchTransportOptions) {}
private async fetchWithAuth(url: string, init: RequestInit): Promise<Response> {
const doFetch = this.options.fetch ?? fetch;
const send = async (header: string | null): Promise<Response> => {
if (!header) throw new TransportError('auth', 'No credentials yet');
const signal = AbortSignal.timeout(REQUEST_TIMEOUT_MS);
try {
return await doFetch(url, { ...init, signal, headers: { ...init.headers, Authorization: header } });
} catch (error) {
throw new TransportError('network', error instanceof Error ? error.message : 'Network failure');
}
};
let response = await send(this.options.auth());
if (response.status === 401) {
response = await send(await this.options.refreshAuth());
if (response.status === 401) throw new TransportError('auth', 'The server refused the credentials', 401);
}
if (response.status === 429 || response.status === 503) {
const seconds = Number(response.headers.get('Retry-After'));
throw new TransportError('rate', `Server busy (${response.status})`, response.status, undefined, Number.isFinite(seconds) && seconds > 0 ? seconds * 1000 : undefined);
}
return response;
}
async request(calls: MethodCall[], using: string[]): Promise<JmapResponse> {
const response = await this.fetchWithAuth(this.options.session().apiUrl, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ using, methodCalls: calls }),
});
if (!response.ok) {
let problem: string | undefined;
try {
problem = (await response.json() as { type?: string }).type;
} catch {
// Not a problem document.
}
throw new TransportError('http', `JMAP request failed: ${response.status}`, response.status, problem);
}
return await response.json() as JmapResponse;
}
async download(jmap: string, blobId: string, type: string, name: string): Promise<Uint8Array> {
const url = fillTemplate(this.options.session().downloadUrl, { accountId: jmap, blobId, type, name });
const response = await this.fetchWithAuth(url, { method: 'GET' });
if (!response.ok) throw new TransportError('http', `Download failed: ${response.status}`, response.status);
return new Uint8Array(await response.arrayBuffer());
}
async upload(jmap: string, data: Uint8Array, type: string): Promise<{ blobId: string; type: string; size: number }> {
const url = fillTemplate(this.options.session().uploadUrl, { accountId: jmap });
const response = await this.fetchWithAuth(url, {
method: 'POST',
headers: { 'Content-Type': type || 'application/octet-stream' },
body: data as BodyInit,
});
if (!response.ok) throw new TransportError('http', `Upload failed: ${response.status}`, response.status);
return await response.json() as { blobId: string; type: string; size: number };
}
}
/** Finds the response to one call. Throws if the server answered it with an error. */
export function responseFor(response: JmapResponse, callId: string): MethodResponse {
const found = response.methodResponses.find(([name, , id]) => id === callId && name !== 'error');
if (found) return found;
const failed = response.methodResponses.find(([, , id]) => id === callId);
if (failed) throw new JmapMethodError(String(failed[1]?.type ?? 'serverFail'), failed[1]?.description);
throw new JmapMethodError('serverFail', `No response for call ${callId}`);
}
/** The server answered a call with a method-level error. */
export class JmapMethodError extends Error {
constructor(readonly type: string, description?: string) {
super(description ? `${type}: ${description}` : type);
this.name = 'JmapMethodError';
}
}
+11
View File
@@ -0,0 +1,11 @@
import type { StoreArea } from '../area';
import { MAIL_METHODS } from './methods';
import { MAIL_MIGRATIONS } from './schema';
import { mailSyncer } from './sync';
export const mailArea: StoreArea = {
name: 'mail',
migrations: MAIL_MIGRATIONS,
methods: MAIL_METHODS,
syncers: [mailSyncer],
};
+273
View File
@@ -0,0 +1,273 @@
/** Mail reads answered from the store. */
import { MethodError, type MethodContext, type MethodHandler, type MethodResult } from '../area';
import { listObjects } from '../engine/core-schema';
import type { SqlValue } from '../storage';
import { textCondition, type TextField } from './search';
type Filter = Record<string, unknown>;
interface Sql { sql: string; params: SqlValue[] }
const TEXT_FIELDS: readonly TextField[] = ['text', 'subject', 'from', 'to', 'body'];
function jmapOf(ctx: MethodContext, args: Record<string, unknown>): string {
return (args.accountId as string | undefined) ?? ctx.defaultJmap;
}
function filterSql(filter: Filter, jmap: string): Sql {
if (typeof filter.operator === 'string') {
const conditions = Array.isArray(filter.conditions) ? filter.conditions as Filter[] : [];
const parts = conditions.map((c) => filterSql(c, jmap));
const params = parts.flatMap((p) => p.params);
const joined = (joiner: string) => parts.map((p) => `(${p.sql})`).join(joiner);
if (filter.operator === 'AND') return { sql: parts.length ? joined(' AND ') : '1', params };
if (filter.operator === 'OR') return { sql: parts.length ? joined(' OR ') : '0', params };
if (filter.operator === 'NOT') return { sql: parts.length ? `NOT (${joined(' OR ')})` : '1', params };
throw new MethodError('unsupportedFilter', `Unknown operator ${String(filter.operator)}`);
}
const conditions: string[] = [];
const params: SqlValue[] = [];
for (const [key, value] of Object.entries(filter)) {
switch (key) {
case 'inMailbox':
conditions.push('EXISTS (SELECT 1 FROM email_mailbox m WHERE m.email_rowid = e.rowid AND m.mailbox_id = ?)');
params.push(String(value));
break;
case 'inMailboxOtherThan': {
const ids = (value as string[]).map(String);
conditions.push(`EXISTS (SELECT 1 FROM email_mailbox m WHERE m.email_rowid = e.rowid AND m.mailbox_id NOT IN (${ids.map(() => '?').join(',') || "''"}))`);
params.push(...ids);
break;
}
case 'hasKeyword':
conditions.push('EXISTS (SELECT 1 FROM email_keyword k WHERE k.email_rowid = e.rowid AND k.keyword = ?)');
params.push(String(value).toLowerCase());
break;
case 'notKeyword':
conditions.push('NOT EXISTS (SELECT 1 FROM email_keyword k WHERE k.email_rowid = e.rowid AND k.keyword = ?)');
params.push(String(value).toLowerCase());
break;
case 'after':
conditions.push('e.received_at >= ?');
params.push(Date.parse(String(value)) || 0);
break;
case 'before':
conditions.push('e.received_at < ?');
params.push(Date.parse(String(value)) || 0);
break;
case 'minSize':
conditions.push('e.size >= ?');
params.push(Number(value));
break;
case 'maxSize':
conditions.push('e.size < ?');
params.push(Number(value));
break;
case 'hasAttachment':
conditions.push(value ? 'e.has_attachment = 1' : 'e.has_attachment = 0');
break;
case 'id':
conditions.push(`e.id IN (${(value as string[]).map(() => '?').join(',') || "''"})`);
params.push(...(value as string[]).map(String));
break;
default:
if ((TEXT_FIELDS as readonly string[]).includes(key)) {
const text = textCondition(key as TextField, String(value));
conditions.push(text.sql);
params.push(...text.params);
break;
}
// cc, bcc, header and the thread keyword filters: the server's job.
throw new MethodError('unsupportedFilter', `${key} isn't available offline`);
}
}
return { sql: conditions.length ? conditions.join(' AND ') : '1', params };
}
interface Comparator { property?: string; isAscending?: boolean; keyword?: string }
function orderSql(ctx: MethodContext, sort: Comparator[]): Sql {
const levels = sort.length > 0 ? sort : [{ property: 'receivedAt', isAscending: false }];
const parts: string[] = [];
const params: SqlValue[] = [];
for (const level of levels) {
// RFC 8621 says isAscending defaults to true.
let ascending = level.isAscending !== false;
switch (level.property) {
case 'receivedAt': parts.push(`e.received_at ${ascending ? 'ASC' : 'DESC'}`); break;
case 'sentAt': parts.push(`COALESCE(e.sent_at, e.received_at) ${ascending ? 'ASC' : 'DESC'}`); break;
case 'size': parts.push(`e.size ${ascending ? 'ASC' : 'DESC'}`); break;
case 'subject': parts.push(`e.subject_sort COLLATE NOCASE ${ascending ? 'ASC' : 'DESC'}`); break;
case 'from': parts.push(`e.from_sort COLLATE NOCASE ${ascending ? 'ASC' : 'DESC'}`); break;
case 'to': parts.push(`e.to_sort COLLATE NOCASE ${ascending ? 'ASC' : 'DESC'}`); break;
case 'hasKeyword': {
// The client flips this on servers that read it the other way, so a
// stored answer has to read it the way that server would.
if (ctx.invertedKeywordSort) ascending = !ascending;
parts.push(`EXISTS (SELECT 1 FROM email_keyword s WHERE s.email_rowid = e.rowid AND s.keyword = ?) ${ascending ? 'ASC' : 'DESC'}`);
params.push(String(level.keyword ?? '').toLowerCase());
break;
}
default:
throw new MethodError('unsupportedSort', `${String(level.property)} isn't available offline`);
}
}
parts.push('e.rowid DESC');
return { sql: parts.join(', '), params };
}
/** The one folder a filter limits itself to, so the query can start from that folder's index. */
function drivingMailbox(filter: Filter | undefined): string | null {
if (!filter) return null;
if (typeof filter.inMailbox === 'string') return filter.inMailbox;
if (filter.operator === 'AND' && Array.isArray(filter.conditions)) {
for (const condition of filter.conditions as Filter[]) {
const found = drivingMailbox(condition);
if (found) return found;
}
}
return null;
}
const emailQuery: MethodHandler = (ctx, args) => {
const jmap = jmapOf(ctx, args);
const filter = (args.filter ?? undefined) as Filter | undefined;
if (args.anchor !== undefined) throw new MethodError('unsupportedFilter', 'anchor needs the server');
const where = filter ? filterSql(filter, jmap) : { sql: '1', params: [] };
const order = orderSql(ctx, (args.sort as Comparator[]) ?? []);
const box = drivingMailbox(filter);
// Starting from the folder's own index keeps small folders fast in a large account.
const from = box
? 'email_mailbox d INDEXED BY email_mailbox_list CROSS JOIN email e ON e.rowid = d.email_rowid'
: 'email e';
const base = box
? { sql: 'd.jmap = ? AND d.mailbox_id = ?', params: [jmap, box] as SqlValue[] }
: { sql: 'e.jmap = ?', params: [jmap] as SqlValue[] };
const position = Math.max(0, Number(args.position ?? 0));
const limit = typeof args.limit === 'number' && args.limit > 0 ? args.limit : -1;
const collapse = Boolean(args.collapseThreads);
const group = collapse ? 'GROUP BY COALESCE(e.thread_id, e.id)' : '';
const ids = ctx.db.all<{ id: string }>(
`SELECT e.id FROM ${from} WHERE ${base.sql} AND (${where.sql}) ${group} ORDER BY ${order.sql} LIMIT ? OFFSET ?`,
[...base.params, ...where.params, ...order.params, limit, position],
).map((row) => String(row.id));
const result: MethodResult = {
accountId: jmap,
canCalculateChanges: false,
// Local answers aren't a position in the server's log, so this is a token only.
queryState: 'local',
position,
ids,
};
if (args.calculateTotal) {
const count = collapse ? 'COUNT(DISTINCT COALESCE(e.thread_id, e.id))' : 'COUNT(*)';
result.total = Number(ctx.db.get<{ n: number }>(
`SELECT ${count} AS n FROM ${from} WHERE ${base.sql} AND (${where.sql})`,
[...base.params, ...where.params],
)?.n ?? 0);
}
return result;
};
const BODY_PROPERTIES = new Set(['bodyValues', 'textBody', 'htmlBody', 'bodyStructure', 'headers', 'attachments', 'bcc', 'replyTo', 'messageId', 'inReplyTo', 'references']);
function pick(email: Record<string, unknown>, properties: string[] | undefined): Record<string, unknown> {
if (!properties) return email;
const out: Record<string, unknown> = { id: email.id };
for (const property of properties) if (property in email) out[property] = email[property];
return out;
}
const emailGet: MethodHandler = (ctx, args) => {
const jmap = jmapOf(ctx, args);
if (!Array.isArray(args.ids)) throw new MethodError('requestTooLarge', 'ids: null needs the server');
const properties = args.properties as string[] | undefined;
const wantsValues = Boolean(args.fetchTextBodyValues || args.fetchHTMLBodyValues || args.fetchAllBodyValues) || Boolean(properties?.includes('bodyValues'));
const wantsBody = wantsValues || !properties || properties.some((p) => BODY_PROPERTIES.has(p));
const list: Record<string, unknown>[] = [];
const notFound: string[] = [];
for (const id of args.ids as string[]) {
const row = ctx.db.get<{ rowid: number; json: string; body_state: number }>(
'SELECT rowid, json, body_state FROM email WHERE jmap = ? AND id = ?',
[jmap, id],
);
if (!row) {
notFound.push(id);
continue;
}
const email = JSON.parse(row.json) as Record<string, unknown>;
if (!wantsBody) {
list.push(pick(email, properties));
continue;
}
const body = ctx.db.get<{ json: string }>('SELECT json FROM email_body WHERE email_rowid = ?', [row.rowid]);
// Asked for the text and it isn't here: as if the message weren't there, not an empty message.
if (!body && wantsValues) {
notFound.push(id);
continue;
}
list.push(pick({ ...(body ? JSON.parse(body.json) as object : {}), ...email }, properties));
}
const state = ctx.db.get<{ state: string | null }>("SELECT state FROM sync_state WHERE jmap = ? AND type = 'Email'", [jmap])?.state;
return { accountId: jmap, ...(state ? { state } : {}), list, notFound };
};
const threadGet: MethodHandler = (ctx, args) => {
const jmap = jmapOf(ctx, args);
const ids = Array.isArray(args.ids) ? [...new Set(args.ids as string[])] : [];
const list: Array<{ id: string; emailIds: string[] }> = [];
const notFound: string[] = [];
for (const id of ids) {
const stored = ctx.db.get<{ email_ids: string }>('SELECT email_ids FROM thread WHERE jmap = ? AND id = ?', [jmap, id]);
if (stored) {
list.push({ id, emailIds: JSON.parse(stored.email_ids) as string[] });
continue;
}
const rows = ctx.db.all<{ id: string }>('SELECT id FROM email WHERE jmap = ? AND thread_id = ? ORDER BY received_at ASC', [jmap, id]);
if (rows.length > 0) list.push({ id, emailIds: rows.map((row) => String(row.id)) });
else notFound.push(id);
}
return { accountId: jmap, list, notFound };
};
const mailboxGet: MethodHandler = (ctx, args) => {
const jmap = jmapOf(ctx, args);
const boxes = listObjects<Record<string, unknown>>(ctx.db, jmap, 'Mailbox');
if (boxes.length === 0) throw new MethodError('serverUnavailable', 'Folders not kept yet');
const ids = Array.isArray(args.ids) ? new Set(args.ids as string[]) : null;
const chosen = ids ? boxes.filter((box) => ids.has(box.id as string)) : boxes;
const state = ctx.db.get<{ state: string | null }>("SELECT state FROM sync_state WHERE jmap = ? AND type = 'Mailbox'", [jmap])?.state;
return {
accountId: jmap,
...(state ? { state } : {}),
list: chosen.map((box) => pick(box, args.properties as string[] | undefined)),
notFound: ids ? [...ids].filter((id) => !boxes.some((box) => box.id === id)) : [],
};
};
const mailboxQuery: MethodHandler = (ctx, args) => {
const jmap = jmapOf(ctx, args);
const boxes = listObjects<Record<string, unknown>>(ctx.db, jmap, 'Mailbox');
if (boxes.length === 0) throw new MethodError('serverUnavailable', 'Folders not kept yet');
const filter = (args.filter ?? {}) as { role?: string };
const chosen = filter.role ? boxes.filter((box) => box.role === filter.role) : boxes;
const position = typeof args.position === 'number' ? args.position : 0;
const limit = typeof args.limit === 'number' ? args.limit : chosen.length;
return {
accountId: jmap,
queryState: 'local',
canCalculateChanges: false,
position,
ids: chosen.slice(position, position + limit).map((box) => box.id),
...(args.calculateTotal ? { total: chosen.length } : {}),
};
};
export const MAIL_METHODS: Readonly<Record<string, MethodHandler>> = {
'Email/query': emailQuery,
'Email/get': emailGet,
'Thread/get': threadGet,
'Mailbox/get': mailboxGet,
'Mailbox/query': mailboxQuery,
};
+147
View File
@@ -0,0 +1,147 @@
/** How an Email object becomes rows, and back. */
import type { StorageAdapter } from '../storage';
export const MAIL_USING = ['urn:ietf:params:jmap:core', 'urn:ietf:params:jmap:mail'];
export const EMAIL_LIST_PROPERTIES = [
'id', 'threadId', 'mailboxIds', 'keywords', 'size', 'receivedAt', 'sentAt', 'from', 'to', 'cc',
'subject', 'preview', 'hasAttachment', 'blobId', 'header:Authentication-Results:all',
];
/** What opening a message asks for, past the list properties. */
export const EMAIL_BODY_PROPERTIES = [
'bcc', 'replyTo', 'textBody', 'htmlBody', 'bodyValues', 'attachments', 'messageId', 'inReplyTo',
'references', 'headers', 'bodyStructure',
];
export type EmailObject = Record<string, unknown> & { id: string };
interface Address { name?: string | null; email?: string | null }
function firstAddress(value: unknown): string | null {
const first = Array.isArray(value) ? value[0] as Address | undefined : undefined;
const text = first?.name || first?.email;
return text ? text.toLowerCase() : null;
}
function addressText(value: unknown): string {
if (!Array.isArray(value)) return '';
return (value as Address[]).map((a) => [a.name, a.email].filter(Boolean).join(' ')).join(' ');
}
function time(value: unknown): number | null {
if (typeof value !== 'string') return null;
const ms = Date.parse(value);
return Number.isFinite(ms) ? ms : null;
}
/** The subject without Re:, Fwd: and the like, as the client sorts it. */
export function baseSubject(subject: string): string {
return subject.replace(/^(\s*(re|aw|fw|fwd|wg)\s*(\[\d+\])?\s*:\s*)+/i, '').trim().toLowerCase();
}
/** Inserts or updates one message from its list properties. Keeps a body already stored. */
export function upsertEmail(db: StorageAdapter, jmap: string, email: EmailObject, gen: number): number {
const receivedAt = time(email.receivedAt) ?? 0;
const subject = typeof email.subject === 'string' ? email.subject : '';
const existing = db.get<{ rowid: number }>('SELECT rowid FROM email WHERE jmap = ? AND id = ?', [jmap, email.id]);
const values = [
typeof email.threadId === 'string' ? email.threadId : null,
receivedAt,
time(email.sentAt),
typeof email.size === 'number' ? email.size : 0,
email.hasAttachment ? 1 : 0,
baseSubject(subject),
firstAddress(email.from),
firstAddress(email.to),
JSON.stringify(email),
gen,
];
let rowid: number;
if (existing) {
rowid = Number(existing.rowid);
db.run(
`UPDATE email SET thread_id = ?, received_at = ?, sent_at = ?, size = ?, has_attachment = ?, subject_sort = ?,
from_sort = ?, to_sort = ?, json = ?, gen = ? WHERE rowid = ?`,
[...values, rowid],
);
db.run('DELETE FROM email_mailbox WHERE email_rowid = ?', [rowid]);
db.run('DELETE FROM email_keyword WHERE email_rowid = ?', [rowid]);
} else {
rowid = db.run(
`INSERT INTO email (thread_id, received_at, sent_at, size, has_attachment, subject_sort, from_sort, to_sort, json, gen, jmap, id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
[...values, jmap, email.id],
).lastInsertRowid;
indexEmail(db, rowid, email, '');
}
for (const [box, on] of Object.entries((email.mailboxIds ?? {}) as Record<string, boolean>)) {
if (on) db.run('INSERT INTO email_mailbox (email_rowid, jmap, mailbox_id, received_at) VALUES (?, ?, ?, ?)', [rowid, jmap, box, receivedAt]);
}
for (const [keyword, on] of Object.entries((email.keywords ?? {}) as Record<string, boolean>)) {
if (on) db.run('INSERT INTO email_keyword (email_rowid, keyword) VALUES (?, ?)', [rowid, keyword.toLowerCase()]);
}
return rowid;
}
export function deleteEmail(db: StorageAdapter, jmap: string, id: string): void {
const row = db.get<{ rowid: number }>('SELECT rowid FROM email WHERE jmap = ? AND id = ?', [jmap, id]);
if (!row) return;
deleteEmailRow(db, Number(row.rowid));
}
export function deleteEmailRow(db: StorageAdapter, rowid: number): void {
db.run('DELETE FROM email_fts_word WHERE rowid = ?', [rowid]);
db.run('DELETE FROM email_fts_tri WHERE rowid = ?', [rowid]);
// Explicit rather than ON DELETE CASCADE: foreign keys stay off for bulk speed.
db.run('DELETE FROM email_mailbox WHERE email_rowid = ?', [rowid]);
db.run('DELETE FROM email_keyword WHERE email_rowid = ?', [rowid]);
db.run('DELETE FROM email_body WHERE email_rowid = ?', [rowid]);
db.run('DELETE FROM email WHERE rowid = ?', [rowid]);
}
/** Plain text of a message for the search index: the text part, or the HTML part without its tags. */
export function bodyText(email: Record<string, unknown>): string {
const values = (email.bodyValues ?? {}) as Record<string, { value?: string }>;
const parts = (key: string) => ((email[key] ?? []) as Array<{ partId?: string; type?: string }>);
const text = parts('textBody')
.filter((part) => part.type !== 'text/html')
.map((part) => values[part.partId ?? '']?.value ?? '')
.join('\n');
if (text.trim()) return text;
return parts('htmlBody')
.map((part) => values[part.partId ?? '']?.value ?? '')
.join('\n')
.replace(/<(style|script)[\s\S]*?<\/\1>/gi, ' ')
.replace(/<[^>]+>/g, ' ')
.replace(/&nbsp;/g, ' ')
.replace(/&amp;/g, '&')
.replace(/&lt;/g, '<')
.replace(/&gt;/g, '>')
.replace(/\s+/g, ' ');
}
/** A long message past this is rare and only bloats the index. */
const INDEX_BODY_CHARS = 200_000;
export function indexEmail(db: StorageAdapter, rowid: number, email: Record<string, unknown>, body: string): void {
const subject = typeof email.subject === 'string' ? email.subject : '';
const from = addressText(email.from);
const to = [addressText(email.to), addressText(email.cc), addressText(email.bcc)].filter(Boolean).join(' ');
const text = body.slice(0, INDEX_BODY_CHARS);
for (const table of ['email_fts_word', 'email_fts_tri']) {
db.run(`INSERT INTO ${table} (rowid, subject, from_addr, to_addr, body) VALUES (?, ?, ?, ?, ?)`, [rowid, subject, from, to, text]);
}
}
export function storeBody(db: StorageAdapter, rowid: number, listJson: string, full: Record<string, unknown>): void {
const list = JSON.parse(listJson) as Record<string, unknown>;
const body: Record<string, unknown> = {};
for (const [key, value] of Object.entries(full)) if (!(key in list)) body[key] = value;
db.run('INSERT INTO email_body (email_rowid, json) VALUES (?, ?) ON CONFLICT (email_rowid) DO UPDATE SET json = excluded.json', [rowid, JSON.stringify(body)]);
db.run('UPDATE email SET body_state = 1 WHERE rowid = ?', [rowid]);
db.run('DELETE FROM email_fts_word WHERE rowid = ?', [rowid]);
db.run('DELETE FROM email_fts_tri WHERE rowid = ?', [rowid]);
indexEmail(db, rowid, { ...list, ...body }, bodyText(full));
}
+65
View File
@@ -0,0 +1,65 @@
import type { Migration } from '../storage';
export const MAIL_MIGRATIONS: readonly Migration[] = [
{
id: 100,
name: 'mail',
sql: `
CREATE TABLE email (
rowid INTEGER PRIMARY KEY,
jmap TEXT NOT NULL,
id TEXT NOT NULL,
thread_id TEXT,
received_at INTEGER NOT NULL,
sent_at INTEGER,
size INTEGER NOT NULL DEFAULT 0,
has_attachment INTEGER NOT NULL DEFAULT 0,
subject_sort TEXT,
from_sort TEXT,
to_sort TEXT,
json TEXT NOT NULL,
-- 0 no body yet, 1 kept, 2 the server had none to give
body_state INTEGER NOT NULL DEFAULT 0,
gen INTEGER NOT NULL DEFAULT 0,
UNIQUE (jmap, id)
);
CREATE INDEX email_received ON email (jmap, received_at DESC);
CREATE INDEX email_thread ON email (jmap, thread_id);
CREATE INDEX email_backfill ON email (jmap, body_state, received_at DESC);
CREATE TABLE email_mailbox (
email_rowid INTEGER NOT NULL REFERENCES email (rowid) ON DELETE CASCADE,
jmap TEXT NOT NULL,
mailbox_id TEXT NOT NULL,
received_at INTEGER NOT NULL,
PRIMARY KEY (email_rowid, mailbox_id)
) WITHOUT ROWID;
CREATE INDEX email_mailbox_list ON email_mailbox (jmap, mailbox_id, received_at DESC);
CREATE TABLE email_keyword (
email_rowid INTEGER NOT NULL REFERENCES email (rowid) ON DELETE CASCADE,
keyword TEXT NOT NULL,
PRIMARY KEY (email_rowid, keyword)
) WITHOUT ROWID;
CREATE INDEX email_keyword_lookup ON email_keyword (keyword, email_rowid);
CREATE TABLE email_body (
email_rowid INTEGER PRIMARY KEY REFERENCES email (rowid) ON DELETE CASCADE,
json TEXT NOT NULL
);
CREATE TABLE thread (
jmap TEXT NOT NULL,
id TEXT NOT NULL,
email_ids TEXT NOT NULL,
PRIMARY KEY (jmap, id)
) WITHOUT ROWID;
CREATE VIRTUAL TABLE email_fts_word USING fts5(
subject, from_addr, to_addr, body,
content = '', contentless_delete = 1,
tokenize = 'unicode61 remove_diacritics 2'
);
CREATE VIRTUAL TABLE email_fts_tri USING fts5(
subject, from_addr, to_addr, body,
content = '', contentless_delete = 1,
tokenize = 'trigram remove_diacritics 1'
);
`,
},
];
+59
View File
@@ -0,0 +1,59 @@
/** Full-text matching over the two FTS indexes. */
import type { SqlValue } from '../storage';
export type TextField = 'text' | 'subject' | 'from' | 'to' | 'body';
const COLUMNS: Record<TextField, string | null> = {
text: null,
subject: 'subject',
from: 'from_addr',
to: 'to_addr',
body: 'body',
};
function quote(term: string): string {
return `"${term.replace(/"/g, '""')}"`;
}
/** The terms of a search box value: quoted phrases stay whole, `word*` keeps its star. */
export function splitTerms(input: string): string[] {
const out: string[] = [];
for (const match of input.matchAll(/"([^"]+)"|(\S+)/g)) out.push(match[1] ?? match[2]);
return out;
}
/**
* One term as SQL over rowids. The word index finds words and word starts,
* the trigram index finds a term inside a longer word, which is how German
* compounds are found ("Rechnung" in "Zahlungserinnerungsrechnung").
*/
function termRowids(field: TextField, term: string): { sql: string; params: SqlValue[] } {
const column = COLUMNS[field];
const star = term.endsWith('*');
const clean = star ? term.slice(0, -1) : term;
const scope = column ? `${column} : ` : '';
const word = `${scope}${quote(clean)}${' *'.slice(1)}`;
const parts = ['SELECT rowid FROM email_fts_word WHERE email_fts_word MATCH ?'];
const params: SqlValue[] = [word];
// Trigrams need three characters. Shorter terms use the word index alone.
if ([...clean].length >= 3) {
parts.push('SELECT rowid FROM email_fts_tri WHERE email_fts_tri MATCH ?');
params.push(`${scope}${quote(clean)}`);
}
return { sql: parts.join(' UNION '), params };
}
/** SQL for a text condition, as a condition on `e.rowid`. Every term has to match. */
export function textCondition(field: TextField, input: string): { sql: string; params: SqlValue[] } {
const terms = splitTerms(input);
if (terms.length === 0) return { sql: '1', params: [] };
const conditions: string[] = [];
const params: SqlValue[] = [];
for (const term of terms) {
const part = termRowids(field, term);
conditions.push(`e.rowid IN (${part.sql})`);
params.push(...part.params);
}
return { sql: conditions.join(' AND '), params };
}
+323
View File
@@ -0,0 +1,323 @@
/**
* Mail sync for one JMAP account.
*
* A full sync captures the state first, then pages through the window, so a
* change that lands during it is picked up by the delta afterwards. Its cursor
* lives in sync_state, written in the same transaction as the page, so a
* reload continues where it stopped. Bodies come after the list, newest first.
*/
import type { SyncContext, Syncer } from '../area';
import { deleteObject, putObject } from '../engine/core-schema';
import { runChangesLoop } from '../engine/changes';
import { responseFor, type MethodCall } from '../engine/transport';
import {
deleteEmail,
deleteEmailRow,
EMAIL_BODY_PROPERTIES,
EMAIL_LIST_PROPERTIES,
MAIL_USING,
storeBody,
upsertEmail,
type EmailObject,
} from './records';
const PAGE = 500;
const PAIRS_PER_REQUEST = 4;
const BODY_BATCH_IDS = 100;
const BODY_BATCH_BYTES = 8 * 1024 * 1024;
const MAX_BODY_VALUE_BYTES = 1_000_000;
const DELTA_BATCH = 500;
function windowFilter(ctx: SyncContext): Record<string, unknown> | undefined {
if (ctx.scope.days === null) return undefined;
return { after: new Date(ctx.now() - ctx.scope.days * 86_400_000).toISOString() };
}
function inWindow(ctx: SyncContext, email: EmailObject): boolean {
if (ctx.scope.days === null) return true;
return (Date.parse(String(email.receivedAt)) || 0) >= ctx.now() - ctx.scope.days * 86_400_000;
}
// ---- Mailboxes ----
async function fullMailboxes(ctx: SyncContext, jmap: string): Promise<void> {
const response = await ctx.transport.request([['Mailbox/get', { accountId: jmap, ids: null }, 'm']], MAIL_USING);
const [, result] = responseFor(response, 'm');
ctx.db.transaction(() => {
ctx.db.run("DELETE FROM object WHERE jmap = ? AND type = 'Mailbox'", [jmap]);
for (const box of result.list as Array<{ id: string }>) putObject(ctx.db, jmap, 'Mailbox', box.id, box);
ctx.setSyncState(jmap, 'Mailbox', { state: String(result.state), phase: null, cursor: null });
});
}
async function syncMailboxes(ctx: SyncContext, jmap: string): Promise<void> {
const { state, target } = ctx.syncState(jmap, 'Mailbox');
if (!state) return fullMailboxes(ctx, jmap);
const outcome = await runChangesLoop({
transport: ctx.transport,
type: 'Mailbox',
jmap,
using: MAIL_USING,
since: state,
target,
stopped: ctx.stopped,
apply: async (page) => {
const ids = [...page.created, ...page.updated];
const list: Array<{ id: string }> = [];
for (let i = 0; i < ids.length; i += PAGE) {
const response = await ctx.transport.request([['Mailbox/get', { accountId: jmap, ids: ids.slice(i, i + PAGE) }, 'm']], MAIL_USING);
list.push(...responseFor(response, 'm')[1].list);
}
ctx.db.transaction(() => {
for (const box of list) putObject(ctx.db, jmap, 'Mailbox', box.id, box);
for (const id of page.destroyed) deleteObject(ctx.db, jmap, 'Mailbox', id);
ctx.setSyncState(jmap, 'Mailbox', { state: page.newState });
});
},
});
if (outcome.outcome === 'resync') await fullMailboxes(ctx, jmap);
}
// ---- Email list ----
interface ListCursor { position: number; gen: number; captured: string }
async function startFullList(ctx: SyncContext, jmap: string): Promise<ListCursor> {
const response = await ctx.transport.request([['Email/get', { accountId: jmap, ids: [], properties: ['id'] }, 's']], MAIL_USING);
const captured = String(responseFor(response, 's')[1].state);
const gen = (ctx.db.get<{ gen: number | null }>('SELECT MAX(gen) AS gen FROM email WHERE jmap = ?', [jmap])?.gen ?? 0) + 1;
const cursor: ListCursor = { position: 0, gen: Number(gen), captured };
ctx.db.transaction(() => ctx.setSyncState(jmap, 'Email', { state: null, phase: 'list', cursor: { ...cursor } }));
return cursor;
}
async function runFullList(ctx: SyncContext, jmap: string, start: ListCursor): Promise<boolean> {
const cursor = { ...start };
const filter = windowFilter(ctx);
let total: number | null = null;
for (;;) {
if (ctx.stopped()) return false;
const calls: MethodCall[] = [];
for (let i = 0; i < PAIRS_PER_REQUEST; i += 1) {
calls.push(['Email/query', {
accountId: jmap,
...(filter ? { filter } : {}),
sort: [{ property: 'receivedAt', isAscending: false }],
position: cursor.position + i * PAGE,
limit: PAGE,
calculateTotal: i === 0,
}, `q${i}`]);
calls.push(['Email/get', {
accountId: jmap,
'#ids': { resultOf: `q${i}`, name: 'Email/query', path: '/ids' },
properties: EMAIL_LIST_PROPERTIES,
}, `g${i}`]);
}
const response = await ctx.transport.request(calls, MAIL_USING);
let done = false;
let got = 0;
const lists: EmailObject[][] = [];
for (let i = 0; i < PAIRS_PER_REQUEST; i += 1) {
const [, query] = responseFor(response, `q${i}`);
if (i === 0 && typeof query.total === 'number') total = query.total;
lists.push(responseFor(response, `g${i}`)[1].list as EmailObject[]);
got += (query.ids as string[]).length;
if ((query.ids as string[]).length < PAGE) {
done = true;
break;
}
}
cursor.position += got;
ctx.db.transaction(() => {
for (const list of lists) for (const email of list) upsertEmail(ctx.db, jmap, email, cursor.gen);
if (done) {
// Whatever this pass didn't see is gone from the server or the window.
const stale = ctx.db.all<{ rowid: number }>('SELECT rowid FROM email WHERE jmap = ? AND gen < ?', [jmap, cursor.gen]);
for (const row of stale) deleteEmailRow(ctx.db, Number(row.rowid));
ctx.setSyncState(jmap, 'Email', { state: cursor.captured, phase: null, cursor: null });
} else {
ctx.setSyncState(jmap, 'Email', { cursor: { ...cursor } });
}
});
ctx.emit({ type: 'progress', account: ctx.account, area: 'mail', phase: 'list', done: cursor.position, total });
if (done) {
ctx.emit({ type: 'changed', account: ctx.account, collections: ['Email'] });
return true;
}
}
}
async function syncEmailDelta(ctx: SyncContext, jmap: string, since: string, target: string | null): Promise<'done' | 'resync' | 'stopped'> {
const outcome = await runChangesLoop({
transport: ctx.transport,
type: 'Email',
jmap,
using: MAIL_USING,
since,
target,
stopped: ctx.stopped,
apply: async (page) => {
const ids = [...page.created, ...page.updated];
const fetched: EmailObject[] = [];
for (let i = 0; i < ids.length; i += DELTA_BATCH) {
const response = await ctx.transport.request([['Email/get', {
accountId: jmap,
ids: ids.slice(i, i + DELTA_BATCH),
properties: EMAIL_LIST_PROPERTIES,
}, 'g']], MAIL_USING);
fetched.push(...responseFor(response, 'g')[1].list as EmailObject[]);
}
const created = new Set(page.created);
const newIds: string[] = [];
ctx.db.transaction(() => {
for (const email of fetched) {
const known = ctx.db.get('SELECT 1 AS x FROM email WHERE jmap = ? AND id = ?', [jmap, email.id]);
// An old message changing doesn't pull it into the window.
if (!known && !inWindow(ctx, email)) continue;
const gen = Number(ctx.db.get<{ gen: number | null }>('SELECT MAX(gen) AS gen FROM email WHERE jmap = ?', [jmap])?.gen ?? 0);
upsertEmail(ctx.db, jmap, email, gen);
if (!known && created.has(email.id)) newIds.push(email.id);
}
for (const id of page.destroyed) deleteEmail(ctx.db, jmap, id);
ctx.setSyncState(jmap, 'Email', { state: page.newState });
});
if (newIds.length > 0) ctx.emit({ type: 'created', account: ctx.account, jmap, collection: 'Email', ids: newIds });
if (ids.length + page.destroyed.length > 0) ctx.emit({ type: 'changed', account: ctx.account, collections: ['Email'] });
},
});
if (outcome.outcome === 'resync') return 'resync';
return outcome.outcome === 'stopped' ? 'stopped' : 'done';
}
// ---- Threads ----
async function fetchThreads(ctx: SyncContext, jmap: string, ids: string[]): Promise<Array<{ id: string; emailIds: string[] }>> {
const out: Array<{ id: string; emailIds: string[] }> = [];
for (let i = 0; i < ids.length; i += PAGE) {
const response = await ctx.transport.request([['Thread/get', { accountId: jmap, ids: ids.slice(i, i + PAGE) }, 't']], MAIL_USING);
out.push(...responseFor(response, 't')[1].list);
}
return out;
}
async function syncThreads(ctx: SyncContext, jmap: string): Promise<void> {
const { state, target } = ctx.syncState(jmap, 'Thread');
const fullThreads = async () => {
const response = await ctx.transport.request([['Thread/get', { accountId: jmap, ids: [] }, 's']], MAIL_USING);
const captured = String(responseFor(response, 's')[1].state);
const known = ctx.db.all<{ id: string }>(
`SELECT DISTINCT e.thread_id AS id FROM email e WHERE e.jmap = ? AND e.thread_id IS NOT NULL
AND NOT EXISTS (SELECT 1 FROM thread t WHERE t.jmap = e.jmap AND t.id = e.thread_id)`,
[jmap],
).map((row) => String(row.id));
for (let i = 0; i < known.length; i += PAGE) {
if (ctx.stopped()) return;
const list = await fetchThreads(ctx, jmap, known.slice(i, i + PAGE));
ctx.db.transaction(() => {
for (const thread of list) ctx.db.run('INSERT OR REPLACE INTO thread (jmap, id, email_ids) VALUES (?, ?, ?)', [jmap, thread.id, JSON.stringify(thread.emailIds)]);
});
ctx.emit({ type: 'progress', account: ctx.account, area: 'mail', phase: 'threads', done: i + list.length, total: known.length });
}
ctx.db.transaction(() => ctx.setSyncState(jmap, 'Thread', { state: captured }));
};
if (!state) return fullThreads();
const outcome = await runChangesLoop({
transport: ctx.transport,
type: 'Thread',
jmap,
using: MAIL_USING,
since: state,
target,
stopped: ctx.stopped,
apply: async (page) => {
const list = await fetchThreads(ctx, jmap, [...page.created, ...page.updated]);
ctx.db.transaction(() => {
for (const thread of list) ctx.db.run('INSERT OR REPLACE INTO thread (jmap, id, email_ids) VALUES (?, ?, ?)', [jmap, thread.id, JSON.stringify(thread.emailIds)]);
for (const id of page.destroyed) ctx.db.run('DELETE FROM thread WHERE jmap = ? AND id = ?', [jmap, id]);
ctx.setSyncState(jmap, 'Thread', { state: page.newState });
});
},
});
if (outcome.outcome === 'resync') {
ctx.db.transaction(() => {
ctx.db.run('DELETE FROM thread WHERE jmap = ?', [jmap]);
ctx.setSyncState(jmap, 'Thread', { state: null });
});
await fullThreads();
}
}
// ---- Bodies ----
async function backfillBodies(ctx: SyncContext, jmap: string): Promise<void> {
let done = 0;
const totalRow = ctx.db.get<{ n: number }>('SELECT COUNT(*) AS n FROM email WHERE jmap = ? AND body_state = 0', [jmap]);
const total = Number(totalRow?.n ?? 0);
for (;;) {
if (ctx.stopped()) return;
const candidates = ctx.db.all<{ rowid: number; id: string; size: number; json: string }>(
'SELECT rowid, id, size, json FROM email WHERE jmap = ? AND body_state = 0 ORDER BY received_at DESC LIMIT ?',
[jmap, BODY_BATCH_IDS],
);
if (candidates.length === 0) return;
let bytes = 0;
const batch: typeof candidates = [];
for (const row of candidates) {
if (batch.length > 0 && bytes + Number(row.size) > BODY_BATCH_BYTES) break;
bytes += Number(row.size);
batch.push(row);
}
const response = await ctx.transport.request([['Email/get', {
accountId: jmap,
ids: batch.map((row) => String(row.id)),
properties: ['id', ...EMAIL_BODY_PROPERTIES],
fetchTextBodyValues: true,
fetchHTMLBodyValues: true,
fetchAllBodyValues: true,
maxBodyValueBytes: MAX_BODY_VALUE_BYTES,
}, 'b']], MAIL_USING);
const list = responseFor(response, 'b')[1].list as EmailObject[];
const byId = new Map(list.map((email) => [email.id, email]));
ctx.db.transaction(() => {
for (const row of batch) {
const full = byId.get(String(row.id));
if (full) storeBody(ctx.db, Number(row.rowid), String(row.json), full);
else ctx.db.run('UPDATE email SET body_state = 2 WHERE rowid = ?', [row.rowid]);
}
});
done += batch.length;
ctx.emit({ type: 'progress', account: ctx.account, area: 'mail', phase: 'bodies', done, total });
}
}
// ---- The whole pass ----
export const mailSyncer: Syncer = {
name: 'mail',
capability: 'urn:ietf:params:jmap:mail',
async run(ctx, jmap) {
await syncMailboxes(ctx, jmap);
if (ctx.stopped()) return;
const email = ctx.syncState(jmap, 'Email');
let cursor = email.phase === 'list' ? email.cursor as unknown as ListCursor : null;
if (!cursor && !email.state) cursor = await startFullList(ctx, jmap);
if (!cursor && email.state) {
const outcome = await syncEmailDelta(ctx, jmap, email.state, email.target);
if (outcome === 'stopped') return;
if (outcome === 'resync') cursor = await startFullList(ctx, jmap);
}
if (cursor) {
const finished = await runFullList(ctx, jmap, cursor);
if (!finished) return;
// Catch what changed while paging.
const after = ctx.syncState(jmap, 'Email');
if (after.state && (await syncEmailDelta(ctx, jmap, after.state, after.target)) === 'stopped') return;
}
if (ctx.stopped()) return;
await syncThreads(ctx, jmap);
if (ctx.stopped()) return;
await backfillBodies(ctx, jmap);
},
};
+3
View File
@@ -0,0 +1,3 @@
/** The vendored SQLite3MultipleCiphers build. Change it with scripts/fetch-sqlite3mc.mjs. */
export const SQLITE3MC_VERSION = '2.6.1-3.54.0';
export const SQLITE3MC_PUBLIC_DIR = `sqlite3mc/${SQLITE3MC_VERSION}`;
+73
View File
@@ -0,0 +1,73 @@
/**
* The small storage interface the local store is written against.
*
* Synchronous on purpose: SQLite3MC in a worker and node:sqlite in the tests
* both are, and a transaction then can't hold an await by accident. A native
* or mobile adapter runs the engine next to its database, not across a
* message boundary.
*/
export type SqlValue = string | number | bigint | null | Uint8Array;
export type SqlRow = Record<string, SqlValue>;
export interface StorageAdapter {
/** Runs one or more statements without parameters. */
exec(sql: string): void;
/** Runs one statement. */
run(sql: string, params?: readonly SqlValue[]): { changes: number; lastInsertRowid: number };
all<T extends object = SqlRow>(sql: string, params?: readonly SqlValue[]): T[];
get<T extends object = SqlRow>(sql: string, params?: readonly SqlValue[]): T | undefined;
/** Commits when fn returns, rolls back when it throws. Nested calls join the outer one. */
transaction<T>(fn: () => T): T;
close(): void;
}
/** Shared by the adapters: BEGIN/COMMIT around fn, and a guard against async bodies. */
export function runTransaction<T>(
state: { depth: number },
exec: (sql: string) => void,
fn: () => T,
): T {
if (state.depth > 0) return fn();
exec('BEGIN IMMEDIATE');
state.depth += 1;
try {
const result = fn();
if (result && typeof (result as { then?: unknown }).then === 'function') {
throw new Error('A store transaction must be synchronous');
}
exec('COMMIT');
return result;
} catch (error) {
exec('ROLLBACK');
throw error;
} finally {
state.depth -= 1;
}
}
export interface Migration {
/** Unique, ascending. Ranges per area are listed in lib/store/area.ts. */
id: number;
name: string;
sql: string;
}
/** Applies what's missing, in id order, each in its own transaction. */
export function migrate(db: StorageAdapter, migrations: readonly Migration[]): number[] {
db.exec('CREATE TABLE IF NOT EXISTS schema_migration (id INTEGER PRIMARY KEY, name TEXT NOT NULL, at INTEGER NOT NULL)');
const done = new Set(db.all<{ id: number }>('SELECT id FROM schema_migration').map((row) => Number(row.id)));
const ids = new Set<number>();
const applied: number[] = [];
for (const migration of [...migrations].sort((a, b) => a.id - b.id)) {
if (ids.has(migration.id)) throw new Error(`Duplicate store migration id ${migration.id}`);
ids.add(migration.id);
if (done.has(migration.id)) continue;
db.transaction(() => {
db.exec(migration.sql);
db.run('INSERT INTO schema_migration (id, name, at) VALUES (?, ?, ?)', [migration.id, migration.name, Date.now()]);
});
applied.push(migration.id);
}
return applied;
}
+168
View File
@@ -0,0 +1,168 @@
/**
* What runs in the owner tab's worker: one database and engine per account.
* The list of areas is the only thing that decides what the store answers.
*/
import type { StoreArea, StoreEvent } from '../area';
import { AccountEngine } from '../engine/account-engine';
import { FetchTransport, type StoreSession } from '../engine/transport';
import type { StorageAdapter } from '../storage';
import type { OpenAccountParams, StoreRpc, StoreStatus } from './protocol';
export interface HostStorage {
open(file: string, keyHex: string): Promise<StorageAdapter>;
remove(file: string): boolean;
}
interface OpenAccount {
engine: AccountEngine;
db: StorageAdapter;
file: string;
auth: string | null;
pendingAuth: Array<(header: string | null) => void>;
session: StoreSession;
}
export interface StoreHostOptions {
areas: readonly StoreArea[];
storage: HostStorage;
emit(event: StoreEvent): void;
needAuth(account: string): void;
}
export class StoreHost {
private readonly accounts = new Map<string, OpenAccount>();
private online = true;
private readonly synced = new Map<string, number>();
constructor(private readonly options: StoreHostOptions) {}
async openAccount(params: OpenAccountParams): Promise<void> {
const existing = this.accounts.get(params.account);
if (existing) {
if (params.session) this.setSession(params.account, params.session);
this.setAuth(params.account, params.auth);
existing.engine.setScope(params.scope);
existing.engine.setInvertedKeywordSort(params.invertedKeywordSort);
void existing.engine.requestSync();
return;
}
const db = await this.options.storage.open(params.file, params.keyHex);
const session = params.session ?? AccountEngine.storedSession(db);
if (!session) {
db.close();
throw new Error('No session for this account yet');
}
const account: OpenAccount = { db, file: params.file, auth: params.auth, pendingAuth: [], session, engine: undefined as unknown as AccountEngine };
const transport = new FetchTransport({
session: () => account.session,
auth: () => account.auth,
refreshAuth: () => new Promise((resolve) => {
account.pendingAuth.push(resolve);
this.options.needAuth(params.account);
}),
});
account.engine = new AccountEngine({
account: params.account,
db,
session,
transport,
areas: this.options.areas,
scope: params.scope,
invertedKeywordSort: params.invertedKeywordSort,
emit: (event) => {
if (event.type === 'synced') this.synced.set(event.account, event.at);
this.options.emit(event);
},
});
account.engine.setOnline(this.online);
this.accounts.set(params.account, account);
void account.engine.requestSync();
}
setAuth(account: string, auth: string | null): void {
const open = this.accounts.get(account);
if (!open) return;
open.auth = auth;
for (const resolve of open.pendingAuth.splice(0)) resolve(auth);
}
setSession(account: string, session: StoreSession): void {
const open = this.accounts.get(account);
if (!open) return;
open.session = session;
open.engine.setSession(session);
}
setScope(account: string, scope: Parameters<AccountEngine['setScope']>[0]): void {
const open = this.accounts.get(account);
if (!open) return;
open.engine.setScope(scope);
void open.engine.requestSync();
}
async closeAccount(account: string): Promise<void> {
const open = this.accounts.get(account);
if (!open) return;
this.accounts.delete(account);
await open.engine.close();
for (const resolve of open.pendingAuth.splice(0)) resolve(null);
open.db.close();
}
async removeAccount(account: string, file: string): Promise<boolean> {
await this.closeAccount(account);
this.synced.delete(account);
return this.options.storage.remove(file);
}
async closeAll(): Promise<void> {
await Promise.all([...this.accounts.keys()].map((account) => this.closeAccount(account)));
}
/** The calls other tabs can make. */
async call(method: keyof StoreRpc, params: unknown): Promise<unknown> {
switch (method) {
case 'answer': {
const { account, calls } = params as Parameters<StoreRpc['answer']>[0];
const open = this.accounts.get(account);
if (!open) throw new Error('The local store is not open for this account');
return open.engine.answer(calls);
}
case 'sync': {
const { account } = params as Parameters<StoreRpc['sync']>[0];
await this.accounts.get(account)?.engine.requestSync();
return undefined;
}
case 'noteStateChange': {
const { account, changed } = params as Parameters<StoreRpc['noteStateChange']>[0];
this.accounts.get(account)?.engine.noteStateChange(changed);
return undefined;
}
case 'setOnline': {
const { online } = params as Parameters<StoreRpc['setOnline']>[0];
this.online = online;
for (const open of this.accounts.values()) open.engine.setOnline(online);
return undefined;
}
case 'status': {
const { account } = params as Parameters<StoreRpc['status']>[0];
return this.status(account);
}
default:
throw new Error(`Unknown store call ${String(method)}`);
}
}
private status(account: string): StoreStatus | null {
const open = this.accounts.get(account);
if (!open) return null;
return {
account,
open: true,
emails: Number(open.db.get<{ n: number }>('SELECT COUNT(*) AS n FROM email')?.n ?? 0),
bodies: Number(open.db.get<{ n: number }>('SELECT COUNT(*) AS n FROM email_body')?.n ?? 0),
lastSyncedAt: this.synced.get(account) ?? null,
};
}
}
+76
View File
@@ -0,0 +1,76 @@
/** Messages between the page and the store worker. */
import type { StoreEvent, StoreScope } from '../area';
import type { JmapResponse, MethodCall, StoreSession } from '../engine/transport';
export const RPC_CHANNEL = 'bulwark-store-rpc';
export const OWNER_LOCK = 'bulwark-store-owner';
export interface OpenAccountParams {
account: string;
/** `/a<hex>.db`, no account data: the name is plaintext in the OPFS slot. */
file: string;
/** 32 bytes in hex. Only ever sent to this tab's own worker, never over the channel. */
keyHex: string;
session: StoreSession | null;
scope: StoreScope;
invertedKeywordSort: boolean;
/** The page's current Authorization header. */
auth: string | null;
}
/** What other tabs can ask for, over the channel. */
export interface StoreRpc {
answer(params: { account: string; calls: MethodCall[] }): JmapResponse | null;
sync(params: { account: string }): void;
noteStateChange(params: { account: string; changed: Record<string, Record<string, string>> }): void;
setOnline(params: { online: boolean }): void;
status(params: { account: string }): StoreStatus | null;
}
export interface StoreStatus {
account: string;
open: boolean;
emails: number;
bodies: number;
lastSyncedAt: number | null;
}
export type RpcRequest<K extends keyof StoreRpc = keyof StoreRpc> = {
kind: 'rpc';
id: string;
from: string;
method: K;
params: Parameters<StoreRpc[K]>[0];
};
export type RpcReply = {
kind: 'reply';
id: string;
to: string;
result?: unknown;
error?: string;
};
export type ChannelMessage =
| RpcRequest
| RpcReply
| { kind: 'event'; event: StoreEvent }
| { kind: 'owner-changed'; owner: string | null };
/** From this tab's page to its own worker, never on the channel. */
export type PageToWorker =
| { kind: 'init'; clientId: string; moduleUrl: string }
| { kind: 'open'; id: string; params: OpenAccountParams }
| { kind: 'close-account'; id: string; account: string }
| { kind: 'remove-account'; id: string; account: string; file: string }
| { kind: 'set-auth'; account: string; auth: string | null }
| { kind: 'set-session'; account: string; session: StoreSession }
| { kind: 'set-scope'; account: string; scope: StoreScope }
| { kind: 'release'; id: string };
export type WorkerToPage =
| { kind: 'owner' }
| { kind: 'result'; id: string; result?: unknown; error?: string; errorName?: string }
| { kind: 'need-auth'; account: string }
| { kind: 'event'; event: StoreEvent };
+147
View File
@@ -0,0 +1,147 @@
/**
* The store worker. Every tab has one, parked on the owner lock, and only the
* holder loads the wasm and touches OPFS. The lock is held for the worker's
* whole life, so handing over means: close, pause the VFS, then let go.
*
* Built to public/store-worker.js by scripts/build-store-worker.mjs.
*/
import { WasmStorage } from '../adapters/wasm';
import { STORE_AREAS } from '../areas';
import { RPC_CHANNEL, OWNER_LOCK, type ChannelMessage, type PageToWorker, type WorkerToPage } from './protocol';
import { StoreHost } from './host';
const scope = self as unknown as {
postMessage(message: WorkerToPage): void;
addEventListener(type: 'message', listener: (event: MessageEvent<PageToWorker>) => void): void;
close(): void;
};
let clientId = '';
let moduleUrl = '';
let storage: WasmStorage | null = null;
let host: StoreHost | null = null;
let channel: BroadcastChannel | null = null;
let releaseLock: (() => void) | null = null;
function reply(id: string, result?: unknown): void {
scope.postMessage({ kind: 'result', id, result });
}
function replyError(id: string, error: unknown): void {
scope.postMessage({
kind: 'result',
id,
error: error instanceof Error ? error.message : String(error),
errorName: error instanceof Error ? error.name : undefined,
});
}
function requireHost(): StoreHost {
if (!host) throw new Error('This tab does not own the local store');
return host;
}
async function becomeOwner(): Promise<void> {
storage = await WasmStorage.install(moduleUrl);
host = new StoreHost({
areas: STORE_AREAS,
storage,
emit: (event) => {
scope.postMessage({ kind: 'event', event });
channel?.postMessage({ kind: 'event', event } satisfies ChannelMessage);
},
needAuth: (account) => scope.postMessage({ kind: 'need-auth', account }),
});
channel = new BroadcastChannel(RPC_CHANNEL);
channel.onmessage = async (event: MessageEvent<ChannelMessage>) => {
const message = event.data;
if (message.kind !== 'rpc') return;
try {
const result = await requireHost().call(message.method, message.params);
channel?.postMessage({ kind: 'reply', id: message.id, to: message.from, result } satisfies ChannelMessage);
} catch (error) {
channel?.postMessage({
kind: 'reply',
id: message.id,
to: message.from,
error: error instanceof Error ? error.message : String(error),
} satisfies ChannelMessage);
}
};
channel.postMessage({ kind: 'owner-changed', owner: clientId } satisfies ChannelMessage);
scope.postMessage({ kind: 'owner' });
}
async function release(): Promise<void> {
// The order matters: the OPFS handles have to be free before the lock is.
channel?.postMessage({ kind: 'owner-changed', owner: null } satisfies ChannelMessage);
await host?.closeAll();
host = null;
storage?.pause();
storage = null;
channel?.close();
channel = null;
releaseLock?.();
releaseLock = null;
}
scope.addEventListener('message', async (event) => {
const message = event.data;
switch (message.kind) {
case 'init':
clientId = message.clientId;
moduleUrl = message.moduleUrl;
void navigator.locks.request(OWNER_LOCK, () => new Promise<void>((resolve) => {
releaseLock = resolve;
becomeOwner().catch((error) => {
scope.postMessage({
kind: 'event',
event: { type: 'error', account: '', area: 'store', message: error instanceof Error ? error.message : String(error) },
});
resolve();
});
}));
break;
case 'open':
try {
await requireHost().openAccount(message.params);
reply(message.id);
} catch (error) {
replyError(message.id, error);
}
break;
case 'close-account':
try {
await requireHost().closeAccount(message.account);
reply(message.id);
} catch (error) {
replyError(message.id, error);
}
break;
case 'remove-account':
try {
reply(message.id, await requireHost().removeAccount(message.account, message.file));
} catch (error) {
replyError(message.id, error);
}
break;
case 'set-auth':
host?.setAuth(message.account, message.auth);
break;
case 'set-session':
host?.setSession(message.account, message.session);
break;
case 'set-scope':
host?.setScope(message.account, message.scope);
break;
case 'release':
try {
await release();
reply(message.id);
} catch (error) {
replyError(message.id, error);
}
break;
}
});
+5
View File
@@ -140,6 +140,11 @@ const nextConfig: NextConfig = {
{ key: "Cache-Control", value: "public, max-age=31536000, immutable" },
],
},
{
// The local store's database engine, in a folder named after its version.
source: "/sqlite3mc/:path*",
headers: [{ key: "Cache-Control", value: "public, max-age=31536000, immutable" }],
},
// The upload routes proxy.ts's matcher leaves out, with the
// headers it sets on every other API response.
...["/api/webdav", "/api/plugin-proxy", "/api/wopi/files/:fileId/contents"].map((source) => ({
+2 -2
View File
@@ -22,8 +22,8 @@
"typescript"
],
"scripts": {
"dev": "node scripts/build-pst-worker.mjs && node scripts/build-plugin-sandbox.mjs --watch -- next dev --turbopack",
"build": "node scripts/build-pst-worker.mjs && node scripts/build-plugin-sandbox.mjs && next build --turbopack && node scripts/standalone-guard.mjs",
"dev": "node scripts/build-pst-worker.mjs && node scripts/build-store-worker.mjs && node scripts/build-plugin-sandbox.mjs --watch -- next dev --turbopack",
"build": "node scripts/build-pst-worker.mjs && node scripts/build-store-worker.mjs && node scripts/build-plugin-sandbox.mjs && next build --turbopack && node scripts/standalone-guard.mjs",
"build:lite": "node scripts/lite/build.mjs",
"lite:serve": "node scripts/lite/serve.mjs out 4173",
"test:lite-smoke": "playwright test -c scripts/lite/smoke/playwright.config.ts",
+1 -1
View File
@@ -86,7 +86,7 @@ export const config = {
// /en/mail/folder/inbox/statement.pdf a real signed-in page, and it rendered
// without a CSP or any other security header (GHSA-xvjh-v9c6-qcvc). Only what
// is genuinely static may skip locale routing and the headers.
const STATIC_ASSET_PATTERN = /^\/[^/]+\.[^/]+$|^\/(?:branding|notification|demo|plugin-sandbox-assets)\//;
const STATIC_ASSET_PATTERN = /^\/[^/]+\.[^/]+$|^\/(?:branding|notification|demo|plugin-sandbox-assets|sqlite3mc)\//;
export function isStaticAssetPath(pathname: string): boolean {
return STATIC_ASSET_PATTERN.test(pathname);
+46
View File
@@ -0,0 +1,46 @@
// Builds the local store's worker (lib/store/worker/worker.ts) into
// public/store-worker.js and copies the vendored SQLite3MultipleCiphers build
// next to it, in a folder named after its version: emscripten looks for the
// literal name sqlite3.wasm, so the files can't have hashed names.
// Both go through proxy.ts's static asset rule, which keeps the page CSP off
// them: wasm compiles under the worker's own response headers.
//
// node scripts/build-store-worker.mjs (npm run build, npm run dev)
import { build } from 'esbuild';
import { copyFileSync, mkdirSync } from 'node:fs';
import path from 'node:path';
import { fileURLToPath } from 'node:url';
import { SQLITE3MC, SQLITE3MC_VENDOR_DIR, verifyVendored } from './fetch-sqlite3mc.mjs';
const root = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '..');
export const STORE_WORKER_OUTFILE = path.join(root, 'public', 'store-worker.js');
export const SQLITE3MC_PUBLIC_DIR = path.join(root, 'public', 'sqlite3mc', SQLITE3MC.version);
export async function buildStoreWorker({ outfile = STORE_WORKER_OUTFILE, publicDir = SQLITE3MC_PUBLIC_DIR } = {}) {
verifyVendored();
mkdirSync(publicDir, { recursive: true });
for (const name of [...Object.keys(SQLITE3MC.files), 'LICENSE']) {
copyFileSync(path.join(SQLITE3MC_VENDOR_DIR, name), path.join(publicDir, name));
}
const result = await build({
entryPoints: [path.join(root, 'lib/store/worker/worker.ts')],
bundle: true,
format: 'esm',
platform: 'browser',
target: 'es2022',
outfile,
minify: true,
legalComments: 'none',
alias: { '@': root },
define: { 'process.env.NODE_ENV': '"production"' },
metafile: true,
logLevel: 'warning',
});
return { bytes: Object.values(result.metafile.outputs).reduce((n, o) => n + o.bytes, 0) };
}
if (process.argv[1] && path.resolve(process.argv[1]) === fileURLToPath(import.meta.url)) {
const { bytes } = await buildStoreWorker();
console.log(`[build-store-worker] ${path.relative(root, STORE_WORKER_OUTFILE)} ${bytes}`);
}
+79
View File
@@ -0,0 +1,79 @@
// Fetches the pinned SQLite3MultipleCiphers WASM build into vendor/sqlite3mc/.
// Run it by hand when moving to a new version, and commit the result, so a
// Docker build needs no access to GitHub. Every download is checked against the
// pinned sha256: a changed artifact fails here.
//
// node scripts/fetch-sqlite3mc.mjs downloads and verifies
// node scripts/fetch-sqlite3mc.mjs --check verifies what is committed
//
// Fixes for bugs #291-#294 are on trunk, not in a release yet. Check the rules
// in lib/store/adapters/wasm.ts when moving to a newer build, and never go below 2.6.0.
import { createHash } from 'node:crypto';
import { mkdirSync, readFileSync, writeFileSync } from 'node:fs';
import path from 'node:path';
import { unzipSync } from 'fflate';
import { fileURLToPath } from 'node:url';
const root = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '..');
export const SQLITE3MC = {
version: '2.6.1-3.54.0',
tag: 'v2.6.1',
zip: 'sqlite3mc-2.6.1-sqlite-3.54.0-wasm.zip',
zipSha256: 'aba121490fd6123bcbef8404ee26df9b26c0301063781e2aa31f05aed6f48384',
innerDir: 'sqlite3mc-wasm-3540000/jswasm',
files: {
'sqlite3.mjs': '5308c7cc857408697c657a8d314e647a6f31c4aee67974ab8efd80c1a07161e0',
'sqlite3.wasm': '0b5a4bfff35f1ec87bf6f68ea293f425b86cbd661b77628a955cb827c9fbf524',
},
};
export const SQLITE3MC_VENDOR_DIR = path.join(root, 'vendor', 'sqlite3mc', SQLITE3MC.version);
function sha256(data) {
return createHash('sha256').update(data).digest('hex');
}
/** Throws unless the committed files match their pins. */
export function verifyVendored(dir = SQLITE3MC_VENDOR_DIR) {
for (const [name, expected] of Object.entries(SQLITE3MC.files)) {
const actual = sha256(readFileSync(path.join(dir, name)));
if (actual !== expected) throw new Error(`[sqlite3mc] ${name}: expected ${expected}, got ${actual}`);
}
readFileSync(path.join(dir, 'LICENSE'));
}
async function download(url) {
const response = await fetch(url);
if (!response.ok) throw new Error(`[sqlite3mc] ${url}: ${response.status}`);
return Buffer.from(await response.arrayBuffer());
}
async function fetchAndVendor() {
const base = `https://github.com/utelle/SQLite3MultipleCiphers/releases/download/${SQLITE3MC.tag}`;
const zip = await download(`${base}/${SQLITE3MC.zip}`);
const actual = sha256(zip);
if (actual !== SQLITE3MC.zipSha256) throw new Error(`[sqlite3mc] zip: expected ${SQLITE3MC.zipSha256}, got ${actual}`);
const entries = unzipSync(zip);
mkdirSync(SQLITE3MC_VENDOR_DIR, { recursive: true });
for (const name of Object.keys(SQLITE3MC.files)) {
const entry = entries[`${SQLITE3MC.innerDir}/${name}`];
if (!entry) throw new Error(`[sqlite3mc] ${name} is not in the zip`);
writeFileSync(path.join(SQLITE3MC_VENDOR_DIR, name), entry);
}
// The zip has no licence file.
const license = await download(`https://raw.githubusercontent.com/utelle/SQLite3MultipleCiphers/${SQLITE3MC.tag}/LICENSE`);
writeFileSync(path.join(SQLITE3MC_VENDOR_DIR, 'LICENSE'), license);
verifyVendored();
console.log(`[fetch-sqlite3mc] vendored ${SQLITE3MC.version}`);
}
if (process.argv[1] && path.resolve(process.argv[1]) === fileURLToPath(import.meta.url)) {
if (process.argv.includes('--check')) {
verifyVendored();
console.log('[fetch-sqlite3mc] ok');
} else {
await fetchAndVendor();
}
}
+3
View File
@@ -43,6 +43,9 @@ export const LITE_REMOVED_PATHS = [
"app/(main)/protocol",
"e2e",
"integration",
// The local store: Lite has no offline mode. A dev checkout may hold these.
"public/store-worker.js",
"public/sqlite3mc",
];
/** Test trees are pruned too: they import the routes removed above. */
+21
View File
@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2019-2026 Ulrich Telle
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
File diff suppressed because it is too large Load Diff
Binary file not shown.