feat(mail-ingest): tell every address holder once when the import stops

A plan without the import switches it off from outside the server, and an
admin can switch it off too; the plan promises a "mail import paused" note.
An hourly job sends mail_ingest_paused to each holder of a live address with
their own import on, notes the pause first, and forgets it once it is back.

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

@@ -94,6 +94,18 @@ export class MailIngestAddressesRepository extends TrekRepository<MailIngestAddr
     return Number(result.numUpdatedRows);
   }
 
+  /** The people who hold at least one address that is not revoked. */
+  async listHolders(): Promise<number[]> {
+    const rows = await this.db()
+      .selectFrom('mail_ingest_addresses')
+      .select('user_id')
+      .distinct()
+      .where('revoked_at', 'is', null)
+      .orderBy('user_id')
+      .execute();
+    return rows.map((row) => row.user_id);
+  }
+
   /** Notes that a mail came in on the address. */
   async touchUsed(id: number, at: DbTimestamp): Promise<void> {
     await this.db().updateTable('mail_ingest_addresses').set({ last_used_at: at }).where('id', '=', id).execute();

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

@@ -4,8 +4,12 @@ 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. */
-type MailIngestCursorTransport = 'imap' | 'cloud' | 'pipeline';
+/**
+ * What reads mail: an IMAP box, the TREK Cloud relay, or the processing
+ * pipeline itself; and the pause notice, whose good run is the pause it
+ * announced.
+ */
+type MailIngestCursorTransport = 'imap' | 'cloud' | 'pipeline' | 'notice';
 
 /** One `mail_ingest_cursor` row as stored. */
 interface MailIngestCursorRow {
@@ -102,6 +106,16 @@ export class MailIngestCursorRepository extends TrekRepository<MailIngestCursor>
       .execute();
   }
 
+  /** Forgets the last good run, as if the reader had never run. */
+  async forgetOk(transport: MailIngestCursorTransport, mailbox: string): Promise<void> {
+    await this.db()
+      .updateTable('mail_ingest_cursor')
+      .set({ last_ok_at: null })
+      .where('transport', '=', transport)
+      .where('mailbox', '=', mailbox)
+      .execute();
+  }
+
   /** A failed run, for the admin's status card; the position stays where the last good run left it. */
   async recordError(transport: MailIngestCursorTransport, mailbox: string, error: string): Promise<void> {
     await this.db()

server/src/nest/mail-ingest/mail-ingest-notify.service.ts

@@ -75,6 +75,19 @@ export class MailIngestNotifyService {
     );
   }
 
+  /** @txStandalone the import stopped; the person's address takes no mail until it is back. */
+  async importPaused(userId: number): Promise<void> {
+    await this.quietly(
+      this.notifications.send({
+        event: 'mail_ingest_paused',
+        actorId: null,
+        scope: 'user',
+        targetId: userId,
+        params: {},
+      }),
+    );
+  }
+
   /** @txStandalone one note for the bookings of a mail that wait in the inbox. */
   async bookingsPending(userId: number, pending: number): Promise<void> {
     this.inboxChanged(userId, pending);

server/src/nest/mail-ingest/mail-ingest-pause.job.ts

@@ -0,0 +1,74 @@
+import { MailIngestAddresses } from '../../db/entities/MailIngestAddresses.entity';
+import { MailIngestCursor } from '../../db/entities/MailIngestCursor.entity';
+import type { MailIngestAddressesRepository } from '../../db/repositories/MailIngestAddresses.repository';
+import type { MailIngestCursorRepository } from '../../db/repositories/MailIngestCursor.repository';
+import { dbNow } from '../../db/types';
+import { logError } from '../audit/audit-log.logger';
+import { CronRegistrarService, LEASE_OWNER } from '../scheduling/cron-registrar.service';
+import { MailIngestNotifyService } from './mail-ingest-notify.service';
+import { MailIngestSettingsService } from './mail-ingest-settings.service';
+import { InjectRepository } from '@mikro-orm/nestjs';
+import { Injectable, type OnApplicationBootstrap } from '@nestjs/common';
+
+/** How long a run holds the notice before another process may take it over. */
+const LEASE_MS = 10 * 60 * 1000;
+/** The notice's row in the readers' cursor table: transport `notice`, this as its mailbox. */
+const PAUSED = 'paused';
+
+/**
+ * Tells every person who holds a forwarding address, once, that the import
+ * stopped: on the TREK Cloud a plan without it switches it off from outside
+ * the server, and an admin can switch it off in the dialog. Checked every
+ * hour. The notice's cursor row notes the pause it announced, and forgets it
+ * once the import is back, so the next pause is announced again. A person
+ * who switched their own import off hears nothing, having nothing to lose.
+ */
+@Injectable()
+export class MailIngestPauseJob implements OnApplicationBootstrap {
+  constructor(
+    @InjectRepository(MailIngestCursor) private readonly cursor: MailIngestCursorRepository,
+    @InjectRepository(MailIngestAddresses) private readonly addresses: MailIngestAddressesRepository,
+    private readonly settings: MailIngestSettingsService,
+    private readonly notify: MailIngestNotifyService,
+    private readonly registrar: CronRegistrarService,
+  ) {}
+
+  onApplicationBootstrap(): void {
+    if (!this.registrar.isEnabled()) return;
+    this.registrar.register('mail-ingest-pause-notice', '25 * * * *', () => this.tick());
+  }
+
+  /**
+   * One run; answers how many people it told.
+   *
+   * @txIndependent the pause is noted before anyone is told, so a run that
+   * dies halfway never tells anyone twice; each notice stands on its own.
+   */
+  async tick(now: Date = new Date()): Promise<number> {
+    if (await this.settings.isAvailable()) {
+      if ((await this.cursor.find('notice', PAUSED))?.last_ok_at) await this.cursor.forgetOk('notice', PAUSED);
+      return 0;
+    }
+    const until = dbNow(new Date(now.getTime() + LEASE_MS));
+    if (!(await this.cursor.acquireLease('notice', PAUSED, LEASE_OWNER, until, dbNow(now)))) return 0;
+    try {
+      if ((await this.cursor.find('notice', PAUSED))?.last_ok_at) return 0;
+      const people = await this.peopleToTell();
+      await this.cursor.recordOk('notice', PAUSED, dbNow(now));
+      await Promise.all(people.map((userId) => this.notify.importPaused(userId)));
+      return people.length;
+    } catch (err) {
+      logError(`[mail-ingest] The pause notice failed: ${err instanceof Error ? err.message : String(err)}`);
+      return 0;
+    } finally {
+      await this.cursor.releaseLease('notice', PAUSED, LEASE_OWNER);
+    }
+  }
+
+  /** The holders of a live address who have their own import on. */
+  private async peopleToTell(): Promise<number[]> {
+    const holders = await this.addresses.listHolders();
+    const enabled = await Promise.all(holders.map(async (userId) => (await this.settings.prefsOf(userId)).enabled));
+    return holders.filter((_, i) => enabled[i]);
+  }
+}

server/src/nest/mail-ingest/mail-ingest.module.ts

@@ -30,6 +30,7 @@ import { MailIngestInboxService } from './mail-ingest-inbox.service';
 import { MailIngestIntakeService } from './mail-ingest-intake.service';
 import { MailIngestNotificationActions } from './mail-ingest-notification-actions';
 import { MailIngestNotifyService } from './mail-ingest-notify.service';
+import { MailIngestPauseJob } from './mail-ingest-pause.job';
 import { MailIngestProcessJob } from './mail-ingest-process.job';
 import { MailIngestProcessService } from './mail-ingest-process.service';
 import { MailIngestReadingService } from './mail-ingest-reading.service';
@@ -109,6 +110,7 @@ import { Module } from '@nestjs/common';
     MailIngestRetentionJob,
     MailIngestImapJob,
     MailIngestRelayJob,
+    MailIngestPauseJob,
   ],
   exports: [
     MailIngestSettingsService,

server/src/nest/notifications/notification-events.ts

@@ -48,6 +48,7 @@ export type NotifEventType =
   | 'wanderer_source_hint'
   | 'mail_booking_added'
   | 'mail_booking_pending'
+  | 'mail_ingest_paused'
   | 'plugin_notification';
 
 /** Every event, in the order the preferences UI lists them. */
@@ -72,6 +73,7 @@ export const ALL_EVENT_TYPES: NotifEventType[] = [
   'wanderer_source_hint',
   'mail_booking_added',
   'mail_booking_pending',
+  'mail_ingest_paused',
   'plugin_notification',
 ];
 

server/src/nest/notifications/notification-preferences.service.ts

@@ -137,6 +137,7 @@ export class NotificationPreferencesService {
         return (await resolveAppSetting(this.appSettings, 'flight_live_enabled')) !== 'false';
       case 'mail_booking_added':
       case 'mail_booking_pending':
+      case 'mail_ingest_paused':
         return (await resolveAppSetting(this.appSettings, 'mail_ingest_enabled')) === 'true';
       default:
         return true;

server/src/nest/notifications/notifications.service.ts

@@ -306,6 +306,14 @@ const EVENT_NOTIFICATION_CONFIG: Record<string, EventNotifConfig> = {
     navigateTextKey: 'notif.action.review',
     navigateTarget: () => '/dashboard?inbox=mail',
   },
+  // The import stopped (a plan without it, or an admin switched it off): every
+  // person with an address hears it once, with nothing to open.
+  mail_ingest_paused: {
+    inAppType: 'simple',
+    titleKey: 'notif.mail_ingest_paused.title',
+    textKey: 'notif.mail_ingest_paused.text',
+    navigateTarget: () => null,
+  },
   synology_session_cleared: {
     inAppType: 'simple',
     titleKey: 'notifications.synologySessionCleared.title',

server/src/nest/notifications/transports/ntfy.service.ts

@@ -41,6 +41,7 @@ const NTFY_EVENT_META: Partial<Record<NotifEventType, { priority: 1 | 2 | 3 | 4
   wanderer_source_hint: { priority: 2, tags: ['hiking_boot'] },
   mail_booking_added: { priority: 3, tags: ['envelope'] },
   mail_booking_pending: { priority: 2, tags: ['envelope'] },
+  mail_ingest_paused: { priority: 3, tags: ['pause_button'] },
 };
 const NTFY_DEFAULT_META = { priority: 3 as const, tags: [] as string[] };
 

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

@@ -1,7 +1,8 @@
 /**
  * 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.
+ * token is still recognised, a mail notes the address it came in on, and the
+ * people who hold a live address are named for the pause notice.
  */
 import { MailIngestAddresses } from '../../../../src/db/entities/MailIngestAddresses.entity';
 import type { MailIngestAddressesRepository } from '../../../../src/db/repositories/MailIngestAddresses.repository';
@@ -93,4 +94,17 @@ describe('MailIngestAddressesRepository', () => {
     await addresses.touchUsed(id, LATER);
     expect((await addresses.findById(id))?.last_used_at).toBe(LATER);
   });
+
+  it('names each person who holds an address that is not revoked, once', async () => {
+    const { user: mia } = await makeUser(t);
+    const { user: noah } = await makeUser(t);
+    const trip = await makeTrip(t, mia.id);
+    await addresses.insertAddress({ userId: mia.id, tripId: null, tokenHash: 'm1', tokenEnc: 'e', createdAt: AT });
+    await addresses.insertAddress({ userId: mia.id, tripId: trip.id, tokenHash: 'm2', tokenEnc: 'e', createdAt: AT });
+    await addresses.insertAddress({ userId: noah.id, tripId: null, tokenHash: 'n1', tokenEnc: 'e', createdAt: AT });
+    expect(await addresses.listHolders()).toEqual([mia.id, noah.id].sort((a, b) => a - b));
+
+    await addresses.revokeLive(noah.id, null, LATER);
+    expect(await addresses.listHolders()).toEqual([mia.id]);
+  });
 });

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

@@ -58,6 +58,9 @@ describe('MailIngestCursorRepository', () => {
     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 });
+
+    await cursor.forgetOk('imap', 'INBOX');
+    expect(await cursor.find('imap', 'INBOX')).toMatchObject({ uid_validity: 7, last_ok_at: null });
     expect(await cursor.find('cloud', '')).toBeNull();
   });
 });

