feat: keep mail sent offline in an Outbox and send it once back online

This commit is contained in:
Linus Rath
2026-10-31 15:34:02 +01:00
parent e1ad5c0945
commit 0ec809eb49
12 changed files with 1412 additions and 17 deletions
@@ -0,0 +1,106 @@
/**
* Live checks of the offline send queue's rule against sending twice: a
* message is found by its Message-ID, a copy that went out stops a resend,
* and the autosaved draft alone does not.
*
* Opt-in like live-stalwart.test.ts: skipped unless `BULWARK_LIVE_JMAP` and
* `BULWARK_LIVE_PASS` are set.
*
* BULWARK_LIVE_JMAP=http://10.10.10.62:8120 \
* BULWARK_LIVE_USER=e2e-perf@example.org BULWARK_LIVE_PASS=… \
* npx vitest run src/api/__tests__/live-stalwart-send-queue.test.ts
*
* Mail goes to the user's own address; everything it creates is removed.
* Verified against Stalwart 0.16.27 and 1.0.1 (2026-10-31).
*/
import { describe, it, expect, beforeAll, afterAll } from 'vitest';
import { JMAPClient } from '../jmap-client';
import { destroyEmails, getMailboxes, saveDraftVersion, sendEmail } from '../email';
import { getIdentities } from '../identity';
import { findCopiesByMessageId, runQueuedSend } from '../queued-send';
import { generateMessageId } from '../../lib/email-threading';
import { newQueuedSend } from '../../lib/send-queue';
import type { Identity, Mailbox } from '../types';
const SERVER = process.env.BULWARK_LIVE_JMAP;
const USER = process.env.BULWARK_LIVE_USER ?? 'e2e-perf@example.org';
const PASS = process.env.BULWARK_LIVE_PASS ?? '';
const live = SERVER && PASS ? describe : describe.skip;
const RUN = Date.now().toString(36);
let singleton: JMAPClient;
live('live Stalwart: offline send queue', () => {
let drafts = '';
let sent = '';
let boxes: Mailbox[] = [];
let identity: Identity;
const tag = `bulwark-sendqueue-${RUN}`;
beforeAll(async () => {
const mod = await import('../jmap-client');
singleton = mod.jmapClient as unknown as JMAPClient;
const client = new JMAPClient();
await client.connect(SERVER!, USER, PASS);
singleton.restoreSnapshot(client.snapshot());
boxes = await getMailboxes();
drafts = boxes.find((b) => b.role === 'drafts')?.id ?? '';
sent = boxes.find((b) => b.role === 'sent')?.id ?? '';
[identity] = await getIdentities();
}, 60_000);
afterAll(async () => {
const res = await singleton.request([['Email/query', { accountId: singleton.accountId, filter: { subject: tag } }, '0']]);
const leftovers = res.methodResponses[0][1].ids as string[] ?? [];
if (leftovers.length) await destroyEmails(leftovers).catch(() => undefined);
}, 60_000);
const message = (subject: string) => ({
from: [{ email: identity.email }],
to: [{ email: identity.email }],
subject: `${subject} ${tag}`,
textBody: 'live',
messageId: generateMessageId(identity.email),
});
const started = Date.now();
const lookup = () => ({ mailboxIds: [sent, drafts], since: started });
const payloadFor = (outgoing: ReturnType<typeof message>, draftId?: string) => ({
outgoing, identity, fromEmail: identity.email, sentMailboxId: sent, draftsMailboxId: drafts, draftId,
});
it('finds a message by its Message-ID, and only that one', async () => {
const outgoing = message('lookup');
const result = await sendEmail(outgoing, identity.id, sent, undefined, { draftsMailboxId: drafts });
const copies = await findCopiesByMessageId(outgoing.messageId, lookup());
expect(copies.map((c) => c.id)).toEqual([result.emailId]);
expect(copies[0].keywords?.$draft).toBeFalsy();
expect(await findCopiesByMessageId(generateMessageId(identity.email), lookup())).toEqual([]);
}, 60_000);
it('does not send again a message an unanswered attempt already sent', async () => {
const outgoing = message('sent-once');
const first = await sendEmail(outgoing, identity.id, sent, undefined, { draftsMailboxId: drafts });
// The answer to that send was lost: the queued entry is still marked.
const outcome = await runQueuedSend(newQueuedSend('q1', payloadFor(outgoing), Date.now(), true), boxes);
expect(outcome.kind).toBe('alreadySent');
const copies = await findCopiesByMessageId(outgoing.messageId, lookup());
expect(copies.map((c) => c.id)).toEqual([first.emailId]);
}, 60_000);
it('sends a marked message whose only copy is its draft, and removes the draft', async () => {
const outgoing = message('draft-only');
const draft = await saveDraftVersion(outgoing, drafts);
const outcome = await runQueuedSend(newQueuedSend('q2', payloadFor(outgoing, draft.id), Date.now(), true), boxes);
expect(outcome.kind).toBe('sent');
const copies = await findCopiesByMessageId(outgoing.messageId, lookup());
expect(copies).toHaveLength(1);
expect(copies[0].id).not.toBe(draft.id);
expect(copies[0].keywords?.$draft).toBeFalsy();
const res = await singleton.request([['Email/get', { accountId: singleton.accountId, ids: [copies[0].id], properties: ['mailboxIds'] }, '0']]);
expect(res.methodResponses[0][1].list[0].mailboxIds[sent]).toBe(true);
}, 60_000);
});
+107
View File
@@ -0,0 +1,107 @@
import { describe, it, expect, vi, beforeEach } from 'vitest';
// Sending from the offline queue: a message an unanswered attempt may have
// sent is looked up by its Message-ID first and not sent again when a copy
// went out.
vi.mock('../jmap-client', () => ({
jmapClient: { accountId: 'acc-1', request: vi.fn(), getAccountCapability: vi.fn(() => undefined) },
}));
const sendEmail = vi.fn(async (..._a: any[]) => ({ scheduled: false, emailId: 'e-new', emailSubmissionId: 's-1' }));
const destroyEmails = vi.fn(async (..._a: any[]) => undefined);
vi.mock('../email', () => ({
sendEmail: (...a: any[]) => sendEmail(...a),
destroyEmails: (...a: any[]) => destroyEmails(...a),
}));
vi.mock('../../lib/compose-send-account', () => ({
sendTargetFor: vi.fn(async (identity: { id: string }) => ({ identityId: identity.id, sentMailboxId: 'sent', draftsMailboxId: 'drafts' })),
}));
import { jmapClient } from '../jmap-client';
import { findCopiesByMessageId, runQueuedSend } from '../queued-send';
import { newQueuedSend, type QueuedSendPayload } from '../../lib/send-queue';
const mockRequest = jmapClient.request as ReturnType<typeof vi.fn>;
const PAYLOAD = {
outgoing: { from: [{ email: 'me@x.de' }], to: [{ email: 'you@x.de' }], subject: 'Hi', textBody: 'x', messageId: 'm1@x.de' },
identity: { id: 'id-1', email: 'me@x.de', name: 'Me' },
fromEmail: 'me@x.de',
sentMailboxId: 'sent',
draftsMailboxId: 'drafts',
draftId: 'draft-1',
requestDsn: true,
} as unknown as QueuedSendPayload;
const found = (list: Array<{ id: string; keywords?: Record<string, boolean>; messageId?: string[] }>) => ({
methodResponses: [
['Email/query', { ids: list.map((e) => e.id) }, '0'],
['Email/get', { list }, '1'],
],
});
beforeEach(() => {
vi.clearAllMocks();
});
describe('runQueuedSend', () => {
it('sends a message no attempt reached the server with, without looking first', async () => {
const outcome = await runQueuedSend(newQueuedSend('q', PAYLOAD, 0, false), []);
expect(outcome.kind).toBe('sent');
expect(mockRequest).not.toHaveBeenCalled();
expect(sendEmail).toHaveBeenCalledWith(
PAYLOAD.outgoing, 'id-1', 'sent', undefined,
// DSN only where the account lists it; this one lists nothing.
expect.objectContaining({ draftsMailboxId: 'drafts', draftId: 'draft-1', requestDsn: false }),
);
});
it('does not send again a message whose earlier attempt went out, and removes its old draft', async () => {
mockRequest.mockResolvedValueOnce(found([
{ id: 'draft-1', keywords: { $draft: true }, messageId: ['m1@x.de'] },
{ id: 'sent-1', keywords: { $seen: true }, messageId: ['<m1@x.de>'] },
]));
const outcome = await runQueuedSend(newQueuedSend('q', PAYLOAD, 0, true), []);
expect(outcome.kind).toBe('alreadySent');
expect(sendEmail).not.toHaveBeenCalled();
expect(destroyEmails).toHaveBeenCalledWith(['draft-1'], 'acc-1');
});
it('removes a copy a refused attempt left behind, then sends', async () => {
mockRequest.mockResolvedValueOnce(found([
{ id: 'draft-1', keywords: { $draft: true }, messageId: ['m1@x.de'] },
{ id: 'left-1', keywords: { $draft: true, $seen: true }, messageId: ['m1@x.de'] },
]));
const outcome = await runQueuedSend(newQueuedSend('q', PAYLOAD, 0, true), []);
expect(outcome.kind).toBe('sent');
expect(destroyEmails).toHaveBeenCalledWith(['left-1'], 'acc-1');
expect(sendEmail).toHaveBeenCalledTimes(1);
});
it('holds a scheduled message for the time that is left', async () => {
const now = Date.parse('2026-10-31T12:00:00Z');
await runQueuedSend(newQueuedSend('q', { ...PAYLOAD, scheduledAt: '2026-10-31T13:00:00Z' }, now, false), [], now);
expect(sendEmail.mock.calls[0][3]).toBe(3600);
});
});
describe('findCopiesByMessageId', () => {
it('reads the recent messages of Sent and Drafts and keeps exact matches only', async () => {
mockRequest.mockResolvedValueOnce(found([
{ id: 'a', keywords: {}, messageId: ['m1@x.de'] },
{ id: 'b', keywords: {}, messageId: ['other-m1@x.de'] },
{ id: 'c', keywords: {}, messageId: [] },
]));
const copies = await findCopiesByMessageId('<m1@x.de>', { mailboxIds: ['sent', 'drafts'], since: Date.parse('2026-10-31T12:00:00Z') });
expect(copies.map((c) => c.id)).toEqual(['a']);
const [query] = mockRequest.mock.calls[0][0];
// Not the header filter: Stalwart answers that from a lagging index.
expect(query[1].filter).toEqual({
operator: 'AND',
conditions: [
{ after: '2026-10-31T11:50:00.000Z' },
{ operator: 'OR', conditions: [{ inMailbox: 'sent' }, { inMailbox: 'drafts' }] },
],
});
});
});
+120
View File
@@ -0,0 +1,120 @@
// Sending a message from the offline send queue (see lib/send-queue.ts for
// the state machine and the rule against sending twice).
import { jmapClient } from './jmap-client';
import { destroyEmails, sendEmail, type SendEmailResult } from './email';
import { requireMethodResult } from './jmap-result';
import type { Mailbox } from './types';
import { sendTargetFor } from '../lib/compose-send-account';
import { supportsSubmissionExtension } from '../lib/submission-options';
import { stripMessageIdBrackets } from '../lib/email-threading';
import { holdForScheduled, verdictFromCopies, type FoundCopy, type QueuedSend } from '../lib/send-queue';
/** How far before the queueing a copy can be dated (the server's clock may differ). */
const CLOCK_SKEW_MS = 10 * 60_000;
/** Messages filed in Sent and Drafts since the queueing that are compared. */
const LOOKUP_LIMIT = 200;
/**
* The messages in the given folders of `accountId` whose Message-ID is
* `messageId`, filed since `since` (ms). Not a header search: Stalwart
* answers `{ header: ['Message-ID', …] }` from its full-text index, which
* lags behind (0.16.27 and 1.0.1 found nothing seconds after the send), so
* the folders' recent messages are read and compared instead.
*/
export async function findCopiesByMessageId(
messageId: string,
opts: { accountId?: string; mailboxIds: string[]; since: number },
): Promise<FoundCopy[]> {
const id = stripMessageIdBrackets(messageId);
const mailboxIds = opts.mailboxIds.filter(Boolean);
if (!id || mailboxIds.length === 0) return [];
const account = opts.accountId ?? jmapClient.accountId;
const inFolders = mailboxIds.length === 1
? { inMailbox: mailboxIds[0] }
: { operator: 'OR', conditions: mailboxIds.map((m) => ({ inMailbox: m })) };
const filter = {
operator: 'AND',
conditions: [{ after: new Date(opts.since - CLOCK_SKEW_MS).toISOString() }, inFolders],
};
const res = await jmapClient.request([
['Email/query', {
accountId: account, filter, sort: [{ property: 'receivedAt', isAscending: false }], limit: LOOKUP_LIMIT,
}, '0'],
['Email/get', {
accountId: account,
'#ids': { resultOf: '0', name: 'Email/query', path: '/ids' },
properties: ['id', 'keywords', 'messageId'],
}, '1'],
]);
requireMethodResult(res, '0', 'Email/query');
const list = (requireMethodResult(res, '1', 'Email/get').list ?? []) as Array<{
id: string; keywords?: Record<string, boolean>; messageId?: string[] | null;
}>;
return list
.filter((e) => (e.messageId ?? []).some((m) => stripMessageIdBrackets(m) === id))
.map((e) => ({ id: e.id, keywords: e.keywords ?? {} }));
}
export type QueuedSendOutcome =
| { kind: 'sent'; result: SendEmailResult; accountId?: string }
/** An earlier attempt went out; nothing was sent now. */
| { kind: 'alreadySent'; accountId?: string };
/**
* Send a queued message. The account, identity and folders are worked out
* as the composer does (sendTargetFor, #1090). A message an earlier attempt
* may have sent is looked up first, and is not sent again when a copy of it
* went out; copies a refused attempt left in Drafts are removed before the
* new attempt.
*/
export async function runQueuedSend(send: QueuedSend, mailboxes: Mailbox[], now = Date.now()): Promise<QueuedSendOutcome> {
const p = send.payload;
const target = await sendTargetFor(p.identity, p.fromEmail, mailboxes);
const accountId = target.accountId ?? jmapClient.accountId;
const draftAccountId = p.draftAccountId ?? jmapClient.accountId;
const sentMailboxId = target.sentMailboxId ?? p.sentMailboxId;
const draftsMailboxId = target.accountId ? target.draftsMailboxId : p.draftsMailboxId;
if (send.maybeSent && p.outgoing.messageId) {
const copies = await findCopiesByMessageId(p.outgoing.messageId, {
accountId: target.accountId,
mailboxIds: [sentMailboxId, draftsMailboxId ?? ''],
since: send.createdAt,
});
// The autosaved draft lives in this account only when the draft did.
const draftHere = p.draftId && draftAccountId === accountId ? p.draftId : undefined;
const verdict = verdictFromCopies(copies, draftHere);
if (verdict.kind === 'sent') {
// The send went out but its answer was lost, so the old draft it would
// have removed is still there.
if (p.draftId && p.draftId !== verdict.copyId) {
await destroyEmails([p.draftId], draftAccountId).catch((err) => {
console.warn('[send-queue] could not remove the draft of a sent message', err);
});
}
return { kind: 'alreadySent', accountId: target.accountId };
}
if (verdict.staleIds.length > 0) {
await destroyEmails(verdict.staleIds, accountId);
}
}
const result = await sendEmail(
p.outgoing,
target.identityId,
sentMailboxId,
holdForScheduled(p.scheduledAt, now),
{
draftsMailboxId,
draftId: p.draftId,
draftAccountId: p.draftId ? p.draftAccountId : undefined,
accountId: target.accountId,
// Only what the submitting account offers, as the composer does.
requestDsn: !!p.requestDsn && supportsSubmissionExtension('DSN', target.accountId),
requireTls: !!p.requireTls && supportsSubmissionExtension('REQUIRETLS', target.accountId),
},
);
return { kind: 'sent', result, accountId: target.accountId };
}