From baa906fb6b40bfa626f9c1a90d035115cc425b0c Mon Sep 17 00:00:00 2001 From: Maurice Date: Fri, 13 Nov 2026 15:30:07 +0100 Subject: [PATCH] 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 | 5 + .../db/entities/MailIngestAddresses.entity.ts | 46 +++ .../entities/MailIngestCandidates.entity.ts | 81 ++++ .../db/entities/MailIngestCursor.entity.ts | 33 ++ .../db/entities/MailIngestMessages.entity.ts | 91 +++++ .../src/db/entities/MailIngestPrefs.entity.ts | 27 ++ server/src/db/entities/Trips.entity.ts | 11 + server/src/db/entities/Users.entity.ts | 12 + server/src/db/entities/index.ts | 10 + .../db/kysely/MailIngestAddresses.table.ts | 14 + .../db/kysely/MailIngestCandidates.table.ts | 27 ++ .../src/db/kysely/MailIngestCursor.table.ts | 15 + .../src/db/kysely/MailIngestMessages.table.ts | 31 ++ server/src/db/kysely/MailIngestPrefs.table.ts | 11 + server/src/db/kysely/db.ts | 10 + ...orwarded_booking_mail_lands_in_its_trip.ts | 163 ++++++++ .../MailIngestAddresses.repository.ts | 101 +++++ .../MailIngestCandidates.repository.ts | 351 ++++++++++++++++++ .../MailIngestCursor.repository.ts | 124 +++++++ .../MailIngestMessages.repository.ts | 263 +++++++++++++ .../MailIngestPrefs.repository.ts | 46 +++ server/tests/unit/db/entities-index.test.ts | 8 +- .../tests/unit/db/entity-class-fields.test.ts | 5 +- server/tests/unit/db/legacy-baseline.test.ts | 1 + .../MailIngestAddresses.repository.test.ts | 84 +++++ .../MailIngestCandidates.repository.test.ts | 250 +++++++++++++ .../MailIngestCursor.repository.test.ts | 63 ++++ .../MailIngestMessages.repository.test.ts | 191 ++++++++++ .../MailIngestPrefs.repository.test.ts | 48 +++ 29 files changed, 2118 insertions(+), 4 deletions(-) create mode 100644 server/src/db/entities/MailIngestAddresses.entity.ts create mode 100644 server/src/db/entities/MailIngestCandidates.entity.ts create mode 100644 server/src/db/entities/MailIngestCursor.entity.ts create mode 100644 server/src/db/entities/MailIngestMessages.entity.ts create mode 100644 server/src/db/entities/MailIngestPrefs.entity.ts create mode 100644 server/src/db/kysely/MailIngestAddresses.table.ts create mode 100644 server/src/db/kysely/MailIngestCandidates.table.ts create mode 100644 server/src/db/kysely/MailIngestCursor.table.ts create mode 100644 server/src/db/kysely/MailIngestMessages.table.ts create mode 100644 server/src/db/kysely/MailIngestPrefs.table.ts create mode 100644 server/src/db/migrations/Migration20200101046410_a_forwarded_booking_mail_lands_in_its_trip.ts create mode 100644 server/src/db/repositories/MailIngestAddresses.repository.ts create mode 100644 server/src/db/repositories/MailIngestCandidates.repository.ts create mode 100644 server/src/db/repositories/MailIngestCursor.repository.ts create mode 100644 server/src/db/repositories/MailIngestMessages.repository.ts create mode 100644 server/src/db/repositories/MailIngestPrefs.repository.ts create mode 100644 server/tests/unit/db/repositories/MailIngestAddresses.repository.test.ts create mode 100644 server/tests/unit/db/repositories/MailIngestCandidates.repository.test.ts create mode 100644 server/tests/unit/db/repositories/MailIngestCursor.repository.test.ts create mode 100644 server/tests/unit/db/repositories/MailIngestMessages.repository.test.ts create mode 100644 server/tests/unit/db/repositories/MailIngestPrefs.repository.test.ts diff --git a/server/scripts/repository-owners.json b/server/scripts/repository-owners.json index fee86efd4..b121a6fb6 100644 --- a/server/scripts/repository-owners.json +++ b/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", diff --git a/server/src/db/entities/MailIngestAddresses.entity.ts b/server/src/db/entities/MailIngestAddresses.entity.ts new file mode 100644 index 000000000..e3a5fe465 --- /dev/null +++ b/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; + user_id!: number; + trip?: Ref | null; + trip_id?: number | null; + created_at!: string & Opt; + revoked_at?: string | null; + last_used_at?: string | null; + mail_ingest_messages_collection = new Collection(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(), + }, +}); diff --git a/server/src/db/entities/MailIngestCandidates.entity.ts b/server/src/db/entities/MailIngestCandidates.entity.ts new file mode 100644 index 000000000..7586559fe --- /dev/null +++ b/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; + message_id!: number; + user!: Ref; + 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 | null; + suggested_trip_id?: number | null; + dedupe_key?: string | null; + content_hash?: string | null; + resolution: string & Opt = 'pending'; + reason?: string | null; + resolvedTrip?: Ref | 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`), + }, +}); diff --git a/server/src/db/entities/MailIngestCursor.entity.ts b/server/src/db/entities/MailIngestCursor.entity.ts new file mode 100644 index 000000000..a64fbdae2 --- /dev/null +++ b/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(), + }, +}); diff --git a/server/src/db/entities/MailIngestMessages.entity.ts b/server/src/db/entities/MailIngestMessages.entity.ts new file mode 100644 index 000000000..6da0cbce3 --- /dev/null +++ b/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 | null; + address_id?: number | null; + user!: Ref; + user_id!: number; + trip?: Ref | 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(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(), + }, +}); diff --git a/server/src/db/entities/MailIngestPrefs.entity.ts b/server/src/db/entities/MailIngestPrefs.entity.ts new file mode 100644 index 000000000..2609d41a9 --- /dev/null +++ b/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; + 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), + }, +}); diff --git a/server/src/db/entities/Trips.entity.ts b/server/src/db/entities/Trips.entity.ts index 163e9ba04..a9d5d0b7a 100644 --- a/server/src/db/entities/Trips.entity.ts +++ b/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(this); journey_entries_collection = new Collection(this); journey_trips_collection = new Collection(this); + mail_ingest_addresses_collection = new Collection(this); + mail_ingest_candidates_collection = new Collection(this); + mail_ingest_candidates_collection1 = new Collection(this); + mail_ingest_messages_collection = new Collection(this); packing_bags_collection = new Collection(this); packing_category_assignees_collection = new Collection(this); packing_items_collection = new Collection(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(), diff --git a/server/src/db/entities/Users.entity.ts b/server/src/db/entities/Users.entity.ts index 1900e7132..299899357 100644 --- a/server/src/db/entities/Users.entity.ts +++ b/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(this); journey_share_tokens_collection = new Collection(this); journeys_collection = new Collection(this); + mail_ingest_addresses_collection = new Collection(this); + mail_ingest_candidates_collection = new Collection(this); + mail_ingest_messages_collection = new Collection(this); + mail_ingest_prefs_collection = new Collection(this); mcp_tokens_collection = new Collection(this); notification_channel_preferences_collection = new Collection(this); notifications_collection = new Collection(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(), diff --git a/server/src/db/entities/index.ts b/server/src/db/entities/index.ts index 584436d9a..0cb3eb08b 100644 --- a/server/src/db/entities/index.ts +++ b/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, diff --git a/server/src/db/kysely/MailIngestAddresses.table.ts b/server/src/db/kysely/MailIngestAddresses.table.ts new file mode 100644 index 000000000..2cf109dda --- /dev/null +++ b/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; + token_hash: string; + token_enc: string; + user_id: number; + trip_id: InsertOptional; + created_at: InsertOptional; + revoked_at: InsertOptional; + last_used_at: InsertOptional; +} diff --git a/server/src/db/kysely/MailIngestCandidates.table.ts b/server/src/db/kysely/MailIngestCandidates.table.ts new file mode 100644 index 000000000..3efba72d9 --- /dev/null +++ b/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; + message_id: number; + user_id: number; + position: InsertOptional; + item_json: InsertOptional; + source: InsertOptional; + checks_json: InsertOptional; + uncertain_fields: InsertOptional; + match_json: InsertOptional; + suggested_trip_id: InsertOptional; + dedupe_key: InsertOptional; + content_hash: InsertOptional; + resolution: InsertOptional; + reason: InsertOptional; + resolved_trip_id: InsertOptional; + created_ids_json: InsertOptional; + attach_error: InsertOptional; + resolved_by: InsertOptional; + resolved_at: InsertOptional; + undo_until: InsertOptional; + created_at: InsertOptional; +} diff --git a/server/src/db/kysely/MailIngestCursor.table.ts b/server/src/db/kysely/MailIngestCursor.table.ts new file mode 100644 index 000000000..07f3a59b6 --- /dev/null +++ b/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; + transport: string; + mailbox: string; + uid_validity: InsertOptional; + last_uid: InsertOptional; + last_ok_at: InsertOptional; + last_error: InsertOptional; + lease_owner: InsertOptional; + lease_until: InsertOptional; +} diff --git a/server/src/db/kysely/MailIngestMessages.table.ts b/server/src/db/kysely/MailIngestMessages.table.ts new file mode 100644 index 000000000..90d653215 --- /dev/null +++ b/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; + transport: string; + message_id_hdr: InsertOptional; + dedupe_hash: string; + address_id: InsertOptional; + user_id: number; + trip_id: InsertOptional; + from_addr: InsertOptional; + from_verified: InsertOptional; + auth_results: InsertOptional; + subject: InsertOptional; + received_at: InsertOptional; + size: number; + raw_eml_enc: InsertOptional; + raw_purged_at: InsertOptional; + status: InsertOptional; + reason: InsertOptional; + hints: InsertOptional; + attachments: InsertOptional; + forwarding_code: InsertOptional; + attempts: InsertOptional; + next_attempt_at: InsertOptional; + lease_until: InsertOptional; + ai_used: InsertOptional; + processed_at: InsertOptional; +} diff --git a/server/src/db/kysely/MailIngestPrefs.table.ts b/server/src/db/kysely/MailIngestPrefs.table.ts new file mode 100644 index 000000000..b2e9c7e57 --- /dev/null +++ b/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; + user_id: number; + enabled: InsertOptional; + ai_enabled: InsertOptional; + auto_add: InsertOptional; +} diff --git a/server/src/db/kysely/db.ts b/server/src/db/kysely/db.ts index 88fd33366..bb4fabea7 100644 --- a/server/src/db/kysely/db.ts +++ b/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; diff --git a/server/src/db/migrations/Migration20200101046410_a_forwarded_booking_mail_lands_in_its_trip.ts b/server/src/db/migrations/Migration20200101046410_a_forwarded_booking_mail_lands_in_its_trip.ts new file mode 100644 index 000000000..e8e913061 --- /dev/null +++ b/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)', + ); + } +} diff --git a/server/src/db/repositories/MailIngestAddresses.repository.ts b/server/src/db/repositories/MailIngestAddresses.repository.ts new file mode 100644 index 000000000..4dfb267a3 --- /dev/null +++ b/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 = 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; + +/** + * `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 { + private db() { + return this.kysely(); + } + + async insertAddress(row: MailIngestAddressInsert): Promise { + 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 { + 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 { + const row = await this.db() + .selectFrom('mail_ingest_addresses') + .selectAll() + .where('token_hash', '=', tokenHash) + .executeTakeFirst(); + return row ?? null; + } + + async findById(id: number): Promise { + 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 { + 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 { + await this.db().updateTable('mail_ingest_addresses').set({ last_used_at: at }).where('id', '=', id).execute(); + } +} diff --git a/server/src/db/repositories/MailIngestCandidates.repository.ts b/server/src/db/repositories/MailIngestCandidates.repository.ts new file mode 100644 index 000000000..d9373197c --- /dev/null +++ b/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 = 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 { + private db() { + return this.kysely(); + } + + async insertCandidate(row: MailIngestCandidateInsert): Promise { + 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 { + 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 { + 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 { + 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 { + 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 { + const row = await this.db() + .selectFrom('mail_ingest_candidates') + .select((eb) => eb.fn.countAll().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 { + 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 { + 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 { + 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> { + 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 { + 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 { + 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 { + 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 { + const { type, startFrom, startTo, confirmation } = probe; + const row = await this.db() + .selectFrom('reservations') + .select('id') + .where('trip_id', '=', tripId) + .where((eb) => { + const either: Expression[] = []; + 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 { + if (budgetItemIds.length === 0) return 0; + const row = await this.db() + .selectFrom('budget_item_payers') + .select((eb) => eb.fn.countAll().as('n')) + .where('budget_item_id', 'in', [...budgetItemIds]) + .executeTakeFirstOrThrow(); + return Number(row.n); + } +} diff --git a/server/src/db/repositories/MailIngestCursor.repository.ts b/server/src/db/repositories/MailIngestCursor.repository.ts new file mode 100644 index 000000000..0c1d5d911 --- /dev/null +++ b/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 = 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; + +/** + * `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 { + private db() { + return this.kysely(); + } + + /** + * 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 { + 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 { + 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 { + 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 { + 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 { + const row = await this.db() + .selectFrom('mail_ingest_cursor') + .selectAll() + .where('transport', '=', transport) + .where('mailbox', '=', mailbox) + .executeTakeFirst(); + return row ?? null; + } +} diff --git a/server/src/db/repositories/MailIngestMessages.repository.ts b/server/src/db/repositories/MailIngestMessages.repository.ts new file mode 100644 index 000000000..6f6a415c3 --- /dev/null +++ b/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 = 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; + +/** + * `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 { + private db() { + return this.kysely(); + } + + /** Stores a new mail, queued for now; null when the same mail is in already. */ + async insertIfNew(row: MailIngestMessageInsert): Promise { + 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 { + 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 { + 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 { + const row = await this.db() + .selectFrom('mail_ingest_messages') + .select((eb) => eb.fn.countAll().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 { + const row = await this.db() + .selectFrom('mail_ingest_messages') + .select((eb) => eb.fn.countAll().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 { + 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 { + 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 { + 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 { + 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 { + 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); + } +} diff --git a/server/src/db/repositories/MailIngestPrefs.repository.ts b/server/src/db/repositories/MailIngestPrefs.repository.ts new file mode 100644 index 000000000..aafd86c93 --- /dev/null +++ b/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 = true; + +/** The switches a write names; the others keep their value. */ +export type MailIngestPrefsPatch = Partial>; + +type PrefsKyselyDB = Pick; + +/** + * `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 { + private db() { + return this.kysely(); + } + + async findForUser(userId: number): Promise { + 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 { + 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(); + } +} diff --git a/server/tests/unit/db/entities-index.test.ts b/server/tests/unit/db/entities-index.test.ts index 32295b7b5..a81562d96 100644 --- a/server/tests/unit/db/entities-index.test.ts +++ b/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); }); diff --git a/server/tests/unit/db/entity-class-fields.test.ts b/server/tests/unit/db/entity-class-fields.test.ts index d7973d94b..a2bf04c54 100644 --- a/server/tests/unit/db/entity-class-fields.test.ts +++ b/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', () => { diff --git a/server/tests/unit/db/legacy-baseline.test.ts b/server/tests/unit/db/legacy-baseline.test.ts index 6d396b33b..20ed490f2 100644 --- a/server/tests/unit/db/legacy-baseline.test.ts +++ b/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. diff --git a/server/tests/unit/db/repositories/MailIngestAddresses.repository.test.ts b/server/tests/unit/db/repositories/MailIngestAddresses.repository.test.ts new file mode 100644 index 000000000..85d4f4b71 --- /dev/null +++ b/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); + }); +}); diff --git a/server/tests/unit/db/repositories/MailIngestCandidates.repository.test.ts b/server/tests/unit/db/repositories/MailIngestCandidates.repository.test.ts new file mode 100644 index 000000000..9765b591a --- /dev/null +++ b/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 { + 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 = {}) { + 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); + }); +}); diff --git a/server/tests/unit/db/repositories/MailIngestCursor.repository.test.ts b/server/tests/unit/db/repositories/MailIngestCursor.repository.test.ts new file mode 100644 index 000000000..22423a68d --- /dev/null +++ b/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(); + }); +}); diff --git a/server/tests/unit/db/repositories/MailIngestMessages.repository.test.ts b/server/tests/unit/db/repositories/MailIngestMessages.repository.test.ts new file mode 100644 index 000000000..d56a7ae11 --- /dev/null +++ b/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(); + }); +}); diff --git a/server/tests/unit/db/repositories/MailIngestPrefs.repository.test.ts b/server/tests/unit/db/repositories/MailIngestPrefs.repository.test.ts new file mode 100644 index 000000000..e0e0f6649 --- /dev/null +++ b/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(); + }); +});