diff --git a/server/src/nest/mail-ingest/mail-ingest-address.service.ts b/server/src/nest/mail-ingest/mail-ingest-address.service.ts index 9396d6d62..9f751b229 100644 --- a/server/src/nest/mail-ingest/mail-ingest-address.service.ts +++ b/server/src/nest/mail-ingest/mail-ingest-address.service.ts @@ -3,7 +3,6 @@ import type { MailIngestAddressesRepository, MailIngestAddressRow, } from '../../db/repositories/MailIngestAddresses.repository'; -import type { MailIngestPersonRow } from '../../db/repositories/MailIngestCandidates.repository'; import { dbNow, utcSuffix } from '../../db/types'; import { decrypt_api_key, encrypt_api_key } from '../common/crypto/apiKeyCrypto'; import { DomainError } from '../common/domain-error'; @@ -20,12 +19,6 @@ import { createHash, randomBytes } from 'node:crypto'; /** RFC 4648 base32, lower case: a mail's local part keeps it as it is. */ const BASE32 = 'abcdefghijklmnopqrstuvwxyz234567'; -/** A live address and the person it belongs to, as a mail that reached it is filed under. */ -interface ResolvedAddress { - address: MailIngestAddressRow; - person: MailIngestPersonRow; -} - /** * The forwarding addresses. A person has one of their own and may make one * per trip they can add bookings to; each address is a random token behind @@ -75,20 +68,20 @@ export class MailIngestAddressService { } /** - * The live address a mail was sent to, from the recipients in the order the - * receiving server names them (Delivered-To, X-Original-To, To). Null for a - * token nobody holds, a revoked one, a person who switched the import off, - * and a trip address whose owner may no longer add bookings to the trip. + * Every live address a mail was sent to, each once: a confirmation two + * co-travellers forward together reaches both. A token nobody holds, a + * revoked one, a person who switched the import off, and a trip address + * whose owner may no longer add bookings to the trip are left out. */ - async resolve(recipients: readonly string[]): Promise { + async resolveAll(recipients: readonly string[]): Promise { const template = await this.settings.addressTemplate(); - const tokens = recipients - .map((recipient) => tokenOf(recipient, template)) - .filter((token): token is string => token !== null); - const found = await Promise.all(tokens.map((token) => this.addresses.findByTokenHash(hashToken(token)))); - const address = found.find((row): row is MailIngestAddressRow => row !== null); - if (!address || !(await this.isLive(address))) return null; - return this.withPerson(address); + const tokens = new Set( + recipients.map((recipient) => tokenOf(recipient, template)).filter((token): token is string => token !== null), + ); + const found = await Promise.all([...tokens].map((token) => this.addresses.findByTokenHash(hashToken(token)))); + const rows = found.filter((row): row is MailIngestAddressRow => row !== null); + const live = await Promise.all(rows.map((row) => this.isLive(row))); + return rows.filter((_, i) => live[i]); } /** The person's own live address, made now when they have none: an uploaded mail goes through it. */ @@ -100,9 +93,9 @@ export class MailIngestAddressService { async isLive(address: MailIngestAddressRow): Promise { if (address.revoked_at) return false; if (!(await this.settings.prefsOf(address.user_id)).enabled) return false; - if (address.trip_id === null) return true; const person = await this.trips.person(address.user_id); - return person !== null && (await this.trips.editableTrip(person, address.trip_id)) !== null; + if (person === null) return false; + return address.trip_id === null || (await this.trips.editableTrip(person, address.trip_id)) !== null; } async findById(id: number): Promise { @@ -114,11 +107,6 @@ export class MailIngestAddressService { await this.addresses.touchUsed(addressId, dbNow()); } - private async withPerson(address: MailIngestAddressRow): Promise { - const person = await this.trips.person(address.user_id); - return person ? { address, person } : null; - } - private async rotate(userId: number, tripId: number | null): Promise { const token = newToken(); const id = await this.uow.transactional(async () => { diff --git a/server/src/nest/mail-ingest/mail-ingest-imap.client.ts b/server/src/nest/mail-ingest/mail-ingest-imap.client.ts index e64add5b4..d9aca2bbb 100644 --- a/server/src/nest/mail-ingest/mail-ingest-imap.client.ts +++ b/server/src/nest/mail-ingest/mail-ingest-imap.client.ts @@ -9,10 +9,11 @@ import { isIP } from 'node:net'; const CONNECT_TIMEOUT_MS = 15_000; const SOCKET_TIMEOUT_MS = 60_000; -/** A mail waiting in the box: its uid and its size, before anything of it is downloaded. */ +/** A mail waiting in the box: its uid, its size and when the box got it, before anything of it is downloaded. */ export interface ImapMailRef { uid: number; size: number; + receivedAt: Date | null; } /** What the import asks of an IMAP box, one connection per run. */ @@ -84,8 +85,11 @@ class ImapFlowSession implements ImapSession { if (this.exists === 0) return []; const found: ImapMailRef[] = []; // `n:*` always names the newest mail, even below n, so the range alone does not filter. - for await (const message of this.client.fetch(`${lastUid + 1}:*`, { uid: true, size: true }, { uid: true })) { - if (message.uid > lastUid) found.push({ uid: message.uid, size: message.size ?? 0 }); + const query = { uid: true, size: true, internalDate: true }; + for await (const message of this.client.fetch(`${lastUid + 1}:*`, query, { uid: true })) { + if (message.uid > lastUid) { + found.push({ uid: message.uid, size: message.size ?? 0, receivedAt: dateOf(message.internalDate) }); + } } return found.sort((a, b) => a.uid - b.uid); } @@ -115,6 +119,13 @@ class ImapFlowSession implements ImapSession { } } +/** The server's arrival time of a mail, which imapflow hands over as a Date or as the server wrote it. */ +function dateOf(value: Date | string | undefined): Date | null { + if (value === undefined) return null; + const date = value instanceof Date ? value : new Date(value); + return Number.isNaN(date.getTime()) ? null : date; +} + /** What an admin reads when the box cannot be read: the reason, never the password. */ export function describeImapFailure(err: unknown): string { if (err instanceof SsrfBlockedError) return err.message; diff --git a/server/src/nest/mail-ingest/mail-ingest-imap.job.ts b/server/src/nest/mail-ingest/mail-ingest-imap.job.ts index 9a50e0e8e..b901a7a86 100644 --- a/server/src/nest/mail-ingest/mail-ingest-imap.job.ts +++ b/server/src/nest/mail-ingest/mail-ingest-imap.job.ts @@ -9,7 +9,7 @@ import { type ImapMailRef, type ImapSession, } from './mail-ingest-imap.client'; -import { MailIngestIntakeService } from './mail-ingest-intake.service'; +import { isPassingRefusal, MailIngestIntakeService } from './mail-ingest-intake.service'; import { MailIngestSettingsService, type MailIngestImapConfig } from './mail-ingest-settings.service'; import { InjectRepository } from '@mikro-orm/nestjs'; import { Injectable, type OnApplicationBootstrap } from '@nestjs/common'; @@ -18,16 +18,30 @@ import { Injectable, type OnApplicationBootstrap } from '@nestjs/common'; const LEASE_MS = 10 * 60 * 1000; /** The mails one run takes from the box; the rest wait for the next run. */ const MAILS_PER_RUN = 20; +/** The mails one run offers the intake again from the deferred folder, oldest first. */ +const DEFERRED_PER_RUN = 5; +/** How long a deferred mail is offered again before it is given up. */ +const DEFER_DAYS = 3; +const DAY_MS = 24 * 60 * 60 * 1000; /** Where read mails go: kept, never deleted, out of the folder the next run reads. */ const PROCESSED_FOLDER = ['TREK', 'processed'] as const; +/** Where a mail the import cannot take for now waits to be offered again. */ +const DEFERRED_FOLDER = ['TREK', 'deferred'] as const; + +/** What became of one mail: taken in, done with otherwise, or kept to be offered again. */ +type ImapOutcome = 'taken' | 'done' | 'deferred'; /** * Reads the admin's IMAP box every two minutes, with a client of its own for * every run. Each new mail is handed to the intake, which stores it once, - * and then moved to TREK/processed. The cursor (uid validity, last uid) - * keeps a mail the move missed from being read twice, and the intake's - * dedupe keeps it so when the server renumbers the box and the run starts - * over. One process holds the box at a time, by the cursor's lease. + * and then moved to TREK/processed. A mail the import cannot take for now + * (an address's share for the day taken, an inbox full) goes to + * TREK/deferred instead, so it holds up no mail behind it, and a few of + * those are offered again on every run until they are taken or + * {@link DEFER_DAYS} old. The cursor (uid validity, last uid) keeps a mail + * a move missed from being read twice, and the intake's dedupe keeps it so + * when the server renumbers the box and the run starts over. One process + * holds the box at a time, by the cursor's lease. */ @Injectable() export class MailIngestImapJob implements OnApplicationBootstrap { @@ -45,8 +59,8 @@ export class MailIngestImapJob implements OnApplicationBootstrap { } /** - * One run; answers how many mails it took from the box. Nothing happens - * unless the import is on, IMAP is its transport and the box is complete. + * One run; answers how many mails it took in. Nothing happens unless the + * import is on, IMAP is its transport and the box is complete. * * @txIndependent each mail is stored on its own as it is read; the cursor * notes where the run got to once the box is done with. @@ -58,7 +72,7 @@ export class MailIngestImapJob implements OnApplicationBootstrap { const until = dbNow(new Date(now.getTime() + LEASE_MS)); if (!(await this.cursor.acquireLease('imap', config.mailbox, LEASE_OWNER, until, dbNow(now)))) return 0; try { - const { taken, uidValidity, lastUid } = await this.poll(config); + const { taken, uidValidity, lastUid } = await this.poll(config, now); await this.cursor.recordOk('imap', config.mailbox, dbNow(new Date()), { uidValidity, lastUid }); return taken; } catch (err) { @@ -71,7 +85,10 @@ export class MailIngestImapJob implements OnApplicationBootstrap { } } - private async poll(config: MailIngestImapConfig): Promise<{ taken: number; uidValidity: number; lastUid: number }> { + private async poll( + config: MailIngestImapConfig, + now: Date, + ): Promise<{ taken: number; uidValidity: number; lastUid: number }> { const session = await this.imap.connect(config); try { const { uidValidity } = await session.open(config.mailbox); @@ -79,28 +96,71 @@ export class MailIngestImapJob implements OnApplicationBootstrap { // A changed uid validity means the server renumbered the box: start over, the dedupe keeps each mail once. const after = position?.uid_validity === uidValidity ? (position.last_uid ?? 0) : 0; const due = (await session.listAfter(after)).slice(0, MAILS_PER_RUN); - if (due.length === 0) return { taken: 0, uidValidity, lastUid: after }; - const folder = await session.ensureFolder(PROCESSED_FOLDER); - await due.reduce(async (previous, mail) => { - await previous; - await this.takeIn(session, mail); - }, Promise.resolve()); - await session.move(due.map((mail) => mail.uid), folder); - return { taken: due.length, uidValidity, lastUid: Math.max(...due.map((mail) => mail.uid)) }; + const fresh = due.length > 0 ? await this.readNew(session, due) : 0; + const lastUid = due.length > 0 ? Math.max(...due.map((mail) => mail.uid)) : after; + const retried = await this.retryDeferred(session, now); + return { taken: fresh + retried, uidValidity, lastUid }; } finally { await session.close(); } } + /** The new mails of the box into the intake, each moved on by what became of it; answers how many were taken. */ + private async readNew(session: ImapSession, due: readonly ImapMailRef[]): Promise { + const outcomes = await this.takeAll(session, due); + const deferred = due.filter((_, i) => outcomes[i] === 'deferred').map((mail) => mail.uid); + const settled = due.filter((_, i) => outcomes[i] !== 'deferred').map((mail) => mail.uid); + if (settled.length > 0) await session.move(settled, await session.ensureFolder(PROCESSED_FOLDER)); + if (deferred.length > 0) await session.move(deferred, await session.ensureFolder(DEFERRED_FOLDER)); + return outcomes.filter((outcome) => outcome === 'taken').length; + } + + /** + * Offers the oldest deferred mails again. One taken now, or done with + * otherwise, moves on to TREK/processed; one still refused stays, unless + * it waited {@link DEFER_DAYS} already, and is then given up the same way. + */ + private async retryDeferred(session: ImapSession, now: Date): Promise { + await session.open(await session.ensureFolder(DEFERRED_FOLDER)); + const due = (await session.listAfter(0)).slice(0, DEFERRED_PER_RUN); + if (due.length === 0) return 0; + const outcomes = await this.takeAll(session, due); + // A mail whose arrival the server does not name cannot be aged, so it gets this one retry only. + const expired = (mail: ImapMailRef) => + !mail.receivedAt || now.getTime() - mail.receivedAt.getTime() > DEFER_DAYS * DAY_MS; + const settled = due.filter((mail, i) => outcomes[i] !== 'deferred' || expired(mail)); + const givenUp = due.filter((mail, i) => outcomes[i] === 'deferred' && expired(mail)).length; + if (givenUp > 0) { + logWarn(`[mail-ingest] Deferred IMAP mails given up after waiting ${DEFER_DAYS} days: ${givenUp}`); + } + if (settled.length > 0) { + await session.move( + settled.map((mail) => mail.uid), + await session.ensureFolder(PROCESSED_FOLDER), + ); + } + return outcomes.filter((outcome) => outcome === 'taken').length; + } + + private takeAll(session: ImapSession, due: readonly ImapMailRef[]): Promise { + return due.reduce>(async (previous, mail) => { + const done = await previous; + return [...done, await this.takeIn(session, mail)]; + }, Promise.resolve([])); + } + /** One mail into the intake. One larger than the import takes is left unread, and moved on with the rest. */ - private async takeIn(session: ImapSession, mail: ImapMailRef): Promise { + private async takeIn(session: ImapSession, mail: ImapMailRef): Promise { if (mail.size > this.settings.maxMailBytes()) { logWarn(`[mail-ingest] An IMAP mail of ${mail.size} bytes is larger than the import takes; left unread`); - return; + return 'done'; } const raw = await session.source(mail.uid); - if (!raw) return; + if (!raw) return 'done'; const result = await this.intake.receive(raw, 'imap'); - if (result.status === 'refused') logWarn(`[mail-ingest] An IMAP mail was not taken in: ${result.reason}`); + if (result.status !== 'refused') return result.status === 'accepted' ? 'taken' : 'done'; + if (isPassingRefusal(result.reason)) return 'deferred'; + logWarn(`[mail-ingest] An IMAP mail was not taken in: ${result.reason}`); + return 'done'; } } diff --git a/server/src/nest/mail-ingest/mail-ingest-intake.service.ts b/server/src/nest/mail-ingest/mail-ingest-intake.service.ts index bc5010c52..460f7ed69 100644 --- a/server/src/nest/mail-ingest/mail-ingest-intake.service.ts +++ b/server/src/nest/mail-ingest/mail-ingest-intake.service.ts @@ -5,6 +5,7 @@ import type { MailIngestCandidatesRepository } from '../../db/repositories/MailI import type { MailIngestMessagesRepository } from '../../db/repositories/MailIngestMessages.repository'; import { dbNow } from '../../db/types'; import { MailReadService } from '../llm-parse/mail-read.service'; +import type { WalkedMail } from '../llm-parse/mail-read.types'; import { MailIngestAddressService } from './mail-ingest-address.service'; import { MailIngestSettingsService } from './mail-ingest-settings.service'; import { sealMail } from './mail-seal'; @@ -24,10 +25,24 @@ export type MailIntakeRefusal = | 'address_daily_limit' | 'inbox_full'; +/** + * Refusals that pass: the import is switched off for now, the address took + * its share for the day, or its person's inbox is full. A transport keeps + * such a mail and offers it again later instead of dropping it. + */ +const PASSING_REFUSALS: ReadonlySet = new Set(['unavailable', 'address_daily_limit', 'inbox_full']); + +export function isPassingRefusal(reason: MailIntakeRefusal): boolean { + return PASSING_REFUSALS.has(reason); +} + type MailIntakeResult = | { status: 'accepted' | 'duplicate'; messageId: number } | { status: 'refused'; reason: MailIntakeRefusal }; +/** A received mail's outcome over every address it reached. */ +type MailReceiveResult = { status: 'accepted' | 'duplicate' } | { status: 'refused'; reason: MailIntakeRefusal }; + const DAY_MS = 24 * 60 * 60 * 1000; /** @@ -48,26 +63,32 @@ export class MailIngestIntakeService { ) {} /** - * A mail a transport received. `envelopeRecipients` are the RCPT TO - * addresses where the transport knows them; otherwise the headers name the - * address, in the order the receiving server adds them. + * A mail a transport received, stored once for every live address it was + * sent to. `envelopeRecipients` are the RCPT TO addresses where the + * transport knows them, and then the only ones read; otherwise the delivery + * header the receiving server put on top names the address. + * + * When one of its addresses refuses it for now, the whole mail answers that + * refusal: the transport offers it again, and the copies stored already + * come back as duplicates. */ async receive( raw: Buffer, transport: Exclude, envelopeRecipients: readonly string[] = [], - ): Promise { + ): Promise { const refused = await this.precheck(raw); if (refused) return { status: 'refused', reason: refused }; const mail = this.reader.walk(raw); if (!mail) return { status: 'refused', reason: 'not_a_mail' }; - const { deliveredTo, xOriginalTo, to } = mail.envelope; - const headerRecipients = [deliveredTo, xOriginalTo, ...(to ?? '').split(',')].filter( - (value): value is string => !!value && value.trim() !== '', - ); - const resolved = await this.addresses.resolve([...envelopeRecipients, ...headerRecipients]); - if (!resolved) return { status: 'refused', reason: 'unknown_address' }; - return this.store(raw, transport, resolved.address, mail.envelope.messageId); + const recipients = envelopeRecipients.length > 0 ? envelopeRecipients : headerRecipients(mail.envelope); + const addresses = await this.addresses.resolveAll(recipients); + if (addresses.length === 0) return { status: 'refused', reason: 'unknown_address' }; + const results = await addresses.reduce>(async (previous, address) => { + const done = await previous; + return [...done, await this.store(raw, transport, address, mail.envelope.messageId)]; + }, Promise.resolve([])); + return combined(results); } /** An .eml the person uploaded, read as if they had forwarded it to their own address. */ @@ -126,3 +147,25 @@ export class MailIngestIntakeService { return { status: 'accepted', messageId: id }; } } + +/** + * The addresses the headers name when the transport knows no envelope: the + * topmost Delivered-To, else the topmost X-Original-To, which the receiving + * server adds above whatever the mail brought along, and only without either + * the To line. A sender who names another token there gains nothing over + * sending to that token directly. + */ +function headerRecipients(envelope: WalkedMail['envelope']): string[] { + const delivered = [envelope.deliveredTo, envelope.xOriginalTo].find((value) => !!value?.trim()); + if (delivered) return [delivered]; + return (envelope.to ?? '').split(',').filter((value) => value.trim() !== ''); +} + +/** One answer for a mail stored per address: a passing refusal first, then accepted, then duplicate. */ +function combined(results: readonly MailIntakeResult[]): MailReceiveResult { + const passing = results.find((result) => result.status === 'refused' && isPassingRefusal(result.reason)); + if (passing) return passing; + if (results.some((result) => result.status === 'accepted')) return { status: 'accepted' }; + if (results.some((result) => result.status === 'duplicate')) return { status: 'duplicate' }; + return results[0] ?? { status: 'refused', reason: 'unknown_address' }; +} diff --git a/server/src/nest/mail-ingest/mail-ingest-relay.job.ts b/server/src/nest/mail-ingest/mail-ingest-relay.job.ts index 296b692dc..847918931 100644 --- a/server/src/nest/mail-ingest/mail-ingest-relay.job.ts +++ b/server/src/nest/mail-ingest/mail-ingest-relay.job.ts @@ -3,7 +3,7 @@ import type { MailIngestCursorRepository } from '../../db/repositories/MailInges import { dbNow } from '../../db/types'; import { logError, logWarn } from '../audit/audit-log.logger'; import { CronRegistrarService, LEASE_OWNER } from '../scheduling/cron-registrar.service'; -import { MailIngestIntakeService } from './mail-ingest-intake.service'; +import { isPassingRefusal, MailIngestIntakeService } from './mail-ingest-intake.service'; import { MailIngestRelayHost, type MailInboxRelay } from './mail-ingest-relay.service'; import { MailIngestSettingsService } from './mail-ingest-settings.service'; import { MAIL_INGEST_LIMITS } from './mail-ingest.constants'; @@ -15,35 +15,51 @@ import { z } from 'zod'; /** How long a run holds the relay before another process may take it over. */ const LEASE_MS = 10 * 60 * 1000; -/** The mails one run asks the relay for. */ +/** The mails one page of the relay's queue holds, and the pages one run reads. */ const PAGE = 10; +const PAGES_PER_RUN = 5; +/** The most recipients one mail names on this instance; the relay sends no more. */ +const MAX_RELAYED_RECIPIENTS = 100; /** The base64 length of the largest mail the relay hands on. */ const MAX_RAW_CHARS = Math.ceil(MAIL_INGEST_LIMITS.cloudMailBytes / 3) * 4; +/** The relay's mail ids are UUIDs. */ +const MAIL_ID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/; -/** The relay's answer, checked before anything of it is used. */ -const relayInboxSchema = z.object({ - mails: z - .array( - z.object({ - id: z.string().regex(/^[A-Za-z0-9_-]{1,64}$/), - /** The addresses it was sent to, as the relay's mail server took them (RCPT TO). */ - recipients: z.array(z.string().min(3).max(320)).min(1).max(5), - sha256: z.string().regex(/^[0-9a-f]{64}$/), - /** The mail itself, base64. */ - raw: z.string().max(MAX_RAW_CHARS), - }), - ) - .max(PAGE), +/** One page of the relay's queue; each mail in it is checked on its own, so an unreadable one blocks no other. */ +const relayPageSchema = z.object({ + messages: z.array(z.unknown()).max(PAGE), + /** Further mails wait beyond `next`. */ + more: z.boolean(), + /** Where the next page starts: after the last mail the relay looked at for this one. */ + next: z + .string() + .regex(/^[A-Za-z0-9_-]{1,200}$/) + .nullable(), }); -type RelayedMail = z.infer['mails'][number]; +const relayedMailSchema = z.object({ + id: z.string().regex(MAIL_ID), + receivedAt: z.string().max(40), + size: z.number().int().nonnegative(), + sha256: z.string().regex(/^[0-9a-f]{64}$/), + /** The addresses on this instance it was sent to, as the relay's mail server took them (RCPT TO). */ + recipients: z.array(z.string().min(3).max(320)).min(1).max(MAX_RELAYED_RECIPIENTS), + /** The mail itself, base64. */ + raw: z.string().max(MAX_RAW_CHARS), +}); +type RelayedMail = z.infer; +const relayedIdSchema = z.object({ id: z.string().regex(MAIL_ID) }); /** * Collects this instance's forwarded mails from the TREK Cloud relay every - * two minutes: a page of at most ten, each checked against its hash, handed - * to the intake under the address it was sent to, and acknowledged once it - * is in (or once it is clear it never will be), so the relay drops it. A - * mail that does not arrive whole is left for the next run. Idle wherever - * no relay is attached, which is every self-hosted install. + * two minutes, page after page while more wait, up to five pages a run. Each + * mail is checked against its hash, handed to the intake under the addresses + * it was sent to, and acknowledged once it is in (or once it is clear it + * never will be), so the relay drops it. A mail that did not arrive whole, + * or that the import cannot take for now (switched off, an address's share + * for the day taken, an inbox full), stays with the relay for a later run, + * and the pages after it are still read. Idle wherever no relay is attached, + * which is every self-hosted install, and where the control plane runs no + * import for this instance. */ @Injectable() export class MailIngestRelayJob implements OnApplicationBootstrap { @@ -71,12 +87,7 @@ export class MailIngestRelayJob implements OnApplicationBootstrap { const until = dbNow(new Date(now.getTime() + LEASE_MS)); if (!(await this.cursor.acquireLease('cloud', '', LEASE_OWNER, until, dbNow(now)))) return 0; try { - const page = relayInboxSchema.safeParse(await relay.list(PAGE)); - if (!page.success) throw new Error('The relay answered with an inbox this server cannot read'); - const taken = await page.data.mails.reduce(async (previous, mail) => { - const count = await previous; - return count + ((await this.takeIn(relay, mail)) ? 1 : 0); - }, Promise.resolve(0)); + const taken = await this.drain(relay, null, PAGES_PER_RUN); await this.cursor.recordOk('cloud', '', dbNow(new Date())); return taken; } catch (err) { @@ -89,16 +100,43 @@ export class MailIngestRelayJob implements OnApplicationBootstrap { } } - /** One relayed mail into the intake; answers whether it was taken in. */ - private async takeIn(relay: MailInboxRelay, mail: RelayedMail): Promise { + /** The pages from `after` on, as many as are left to the run; answers the mails taken in. */ + private async drain(relay: MailInboxRelay, after: string | null, pagesLeft: number): Promise { + const answer = await relay.list(PAGE, after); + if (answer === null) return 0; + const page = relayPageSchema.safeParse(answer); + if (!page.success) throw new Error('The relay answered with an inbox this server cannot read'); + const taken = await page.data.messages.reduce(async (previous, item) => { + const count = await previous; + return count + ((await this.takeIn(relay, item)) ? 1 : 0); + }, Promise.resolve(0)); + const { more, next } = page.data; + if (!more || next === null || next === after || pagesLeft <= 1) return taken; + return taken + (await this.drain(relay, next, pagesLeft - 1)); + } + + /** One mail of a page into the intake; answers whether it was taken in. */ + private async takeIn(relay: MailInboxRelay, item: unknown): Promise { + const parsed = relayedMailSchema.safeParse(item); + if (parsed.success) return this.store(relay, parsed.data); + // It would read the same on every run: dropped, where it names an id to drop it by. + const id = relayedIdSchema.safeParse(item); + logWarn(`[mail-ingest] The relay sent a mail this server cannot read${id.success ? `: ${id.data.id}` : ''}`); + if (id.success) await relay.ack(id.data.id); + return false; + } + + private async store(relay: MailInboxRelay, mail: RelayedMail): Promise { const raw = Buffer.from(mail.raw, 'base64'); if (createHash('sha256').update(raw).digest('hex') !== mail.sha256) { logWarn(`[mail-ingest] Relayed mail ${mail.id} did not arrive whole; left for the next run`); return false; } const result = await this.intake.receive(raw, 'cloud', mail.recipients); - // Switched off in between: the relay keeps the mail until the import is back or the mail expires. - if (result.status === 'refused' && result.reason === 'unavailable') return false; + if (result.status === 'refused' && isPassingRefusal(result.reason)) { + logWarn(`[mail-ingest] Relayed mail ${mail.id} waits with the relay: ${result.reason}`); + return false; + } if (result.status === 'refused') { logWarn(`[mail-ingest] Relayed mail ${mail.id} was not taken in: ${result.reason}`); } diff --git a/server/src/nest/mail-ingest/mail-ingest-relay.service.ts b/server/src/nest/mail-ingest/mail-ingest-relay.service.ts index 5fa7f5a82..f3724dc94 100644 --- a/server/src/nest/mail-ingest/mail-ingest-relay.service.ts +++ b/server/src/nest/mail-ingest/mail-ingest-relay.service.ts @@ -7,8 +7,13 @@ import { Injectable } from '@nestjs/common'; * encrypted, and hands it over here; nothing of a person reaches it. */ export interface MailInboxRelay { - /** `GET /v1/instance/inbox?limit=n`: the waiting mails, the answer's JSON as parsed, checked by the caller. */ - list(limit: number): Promise; + /** + * `GET /v1/instance/inbox?limit=n&after=`: the waiting mails after + * the cursor a previous page named (from the oldest without one), the + * answer's JSON as parsed, checked by the caller. Null when the control + * plane has no import for this instance (its 404), which is no failure. + */ + list(limit: number, after: string | null): Promise; /** `POST /v1/instance/inbox/:id/ack`: the mail is taken in (or never will be), and the relay drops it. */ ack(id: string): Promise; } diff --git a/server/tests/unit/nest/mail-ingest/mail-ingest-imap.client.test.ts b/server/tests/unit/nest/mail-ingest/mail-ingest-imap.client.test.ts index 473bf5984..eea2bf6ab 100644 --- a/server/tests/unit/nest/mail-ingest/mail-ingest-imap.client.test.ts +++ b/server/tests/unit/nest/mail-ingest/mail-ingest-imap.client.test.ts @@ -42,8 +42,10 @@ const CONFIG = { mailbox: 'INBOX', } as const; +const ARRIVED = new Date('2030-01-15T10:00:00Z'); + async function* messages(...uids: number[]) { - for (const uid of uids) yield { uid, size: uid * 10 }; + for (const uid of uids) yield { uid, size: uid * 10, internalDate: uid === 42 ? 'not a date' : ARRIVED }; } beforeEach(() => { @@ -84,10 +86,10 @@ describe('MailIngestImapClient', () => { // `41:*` names the newest mail even when it is older than 41. flow.fetch.mockReturnValue(messages(40, 42, 41)); expect(await session.listAfter(40)).toEqual([ - { uid: 41, size: 410 }, - { uid: 42, size: 420 }, + { uid: 41, size: 410, receivedAt: ARRIVED }, + { uid: 42, size: 420, receivedAt: null }, ]); - expect(flow.fetch).toHaveBeenCalledWith('41:*', { uid: true, size: true }, { uid: true }); + expect(flow.fetch).toHaveBeenCalledWith('41:*', { uid: true, size: true, internalDate: true }, { uid: true }); flow.fetchOne.mockResolvedValueOnce({ uid: 41, source: Buffer.from('raw') }).mockResolvedValueOnce(false); expect(await session.source(41)).toEqual(Buffer.from('raw')); diff --git a/server/tests/unit/nest/mail-ingest/mail-ingest-transports.test.ts b/server/tests/unit/nest/mail-ingest/mail-ingest-transports.test.ts index 848792ca4..30bcd7c2a 100644 --- a/server/tests/unit/nest/mail-ingest/mail-ingest-transports.test.ts +++ b/server/tests/unit/nest/mail-ingest/mail-ingest-transports.test.ts @@ -2,8 +2,10 @@ * How mail reaches the import and how the admin sees it run, over stubs: the * signed webhook (its window, its signature over the recipient and the raw * mail, its refusals), the IMAP run (cursor, renumbered box, size limit, the - * move to TREK/processed, a failure the status card reads), the TREK Cloud - * relay (hash check, acknowledgement, idle without a relay) and the status + * move to TREK/processed, a mail deferred for now and offered again, a + * failure the status card reads), the TREK Cloud relay (its page format, + * paging, hash check, a mail it cannot read, a mail kept for now, + * acknowledgement, idle without a relay or an import there) and the status * card and connection test. */ import { MailIngestCandidates } from '../../../../src/db/entities/MailIngestCandidates.entity'; @@ -41,6 +43,25 @@ vi.mock('../../../../src/nest/audit/audit-log.logger', () => ({ logWarn: vi.fn(), })); +/** + * The control plane's inbox answer, byte for byte the fixture its own tests + * hold the collection to (trek-hosted control-plane/src/fixtures/instance-inbox.json). + */ +const RELAY_FIXTURE = { + messages: [ + { + id: '0b9f2c4e-7d1a-4c3b-9e8f-1a2b3c4d5e6f', + receivedAt: '2026-11-13T10:00:00.000Z', + size: 37, + sha256: 'd9f4a3ee48bcaa66f8c8ddd7f348385bb63034211431d0e5f921d7cd131e83e7', + recipients: ['k3m7mfrggzdfmztwq2lknnwg23tp@in.liketrek.com'], + raw: 'U3ViamVjdDogWW91ciBib29raW5nDQoNClBOUiBBQkMxMjMNCg==', + }, + ], + more: false, + next: 'MjAyNi0xMS0xM1QxMDowMDowMC4wMDBaCjBiOWYyYzRlLTdkMWEtNGMzYi05ZThmLTFhMmIzYzRkNWU2Zg', +}; + const IMAP = { host: 'imap.example.com', port: 993, tls: 'tls', user: 'u', pass: 'p', mailbox: 'INBOX' } as const; const settings = { @@ -64,7 +85,7 @@ const session = { open: vi.fn(), listAfter: vi.fn(), source: vi.fn(), - ensureFolder: vi.fn(async () => 'TREK.processed'), + ensureFolder: vi.fn(async (parts: readonly string[]) => parts.join('.')), move: vi.fn(async () => undefined), close: vi.fn(async () => undefined), }; @@ -131,10 +152,10 @@ describe('the signed webhook', () => { }); it('takes a mail whose signature covers it, its recipient and a fresh timestamp', async () => { - intake.receive.mockResolvedValue({ status: 'accepted', messageId: 1 }); + intake.receive.mockResolvedValue({ status: 'accepted' }); expect(await service().receive(SECRET, RAW, signed(), NOW)).toEqual({ accepted: true, duplicate: false }); expect(intake.receive).toHaveBeenCalledWith(RAW, 'webhook', ['trek+tok@example.com']); - intake.receive.mockResolvedValue({ status: 'duplicate', messageId: 1 }); + intake.receive.mockResolvedValue({ status: 'duplicate' }); const noRecipient = signed({ recipient: '' }); expect(await service().receive(SECRET, RAW, noRecipient, NOW)).toEqual({ accepted: true, duplicate: true }); expect(intake.receive).toHaveBeenLastCalledWith(RAW, 'webhook', []); @@ -172,23 +193,26 @@ describe('the signed webhook', () => { describe('the IMAP run', () => { const job = () => app.get(MailIngestImapJob); + const NOW = new Date('2030-01-15T10:00:00Z'); + const ref = (uid: number, size = 10, receivedAt: Date | null = NOW) => ({ uid, size, receivedAt }); + /** The box's new mails, then the deferred folder's, as the run opens one after the other. */ + const boxes = (fresh: ReturnType[], deferred: ReturnType[] = []) => { + session.listAfter.mockResolvedValueOnce(fresh).mockResolvedValueOnce(deferred); + }; beforeEach(() => { settings.transport.mockResolvedValue('imap'); session.open.mockResolvedValue({ uidValidity: 9 }); session.listAfter.mockResolvedValue([]); session.source.mockImplementation(async (uid: number) => Buffer.from(`mail ${uid}`)); - intake.receive.mockResolvedValue({ status: 'accepted', messageId: 1 }); + intake.receive.mockResolvedValue({ status: 'accepted' }); }); it('takes the new mails after the cursor, moves them on, and notes where it got to', async () => { cursor.find.mockResolvedValue({ uid_validity: 9, last_uid: 40 }); - session.listAfter.mockResolvedValue([ - { uid: 41, size: 10 }, - { uid: 42, size: 10 }, - ]); - expect(await job().tick()).toBe(2); - expect(session.listAfter).toHaveBeenCalledWith(40); + boxes([ref(41), ref(42)]); + expect(await job().tick(NOW)).toBe(2); + expect(session.listAfter).toHaveBeenNthCalledWith(1, 40); expect(intake.receive.mock.calls.map(([raw]) => String(raw))).toEqual(['mail 41', 'mail 42']); expect(session.ensureFolder).toHaveBeenCalledWith(['TREK', 'processed']); expect(session.move).toHaveBeenCalledWith([41, 42], 'TREK.processed'); @@ -199,33 +223,58 @@ describe('the IMAP run', () => { it('starts over when the server renumbered the box, and leaves a mail over the limit unread', async () => { cursor.find.mockResolvedValue({ uid_validity: 3, last_uid: 400 }); - session.listAfter.mockResolvedValue([ - { uid: 1, size: 5000 }, - { uid: 2, size: 10 }, - ]); - expect(await job().tick()).toBe(2); - expect(session.listAfter).toHaveBeenCalledWith(0); + boxes([ref(1, 5000), ref(2)]); + expect(await job().tick(NOW)).toBe(1); + expect(session.listAfter).toHaveBeenNthCalledWith(1, 0); expect(session.source).toHaveBeenCalledTimes(1); expect(session.move).toHaveBeenCalledWith([1, 2], 'TREK.processed'); }); - it('touches nothing with an empty box, and nothing at all unless IMAP is the transport', async () => { - expect(await job().tick()).toBe(0); - expect(session.ensureFolder).not.toHaveBeenCalled(); + it('defers a mail the import cannot take for now, so the mails behind it go on', async () => { + boxes([ref(5), ref(6), ref(7)]); + intake.receive + .mockResolvedValueOnce({ status: 'refused', reason: 'inbox_full' }) + .mockResolvedValueOnce({ status: 'refused', reason: 'unknown_address' }) + .mockResolvedValueOnce({ status: 'duplicate' }); + expect(await job().tick(NOW)).toBe(0); + expect(session.move).toHaveBeenCalledWith([6, 7], 'TREK.processed'); + expect(session.move).toHaveBeenCalledWith([5], 'TREK.deferred'); + expect(cursor.recordOk).toHaveBeenCalledWith('imap', 'INBOX', expect.any(String), { uidValidity: 9, lastUid: 7 }); + }); + + it('offers deferred mails again, and gives up one that waited too long', async () => { + const old = new Date(NOW.getTime() - 4 * 24 * 60 * 60 * 1000); + boxes([], [ref(1), ref(2), ref(3, 10, old)]); + intake.receive + .mockResolvedValueOnce({ status: 'accepted' }) + .mockResolvedValueOnce({ status: 'refused', reason: 'address_daily_limit' }) + .mockResolvedValueOnce({ status: 'refused', reason: 'address_daily_limit' }); + expect(await job().tick(NOW)).toBe(1); + expect(session.open).toHaveBeenLastCalledWith('TREK.deferred'); + expect(session.listAfter).toHaveBeenLastCalledWith(0); + // 2 is kept for the next run; 3 waited past its days and goes on with 1. + expect(session.move).toHaveBeenCalledTimes(1); + expect(session.move).toHaveBeenCalledWith([1, 3], 'TREK.processed'); + }); + + it('moves nothing with an empty box, and touches nothing unless IMAP is the transport', async () => { + expect(await job().tick(NOW)).toBe(0); + expect(session.move).not.toHaveBeenCalled(); + expect(intake.receive).not.toHaveBeenCalled(); expect(cursor.recordOk).toHaveBeenCalledWith('imap', 'INBOX', expect.any(String), { uidValidity: 9, lastUid: 0 }); settings.transport.mockResolvedValue('webhook'); vi.clearAllMocks(); - expect(await job().tick()).toBe(0); + expect(await job().tick(NOW)).toBe(0); expect(imap.connect).not.toHaveBeenCalled(); settings.transport.mockResolvedValue('imap'); cursor.acquireLease.mockResolvedValue(false); - expect(await job().tick()).toBe(0); + expect(await job().tick(NOW)).toBe(0); expect(imap.connect).not.toHaveBeenCalled(); }); it('records why a run failed for the status card, and keeps the cursor where it was', async () => { imap.connect.mockRejectedValueOnce(Object.assign(new Error('Command failed'), { authenticationFailed: true })); - expect(await job().tick()).toBe(0); + expect(await job().tick(NOW)).toBe(0); const refused = 'The IMAP server refused the user name or password'; expect(cursor.recordError).toHaveBeenCalledWith('imap', 'INBOX', refused); expect(cursor.recordOk).not.toHaveBeenCalled(); @@ -248,50 +297,105 @@ describe('the IMAP run', () => { describe('the TREK Cloud relay', () => { const job = () => app.get(MailIngestRelayJob); const relay = { list: vi.fn(), ack: vi.fn(async () => undefined) }; - const mail = (id: string, body: string, sha = createHash('sha256').update(body).digest('hex')) => ({ - id, - recipients: ['ab12tok@in.example.com'], + const uuid = (n: number) => `00000000-0000-4000-8000-${String(n).padStart(12, '0')}`; + const mail = (n: number, body: string, sha = createHash('sha256').update(body).digest('hex')) => ({ + id: uuid(n), + receivedAt: '2030-01-15T09:58:00.000Z', + size: Buffer.byteLength(body), sha256: sha, + recipients: ['ab12tok@in.example.com'], raw: Buffer.from(body).toString('base64'), }); + const page = (messages: unknown[], more = false, next: string | null = null) => ({ messages, more, next }); beforeEach(() => { settings.transport.mockResolvedValue('cloud'); relays.attach(relay); }); - it('stays idle without a relay attached', async () => { + it('stays idle without a relay attached, and where the control plane runs no import', async () => { relays = new MailIngestRelayHost(); app = await build(); expect(await job().tick()).toBe(0); expect(cursor.acquireLease).not.toHaveBeenCalled(); - }); - it('takes each whole mail in under the address it was sent to and acknowledges it', async () => { - const broken = mail('m2', 'two', 'f'.repeat(64)); - relay.list.mockResolvedValue({ mails: [mail('m1', 'one'), broken, mail('m3', 'three')] }); - intake.receive - .mockResolvedValueOnce({ status: 'accepted', messageId: 1 }) - .mockResolvedValueOnce({ status: 'refused', reason: 'unknown_address' }); - expect(await job().tick()).toBe(1); - expect(relay.list).toHaveBeenCalledWith(10); - expect(intake.receive).toHaveBeenCalledWith(Buffer.from('one'), 'cloud', ['ab12tok@in.example.com']); - // m2 did not arrive whole and stays with the relay; m3 is refused for good and dropped. - expect(relay.ack.mock.calls).toEqual([['m1'], ['m3']]); + relays.attach(relay); + relay.list.mockResolvedValue(null); + expect(await job().tick()).toBe(0); + expect(cursor.recordError).not.toHaveBeenCalled(); expect(cursor.recordOk).toHaveBeenCalledWith('cloud', '', expect.any(String)); }); - it('keeps a mail the import cannot take right now, and refuses an inbox it cannot read', async () => { - relay.list.mockResolvedValue({ mails: [mail('m1', 'one')] }); + it('reads the page the control plane sends, as its own fixture has it', async () => { + relay.list.mockResolvedValue(RELAY_FIXTURE); + intake.receive.mockResolvedValue({ status: 'accepted' }); + expect(await job().tick()).toBe(1); + const [fixture] = RELAY_FIXTURE.messages; + expect(intake.receive).toHaveBeenCalledWith(Buffer.from(fixture.raw, 'base64'), 'cloud', fixture.recipients); + expect(relay.ack).toHaveBeenCalledWith(fixture.id); + }); + + it('takes each whole mail in under the addresses it was sent to and acknowledges it', async () => { + const broken = mail(2, 'two', 'f'.repeat(64)); + relay.list.mockResolvedValue(page([mail(1, 'one'), broken, mail(3, 'three')])); + intake.receive + .mockResolvedValueOnce({ status: 'accepted' }) + .mockResolvedValueOnce({ status: 'refused', reason: 'unknown_address' }); + expect(await job().tick()).toBe(1); + expect(relay.list).toHaveBeenCalledWith(10, null); + expect(intake.receive).toHaveBeenCalledWith(Buffer.from('one'), 'cloud', ['ab12tok@in.example.com']); + // 2 did not arrive whole and stays with the relay; 3 is refused for good and dropped. + expect(relay.ack.mock.calls).toEqual([[uuid(1)], [uuid(3)]]); + expect(cursor.recordOk).toHaveBeenCalledWith('cloud', '', expect.any(String)); + }); + + it('keeps a mail the import cannot take for now, and reads the mails after it', async () => { + relay.list.mockResolvedValue(page([mail(1, 'one'), mail(2, 'two')])); + intake.receive + .mockResolvedValueOnce({ status: 'refused', reason: 'inbox_full' }) + .mockResolvedValueOnce({ status: 'accepted' }); + expect(await job().tick()).toBe(1); + expect(relay.ack.mock.calls).toEqual([[uuid(2)]]); + + intake.receive.mockResolvedValue({ status: 'refused', reason: 'address_daily_limit' }); + await job().tick(); intake.receive.mockResolvedValue({ status: 'refused', reason: 'unavailable' }); await job().tick(); - expect(relay.ack).not.toHaveBeenCalled(); + expect(relay.ack).toHaveBeenCalledTimes(1); + }); - relay.list.mockResolvedValue({ mails: [{ id: '../../etc', raw: 'x' }] }); + it('drops a mail it can never read without holding up the rest, and refuses a page it cannot read', async () => { + const recipients = Array.from({ length: 101 }, (_, i) => `ab12t${i}@in.example.com`); + relay.list.mockResolvedValue(page([{ ...mail(1, 'one'), recipients }, { id: '../../etc' }, mail(3, 'three')])); + intake.receive.mockResolvedValue({ status: 'accepted' }); + expect(await job().tick()).toBe(1); + expect(relay.ack.mock.calls).toEqual([[uuid(1)], [uuid(3)]]); + + relay.list.mockResolvedValue({ mails: [mail(4, 'four')] }); expect(await job().tick()).toBe(0); const unreadable = 'The relay answered with an inbox this server cannot read'; expect(cursor.recordError).toHaveBeenCalledWith('cloud', '', unreadable); }); + + it('pages on while more wait, from where the last page ended, up to five pages a run', async () => { + relay.list + .mockResolvedValueOnce(page([mail(1, 'one')], true, 'c1')) + .mockResolvedValueOnce(page([mail(2, 'two')], false, 'c2')); + intake.receive.mockResolvedValue({ status: 'accepted' }); + expect(await job().tick()).toBe(2); + expect(relay.list.mock.calls).toEqual([ + [10, null], + [10, 'c1'], + ]); + + relay.list.mockReset(); + const position = (after: string | null) => Number(after?.slice(1) ?? 0) + 1; + relay.list.mockImplementation(async (_limit: number, after: string | null) => + page([mail(position(after), 'x')], true, `c${position(after)}`), + ); + await job().tick(); + expect(relay.list).toHaveBeenCalledTimes(5); + }); }); describe('the status card and the connection test', () => { diff --git a/server/tests/unit/nest/mail-ingest/mail-ingest.addresses.test.ts b/server/tests/unit/nest/mail-ingest/mail-ingest.addresses.test.ts index 0ce5649e9..601cb31b5 100644 --- a/server/tests/unit/nest/mail-ingest/mail-ingest.addresses.test.ts +++ b/server/tests/unit/nest/mail-ingest/mail-ingest.addresses.test.ts @@ -1,10 +1,12 @@ /** * The forwarding addresses inside MailIngestModule over a snapshot test db: a * person's own address and a trip's, renewing that retires the old token at - * once, and a trip address that stops working when its owner can no longer - * add bookings to the trip. + * once, a mail to several addresses that reaches each live one, the delivery + * header a mail is filed by without an envelope, and a trip address that + * stops working when its owner can no longer add bookings to the trip. */ import { MailIngestAddresses } from '../../../../src/db/entities/MailIngestAddresses.entity'; +import { MailIngestMessages } from '../../../../src/db/entities/MailIngestMessages.entity'; import { TripMembers } from '../../../../src/db/entities/TripMembers.entity'; import { DomainError } from '../../../../src/nest/common/domain-error'; import { MailIngestAddressService } from '../../../../src/nest/mail-ingest/mail-ingest-address.service'; @@ -89,13 +91,55 @@ describe('a person’s own address', () => { const { user } = await makeUser(orm()); const own = await addresses().renewOwn(user.id); await t.get(MailIngestSettingsService).updatePrefs(user.id, { enabled: false }); - expect(await addresses().resolve([own?.address ?? ''])).toBeNull(); + expect(await addresses().resolveAll([own?.address ?? ''])).toEqual([]); await t.get(MailIngestSettingsService).updatePrefs(user.id, { enabled: true }); - expect((await addresses().resolve([`Mia <${own?.address}>`]))?.person.id).toBe(user.id); + expect((await addresses().resolveAll([`Mia <${own?.address}>`]))[0]?.user_id).toBe(user.id); expect(tokenIn(own?.address ?? '')).toHaveLength(24); }); }); +describe('a mail to several addresses', () => { + it('reaches every live one once, and leaves out the rest', async () => { + const { user: mia } = await makeUser(orm()); + const { user: noah } = await makeUser(orm()); + const { user: off } = await makeUser(orm()); + const miaOwn = (await addresses().renewOwn(mia.id))?.address ?? ''; + const noahOwn = (await addresses().renewOwn(noah.id))?.address ?? ''; + const offOwn = (await addresses().renewOwn(off.id))?.address ?? ''; + await t.get(MailIngestSettingsService).updatePrefs(off.id, { enabled: false }); + + const found = await addresses().resolveAll([miaOwn, noahOwn, offOwn, miaOwn.toUpperCase(), 'nobody@example.com']); + expect(found.map((row) => row.user_id).sort()).toEqual([mia.id, noah.id].sort()); + }); + + it('is stored for each of them, and a second delivery is a duplicate for both', async () => { + const { user: mia } = await makeUser(orm()); + const { user: noah } = await makeUser(orm()); + const miaOwn = (await addresses().renewOwn(mia.id))?.address ?? ''; + const noahOwn = (await addresses().renewOwn(noah.id))?.address ?? ''; + const intake = t.get(MailIngestIntakeService); + const raw = mailTo('someone@example.com', 'both'); + + expect(await intake.receive(raw, 'cloud', [miaOwn, noahOwn])).toEqual({ status: 'accepted' }); + expect(await findRow(orm(), MailIngestMessages, { user: mia.id })).not.toBeNull(); + expect(await findRow(orm(), MailIngestMessages, { user: noah.id })).not.toBeNull(); + expect(await intake.receive(raw, 'cloud', [miaOwn, noahOwn])).toEqual({ status: 'duplicate' }); + }); + + it('without an envelope, goes by the delivery header the receiving server put on top', async () => { + const { user: mia } = await makeUser(orm()); + const { user: noah } = await makeUser(orm()); + const miaOwn = (await addresses().renewOwn(mia.id))?.address ?? ''; + const noahOwn = (await addresses().renewOwn(noah.id))?.address ?? ''; + const headers = [`Delivered-To: ${miaOwn}`, `Delivered-To: ${noahOwn}`, `To: ${noahOwn}`, 'From: x@example.com']; + const raw = Buffer.from(`${headers.join('\r\n')}\r\nSubject: Booking\r\nMessage-ID: \r\n\r\nBooking ABC123`); + + expect(await t.get(MailIngestIntakeService).receive(raw, 'imap')).toEqual({ status: 'accepted' }); + expect(await findRow(orm(), MailIngestMessages, { user: mia.id })).not.toBeNull(); + expect(await findRow(orm(), MailIngestMessages, { user: noah.id })).toBeNull(); + }); +}); + describe('a trip address', () => { it('belongs to the member who made it and stops working when they leave the trip', async () => { const { user: owner } = await makeUser(orm()); @@ -107,10 +151,10 @@ describe('a trip address', () => { expect(made?.trip_id).toBe(trip.id); expect((await addresses().tripAddress(member.id, trip.id))?.address).toBe(made?.address); expect(await addresses().tripAddress(owner.id, trip.id)).toBeNull(); - expect((await addresses().resolve([made?.address ?? '']))?.address.trip_id).toBe(trip.id); + expect((await addresses().resolveAll([made?.address ?? '']))[0]?.trip_id).toBe(trip.id); await deleteRows(orm(), TripMembers, { trip: trip.id, user: member.id }); - expect(await addresses().resolve([made?.address ?? ''])).toBeNull(); + expect(await addresses().resolveAll([made?.address ?? ''])).toEqual([]); await expect(addresses().tripAddress(member.id, trip.id)).rejects.toThrow('Trip not found'); }); @@ -120,7 +164,7 @@ describe('a trip address', () => { const made = await addresses().renewTripAddress(owner.id, trip.id); await addresses().revokeTripAddress(owner.id, trip.id); expect(await addresses().tripAddress(owner.id, trip.id)).toBeNull(); - expect(await addresses().resolve([made?.address ?? ''])).toBeNull(); + expect(await addresses().resolveAll([made?.address ?? ''])).toEqual([]); permissions.checkPermission.mockResolvedValue(false); await expect(addresses().renewTripAddress(owner.id, trip.id)).rejects.toThrow('Trip not found');