server/tests/unit/nest/mail-ingest/mail-ingest-notify.test.ts

@@ -74,6 +74,18 @@ describe('MailIngestNotifyService', () => {
     notify.settingsChanged(3);
     expect(realtime.broadcastToUser).toHaveBeenCalledWith(3, { type: 'mailInbox:settings' });
   });
+
+  it('tells a person the import paused, and survives a failed send', async () => {
+    notifications.send.mockRejectedValue(new Error('smtp down'));
+    await expect(notify.importPaused(4)).resolves.toBeUndefined();
+    expect(notifications.send).toHaveBeenCalledWith({
+      event: 'mail_ingest_paused',
+      actorId: null,
+      scope: 'user',
+      targetId: 4,
+      params: {},
+    });
+  });
 });
 
 describe('MailIngestNotificationActions', () => {

server/tests/unit/nest/mail-ingest/mail-ingest-pause.test.ts

@@ -0,0 +1,98 @@
+/**
+ * The pause notice over stubs: when the import stops, every holder of a live
+ * address with their own import on hears it once; the pause is noted before
+ * anyone is told, a second run stays quiet, another process holding the
+ * notice is left alone, and the import coming back forgets the pause so the
+ * next one is announced again.
+ */
+import { MailIngestAddresses } from '../../../../src/db/entities/MailIngestAddresses.entity';
+import { MailIngestCursor } from '../../../../src/db/entities/MailIngestCursor.entity';
+import { MailIngestNotifyService } from '../../../../src/nest/mail-ingest/mail-ingest-notify.service';
+import { MailIngestPauseJob } from '../../../../src/nest/mail-ingest/mail-ingest-pause.job';
+import { MailIngestSettingsService } from '../../../../src/nest/mail-ingest/mail-ingest-settings.service';
+import { CronRegistrarService } from '../../../../src/nest/scheduling/cron-registrar.service';
+import { getRepositoryToken } from '@mikro-orm/nestjs';
+import { Test } from '@nestjs/testing';
+
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+vi.mock('../../../../src/nest/audit/audit-log.logger', () => ({
+  LOG_LEVEL: 'error',
+  logInfo: vi.fn(),
+  logDebug: vi.fn(),
+  logError: vi.fn(),
+  logWarn: vi.fn(),
+}));
+
+const NOW = new Date('2030-01-15T10:00:00Z');
+const settings = {
+  isAvailable: vi.fn(async () => false),
+  prefsOf: vi.fn(async (userId: number) => ({ enabled: userId !== 3, ai_enabled: true, auto_add: true })),
+};
+const cursor = {
+  find: vi.fn(),
+  acquireLease: vi.fn(async () => true),
+  releaseLease: vi.fn(async () => undefined),
+  recordOk: vi.fn(async () => undefined),
+  forgetOk: vi.fn(async () => undefined),
+};
+const addresses = { listHolders: vi.fn(async () => [1, 2, 3]) };
+const notify = { importPaused: vi.fn(async () => undefined) };
+const registrar = { isEnabled: () => false, register: vi.fn() };
+
+let job: MailIngestPauseJob;
+beforeEach(async () => {
+  vi.clearAllMocks();
+  settings.isAvailable.mockResolvedValue(false);
+  cursor.find.mockResolvedValue(null);
+  cursor.acquireLease.mockResolvedValue(true);
+  const moduleRef = await Test.createTestingModule({
+    providers: [
+      MailIngestPauseJob,
+      { provide: MailIngestSettingsService, useValue: settings },
+      { provide: MailIngestNotifyService, useValue: notify },
+      { provide: CronRegistrarService, useValue: registrar },
+      { provide: getRepositoryToken(MailIngestCursor), useValue: cursor },
+      { provide: getRepositoryToken(MailIngestAddresses), useValue: addresses },
+    ],
+  }).compile();
+  job = moduleRef.get(MailIngestPauseJob);
+});
+
+describe('the pause notice', () => {
+  it('tells each holder of a live address with their import on, once, after noting the pause', async () => {
+    expect(await job.tick(NOW)).toBe(2);
+    expect(notify.importPaused.mock.calls).toEqual([[1], [2]]);
+    expect(cursor.recordOk).toHaveBeenCalledWith('notice', 'paused', '2030-01-15 10:00:00');
+    expect(cursor.recordOk.mock.invocationCallOrder[0]).toBeLessThan(notify.importPaused.mock.invocationCallOrder[0]);
+    expect(cursor.releaseLease).toHaveBeenCalled();
+
+    cursor.find.mockResolvedValue({ last_ok_at: '2030-01-15 10:00:00' });
+    expect(await job.tick(NOW)).toBe(0);
+    expect(notify.importPaused).toHaveBeenCalledTimes(2);
+  });
+
+  it('leaves the notice to the process that holds it', async () => {
+    cursor.acquireLease.mockResolvedValue(false);
+    expect(await job.tick(NOW)).toBe(0);
+    expect(addresses.listHolders).not.toHaveBeenCalled();
+  });
+
+  it('forgets the pause once the import is back, and tells nobody while it runs', async () => {
+    settings.isAvailable.mockResolvedValue(true);
+    expect(await job.tick(NOW)).toBe(0);
+    expect(cursor.forgetOk).not.toHaveBeenCalled();
+    cursor.find.mockResolvedValue({ last_ok_at: '2030-01-15 10:00:00' });
+    expect(await job.tick(NOW)).toBe(0);
+    expect(cursor.forgetOk).toHaveBeenCalledWith('notice', 'paused');
+    expect(notify.importPaused).not.toHaveBeenCalled();
+    expect(cursor.acquireLease).not.toHaveBeenCalled();
+  });
+
+  it('logs a failed run and gives the lease back', async () => {
+    addresses.listHolders.mockRejectedValueOnce(new Error('db gone'));
+    expect(await job.tick(NOW)).toBe(0);
+    expect(cursor.recordOk).not.toHaveBeenCalled();
+    expect(cursor.releaseLease).toHaveBeenCalledWith('notice', 'paused', expect.any(String));
+  });
+});

server/tests/unit/nest/notification-preferences.service.test.ts

@@ -501,5 +501,6 @@ describe('the mail import’s events follow its switch', () => {
     const { event_types } = await svc.getPreferencesMatrix(user.id, 'user');
     expect(event_types).toContain('mail_booking_added');
     expect(event_types).toContain('mail_booking_pending');
+    expect(event_types).toContain('mail_ingest_paused');
   });
 });

server/tests/unit/nest/notifications.service.test.ts

@@ -652,6 +652,7 @@ describe('send() — in-app notification content', () => {
       targetId: user.id,
       params: { count: '2' },
     });
+    await send({ event: 'mail_ingest_paused', actorId: null, scope: 'user', targetId: user.id, params: {} });
 
     const notifs = await getInAppNotifications(user.id);
     expect(notifs.map((n) => [n.text_key, n.navigate_target])).toEqual(
@@ -659,6 +660,7 @@ describe('send() — in-app notification content', () => {
         ['notif.mail_booking_added.text', '/trips/7?tab=bookings'],
         ['notif.mail_booking_added.shared', '/trips/7?tab=bookings'],
         ['notif.mail_booking_pending.text', '/dashboard?inbox=mail'],
+        ['notif.mail_ingest_paused.text', null],
       ]),
     );
   });