feat(mail-ingest): tables for forwarded booking mails and their bookings

Five tables carry the mail import: forwarding addresses (token hash plus the
encrypted token), a person's switches, the readers' cursor and lease, every
mail as it arrived, and each booking read from it with its checks and fate.
Repositories read the trips a booking may go to by column, so the import
weighs a booking without reaching into five other domains.

server/scripts/repository-owners.json

@@ -59,6 +59,11 @@
   "JourneyShareTokens": "journey",
   "JourneyTrips": "journey",
   "Journeys": "journey",
+  "MailIngestAddresses": "mail-ingest",
+  "MailIngestCandidates": "mail-ingest",
+  "MailIngestCursor": "mail-ingest",
+  "MailIngestMessages": "mail-ingest",
+  "MailIngestPrefs": "mail-ingest",
   "McpTokens": "tokens",
   "NotificationChannelPreferences": "notifications",
   "Notifications": "notifications",

server/src/db/entities/MailIngestAddresses.entity.ts

@@ -0,0 +1,46 @@
+import { Collection, EntityRepositoryType, type Opt, type Ref, defineEntity, p } from '@mikro-orm/core';
+import { MailIngestAddressesRepository } from '../repositories/MailIngestAddresses.repository';
+import { DbTimestampType } from '../types';
+import { MailIngestMessages } from './MailIngestMessages.entity';
+import { Trips } from './Trips.entity';
+import { Users } from './Users.entity';
+
+export class MailIngestAddresses {
+  [EntityRepositoryType]?: MailIngestAddressesRepository;
+  id!: number & Opt;
+  token_hash!: string;
+  token_enc!: string;
+  user!: Ref<Users>;
+  user_id!: number;
+  trip?: Ref<Trips> | null;
+  trip_id?: number | null;
+  created_at!: string & Opt;
+  revoked_at?: string | null;
+  last_used_at?: string | null;
+  mail_ingest_messages_collection = new Collection<MailIngestMessages>(this);
+}
+
+export const MailIngestAddressesSchema = defineEntity({
+  class: MailIngestAddresses,
+  repository: () => MailIngestAddressesRepository,
+  indexes: [
+    {
+      name: 'idx_mail_ingest_addresses_user',
+      properties: ['user_id', 'trip_id'],
+    },
+  ],
+  uniques: [{ properties: ['token_hash'] }],
+  properties: {
+    id: p.integer().primary(),
+    token_hash: p.text(),
+    token_enc: p.text(),
+    user: () => p.manyToOne(Users).ref().deleteRule('cascade').hidden(),
+    user_id: p.integer().persist(false),
+    trip: () => p.manyToOne(Trips).ref().deleteRule('cascade').nullable().hidden().index('idx_mail_ingest_addresses_trip'),
+    trip_id: p.integer().nullable().persist(false).index('idx_mail_ingest_addresses_trip'),
+    created_at: p.type(DbTimestampType).defaultRaw(`CURRENT_TIMESTAMP`),
+    revoked_at: p.type(DbTimestampType).nullable(),
+    last_used_at: p.type(DbTimestampType).nullable(),
+    mail_ingest_messages_collection: () => p.oneToMany(MailIngestMessages).mappedBy('address').hidden(),
+  },
+});

server/src/db/entities/MailIngestCandidates.entity.ts

@@ -0,0 +1,81 @@
+import { EntityRepositoryType, type Opt, type Ref, defineEntity, p } from '@mikro-orm/core';
+import { MailIngestCandidatesRepository } from '../repositories/MailIngestCandidates.repository';
+import { DbTimestampType } from '../types';
+import { MailIngestMessages } from './MailIngestMessages.entity';
+import { Trips } from './Trips.entity';
+import { Users } from './Users.entity';
+
+export class MailIngestCandidates {
+  [EntityRepositoryType]?: MailIngestCandidatesRepository;
+  id!: number & Opt;
+  message!: Ref<MailIngestMessages>;
+  message_id!: number;
+  user!: Ref<Users>;
+  user_id!: number;
+  position: number & Opt = 0;
+  item_json?: string | null;
+  source?: string | null;
+  checks_json?: string | null;
+  uncertain_fields?: string | null;
+  match_json?: string | null;
+  suggestedTrip?: Ref<Trips> | null;
+  suggested_trip_id?: number | null;
+  dedupe_key?: string | null;
+  content_hash?: string | null;
+  resolution: string & Opt = 'pending';
+  reason?: string | null;
+  resolvedTrip?: Ref<Trips> | null;
+  resolved_trip_id?: number | null;
+  created_ids_json?: string | null;
+  attach_error?: string | null;
+  resolved_by?: number | null;
+  resolved_at?: string | null;
+  undo_until?: string | null;
+  created_at!: string & Opt;
+}
+
+export const MailIngestCandidatesSchema = defineEntity({
+  class: MailIngestCandidates,
+  repository: () => MailIngestCandidatesRepository,
+  indexes: [
+    {
+      name: 'idx_mail_ingest_candidates_message',
+      properties: ['message_id', 'position'],
+    },
+    {
+      name: 'idx_mail_ingest_candidates_resolved',
+      properties: ['resolved_trip_id', 'content_hash'],
+    },
+    {
+      name: 'idx_mail_ingest_candidates_user',
+      properties: ['user_id', 'resolution'],
+    },
+  ],
+  properties: {
+    id: p.integer().primary(),
+    message: () => p.manyToOne(MailIngestMessages).ref().deleteRule('cascade').hidden(),
+    message_id: p.integer().persist(false),
+    user: () => p.manyToOne(Users).ref().deleteRule('cascade').hidden(),
+    user_id: p.integer().persist(false),
+    position: p.integer().default(0),
+    item_json: p.text().nullable(),
+    source: p.text().nullable(),
+    checks_json: p.text().nullable(),
+    uncertain_fields: p.text().nullable(),
+    match_json: p.text().nullable(),
+    suggestedTrip: () => p.manyToOne(Trips).ref().nullable().hidden().index('idx_mail_ingest_candidates_suggested'),
+    suggested_trip_id: p.integer().nullable().persist(false).index('idx_mail_ingest_candidates_suggested'),
+    dedupe_key: p.text().nullable(),
+    content_hash: p.text().nullable(),
+    resolution: p.text().default('pending'),
+    reason: p.text().nullable(),
+    resolvedTrip: () => p.manyToOne(Trips).ref().nullable().hidden(),
+    resolved_trip_id: p.integer().nullable().persist(false),
+    created_ids_json: p.text().nullable(),
+    attach_error: p.text().nullable(),
+    resolved_by: p.integer().nullable(),
+    resolved_at: p.type(DbTimestampType).nullable(),
+    undo_until: p.type(DbTimestampType).nullable(),
+    created_at: p.type(DbTimestampType).defaultRaw(`CURRENT_TIMESTAMP`),
+  },
+});

server/src/db/entities/MailIngestCursor.entity.ts

@@ -0,0 +1,33 @@
+import { EntityRepositoryType, type Opt, defineEntity, p } from '@mikro-orm/core';
+import { MailIngestCursorRepository } from '../repositories/MailIngestCursor.repository';
+import { DbTimestampType } from '../types';
+
+export class MailIngestCursor {
+  [EntityRepositoryType]?: MailIngestCursorRepository;
+  id!: number & Opt;
+  transport!: string;
+  mailbox!: string;
+  uid_validity?: number | null;
+  last_uid?: number | null;
+  last_ok_at?: string | null;
+  last_error?: string | null;
+  lease_owner?: string | null;
+  lease_until?: string | null;
+}
+
+export const MailIngestCursorSchema = defineEntity({
+  class: MailIngestCursor,
+  repository: () => MailIngestCursorRepository,
+  uniques: [{ properties: ['transport', 'mailbox'] }],
+  properties: {
+    id: p.integer().primary(),
+    transport: p.text(),
+    mailbox: p.text(),
+    uid_validity: p.integer().nullable(),
+    last_uid: p.integer().nullable(),
+    last_ok_at: p.type(DbTimestampType).nullable(),
+    last_error: p.text().nullable(),
+    lease_owner: p.text().nullable(),
+    lease_until: p.type(DbTimestampType).nullable(),
+  },
+});

server/src/db/entities/MailIngestMessages.entity.ts

@@ -0,0 +1,91 @@
+import { Collection, EntityRepositoryType, type Opt, type Ref, defineEntity, p } from '@mikro-orm/core';
+import { MailIngestMessagesRepository } from '../repositories/MailIngestMessages.repository';
+import { DbTimestampType } from '../types';
+import { MailIngestAddresses } from './MailIngestAddresses.entity';
+import { MailIngestCandidates } from './MailIngestCandidates.entity';
+import { Trips } from './Trips.entity';
+import { Users } from './Users.entity';
+
+export class MailIngestMessages {
+  [EntityRepositoryType]?: MailIngestMessagesRepository;
+  id!: number & Opt;
+  transport!: string;
+  message_id_hdr?: string | null;
+  dedupe_hash!: string;
+  address?: Ref<MailIngestAddresses> | null;
+  address_id?: number | null;
+  user!: Ref<Users>;
+  user_id!: number;
+  trip?: Ref<Trips> | null;
+  trip_id?: number | null;
+  from_addr?: string | null;
+  from_verified: number & Opt = 0;
+  auth_results?: string | null;
+  subject?: string | null;
+  received_at!: string & Opt;
+  size!: number;
+  raw_eml_enc?: string | null;
+  raw_purged_at?: string | null;
+  status: string & Opt = 'queued';
+  reason?: string | null;
+  hints?: string | null;
+  attachments?: string | null;
+  forwarding_code?: string | null;
+  attempts: number & Opt = 0;
+  next_attempt_at!: string & Opt;
+  lease_until?: string | null;
+  ai_used: number & Opt = 0;
+  processed_at?: string | null;
+  mail_ingest_candidates_collection = new Collection<MailIngestCandidates>(this);
+}
+
+export const MailIngestMessagesSchema = defineEntity({
+  class: MailIngestMessages,
+  repository: () => MailIngestMessagesRepository,
+  indexes: [
+    {
+      name: 'idx_mail_ingest_messages_address',
+      properties: ['address_id', 'received_at'],
+    },
+    {
+      name: 'idx_mail_ingest_messages_due',
+      properties: ['status', 'next_attempt_at'],
+    },
+    {
+      name: 'idx_mail_ingest_messages_user',
+      properties: ['user_id', 'received_at'],
+    },
+  ],
+  uniques: [{ properties: ['dedupe_hash'] }],
+  properties: {
+    id: p.integer().primary(),
+    transport: p.text(),
+    message_id_hdr: p.text().nullable(),
+    dedupe_hash: p.text(),
+    address: () => p.manyToOne(MailIngestAddresses).ref().nullable().hidden(),
+    address_id: p.integer().nullable().persist(false),
+    user: () => p.manyToOne(Users).ref().deleteRule('cascade').hidden(),
+    user_id: p.integer().persist(false),
+    trip: () => p.manyToOne(Trips).ref().nullable().hidden().index('idx_mail_ingest_messages_trip'),
+    trip_id: p.integer().nullable().persist(false).index('idx_mail_ingest_messages_trip'),
+    from_addr: p.text().nullable(),
+    from_verified: p.integer().default(0),
+    auth_results: p.text().nullable(),
+    subject: p.text().nullable(),
+    received_at: p.type(DbTimestampType).defaultRaw(`CURRENT_TIMESTAMP`),
+    size: p.integer(),
+    raw_eml_enc: p.text().nullable(),
+    raw_purged_at: p.type(DbTimestampType).nullable(),
+    status: p.text().default('queued'),
+    reason: p.text().nullable(),
+    hints: p.text().nullable(),
+    attachments: p.text().nullable(),
+    forwarding_code: p.text().nullable(),
+    attempts: p.integer().default(0),
+    next_attempt_at: p.type(DbTimestampType).defaultRaw(`CURRENT_TIMESTAMP`),
+    lease_until: p.type(DbTimestampType).nullable(),
+    ai_used: p.integer().default(0),
+    processed_at: p.type(DbTimestampType).nullable(),
+    mail_ingest_candidates_collection: () => p.oneToMany(MailIngestCandidates).mappedBy('message').hidden(),
+  },
+});

server/src/db/entities/MailIngestPrefs.entity.ts

