fix(mail-ingest): read the Cloud inbox as it is sent, and keep mail refused for now

The relay job reads the control plane's { messages, more, next } page, checks
each mail on its own, pages on while more wait and is idle on a 404. A mail
for several addresses is stored for each, and a passing refusal (inbox full,
daily share taken) leaves it with the relay or in IMAP's TREK/deferred.

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<ResolvedAddress | null> {
+  async resolveAll(recipients: readonly string[]): Promise<MailIngestAddressRow[]> {
     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<boolean> {
     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<MailIngestAddressRow | null> {
@@ -114,11 +107,6 @@ export class MailIngestAddressService {
     await this.addresses.touchUsed(addressId, dbNow());
   }
 
-  private async withPerson(address: MailIngestAddressRow): Promise<ResolvedAddress | null> {
-    const person = await this.trips.person(address.user_id);
-    return person ? { address, person } : null;
-  }
-
   private async rotate(userId: number, tripId: number | null): Promise<MailIngestAddressRow> {
     const token = newToken();
     const id = await this.uow.transactional(async () => {

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;

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<number> {
+    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<number> {
+    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<ImapOutcome[]> {
+    return due.reduce<Promise<ImapOutcome[]>>(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<void> {
+  private async takeIn(session: ImapSession, mail: ImapMailRef): Promise<ImapOutcome> {
     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';
   }
 }

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<MailIntakeRefusal> = 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<MailIngestTransport, 'upload'>,
     envelopeRecipients: readonly string[] = [],
-  ): Promise<MailIntakeResult> {
+  ): Promise<MailReceiveResult> {
     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<Promise<MailIntakeResult[]>>(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' };
+}

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<typeof relayInboxSchema>['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<typeof relayedMailSchema>;
+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<boolean> {
+  /** 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<number> {
+    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<boolean> {
+    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<boolean> {
     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}`);
     }

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<unknown>;
+  /**
+   * `GET /v1/instance/inbox?limit=n&after=<cursor>`: 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<unknown>;
   /** `POST /v1/instance/inbox/:id/ack`: the mail is taken in (or never will be), and the relay drops it. */
   ack(id: string): Promise<void>;
 }

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'));

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<typeof ref>[], deferred: ReturnType<typeof ref>[] = []) => {
+    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', () => {

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: <top@x>\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');