@@ -0,0 +1,27 @@
+import { EntityRepositoryType, type Opt, type Ref, defineEntity, p } from '@mikro-orm/core';
+import { MailIngestPrefsRepository } from '../repositories/MailIngestPrefs.repository';
+import { Users } from './Users.entity';
+
+export class MailIngestPrefs {
+  [EntityRepositoryType]?: MailIngestPrefsRepository;
+  id!: number & Opt;
+  user!: Ref<Users>;
+  user_id!: number;
+  enabled: number & Opt = 1;
+  ai_enabled: number & Opt = 1;
+  auto_add: number & Opt = 1;
+}
+
+export const MailIngestPrefsSchema = defineEntity({
+  class: MailIngestPrefs,
+  repository: () => MailIngestPrefsRepository,
+  uniques: [{ properties: ['user'] }],
+  properties: {
+    id: p.integer().primary(),
+    user: () => p.manyToOne(Users).ref().deleteRule('cascade').hidden(),
+    user_id: p.integer().persist(false),
+    enabled: p.integer().default(1),
+    ai_enabled: p.integer().default(1),
+    auto_add: p.integer().default(1),
+  },
+});

server/src/db/entities/Trips.entity.ts

@@ -19,6 +19,9 @@ import { DocumentSyncItems } from './DocumentSyncItems.entity';
 import { InviteTokens } from './InviteTokens.entity';
 import { JourneyEntries } from './JourneyEntries.entity';
 import { JourneyTrips } from './JourneyTrips.entity';
+import { MailIngestAddresses } from './MailIngestAddresses.entity';
+import { MailIngestCandidates } from './MailIngestCandidates.entity';
+import { MailIngestMessages } from './MailIngestMessages.entity';
 import { PackingBags } from './PackingBags.entity';
 import { PackingCategoryAssignees } from './PackingCategoryAssignees.entity';
 import { PackingItems } from './PackingItems.entity';
@@ -77,6 +80,10 @@ export class Trips {
   invite_tokens_collection = new Collection<InviteTokens>(this);
   journey_entries_collection = new Collection<JourneyEntries>(this);
   journey_trips_collection = new Collection<JourneyTrips>(this);
+  mail_ingest_addresses_collection = new Collection<MailIngestAddresses>(this);
+  mail_ingest_candidates_collection = new Collection<MailIngestCandidates>(this);
+  mail_ingest_candidates_collection1 = new Collection<MailIngestCandidates>(this);
+  mail_ingest_messages_collection = new Collection<MailIngestMessages>(this);
   packing_bags_collection = new Collection<PackingBags>(this);
   packing_category_assignees_collection = new Collection<PackingCategoryAssignees>(this);
   packing_items_collection = new Collection<PackingItems>(this);
@@ -151,6 +158,10 @@ export const TripsSchema = defineEntity({
     invite_tokens_collection: () => p.oneToMany(InviteTokens).mappedBy('trip').hidden(),
     journey_entries_collection: () => p.oneToMany(JourneyEntries).mappedBy('sourceTrip').hidden(),
     journey_trips_collection: () => p.oneToMany(JourneyTrips).mappedBy('trip').hidden(),
+    mail_ingest_addresses_collection: () => p.oneToMany(MailIngestAddresses).mappedBy('trip').hidden(),
+    mail_ingest_candidates_collection: () => p.oneToMany(MailIngestCandidates).mappedBy('suggestedTrip').hidden(),
+    mail_ingest_candidates_collection1: () => p.oneToMany(MailIngestCandidates).mappedBy('resolvedTrip').hidden(),
+    mail_ingest_messages_collection: () => p.oneToMany(MailIngestMessages).mappedBy('trip').hidden(),
     packing_bags_collection: () => p.oneToMany(PackingBags).mappedBy('trip').hidden(),
     packing_category_assignees_collection: () => p.oneToMany(PackingCategoryAssignees).mappedBy('trip').hidden(),
     packing_items_collection: () => p.oneToMany(PackingItems).mappedBy('trip').hidden(),

server/src/db/entities/Users.entity.ts

@@ -39,6 +39,10 @@ import { JourneyRecordedDays } from './JourneyRecordedDays.entity';
 import { JourneyRecordingPrep } from './JourneyRecordingPrep.entity';
 import { JourneyShareTokens } from './JourneyShareTokens.entity';
 import { Journeys } from './Journeys.entity';
+import { MailIngestAddresses } from './MailIngestAddresses.entity';
+import { MailIngestCandidates } from './MailIngestCandidates.entity';
+import { MailIngestMessages } from './MailIngestMessages.entity';
+import { MailIngestPrefs } from './MailIngestPrefs.entity';
 import { McpTokens } from './McpTokens.entity';
 import { NotificationChannelPreferences } from './NotificationChannelPreferences.entity';
 import { Notifications } from './Notifications.entity';
@@ -176,6 +180,10 @@ export class Users {
   journey_recording_prep_collection = new Collection<JourneyRecordingPrep>(this);
   journey_share_tokens_collection = new Collection<JourneyShareTokens>(this);
   journeys_collection = new Collection<Journeys>(this);
+  mail_ingest_addresses_collection = new Collection<MailIngestAddresses>(this);
+  mail_ingest_candidates_collection = new Collection<MailIngestCandidates>(this);
+  mail_ingest_messages_collection = new Collection<MailIngestMessages>(this);
+  mail_ingest_prefs_collection = new Collection<MailIngestPrefs>(this);
   mcp_tokens_collection = new Collection<McpTokens>(this);
   notification_channel_preferences_collection = new Collection<NotificationChannelPreferences>(this);
   notifications_collection = new Collection<Notifications>(this);
@@ -335,6 +343,10 @@ export const UsersSchema = defineEntity({
     journey_recording_prep_collection: () => p.oneToMany(JourneyRecordingPrep).mappedBy('user').hidden(),
     journey_share_tokens_collection: () => p.oneToMany(JourneyShareTokens).mappedBy('createdByRef').hidden(),
     journeys_collection: () => p.oneToMany(Journeys).mappedBy('user').hidden(),
+    mail_ingest_addresses_collection: () => p.oneToMany(MailIngestAddresses).mappedBy('user').hidden(),
+    mail_ingest_candidates_collection: () => p.oneToMany(MailIngestCandidates).mappedBy('user').hidden(),
+    mail_ingest_messages_collection: () => p.oneToMany(MailIngestMessages).mappedBy('user').hidden(),
+    mail_ingest_prefs_collection: () => p.oneToMany(MailIngestPrefs).mappedBy('user').hidden(),
     mcp_tokens_collection: () => p.oneToMany(McpTokens).mappedBy('user').hidden(),
     notification_channel_preferences_collection: () => p.oneToMany(NotificationChannelPreferences).mappedBy('user').hidden(),
     notifications_collection: () => p.oneToMany(Notifications).mappedBy('sender').hidden(),

server/src/db/entities/index.ts

@@ -59,6 +59,11 @@ import { JourneyRecordingPrepSchema } from './JourneyRecordingPrep.entity';
 import { JourneyShareTokensSchema } from './JourneyShareTokens.entity';
 import { JourneyTripsSchema } from './JourneyTrips.entity';
 import { JourneysSchema } from './Journeys.entity';
+import { MailIngestAddressesSchema } from './MailIngestAddresses.entity';
+import { MailIngestCandidatesSchema } from './MailIngestCandidates.entity';
+import { MailIngestCursorSchema } from './MailIngestCursor.entity';
+import { MailIngestMessagesSchema } from './MailIngestMessages.entity';
+import { MailIngestPrefsSchema } from './MailIngestPrefs.entity';
 import { McpTokensSchema } from './McpTokens.entity';
 import { MigrationsSchema } from './Migrations.entity';
 import { NotificationChannelPreferencesSchema } from './NotificationChannelPreferences.entity';
@@ -229,6 +234,11 @@ export const ALL_ENTITIES: readonly EntitySchema[] = [
   JourneyShareTokensSchema,
   JourneyTripsSchema,
   JourneysSchema,
+  MailIngestAddressesSchema,
+  MailIngestCandidatesSchema,
+  MailIngestCursorSchema,
+  MailIngestMessagesSchema,
+  MailIngestPrefsSchema,
   McpTokensSchema,
   MigrationsSchema,
   NotificationChannelPreferencesSchema,

server/src/db/kysely/MailIngestAddresses.table.ts

@@ -0,0 +1,14 @@
+// Generated by scripts/generate-db-types.ts from the migrated schema. Do not edit by hand:
+// change the migration, then run `npm run gen:db-types`.
+import type { InsertOptional } from './columns';
+
+export interface MailIngestAddressesTable {
+  id: InsertOptional<number>;
+  token_hash: string;
+  token_enc: string;
+  user_id: number;
+  trip_id: InsertOptional<number | null>;
+  created_at: InsertOptional<string>;
+  revoked_at: InsertOptional<string | null>;
+  last_used_at: InsertOptional<string | null>;
+}

server/src/db/kysely/MailIngestCandidates.table.ts

@@ -0,0 +1,27 @@
+// Generated by scripts/generate-db-types.ts from the migrated schema. Do not edit by hand:
+// change the migration, then run `npm run gen:db-types`.
+import type { InsertOptional } from './columns';
+
+export interface MailIngestCandidatesTable {
+  id: InsertOptional<number>;
+  message_id: number;
+  user_id: number;
+  position: InsertOptional<number>;
+  item_json: InsertOptional<string | null>;
+  source: InsertOptional<string | null>;
+  checks_json: InsertOptional<string | null>;
+  uncertain_fields: InsertOptional<string | null>;
+  match_json: InsertOptional<string | null>;
+  suggested_trip_id: InsertOptional<number | null>;
+  dedupe_key: InsertOptional<string | null>;
+  content_hash: InsertOptional<string | null>;
+  resolution: InsertOptional<string>;
+  reason: InsertOptional<string | null>;
+  resolved_trip_id: InsertOptional<number | null>;
+  created_ids_json: InsertOptional<string | null>;
+  attach_error: InsertOptional<string | null>;
+  resolved_by: InsertOptional<number | null>;
+  resolved_at: InsertOptional<string | null>;
+  undo_until: InsertOptional<string | null>;
+  created_at: InsertOptional<string>;
+}

server/src/db/kysely/MailIngestCursor.table.ts

@@ -0,0 +1,15 @@
+// Generated by scripts/generate-db-types.ts from the migrated schema. Do not edit by hand:
+// change the migration, then run `npm run gen:db-types`.
+import type { InsertOptional } from './columns';
+
+export interface MailIngestCursorTable {
+  id: InsertOptional<number>;
+  transport: string;
+  mailbox: string;
+  uid_validity: InsertOptional<number | null>;
+  last_uid: InsertOptional<number | null>;
+  last_ok_at: InsertOptional<string | null>;
+  last_error: InsertOptional<string | null>;
+  lease_owner: InsertOptional<string | null>;
+  lease_until: InsertOptional<string | null>;
+}

server/src/db/kysely/MailIngestMessages.table.ts

@@ -0,0 +1,31 @@
+// Generated by scripts/generate-db-types.ts from the migrated schema. Do not edit by hand:
+// change the migration, then run `npm run gen:db-types`.
+import type { InsertOptional } from './columns';
+
+export interface MailIngestMessagesTable {
+  id: InsertOptional<number>;
+  transport: string;
+  message_id_hdr: InsertOptional<string | null>;
+  dedupe_hash: string;
+  address_id: InsertOptional<number | null>;
+  user_id: number;
+  trip_id: InsertOptional<number | null>;
+  from_addr: InsertOptional<string | null>;
+  from_verified: InsertOptional<number>;
+  auth_results: InsertOptional<string | null>;
+  subject: InsertOptional<string | null>;
+  received_at: InsertOptional<string>;
+  size: number;
+  raw_eml_enc: InsertOptional<string | null>;
+  raw_purged_at: InsertOptional<string | null>;
+  status: InsertOptional<string>;
+  reason: InsertOptional<string | null>;
+  hints: InsertOptional<string | null>;
+  attachments: InsertOptional<string | null>;
+  forwarding_code: InsertOptional<string | null>;
+  attempts: InsertOptional<number>;
+  next_attempt_at: InsertOptional<string>;
+  lease_until: InsertOptional<string | null>;
+  ai_used: InsertOptional<number>;
+  processed_at: InsertOptional<string | null>;
+}

server/src/db/kysely/MailIngestPrefs.table.ts

@@ -0,0 +1,11 @@
+// Generated by scripts/generate-db-types.ts from the migrated schema. Do not edit by hand:
+// change the migration, then run `npm run gen:db-types`.
+import type { InsertOptional } from './columns';
+
+export interface MailIngestPrefsTable {
+  id: InsertOptional<number>;
+  user_id: number;
+  enabled: InsertOptional<number>;
+  ai_enabled: InsertOptional<number>;
+  auto_add: InsertOptional<number>;
+}

server/src/db/kysely/db.ts

@@ -62,6 +62,11 @@ import type { JourneyRecordingPrepTable } from './JourneyRecordingPrep.table';
 import type { JourneyShareTokensTable } from './JourneyShareTokens.table';
 import type { JourneyTripsTable } from './JourneyTrips.table';
 import type { JourneysTable } from './Journeys.table';
+import type { MailIngestAddressesTable } from './MailIngestAddresses.table';
+import type { MailIngestCandidatesTable } from './MailIngestCandidates.table';
+import type { MailIngestCursorTable } from './MailIngestCursor.table';
+import type { MailIngestMessagesTable } from './MailIngestMessages.table';
+import type { MailIngestPrefsTable } from './MailIngestPrefs.table';
 import type { McpTokensTable } from './McpTokens.table';
 import type { MigrationsTable } from './Migrations.table';
 import type { NotificationChannelPreferencesTable } from './NotificationChannelPreferences.table';
@@ -238,6 +243,11 @@ export interface DB {
   journey_share_tokens: JourneyShareTokensTable;
   journey_trips: JourneyTripsTable;
   journeys: JourneysTable;
+  mail_ingest_addresses: MailIngestAddressesTable;
+  mail_ingest_candidates: MailIngestCandidatesTable;
+  mail_ingest_cursor: MailIngestCursorTable;
+  mail_ingest_messages: MailIngestMessagesTable;
+  mail_ingest_prefs: MailIngestPrefsTable;
   mcp_tokens: McpTokensTable;
   migrations: MigrationsTable;
   notification_channel_preferences: NotificationChannelPreferencesTable;

server/src/db/migrations/Migration20200101046410_a_forwarded_booking_mail_lands_in_its_trip.ts

@@ -0,0 +1,163 @@
+import { Migration } from '@mikro-orm/migrations';
+
+/**
+ * A person forwards a booking mail to an address of their own, and TREK adds
+ * the booking to the matching trip, or keeps it in an inbox with a reason.
+ *
+ * `mail_ingest_addresses` holds one row per forwarding address: the person's
+ * own (`trip_id` null) or one for a single trip, which belongs to the member
+ * who made it (`user_id`). The token is stored twice: its sha256 for the
+ * lookup, and encrypted so its owner can see the address again. A renewed or
+ * revoked address keeps its row with `revoked_at` set, so a mail to it is
+ * recognised and refused.
+ *
+ * `mail_ingest_prefs` holds a person's switches (import, AI reading, adding
+ * on its own or asking first); without a row every switch is on.
+ *
+ * `mail_ingest_cursor` is the position and the lease of a reader: an IMAP box
+ * by (uid_validity, last_uid), and the processing pipeline itself, so two
+ * processes on one database never read the same mail at once.
+ *
+ * `mail_ingest_messages` holds each mail as it arrived, encrypted
+ * (`raw_eml_enc`) until the retention job purges it, with what the
+ * pipeline made of it. `dedupe_hash` keeps a mail delivered twice from
+ * being read twice.
+ *
+ * `mail_ingest_candidates` holds each booking read from a mail, the server's
+ * checks on it, the trip it went to or was suggested for, and what became of
+ * it: the ids it created (for the undo, open until `undo_until`), or why it
+ * waits in the inbox.
+ *
+ * Every row goes with its person; a trip that goes clears its references, so
+ * the inbox entry stays and can be filed elsewhere. Every foreign key leads
+ * an index.
+ */
+export class Migration20200101046410_a_forwarded_booking_mail_lands_in_its_trip extends Migration {
+  override name = 'Migration20200101046410_a_forwarded_booking_mail_lands_in_its_trip';
+
+  override up(): void {
+    this.addSql(`
+      CREATE TABLE IF NOT EXISTS mail_ingest_addresses (
+        id INTEGER PRIMARY KEY,
+        token_hash TEXT NOT NULL,
+        token_enc TEXT NOT NULL,
+        user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+        trip_id INTEGER REFERENCES trips(id) ON DELETE CASCADE,
+        created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+        revoked_at DATETIME,
+        last_used_at DATETIME,
+        UNIQUE(token_hash)
+      )
+    `);
+    this.addSql('CREATE INDEX IF NOT EXISTS idx_mail_ingest_addresses_user ON mail_ingest_addresses(user_id, trip_id)');
+    this.addSql('CREATE INDEX IF NOT EXISTS idx_mail_ingest_addresses_trip ON mail_ingest_addresses(trip_id)');
+
+    this.addSql(`
+      CREATE TABLE IF NOT EXISTS mail_ingest_prefs (
+        id INTEGER PRIMARY KEY,
+        user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+        enabled INTEGER NOT NULL DEFAULT 1,
+        ai_enabled INTEGER NOT NULL DEFAULT 1,
+        auto_add INTEGER NOT NULL DEFAULT 1,
+        UNIQUE(user_id)
+      )
+    `);
+
+    this.addSql(`
+      CREATE TABLE IF NOT EXISTS mail_ingest_cursor (
+        id INTEGER PRIMARY KEY,
+        transport TEXT NOT NULL CHECK(transport IN ('imap', 'cloud', 'pipeline')),
+        mailbox TEXT NOT NULL,
+        uid_validity INTEGER,
+        last_uid INTEGER,
+        last_ok_at DATETIME,
+        last_error TEXT,
+        lease_owner TEXT,
+        lease_until DATETIME,
+        UNIQUE(transport, mailbox)
+      )
+    `);
+
+    this.addSql(`
+      CREATE TABLE IF NOT EXISTS mail_ingest_messages (
+        id INTEGER PRIMARY KEY,
+        transport TEXT NOT NULL CHECK(transport IN ('imap', 'webhook', 'cloud', 'upload')),
+        message_id_hdr TEXT,
+        dedupe_hash TEXT NOT NULL,
+        address_id INTEGER REFERENCES mail_ingest_addresses(id) ON DELETE SET NULL,
+        user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+        trip_id INTEGER REFERENCES trips(id) ON DELETE SET NULL,
+        from_addr TEXT,
+        from_verified INTEGER NOT NULL DEFAULT 0,
+        auth_results TEXT,
+        subject TEXT,
+        received_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+        size INTEGER NOT NULL,
+        raw_eml_enc TEXT,
+        raw_purged_at DATETIME,
+        status TEXT NOT NULL DEFAULT 'queued'
+          CHECK(status IN ('queued', 'processing', 'processed', 'discarded', 'failed')),
+        reason TEXT,
+        hints TEXT,
+        attachments TEXT,
+        forwarding_code TEXT,
+        attempts INTEGER NOT NULL DEFAULT 0,
+        next_attempt_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+        lease_until DATETIME,
+        ai_used INTEGER NOT NULL DEFAULT 0,
+        processed_at DATETIME,
+        UNIQUE(dedupe_hash)
+      )
+    `);
+    this.addSql(
+      'CREATE INDEX IF NOT EXISTS idx_mail_ingest_messages_due ON mail_ingest_messages(status, next_attempt_at)',
+    );
+    this.addSql(
+      'CREATE INDEX IF NOT EXISTS idx_mail_ingest_messages_user ON mail_ingest_messages(user_id, received_at)',
+    );
+    this.addSql(
+      'CREATE INDEX IF NOT EXISTS idx_mail_ingest_messages_address ON mail_ingest_messages(address_id, received_at)',
+    );
+    this.addSql('CREATE INDEX IF NOT EXISTS idx_mail_ingest_messages_trip ON mail_ingest_messages(trip_id)');
+
+    this.addSql(`
+      CREATE TABLE IF NOT EXISTS mail_ingest_candidates (
+        id INTEGER PRIMARY KEY,
+        message_id INTEGER NOT NULL REFERENCES mail_ingest_messages(id) ON DELETE CASCADE,
+        user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+        position INTEGER NOT NULL DEFAULT 0,
+        item_json TEXT,
+        source TEXT CHECK(source IN ('ai', 'kitinerary', 'json_ld', 'ics', 'pkpass', 'text')),
+        checks_json TEXT,
+        uncertain_fields TEXT,
+        match_json TEXT,
+        suggested_trip_id INTEGER REFERENCES trips(id) ON DELETE SET NULL,
+        dedupe_key TEXT,
+        content_hash TEXT,
+        resolution TEXT NOT NULL DEFAULT 'pending'
+          CHECK(resolution IN ('pending', 'auto_applied', 'applied', 'filed', 'discarded', 'duplicate', 'conflict',
+            'undone')),
+        reason TEXT,
+        resolved_trip_id INTEGER REFERENCES trips(id) ON DELETE SET NULL,
+        created_ids_json TEXT,
+        attach_error TEXT,
+        resolved_by INTEGER,
+        resolved_at DATETIME,
+        undo_until DATETIME,
+        created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
+      )
+    `);
+    this.addSql(
+      'CREATE INDEX IF NOT EXISTS idx_mail_ingest_candidates_message ON mail_ingest_candidates(message_id, position)',
+    );
+    this.addSql(
+      'CREATE INDEX IF NOT EXISTS idx_mail_ingest_candidates_user ON mail_ingest_candidates(user_id, resolution)',
+    );
+    this.addSql(
+      'CREATE INDEX IF NOT EXISTS idx_mail_ingest_candidates_suggested ON mail_ingest_candidates(suggested_trip_id)',
+    );
+    this.addSql(
+      'CREATE INDEX IF NOT EXISTS idx_mail_ingest_candidates_resolved ON mail_ingest_candidates(resolved_trip_id, content_hash)',
+    );
+  }
+}

server/src/db/repositories/MailIngestAddresses.repository.ts

@@ -0,0 +1,101 @@
+import type { MailIngestAddresses } from '../entities/MailIngestAddresses.entity';
+import type { DB } from '../kysely/db';
+import type { DbTimestamp } from '../types';
+import type { AssertRowKeys } from './_shared/rows';
+import { TrekRepository } from './_shared/trek-repository';
+
+/** One `mail_ingest_addresses` row as stored. */
+export interface MailIngestAddressRow {
+  id: number;
+  token_hash: string;
+  token_enc: string;
+  user_id: number;
+  trip_id: number | null;
+  created_at: string;
+  revoked_at: string | null;
+  last_used_at: string | null;
+}
+
+const _mailIngestAddressRowKeys: AssertRowKeys<MailIngestAddressRow, MailIngestAddresses> = true;
+
+/** A new address: whose it is, for which trip (null for the person's own), and its token in both forms. */
+interface MailIngestAddressInsert {
+  userId: number;
+  tripId: number | null;
+  tokenHash: string;
+  tokenEnc: string;
+  createdAt: DbTimestamp;
+}
+
+type AddressesKyselyDB = Pick<DB, 'mail_ingest_addresses'>;
+
+/**
+ * `mail_ingest_addresses`: the forwarding addresses. A person has at most one
+ * live address of their own and one per trip; renewing one revokes the old row
+ * instead of deleting it, so a mail that still reaches the old token is
+ * recognised and refused rather than taken for a stranger's.
+ */
+export class MailIngestAddressesRepository extends TrekRepository<MailIngestAddresses> {
+  private db() {
+    return this.kysely<AddressesKyselyDB>();
+  }
+
+  async insertAddress(row: MailIngestAddressInsert): Promise<number> {
+    const inserted = await this.db()
+      .insertInto('mail_ingest_addresses')
+      .values({
+        user_id: row.userId,
+        trip_id: row.tripId,
+        token_hash: row.tokenHash,
+        token_enc: row.tokenEnc,
+        created_at: row.createdAt,
+      })
+      .returning('id')
+      .executeTakeFirstOrThrow();
+    return inserted.id;
+  }
+
+  /** The person's live address for the trip, or their own one with `tripId` null. */
+  async findLive(userId: number, tripId: number | null): Promise<MailIngestAddressRow | null> {
+    let query = this.db()
+      .selectFrom('mail_ingest_addresses')
+      .selectAll()
+      .where('user_id', '=', userId)
+      .where('revoked_at', 'is', null);
+    query = tripId === null ? query.where('trip_id', 'is', null) : query.where('trip_id', '=', tripId);
+    const row = await query.orderBy('id', 'desc').executeTakeFirst();
+    return row ?? null;
+  }
+
+  /** The address a token names, revoked or not. */
+  async findByTokenHash(tokenHash: string): Promise<MailIngestAddressRow | null> {
+    const row = await this.db()
+      .selectFrom('mail_ingest_addresses')
+      .selectAll()
+      .where('token_hash', '=', tokenHash)
+      .executeTakeFirst();
+    return row ?? null;
+  }
+
+  async findById(id: number): Promise<MailIngestAddressRow | null> {
+    const row = await this.db().selectFrom('mail_ingest_addresses').selectAll().where('id', '=', id).executeTakeFirst();
+    return row ?? null;
+  }
+
+  /** Revokes the person's live address for the trip (their own with `tripId` null); answers how many went. */
+  async revokeLive(userId: number, tripId: number | null, at: DbTimestamp): Promise<number> {
+    let query = this.db()
+      .updateTable('mail_ingest_addresses')
+      .set({ revoked_at: at })
+      .where('user_id', '=', userId)
+      .where('revoked_at', 'is', null);
+    query = tripId === null ? query.where('trip_id', 'is', null) : query.where('trip_id', '=', tripId);
+    const result = await query.executeTakeFirst();
+    return Number(result.numUpdatedRows);
+  }
+
+  /** Notes that a mail came in on the address. */
+  async touchUsed(id: number, at: DbTimestamp): Promise<void> {
+    await this.db().updateTable('mail_ingest_addresses').set({ last_used_at: at }).where('id', '=', id).execute();
+  }
+}

server/src/db/repositories/MailIngestCandidates.repository.ts

@@ -0,0 +1,351 @@
+import type { MailIngestCandidates } from '../entities/MailIngestCandidates.entity';
+import type { DB } from '../kysely/db';
+import type { DbTimestamp } from '../types';
+import type { AssertRowKeys } from './_shared/rows';
+import { TrekRepository } from './_shared/trek-repository';
+
+import type { Expression, SqlBool } from 'kysely';
+
+/** One `mail_ingest_candidates` row as stored: the JSON columns as text. */
+export interface MailIngestCandidateRow {
+  id: number;
+  message_id: number;
+  user_id: number;
+  position: number;
+  item_json: string | null;
+  source: string | null;
+  checks_json: string | null;
+  uncertain_fields: string | null;
+  match_json: string | null;
+  suggested_trip_id: number | null;
+  dedupe_key: string | null;
+  content_hash: string | null;
+  resolution: string;
+  reason: string | null;
+  resolved_trip_id: number | null;
+  created_ids_json: string | null;
+  attach_error: string | null;
+  resolved_by: number | null;
+  resolved_at: string | null;
+  undo_until: string | null;
+  created_at: string;
+}
+
+const _mailIngestCandidateRowKeys: AssertRowKeys<MailIngestCandidateRow, MailIngestCandidates> = true;
+
+/** A booking read from a mail, as the pipeline writes it: waiting, or already decided in the same write. */
+export interface MailIngestCandidateInsert {
+  messageId: number;
+  userId: number;
+  position: number;
+  itemJson: string | null;
+  source: string | null;
+  checksJson: string | null;
+  uncertainFields: string | null;
+  matchJson: string | null;
+  suggestedTripId: number | null;
+  dedupeKey: string | null;
+  contentHash: string | null;
+  resolution: string;
+  reason: string | null;
+  createdAt: DbTimestamp;
+}
+
+/** What deciding an entry writes. */
+export interface MailIngestResolutionPatch {
+  resolution: string;
+  reason: string | null;
+  resolvedTripId: number | null;
+  createdIdsJson: string | null;
+  resolvedBy: number | null;
+  resolvedAt: DbTimestamp;
+  undoUntil: DbTimestamp | null;
+}
+
+/** A trip a booking from a mail may go to, as the import weighs it. */
+export interface MailIngestTripRow {
+  id: number;
+  title: string;
+  start_date: string | null;
+  end_date: string | null;
+  /** The owner, for the permission check. */
+  user_id: number;
+  currency: string | null;
+}
+
+/** A place of a trip with a position, for naming its destinations and checking a booking lies near them. */
+export interface MailIngestTripPlaceRow {
+  trip_id: number;
+  name: string;
+  address: string | null;
+  lat: number;
+  lng: number;
+}
+
+/** The person a mail is for, as the pipeline needs them. */
+export interface MailIngestPersonRow {
+  id: number;
+  username: string;
+  email: string;
+  role: string;
+}
+
+/** What tells one booking from another already in a trip: its type, its start window and its reference. */
+export interface MailIngestSimilarProbe {
+  type: string;
+  startFrom: string | null;
+  startTo: string | null;
+  confirmation: string | null;
+}
+
+type CandidatesKyselyDB = Pick<
+  DB,
+  | 'mail_ingest_candidates'
+  | 'trips'
+  | 'trip_members'
+  | 'days'
+  | 'places'
+  | 'users'
+  | 'reservations'
+  | 'budget_item_payers'
+>;
+
+/** The most places a trip lends its destinations and its geography to the import. */
+const PLACES_PER_LOOKUP = 2000;
+
+/**
+ * `mail_ingest_candidates`: the bookings read from forwarded mails, and the
+ * reads the import makes of the trips they may go to. Those reads touch the
+ * trips, days, places, users and reservations tables directly, by column, so
+ * the import weighs a booking without reaching into five other domains for
+ * one projection each; every write it makes to a trip goes through the
+ * owning services.
+ */
+export class MailIngestCandidatesRepository extends TrekRepository<MailIngestCandidates> {
+  private db() {
+    return this.kysely<CandidatesKyselyDB>();
+  }
+
+  async insertCandidate(row: MailIngestCandidateInsert): Promise<number> {
+    const inserted = await this.db()
+      .insertInto('mail_ingest_candidates')
+      .values({
+        message_id: row.messageId,
+        user_id: row.userId,
+        position: row.position,
+        item_json: row.itemJson,
+        source: row.source,
+        checks_json: row.checksJson,
+        uncertain_fields: row.uncertainFields,
+        match_json: row.matchJson,
+        suggested_trip_id: row.suggestedTripId,
+        dedupe_key: row.dedupeKey,
+        content_hash: row.contentHash,
+        resolution: row.resolution,
+        reason: row.reason,
+        created_at: row.createdAt,
+      })
+      .returning('id')
+      .executeTakeFirstOrThrow();
+    return inserted.id;
+  }
+
+  async findById(id: number): Promise<MailIngestCandidateRow | null> {
+    const row = await this.db().selectFrom('mail_ingest_candidates').selectAll().where('id', '=', id).executeTakeFirst();
+    return row ?? null;
+  }
+
+  /** The entries read from one mail, in the order the mail listed them. */
+  async listForMessage(messageId: number): Promise<MailIngestCandidateRow[]> {
+    return this.db()
+      .selectFrom('mail_ingest_candidates')
+      .selectAll()
+      .where('message_id', '=', messageId)
+      .orderBy('position', 'asc')
+      .orderBy('id', 'asc')
+      .execute();
+  }
+
+  /**
+   * Decides an entry, if it is still in one of the `from` states: one
+   * conditional update, so of two decisions at once only the first lands.
+   */
+  async resolve(id: number, from: readonly string[], patch: MailIngestResolutionPatch): Promise<boolean> {
+    const result = await this.db()
+      .updateTable('mail_ingest_candidates')
+      .set({
+        resolution: patch.resolution,
+        reason: patch.reason,
+        resolved_trip_id: patch.resolvedTripId,
+        created_ids_json: patch.createdIdsJson,
+        resolved_by: patch.resolvedBy,
+        resolved_at: patch.resolvedAt,
+        undo_until: patch.undoUntil,
+      })
+      .where('id', '=', id)
+      .where('resolution', 'in', [...from])
+      .executeTakeFirst();
+    return Number(result.numUpdatedRows) > 0;
+  }
+
+  /** Records the documents attached after the booking was written, or why that failed. */
+  async setAttachments(id: number, createdIdsJson: string, attachError: string | null): Promise<void> {
+    await this.db()
+      .updateTable('mail_ingest_candidates')
+      .set({ created_ids_json: createdIdsJson, attach_error: attachError })
+      .where('id', '=', id)
+      .execute();
+  }
+
+  /** The person's entries waiting in the inbox. */
+  async countPending(userId: number): Promise<number> {
+    const row = await this.db()
+      .selectFrom('mail_ingest_candidates')
+      .select((eb) => eb.fn.countAll<number>().as('n'))
+      .where('user_id', '=', userId)
+      .where('resolution', '=', 'pending')
+      .executeTakeFirstOrThrow();
+    return Number(row.n);
+  }
+
+  /** Whether a booking of this content was added to the trip from a mail and undone: it is not added on its own again. */
+  async wasUndoneInTrip(tripId: number, contentHash: string): Promise<boolean> {
+    const row = await this.db()
+      .selectFrom('mail_ingest_candidates')
+      .select('id')
+      .where('resolved_trip_id', '=', tripId)
+      .where('content_hash', '=', contentHash)
+      .where('resolution', '=', 'undone')
+      .executeTakeFirst();
+    return row !== undefined;
+  }
+
+  /** Retention: deletes the entries decided before `cutoff`; answers how many went. */
+  async deleteResolvedBefore(cutoff: DbTimestamp): Promise<number> {
+    const result = await this.db()
+      .deleteFrom('mail_ingest_candidates')
+      .where('resolution', '!=', 'pending')
+      .where('resolved_at', '<', cutoff)
+      .executeTakeFirst();
+    return Number(result.numDeletedRows);
+  }
+
+  // ── What the import reads of the trips ────────────────────────────────────
+
+  /**
+   * The trips the person owns or is a member of, not archived, that end on or
+   * after `endCutoff` or have no end date, most recent start first. Whether
+   * the person may add bookings to each is the service's check.
+   */
+  async listOpenTrips(userId: number, endCutoff: string, limit: number): Promise<MailIngestTripRow[]> {
+    return this.db()
+      .selectFrom('trips as t')
+      .select(['t.id', 't.title', 't.start_date', 't.end_date', 't.user_id', 't.currency'])
+      .where((eb) =>
+        eb.or([
+          eb('t.user_id', '=', userId),
+          eb.exists(
+            eb
+              .selectFrom('trip_members as m')
+              .select('m.id')
+              .whereRef('m.trip_id', '=', 't.id')
+              .where('m.user_id', '=', userId),
+          ),
+        ]),
+      )
+      .where((eb) => eb.or([eb('t.is_archived', 'is', null), eb('t.is_archived', '=', 0)]))
+      .where((eb) => eb.or([eb('t.end_date', 'is', null), eb('t.end_date', '>=', endCutoff)]))
+      .orderBy('t.start_date', 'desc')
+      .orderBy('t.id', 'desc')
+      .limit(limit)
+      .execute();
+  }
+
+  /** The dated days of these trips. */
+  async listDayDates(tripIds: readonly number[]): Promise<Array<{ trip_id: number; date: string }>> {
+    if (tripIds.length === 0) return [];
+    const rows = await this.db()
+      .selectFrom('days')
+      .select(['trip_id', 'date'])
+      .where('trip_id', 'in', [...tripIds])
+      .where('date', 'is not', null)
+      .orderBy('trip_id', 'asc')
+      .orderBy('date', 'asc')
+      .execute();
+    return rows.flatMap((row) => (row.date === null ? [] : [{ trip_id: row.trip_id, date: row.date }]));
+  }
+
+  /** The places of these trips that have a position, oldest first. */
+  async listPlacesWithPosition(tripIds: readonly number[]): Promise<MailIngestTripPlaceRow[]> {
+    if (tripIds.length === 0) return [];
+    const rows = await this.db()
+      .selectFrom('places')
+      .select(['trip_id', 'name', 'address', 'lat', 'lng'])
+      .where('trip_id', 'in', [...tripIds])
+      .where('lat', 'is not', null)
+      .where('lng', 'is not', null)
+      .orderBy('id', 'asc')
+      .limit(PLACES_PER_LOOKUP)
+      .execute();
+    return rows.flatMap((row) =>
+      row.lat === null || row.lng === null ? [] : [{ ...row, lat: row.lat, lng: row.lng }],
+    );
+  }
+
+  async findPerson(userId: number): Promise<MailIngestPersonRow | null> {
+    const row = await this.db()
+      .selectFrom('users')
+      .select(['id', 'username', 'email', 'role'])
+      .where('id', '=', userId)
+      .executeTakeFirst();
+    return row ?? null;
+  }
+
+  /** Whether the trip holds a booking an import stored under one of these sources and this key. */
+  async tripHasExternal(tripId: number, sources: readonly string[], externalId: string): Promise<boolean> {
+    const row = await this.db()
+      .selectFrom('reservations')
+      .select('id')
+      .where('trip_id', '=', tripId)
+      .where('external_source', 'in', [...sources])
+      .where('external_id', '=', externalId)
+      .executeTakeFirst();
+    return row !== undefined;
+  }
+
+  /**
+   * Whether the trip holds a booking that looks like this one: the same
+   * reference, or the same type starting in the same window.
+   */
+  async tripHasSimilar(tripId: number, probe: MailIngestSimilarProbe): Promise<boolean> {
+    const { type, startFrom, startTo, confirmation } = probe;
+    const row = await this.db()
+      .selectFrom('reservations')
+      .select('id')
+      .where('trip_id', '=', tripId)
+      .where((eb) => {
+        const either: Expression<SqlBool>[] = [];
+        if (confirmation) either.push(eb('confirmation_number', '=', confirmation));
+        if (startFrom && startTo) {
+          either.push(
+            eb.and([eb('type', '=', type), eb('reservation_time', '>=', startFrom), eb('reservation_time', '<=', startTo)]),
+          );
+        }
+        // An empty list reads as false: a probe with neither names nothing.
+        return eb.or(either);
+      })
+      .executeTakeFirst();
+    return row !== undefined;
+  }
+
+  /** How many payers the costs name: a cost someone settled with moves balances. */
+  async countPayers(budgetItemIds: readonly number[]): Promise<number> {
+    if (budgetItemIds.length === 0) return 0;
+    const row = await this.db()
+      .selectFrom('budget_item_payers')
+      .select((eb) => eb.fn.countAll<number>().as('n'))
+      .where('budget_item_id', 'in', [...budgetItemIds])
+      .executeTakeFirstOrThrow();
+    return Number(row.n);
+  }
+}

server/src/db/repositories/MailIngestCursor.repository.ts

@@ -0,0 +1,124 @@
+import type { MailIngestCursor } from '../entities/MailIngestCursor.entity';
+import type { DB } from '../kysely/db';
+import type { DbTimestamp } from '../types';
+import type { AssertRowKeys } from './_shared/rows';
+import { TrekRepository } from './_shared/trek-repository';
+
+/** What reads mail: an IMAP box, the TREK Cloud relay, or the processing pipeline itself. */
+export type MailIngestCursorTransport = 'imap' | 'cloud' | 'pipeline';
+
+/** One `mail_ingest_cursor` row as stored. */
+interface MailIngestCursorRow {
+  id: number;
+  transport: string;
+  mailbox: string;
+  uid_validity: number | null;
+  last_uid: number | null;
+  last_ok_at: string | null;
+  last_error: string | null;
+  lease_owner: string | null;
+  lease_until: string | null;
+}
+
+const _mailIngestCursorRowKeys: AssertRowKeys<MailIngestCursorRow, MailIngestCursor> = true;
+
+/** A reader's position after a good run; an IMAP reader names its box's uid validity and the last uid it took. */
+interface MailIngestCursorPosition {
+  uidValidity?: number | null;
+  lastUid?: number | null;
+}
+
+type CursorKyselyDB = Pick<DB, 'mail_ingest_cursor'>;
+
+/**
+ * `mail_ingest_cursor`: where each reader stands and who holds it. The lease
+ * is one conditional update, so of two processes on one database exactly one
+ * wins; a lease whose holder died runs out and the next run takes it.
+ */
+export class MailIngestCursorRepository extends TrekRepository<MailIngestCursor> {
+  private db() {
+    return this.kysely<CursorKyselyDB>();
+  }
+
+  /**
+   * Takes the reader's lease until `until` when it is free, run out at `now`,
+   * or already this owner's. The row is created on first use.
+   *
+   * @txIndependent the row is provisioned idempotently first; the lease itself
+   * is the one conditional update that decides the race.
+   */
+  async acquireLease(
+    transport: MailIngestCursorTransport,
+    mailbox: string,
+    owner: string,
+    until: DbTimestamp,
+    now: DbTimestamp,
+  ): Promise<boolean> {
+    await this.db()
+      .insertInto('mail_ingest_cursor')
+      .values({ transport, mailbox })
+      .onConflict((oc) => oc.columns(['transport', 'mailbox']).doNothing())
+      .execute();
+    const result = await this.db()
+      .updateTable('mail_ingest_cursor')
+      .set({ lease_owner: owner, lease_until: until })
+      .where('transport', '=', transport)
+      .where('mailbox', '=', mailbox)
+      .where((eb) =>
+        eb.or([eb('lease_until', 'is', null), eb('lease_until', '<', now), eb('lease_owner', '=', owner)]),
+      )
+      .executeTakeFirst();
+    return Number(result.numUpdatedRows) > 0;
+  }
+
+  /** Gives the lease back, if this owner still holds it. */
+  async releaseLease(transport: MailIngestCursorTransport, mailbox: string, owner: string): Promise<void> {
+    await this.db()
+      .updateTable('mail_ingest_cursor')
+      .set({ lease_owner: null, lease_until: null })
+      .where('transport', '=', transport)
+      .where('mailbox', '=', mailbox)
+      .where('lease_owner', '=', owner)
+      .execute();
+  }
+
+  /** A good run: its time, the reader's position where it has one, and no error. */
+  async recordOk(
+    transport: MailIngestCursorTransport,
+    mailbox: string,
+    at: DbTimestamp,
+    position: MailIngestCursorPosition = {},
+  ): Promise<void> {
+    await this.db()
+      .updateTable('mail_ingest_cursor')
+      .set({
+        last_ok_at: at,
+        last_error: null,
+        ...(position.uidValidity !== undefined && { uid_validity: position.uidValidity }),
+        ...(position.lastUid !== undefined && { last_uid: position.lastUid }),
+      })
+      .where('transport', '=', transport)
+      .where('mailbox', '=', mailbox)
+      .execute();
+  }
+
+  /** A failed run, for the admin's status card; the position stays where the last good run left it. */
+  async recordError(transport: MailIngestCursorTransport, mailbox: string, error: string): Promise<void> {
+    await this.db()
+      .updateTable('mail_ingest_cursor')
+      .set({ last_error: error.slice(0, 500) })
+      .where('transport', '=', transport)
+      .where('mailbox', '=', mailbox)
+      .execute();
+  }
+
+  async find(transport: MailIngestCursorTransport, mailbox: string): Promise<MailIngestCursorRow | null> {
+    const row = await this.db()
+      .selectFrom('mail_ingest_cursor')
+      .selectAll()
+      .where('transport', '=', transport)
+      .where('mailbox', '=', mailbox)
+      .executeTakeFirst();
+    return row ?? null;
+  }
+}

server/src/db/repositories/MailIngestMessages.repository.ts

@@ -0,0 +1,263 @@
+import type { MailIngestMessages } from '../entities/MailIngestMessages.entity';
+import type { DB } from '../kysely/db';
+import type { DbTimestamp } from '../types';
+import type { AssertRowKeys } from './_shared/rows';
+import { TrekRepository } from './_shared/trek-repository';
+
+/** One `mail_ingest_messages` row as stored, the encrypted mail included. */
+export interface MailIngestMessageRow {
+  id: number;
+  transport: string;
+  message_id_hdr: string | null;
+  dedupe_hash: string;
+  address_id: number | null;
+  user_id: number;
+  trip_id: number | null;
+  from_addr: string | null;
+  from_verified: number;
+  auth_results: string | null;
+  subject: string | null;
+  received_at: string;
+  size: number;
+  raw_eml_enc: string | null;
+  raw_purged_at: string | null;
+  status: string;
+  reason: string | null;
+  hints: string | null;
+  attachments: string | null;
+  forwarding_code: string | null;
+  attempts: number;
+  next_attempt_at: string;
+  lease_until: string | null;
+  ai_used: number;
+  processed_at: string | null;
+}
+
+const _mailIngestMessageRowKeys: AssertRowKeys<MailIngestMessageRow, MailIngestMessages> = true;
+
+/** A mail as a transport hands it in, before anything has read it. */
+interface MailIngestMessageInsert {
+  transport: string;
+  messageIdHdr: string | null;
+  dedupeHash: string;
+  addressId: number;
+  userId: number;
+  tripId: number | null;
+  size: number;
+  rawEnc: string;
+  receivedAt: DbTimestamp;
+}
+
+/** What the pipeline made of a mail, written once it is done with it. */
+export interface MailIngestMessageOutcome {
+  status: 'processed' | 'discarded' | 'failed';
+  reason: string | null;
+  subject: string | null;
+  fromAddr: string | null;
+  fromVerified: boolean;
+  authResults: string | null;
+  hints: string | null;
+  attachments: string | null;
+  forwardingCode: string | null;
+  aiUsed: boolean;
+  processedAt: DbTimestamp;
+}
+
+type MessagesKyselyDB = Pick<DB, 'mail_ingest_messages' | 'mail_ingest_candidates'>;
+
+/**
+ * `mail_ingest_messages`: every forwarded mail, from its arrival to its
+ * purge. A transport only ever inserts (idempotently, by `dedupe_hash`); the
+ * pipeline claims due rows one by one with a conditional update, so a mail is
+ * read by one run at a time, and a run that died lets its claim run out.
+ */
+export class MailIngestMessagesRepository extends TrekRepository<MailIngestMessages> {
+  private db() {
+    return this.kysely<MessagesKyselyDB>();
+  }
+
+  /** Stores a new mail, queued for now; null when the same mail is in already. */
+  async insertIfNew(row: MailIngestMessageInsert): Promise<number | null> {
+    const inserted = await this.db()
+      .insertInto('mail_ingest_messages')
+      .values({
+        transport: row.transport,
+        message_id_hdr: row.messageIdHdr,
+        dedupe_hash: row.dedupeHash,
+        address_id: row.addressId,
+        user_id: row.userId,
+        trip_id: row.tripId,
+        size: row.size,
+        raw_eml_enc: row.rawEnc,
+        received_at: row.receivedAt,
+        next_attempt_at: row.receivedAt,
+      })
+      .onConflict((oc) => oc.column('dedupe_hash').doNothing())
+      .returning('id')
+      .executeTakeFirst();
+    return inserted?.id ?? null;
+  }
+
+  async findIdByDedupeHash(dedupeHash: string): Promise<number | null> {
+    const row = await this.db()
+      .selectFrom('mail_ingest_messages')
+      .select('id')
+      .where('dedupe_hash', '=', dedupeHash)
+      .executeTakeFirst();
+    return row?.id ?? null;
+  }
+
+  async findById(id: number): Promise<MailIngestMessageRow | null> {
+    const row = await this.db().selectFrom('mail_ingest_messages').selectAll().where('id', '=', id).executeTakeFirst();
+    return row ?? null;
+  }
+
+  /** How many mails came in on the address since `since`, for its daily limit. */
+  async countForAddressSince(addressId: number, since: DbTimestamp): Promise<number> {
+    const row = await this.db()
+      .selectFrom('mail_ingest_messages')
+      .select((eb) => eb.fn.countAll<number>().as('n'))
+      .where('address_id', '=', addressId)
+      .where('received_at', '>=', since)
+      .executeTakeFirstOrThrow();
+    return Number(row.n);
+  }
+
+  /** How many of the person's mails an AI read since `since`, for their daily limit. */
+  async countAiReadSince(userId: number, since: DbTimestamp): Promise<number> {
+    const row = await this.db()
+      .selectFrom('mail_ingest_messages')
+      .select((eb) => eb.fn.countAll<number>().as('n'))
+      .where('user_id', '=', userId)
+      .where('ai_used', '=', 1)
+      .where('received_at', '>=', since)
+      .executeTakeFirstOrThrow();
+    return Number(row.n);
+  }
+
+  /**
+   * Claims up to `limit` mails that are due: queued and past their next try,
+   * or claimed by a run whose lease ran out. Each claim is one conditional
+   * update, so a mail another run took in between is skipped, never read twice.
+   *
+   * @txIndependent one claim per mail on purpose: each is complete on its own,
+   * and a mail claimed stays claimed even when the next one is taken first by
+   * another run.
+   */
+  async claimDue(now: DbTimestamp, leaseUntil: DbTimestamp, limit: number): Promise<number[]> {
+    const due = await this.db()
+      .selectFrom('mail_ingest_messages')
+      .select('id')
+      .where((eb) =>
+        eb.or([
+          eb.and([eb('status', '=', 'queued'), eb('next_attempt_at', '<=', now)]),
+          eb.and([eb('status', '=', 'processing'), eb('lease_until', '<', now)]),
+        ]),
+      )
+      .orderBy('next_attempt_at', 'asc')
+      .orderBy('id', 'asc')
+      .limit(limit)
+      .execute();
+    const claimed: number[] = [];
+    for (const { id } of due) {
+      const result = await this.db()
+        .updateTable('mail_ingest_messages')
+        .set((eb) => ({ status: 'processing', lease_until: leaseUntil, attempts: eb('attempts', '+', 1) }))
+        .where('id', '=', id)
+        .where((eb) =>
+          eb.or([
+            eb('status', '=', 'queued'),
+            eb.and([eb('status', '=', 'processing'), eb('lease_until', '<', now)]),
+          ]),
+        )
+        .executeTakeFirst();
+      if (Number(result.numUpdatedRows) > 0) claimed.push(id);
+    }
+    return claimed;
+  }
+
+  /** Writes what the pipeline made of the mail and lets go of its claim. */
+  async finish(id: number, outcome: MailIngestMessageOutcome): Promise<void> {
+    await this.db()
+      .updateTable('mail_ingest_messages')
+      .set({
+        status: outcome.status,
+        reason: outcome.reason,
+        subject: outcome.subject,
+        from_addr: outcome.fromAddr,
+        from_verified: outcome.fromVerified ? 1 : 0,
+        auth_results: outcome.authResults,
+        hints: outcome.hints,
+        attachments: outcome.attachments,
+        forwarding_code: outcome.forwardingCode,
+        ai_used: outcome.aiUsed ? 1 : 0,
+        processed_at: outcome.processedAt,
+        lease_until: null,
+      })
+      .where('id', '=', id)
+      .execute();
+  }
+
+  /** Puts a mail whose reading failed back in the queue, to be tried again at `nextAttemptAt`. */
+  async retryLater(id: number, nextAttemptAt: DbTimestamp): Promise<void> {
+    await this.db()
+      .updateTable('mail_ingest_messages')
+      .set({ status: 'queued', next_attempt_at: nextAttemptAt, lease_until: null })
+      .where('id', '=', id)
+      .execute();
+  }
+
+  /**
+   * Drops the stored mail of every message that no longer needs it: one that
+   * came in before `hardCutoff`, or one done before `doneCutoff` that has no
+   * entry left waiting in the inbox (the ways out read the mail). Answers how
+   * many were purged.
+   */
+  async purgeRaw(doneCutoff: DbTimestamp, hardCutoff: DbTimestamp, at: DbTimestamp): Promise<number> {
+    const result = await this.db()
+      .updateTable('mail_ingest_messages')
+      .set({ raw_eml_enc: null, raw_purged_at: at })
+      .where('raw_eml_enc', 'is not', null)
+      .where((eb) =>
+        eb.or([
+          eb('received_at', '<', hardCutoff),
+          eb.and([
+            eb('status', 'in', ['processed', 'discarded', 'failed']),
+            eb('processed_at', '<', doneCutoff),
+            eb.not(
+              eb.exists(
+                eb
+                  .selectFrom('mail_ingest_candidates as c')
+                  .select('c.id')
+                  .whereRef('c.message_id', '=', 'mail_ingest_messages.id')
+                  .where('c.resolution', '=', 'pending'),
+              ),
+            ),
+          ]),
+        ]),
+      )
+      .executeTakeFirst();
+    return Number(result.numUpdatedRows);
+  }
+
+  /** Deletes the mails that came in before `cutoff` and have no inbox entry left; answers how many went. */
+  async deleteSettledBefore(cutoff: DbTimestamp): Promise<number> {
+    const result = await this.db()
+      .deleteFrom('mail_ingest_messages')
+      .where('received_at', '<', cutoff)
+      .where('status', '!=', 'queued')
+      .where('status', '!=', 'processing')
+      .where((eb) =>
+        eb.not(
+          eb.exists(
+            eb
+              .selectFrom('mail_ingest_candidates as c')
+              .select('c.id')
+              .whereRef('c.message_id', '=', 'mail_ingest_messages.id'),
+          ),
+        ),
+      )
+      .executeTakeFirst();
+    return Number(result.numDeletedRows);
+  }
+}

server/src/db/repositories/MailIngestPrefs.repository.ts

@@ -0,0 +1,46 @@
+import type { MailIngestPrefs } from '../entities/MailIngestPrefs.entity';
+import type { DB } from '../kysely/db';
+import type { AssertRowKeys } from './_shared/rows';
+import { TrekRepository } from './_shared/trek-repository';
+
+/** One `mail_ingest_prefs` row as stored: 0/1 flags. */
+export interface MailIngestPrefsRow {
+  id: number;
+  user_id: number;
+  enabled: number;
+  ai_enabled: number;
+  auto_add: number;
+}
+
+const _mailIngestPrefsRowKeys: AssertRowKeys<MailIngestPrefsRow, MailIngestPrefs> = true;
+
+/** The switches a write names; the others keep their value. */
+export type MailIngestPrefsPatch = Partial<Pick<MailIngestPrefsRow, 'enabled' | 'ai_enabled' | 'auto_add'>>;
+
+type PrefsKyselyDB = Pick<DB, 'mail_ingest_prefs'>;
+
+/**
+ * `mail_ingest_prefs`: a person's switches for the mail import. A person
+ * without a row has every switch on, the column defaults, so only a change
+ * writes one.
+ */
+export class MailIngestPrefsRepository extends TrekRepository<MailIngestPrefs> {
+  private db() {
+    return this.kysely<PrefsKyselyDB>();
+  }
+
+  async findForUser(userId: number): Promise<MailIngestPrefsRow | null> {
+    const row = await this.db().selectFrom('mail_ingest_prefs').selectAll().where('user_id', '=', userId).executeTakeFirst();
+    return row ?? null;
+  }
+
+  /** Writes the named switches, creating the person's row with the defaults for the rest. */
+  async upsertForUser(userId: number, patch: MailIngestPrefsPatch): Promise<void> {
+    if (Object.keys(patch).length === 0) return;
+    await this.db()
+      .insertInto('mail_ingest_prefs')
+      .values({ user_id: userId, ...patch })
+      .onConflict((oc) => oc.column('user_id').doUpdateSet(patch))
+      .execute();
+  }
+}

server/tests/unit/db/entities-index.test.ts

@@ -57,14 +57,16 @@ describe('ALL_ENTITIES', () => {
    * + trek_photo_signals, what a photo's pixels say, for picking photos for a book
    * + journey_book_revisions, the books a whole-book design replaced, for its undo
    * + trip_member_removals, who another member took off a trip, so its invite link refuses them
-   * + assistant_attachments, the files attached in the trip chat, kept with their conversation)
+   * + assistant_attachments, the files attached in the trip chat, kept with their conversation
+   * + mail_ingest_addresses, _prefs, _cursor, _messages and _candidates, the forwarded
+   *   booking mails, the bookings read from them and the switches and addresses behind them)
    * is registered, and no two share a class name (a silent duplicate would
    * shadow one of them in `orm.getMetadata()`, exactly the kind of drift
    * `entity-schema-parity.test.ts` can't see — it iterates the metadata map,
    * which de-duplicates by class name).
    */
-  it('ENT-003: exactly 168 entities, every class name unique', () => {
-    expect(ALL_ENTITIES.length).toBe(168);
+  it('ENT-003: exactly 173 entities, every class name unique', () => {
+    expect(ALL_ENTITIES.length).toBe(173);
     const names = ALL_ENTITIES.map((s) => s.meta.className);
     expect(new Set(names).size).toBe(names.length);
   });

server/tests/unit/db/entity-class-fields.test.ts

@@ -63,7 +63,10 @@ import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest';
 // + storage_release_queue: category, key, due_at (a queued release is only ever written whole, by the release queue).
 // + flight_live_states: callsign, service_date, origin, destination (a flight's state is always found and
 //   written by its key).
-const KNOWN_UNINITIALISED_NOT_NULL_SCALARS = 615;
+// + mail_ingest_addresses: id, token_hash, token_enc, user_id; mail_ingest_prefs: id, user_id; mail_ingest_cursor:
+//   id, transport, mailbox; mail_ingest_messages: id, transport, dedupe_hash, user_id, size; mail_ingest_candidates:
+//   id, message_id, user_id (each row is written whole by the mail import, which always knows all of them).
+const KNOWN_UNINITIALISED_NOT_NULL_SCALARS = 632;
 
 describe('entity class fields never carry a NaN initialiser (task-4-review-shape.md, Important 1)', () => {
   it('CLASSFIELD-001: new X() has no own property whose value is NaN, across all 133 entities', () => {

server/tests/unit/db/legacy-baseline.test.ts

@@ -108,6 +108,7 @@ const UNNUMBERED = [
   'Migration20200101046380_ireland_gained_its_counties',
   'Migration20200101046390_a_trip_can_keep_its_costs_as_totals_only',
   'Migration20200101046400_a_booked_flight_shows_where_it_is',
+  'Migration20200101046410_a_forwarded_booking_mail_lands_in_its_trip',
 ];
 // The last step the positional runner ever had; planLegacyBaseline refuses a
 // schema_version past it.

server/tests/unit/db/repositories/MailIngestAddresses.repository.test.ts

@@ -0,0 +1,84 @@
+/**
+ * The forwarding addresses through their repository: a person's own address
+ * and a trip's are told apart, renewing revokes instead of deleting so the old
+ * token is still recognised, and a mail notes the address it came in on.
+ */
+import { MailIngestAddresses } from '../../../../src/db/entities/MailIngestAddresses.entity';
+import type { MailIngestAddressesRepository } from '../../../../src/db/repositories/MailIngestAddresses.repository';
+import { dbNow } from '../../../../src/db/types';
+import { createSnapshotTestDb } from '../../../helpers/db-mock';
+import { makeTrip } from '../../../helpers/factories/trips';
+import { makeUser } from '../../../helpers/factories/users';
+import { resetTestDb } from '../../../helpers/test-db';
+import { createTestOrm, type TestOrm } from '../../../helpers/test-orm';
+
+import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest';
+
+const testDb = createSnapshotTestDb();
+let t: TestOrm;
+let addresses: MailIngestAddressesRepository;
+
+beforeAll(async () => {
+  t = await createTestOrm(testDb);
+  addresses = t.repo(MailIngestAddresses);
+});
+beforeEach(() => {
+  resetTestDb(testDb);
+  t.clear();
+});
+afterAll(async () => {
+  await t.close();
+  testDb.close();
+});
+
+const AT = dbNow(new Date('2026-11-01T10:00:00Z'));
+const LATER = dbNow(new Date('2026-11-02T10:00:00Z'));
+
+describe('MailIngestAddressesRepository', () => {
+  it('keeps a person’s own address apart from a trip’s', async () => {
+    const { user } = await makeUser(t);
+    const trip = await makeTrip(t, user.id);
+    const own = await addresses.insertAddress({
+      userId: user.id,
+      tripId: null,
+      tokenHash: 'h-own',
+      tokenEnc: 'e-own',
+      createdAt: AT,
+    });
+    const forTrip = await addresses.insertAddress({
+      userId: user.id,
+      tripId: trip.id,
+      tokenHash: 'h-trip',
+      tokenEnc: 'e-trip',
+      createdAt: AT,
+    });
+
+    expect((await addresses.findLive(user.id, null))?.id).toBe(own);
+    expect((await addresses.findLive(user.id, trip.id))?.id).toBe(forTrip);
+    expect(await addresses.findById(forTrip)).toMatchObject({ trip_id: trip.id, token_enc: 'e-trip' });
+    expect(await addresses.findById(forTrip + 100)).toBeNull();
+  });
+
+  it('revokes the live address of one scope only, and still finds a revoked token', async () => {
+    const { user } = await makeUser(t);
+    const trip = await makeTrip(t, user.id);
+    const own = await addresses.insertAddress({ userId: user.id, tripId: null, tokenHash: 'a', tokenEnc: 'x', createdAt: AT });
+    await addresses.insertAddress({ userId: user.id, tripId: trip.id, tokenHash: 'b', tokenEnc: 'y', createdAt: AT });
+
+    expect(await addresses.revokeLive(user.id, null, LATER)).toBe(1);
+    expect(await addresses.findLive(user.id, null)).toBeNull();
+    expect(await addresses.findLive(user.id, trip.id)).not.toBeNull();
+    expect(await addresses.findByTokenHash('a')).toMatchObject({ id: own, revoked_at: LATER });
+    expect(await addresses.findByTokenHash('none')).toBeNull();
+
+    expect(await addresses.revokeLive(user.id, trip.id, LATER)).toBe(1);
+    expect(await addresses.findLive(user.id, trip.id)).toBeNull();
+  });
+
+  it('notes when a mail came in', async () => {
+    const { user } = await makeUser(t);
+    const id = await addresses.insertAddress({ userId: user.id, tripId: null, tokenHash: 'c', tokenEnc: 'z', createdAt: AT });
+    await addresses.touchUsed(id, LATER);
+    expect((await addresses.findById(id))?.last_used_at).toBe(LATER);
+  });
+});

server/tests/unit/db/repositories/MailIngestCandidates.repository.test.ts

@@ -0,0 +1,250 @@
+/**
+ * The bookings read from mails through their repository, and what it reads
+ * of the trips they may go to: a decision lands once, an undone booking is
+ * remembered per trip, and only the person's open trips are weighed.
+ */
+import { MailIngestAddresses } from '../../../../src/db/entities/MailIngestAddresses.entity';
+import { MailIngestCandidates } from '../../../../src/db/entities/MailIngestCandidates.entity';
+import { MailIngestMessages } from '../../../../src/db/entities/MailIngestMessages.entity';
+import type { MailIngestAddressesRepository } from '../../../../src/db/repositories/MailIngestAddresses.repository';
+import type {
+  MailIngestCandidateInsert,
+  MailIngestCandidatesRepository,
+} from '../../../../src/db/repositories/MailIngestCandidates.repository';
+import type { MailIngestMessagesRepository } from '../../../../src/db/repositories/MailIngestMessages.repository';
+import { dbNow } from '../../../../src/db/types';
+import { createSnapshotTestDb } from '../../../helpers/db-mock';
+import { addBudgetItemPayer, makeBudgetItem } from '../../../helpers/factories/budget';
+import { makePlace } from '../../../helpers/factories/places';
+import { makeReservation } from '../../../helpers/factories/reservations';
+import { addTripMember, makeTrip } from '../../../helpers/factories/trips';
+import { makeUser } from '../../../helpers/factories/users';
+import { resetTestDb } from '../../../helpers/test-db';
+import { createTestOrm, type TestOrm } from '../../../helpers/test-orm';
+
+import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest';
+
+const testDb = createSnapshotTestDb();
+let t: TestOrm;
+let candidates: MailIngestCandidatesRepository;
+let messages: MailIngestMessagesRepository;
+let addresses: MailIngestAddressesRepository;
+
+beforeAll(async () => {
+  t = await createTestOrm(testDb);
+  candidates = t.repo(MailIngestCandidates);
+  messages = t.repo(MailIngestMessages);
+  addresses = t.repo(MailIngestAddresses);
+});
+beforeEach(() => {
+  resetTestDb(testDb);
+  t.clear();
+});
+afterAll(async () => {
+  await t.close();
+  testDb.close();
+});
+
+const at = (iso: string) => dbNow(new Date(iso));
+const DAY1 = at('2026-11-01T10:00:00Z');
+
+async function mailOf(userId: number): Promise<number> {
+  const addressId = await addresses.insertAddress({
+    userId,
+    tripId: null,
+    tokenHash: `h${userId}`,
+    tokenEnc: 'e',
+    createdAt: DAY1,
+  });
+  const id = await messages.insertIfNew({
+    transport: 'upload',
+    messageIdHdr: null,
+    dedupeHash: `d${userId}`,
+    addressId,
+    userId,
+    tripId: null,
+    size: 1,
+    rawEnc: 'enc',
+    receivedAt: DAY1,
+  });
+  return id!;
+}
+
+function entry(messageId: number, userId: number, over: Partial<MailIngestCandidateInsert> = {}) {
+  return candidates.insertCandidate({
+    messageId,
+    userId,
+    position: 0,
+    itemJson: '{}',
+    source: 'ai',
+    checksJson: '{}',
+    uncertainFields: '[]',
+    matchJson: null,
+    suggestedTripId: null,
+    dedupeKey: null,
+    contentHash: 'c1',
+    resolution: 'pending',
+    reason: 'trip_ambiguous',
+    createdAt: DAY1,
+    ...over,
+  });
+}
+
+describe('MailIngestCandidatesRepository: the entries', () => {
+  it('lists a mail’s entries in order and counts the waiting ones', async () => {
+    const { user } = await makeUser(t);
+    const mail = await mailOf(user.id);
+    const second = await entry(mail, user.id, { position: 1 });
+    const first = await entry(mail, user.id, { position: 0, resolution: 'duplicate', reason: null });
+    expect((await candidates.listForMessage(mail)).map((row) => row.id)).toEqual([first, second]);
+    expect(await candidates.countPending(user.id)).toBe(1);
+    expect(await candidates.findById(second)).toMatchObject({ source: 'ai', reason: 'trip_ambiguous' });
+    expect(await candidates.findById(second + 10)).toBeNull();
+  });
+
+  it('decides an entry once, and records its documents afterwards', async () => {
+    const { user } = await makeUser(t);
+    const trip = await makeTrip(t, user.id);
+    const mail = await mailOf(user.id);
+    const id = await entry(mail, user.id);
+    const patch = {
+      resolution: 'applied',
+      reason: null,
+      resolvedTripId: trip.id,
+      createdIdsJson: '{"reservation_ids":[1]}',
+      resolvedBy: user.id,
+      resolvedAt: DAY1,
+      undoUntil: at('2026-11-08T10:00:00Z'),
+    };
+    expect(await candidates.resolve(id, ['pending'], patch)).toBe(true);
+    expect(await candidates.resolve(id, ['pending'], patch)).toBe(false);
+
+    await candidates.setAttachments(id, '{"reservation_ids":[1],"file_ids":[2]}', null);
+    expect(await candidates.findById(id)).toMatchObject({
+      resolution: 'applied',
+      resolved_trip_id: trip.id,
+      created_ids_json: '{"reservation_ids":[1],"file_ids":[2]}',
+      attach_error: null,
+    });
+  });
+
+  it('remembers a booking undone in a trip by its content', async () => {
+    const { user } = await makeUser(t);
+    const trip = await makeTrip(t, user.id);
+    const id = await entry(await mailOf(user.id), user.id, { contentHash: 'same' });
+    expect(await candidates.wasUndoneInTrip(trip.id, 'same')).toBe(false);
+    await candidates.resolve(id, ['pending'], {
+      resolution: 'undone',
+      reason: null,
+      resolvedTripId: trip.id,
+      createdIdsJson: null,
+      resolvedBy: user.id,
+      resolvedAt: DAY1,
+      undoUntil: null,
+    });
+    expect(await candidates.wasUndoneInTrip(trip.id, 'same')).toBe(true);
+    expect(await candidates.wasUndoneInTrip(trip.id, 'other')).toBe(false);
+  });
+
+  it('deletes decided entries past the cutoff and keeps the waiting ones', async () => {
+    const { user } = await makeUser(t);
+    const mail = await mailOf(user.id);
+    const waiting = await entry(mail, user.id);
+    const decided = await entry(mail, user.id, { position: 1 });
+    await candidates.resolve(decided, ['pending'], {
+      resolution: 'discarded',
+      reason: null,
+      resolvedTripId: null,
+      createdIdsJson: null,
+      resolvedBy: user.id,
+      resolvedAt: at('2026-07-01T10:00:00Z'),
+      undoUntil: null,
+    });
+    expect(await candidates.deleteResolvedBefore(at('2026-08-01T00:00:00Z'))).toBe(1);
+    expect(await candidates.findById(waiting)).not.toBeNull();
+    expect(await candidates.findById(decided)).toBeNull();
+  });
+});
+
+describe('MailIngestCandidatesRepository: what it reads of the trips', () => {
+  it('lists the person’s open trips, owned or joined, newest start first', async () => {
+    const { user } = await makeUser(t);
+    const { user: other } = await makeUser(t);
+    const owned = await makeTrip(t, user.id, { start_date: '2027-03-01', end_date: '2027-03-03', currency: 'JPY' });
+    const joined = await makeTrip(t, other.id, { start_date: '2027-05-01', end_date: '2027-05-02' });
+    await addTripMember(t, joined.id, user.id);
+    const undated = await makeTrip(t, user.id);
+    await makeTrip(t, user.id, { start_date: '2026-01-01', end_date: '2026-01-02' });
+    await makeTrip(t, user.id, { start_date: '2027-06-01', end_date: '2027-06-02', is_archived: 1 });
+    await makeTrip(t, other.id, { start_date: '2027-07-01', end_date: '2027-07-02' });
+
+    const trips = await candidates.listOpenTrips(user.id, '2026-10-01', 20);
+    expect(trips.map((trip) => trip.id).sort()).toEqual([owned.id, joined.id, undated.id].sort());
+    expect(trips.find((trip) => trip.id === owned.id)).toMatchObject({ user_id: user.id, currency: 'JPY' });
+    expect(await candidates.listOpenTrips(user.id, '2026-10-01', 1)).toHaveLength(1);
+  });
+
+  it('reads the dated days and the placed places of trips', async () => {
+    const { user } = await makeUser(t);
+    const trip = await makeTrip(t, user.id, { start_date: '2027-03-01', end_date: '2027-03-02' });
+    await makePlace(t, trip.id, { name: 'Hotel Sakura', address: 'Tokyo', lat: 35.68, lng: 139.76 });
+    await makePlace(t, trip.id, { name: 'Somewhere', lat: null, lng: null });
+
+    expect(await candidates.listDayDates([trip.id])).toEqual([
+      { trip_id: trip.id, date: '2027-03-01' },
+      { trip_id: trip.id, date: '2027-03-02' },
+    ]);
+    expect(await candidates.listDayDates([])).toEqual([]);
+    expect(await candidates.listPlacesWithPosition([trip.id])).toEqual([
+      { trip_id: trip.id, name: 'Hotel Sakura', address: 'Tokyo', lat: 35.68, lng: 139.76 },
+    ]);
+    expect(await candidates.listPlacesWithPosition([])).toEqual([]);
+  });
+
+  it('reads the person a mail is for', async () => {
+    const { user } = await makeUser(t, { username: 'mia', email: 'mia@example.com' });
+    expect(await candidates.findPerson(user.id)).toEqual({ id: user.id, username: 'mia', email: 'mia@example.com', role: 'user' });
+    expect(await candidates.findPerson(user.id + 100)).toBeNull();
+  });
+
+  it('finds a booking an import stored under a key, and one that looks the same', async () => {
+    const { user } = await makeUser(t);
+    const trip = await makeTrip(t, user.id);
+    await makeReservation(t, trip.id, {
+      type: 'flight',
+      reservation_time: '2027-03-01T10:00',
+      confirmation_number: 'ABC123',
+      external_source: 'mail',
+      external_id: 'ABC123|flight|2027-03-01|hnd-itm',
+    });
+
+    expect(await candidates.tripHasExternal(trip.id, ['import', 'mail'], 'ABC123|flight|2027-03-01|hnd-itm')).toBe(true);
+    expect(await candidates.tripHasExternal(trip.id, ['import'], 'ABC123|flight|2027-03-01|hnd-itm')).toBe(false);
+
+    const none = { type: 'flight', startFrom: null, startTo: null, confirmation: null };
+    expect(await candidates.tripHasSimilar(trip.id, none)).toBe(false);
+    expect(await candidates.tripHasSimilar(trip.id, { ...none, confirmation: 'ABC123' })).toBe(true);
+    expect(
+      await candidates.tripHasSimilar(trip.id, { ...none, startFrom: '2027-03-01T09:30', startTo: '2027-03-01T10:30' }),
+    ).toBe(true);
+    expect(
+      await candidates.tripHasSimilar(trip.id, {
+        ...none,
+        type: 'train',
+        startFrom: '2027-03-01T09:30',
+        startTo: '2027-03-01T10:30',
+      }),
+    ).toBe(false);
+  });
+
+  it('counts the payers of costs', async () => {
+    const { user } = await makeUser(t);
+    const trip = await makeTrip(t, user.id);
+    const paid = await makeBudgetItem(t, trip.id);
+    const open = await makeBudgetItem(t, trip.id);
+    await addBudgetItemPayer(t, paid.id, user.id, 100);
+    expect(await candidates.countPayers([paid.id, open.id])).toBe(1);
+    expect(await candidates.countPayers([open.id])).toBe(0);
+    expect(await candidates.countPayers([])).toBe(0);
+  });
+});

server/tests/unit/db/repositories/MailIngestCursor.repository.test.ts

@@ -0,0 +1,63 @@
+/**
+ * The readers' cursor: one holder of a lease at a time, a lease that ran out
+ * is taken over, and a run's position and error are kept for the admin.
+ */
+import { MailIngestCursor } from '../../../../src/db/entities/MailIngestCursor.entity';
+import type { MailIngestCursorRepository } from '../../../../src/db/repositories/MailIngestCursor.repository';
+import { dbNow } from '../../../../src/db/types';
+import { createSnapshotTestDb } from '../../../helpers/db-mock';
+import { resetTestDb } from '../../../helpers/test-db';
+import { createTestOrm, type TestOrm } from '../../../helpers/test-orm';
+
+import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest';
+
+const testDb = createSnapshotTestDb();
+let t: TestOrm;
+let cursor: MailIngestCursorRepository;
+
+beforeAll(async () => {
+  t = await createTestOrm(testDb);
+  cursor = t.repo(MailIngestCursor);
+});
+beforeEach(() => {
+  resetTestDb(testDb);
+  t.clear();
+});
+afterAll(async () => {
+  await t.close();
+  testDb.close();
+});
+
+const T0 = dbNow(new Date('2026-11-01T10:00:00Z'));
+const T1 = dbNow(new Date('2026-11-01T10:05:00Z'));
+const T2 = dbNow(new Date('2026-11-01T10:10:00Z'));
+
+describe('MailIngestCursorRepository', () => {
+  it('gives the lease to one owner until it runs out or is given back', async () => {
+    expect(await cursor.acquireLease('pipeline', '', 'a', T1, T0)).toBe(true);
+    expect(await cursor.acquireLease('pipeline', '', 'b', T1, T0)).toBe(false);
+    // The holder renews its own lease.
+    expect(await cursor.acquireLease('pipeline', '', 'a', T2, T0)).toBe(true);
+    // Run out: another takes it.
+    expect(await cursor.acquireLease('pipeline', '', 'b', T2, dbNow(new Date('2026-11-01T10:11:00Z')))).toBe(true);
+
+    await cursor.releaseLease('pipeline', '', 'a');
+    expect(await cursor.find('pipeline', '')).toMatchObject({ lease_owner: 'b' });
+    await cursor.releaseLease('pipeline', '', 'b');
+    expect(await cursor.find('pipeline', '')).toMatchObject({ lease_owner: null, lease_until: null });
+  });
+
+  it('keeps the position of a good run and the error of a bad one', async () => {
+    await cursor.acquireLease('imap', 'INBOX', 'a', T1, T0);
+    await cursor.recordOk('imap', 'INBOX', T0, { uidValidity: 7, lastUid: 42 });
+    await cursor.recordError('imap', 'INBOX', 'x'.repeat(600));
+    let row = await cursor.find('imap', 'INBOX');
+    expect(row).toMatchObject({ uid_validity: 7, last_uid: 42, last_ok_at: T0 });
+    expect(row?.last_error).toHaveLength(500);
+
+    await cursor.recordOk('imap', 'INBOX', T1);
+    row = await cursor.find('imap', 'INBOX');
+    expect(row).toMatchObject({ uid_validity: 7, last_uid: 42, last_ok_at: T1, last_error: null });
+    expect(await cursor.find('cloud', '')).toBeNull();
+  });
+});

server/tests/unit/db/repositories/MailIngestMessages.repository.test.ts

@@ -0,0 +1,191 @@
+/**
+ * Forwarded mails through their repository: a mail delivered twice is stored
+ * once, a due mail is claimed by one run, a run whose claim ran out hands the
+ * mail on, and the retention keeps a mail while its inbox entry waits.
+ */
+import { MailIngestAddresses } from '../../../../src/db/entities/MailIngestAddresses.entity';
+import { MailIngestCandidates } from '../../../../src/db/entities/MailIngestCandidates.entity';
+import { MailIngestMessages } from '../../../../src/db/entities/MailIngestMessages.entity';
+import type { MailIngestAddressesRepository } from '../../../../src/db/repositories/MailIngestAddresses.repository';
+import type { MailIngestCandidatesRepository } from '../../../../src/db/repositories/MailIngestCandidates.repository';
+import type {
+  MailIngestMessageOutcome,
+  MailIngestMessagesRepository,
+} from '../../../../src/db/repositories/MailIngestMessages.repository';
+import { dbNow } from '../../../../src/db/types';
+import { createSnapshotTestDb } from '../../../helpers/db-mock';
+import { makeUser } from '../../../helpers/factories/users';
+import { resetTestDb } from '../../../helpers/test-db';
+import { createTestOrm, type TestOrm } from '../../../helpers/test-orm';
+
+import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest';
+
+const testDb = createSnapshotTestDb();
+let t: TestOrm;
+let messages: MailIngestMessagesRepository;
+let addresses: MailIngestAddressesRepository;
+let candidates: MailIngestCandidatesRepository;
+
+beforeAll(async () => {
+  t = await createTestOrm(testDb);
+  messages = t.repo(MailIngestMessages);
+  addresses = t.repo(MailIngestAddresses);
+  candidates = t.repo(MailIngestCandidates);
+});
+beforeEach(() => {
+  resetTestDb(testDb);
+  t.clear();
+});
+afterAll(async () => {
+  await t.close();
+  testDb.close();
+});
+
+const at = (iso: string) => dbNow(new Date(iso));
+const DAY1 = at('2026-11-01T10:00:00Z');
+
+async function inbox() {
+  const { user } = await makeUser(t);
+  const addressId = await addresses.insertAddress({
+    userId: user.id,
+    tripId: null,
+    tokenHash: `h${user.id}`,
+    tokenEnc: 'e',
+    createdAt: DAY1,
+  });
+  return { userId: user.id, addressId };
+}
+
+function store(box: { userId: number; addressId: number }, hash: string, receivedAt = DAY1) {
+  return messages.insertIfNew({
+    transport: 'upload',
+    messageIdHdr: `<${hash}@x>`,
+    dedupeHash: hash,
+    addressId: box.addressId,
+    userId: box.userId,
+    tripId: null,
+    size: 10,
+    rawEnc: 'enc',
+    receivedAt,
+  });
+}
+
+const DONE: MailIngestMessageOutcome = {
+  status: 'processed',
+  reason: null,
+  subject: 'Booking',
+  fromAddr: 'a@b.c',
+  fromVerified: true,
+  authResults: 'dkim=pass',
+  hints: '[]',
+  attachments: '[]',
+  forwardingCode: null,
+  aiUsed: true,
+  processedAt: DAY1,
+};
+
+describe('MailIngestMessagesRepository', () => {
+  it('stores a mail once and finds it by its hash', async () => {
+    const box = await inbox();
+    const id = await store(box, 'm1');
+    expect(id).not.toBeNull();
+    expect(await store(box, 'm1')).toBeNull();
+    expect(await messages.findIdByDedupeHash('m1')).toBe(id);
+    expect(await messages.findIdByDedupeHash('m2')).toBeNull();
+    expect(await messages.findById(id!)).toMatchObject({ status: 'queued', attempts: 0, raw_eml_enc: 'enc' });
+    expect(await messages.findById(id! + 1)).toBeNull();
+  });
+
+  it('counts the mails of an address and the AI-read mails of a person since a time', async () => {
+    const box = await inbox();
+    const first = await store(box, 'c1', at('2026-10-31T10:00:00Z'));
+    await store(box, 'c2', DAY1);
+    expect(await messages.countForAddressSince(box.addressId, at('2026-11-01T00:00:00Z'))).toBe(1);
+    await messages.finish(first!, DONE);
+    expect(await messages.countAiReadSince(box.userId, at('2026-10-31T00:00:00Z'))).toBe(1);
+    expect(await messages.countAiReadSince(box.userId, at('2026-11-01T00:00:00Z'))).toBe(0);
+  });
+
+  it('claims a due mail for one run and lets a claim that ran out go to the next', async () => {
+    const box = await inbox();
+    const id = await store(box, 'd1');
+    await store(box, 'd2', at('2026-11-05T10:00:00Z'));
+
+    const now = at('2026-11-01T11:00:00Z');
+    const lease = at('2026-11-01T11:10:00Z');
+    expect(await messages.claimDue(now, lease, 3)).toEqual([id]);
+    expect(await messages.claimDue(now, lease, 3)).toEqual([]);
+
+    const later = at('2026-11-01T11:20:00Z');
+    expect(await messages.claimDue(later, at('2026-11-01T11:30:00Z'), 3)).toEqual([id]);
+    expect(await messages.findById(id!)).toMatchObject({ status: 'processing', attempts: 2 });
+
+    await messages.retryLater(id!, at('2026-11-01T12:00:00Z'));
+    expect(await messages.claimDue(later, lease, 3)).toEqual([]);
+    expect(await messages.claimDue(at('2026-11-01T12:00:00Z'), lease, 3)).toEqual([id]);
+
+    await messages.finish(id!, DONE);
+    expect(await messages.findById(id!)).toMatchObject({ status: 'processed', ai_used: 1, from_verified: 1, lease_until: null });
+  });
+
+  it('purges a mail once it is done and nothing waits on it, and always after the hard limit', async () => {
+    const box = await inbox();
+    const waiting = await store(box, 'p1');
+    const settled = await store(box, 'p2');
+    const old = await store(box, 'p3', at('2026-09-01T10:00:00Z'));
+    await messages.finish(waiting!, DONE);
+    await messages.finish(settled!, DONE);
+    await candidates.insertCandidate({
+      messageId: waiting!,
+      userId: box.userId,
+      position: 0,
+      itemJson: null,
+      source: null,
+      checksJson: null,
+      uncertainFields: null,
+      matchJson: null,
+      suggestedTripId: null,
+      dedupeKey: null,
+      contentHash: null,
+      resolution: 'pending',
+      reason: 'no_trip_in_range',
+      createdAt: DAY1,
+    });
+
+    const purgedAt = at('2026-11-20T10:00:00Z');
+    expect(await messages.purgeRaw(at('2026-11-10T10:00:00Z'), at('2026-10-01T10:00:00Z'), purgedAt)).toBe(2);
+    expect(await messages.findById(waiting!)).toMatchObject({ raw_eml_enc: 'enc', raw_purged_at: null });
+    expect(await messages.findById(settled!)).toMatchObject({ raw_eml_enc: null, raw_purged_at: purgedAt });
+    expect(await messages.findById(old!)).toMatchObject({ raw_eml_enc: null });
+  });
+
+  it('deletes settled mails without entries, and keeps queued ones and those with entries', async () => {
+    const box = await inbox();
+    const queued = await store(box, 'x1', at('2026-08-01T10:00:00Z'));
+    const bare = await store(box, 'x2', at('2026-08-01T10:00:00Z'));
+    const withEntry = await store(box, 'x3', at('2026-08-01T10:00:00Z'));
+    await messages.finish(bare!, { ...DONE, status: 'discarded' });
+    await messages.finish(withEntry!, DONE);
+    await candidates.insertCandidate({
+      messageId: withEntry!,
+      userId: box.userId,
+      position: 0,
+      itemJson: null,
+      source: null,
+      checksJson: null,
+      uncertainFields: null,
+      matchJson: null,
+      suggestedTripId: null,
+      dedupeKey: null,
+      contentHash: null,
+      resolution: 'discarded',
+      reason: null,
+      createdAt: DAY1,
+    });
+
+    expect(await messages.deleteSettledBefore(at('2026-09-01T00:00:00Z'))).toBe(1);
+    expect(await messages.findById(bare!)).toBeNull();
+    expect(await messages.findById(queued!)).not.toBeNull();
+    expect(await messages.findById(withEntry!)).not.toBeNull();
+  });
+});

server/tests/unit/db/repositories/MailIngestPrefs.repository.test.ts

@@ -0,0 +1,48 @@
+/**
+ * A person's switches for the mail import: no row until something changes,
+ * and a write names only the switches it changes.
+ */
+import { MailIngestPrefs } from '../../../../src/db/entities/MailIngestPrefs.entity';
+import type { MailIngestPrefsRepository } from '../../../../src/db/repositories/MailIngestPrefs.repository';
+import { createSnapshotTestDb } from '../../../helpers/db-mock';
+import { makeUser } from '../../../helpers/factories/users';
+import { resetTestDb } from '../../../helpers/test-db';
+import { createTestOrm, type TestOrm } from '../../../helpers/test-orm';
+
+import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest';
+
+const testDb = createSnapshotTestDb();
+let t: TestOrm;
+let prefs: MailIngestPrefsRepository;
+
+beforeAll(async () => {
+  t = await createTestOrm(testDb);
+  prefs = t.repo(MailIngestPrefs);
+});
+beforeEach(() => {
+  resetTestDb(testDb);
+  t.clear();
+});
+afterAll(async () => {
+  await t.close();
+  testDb.close();
+});
+
+describe('MailIngestPrefsRepository', () => {
+  it('has no row until a switch changes, then the defaults for the others', async () => {
+    const { user } = await makeUser(t);
+    expect(await prefs.findForUser(user.id)).toBeNull();
+
+    await prefs.upsertForUser(user.id, { auto_add: 0 });
+    expect(await prefs.findForUser(user.id)).toMatchObject({ enabled: 1, ai_enabled: 1, auto_add: 0 });
+
+    await prefs.upsertForUser(user.id, { ai_enabled: 0 });
+    expect(await prefs.findForUser(user.id)).toMatchObject({ enabled: 1, ai_enabled: 0, auto_add: 0 });
+  });
+
+  it('writes nothing for an empty change', async () => {
+    const { user } = await makeUser(t);
+    await prefs.upsertForUser(user.id, {});
+    expect(await prefs.findForUser(user.id)).toBeNull();
+  });
+});