feat(mail-ingest): tell open clients when the inbox changes

Every new waiting entry and every decision sends mailInbox:changed with the count, so a badge follows without a reload.

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

@@ -20,6 +20,7 @@ import { UnitOfWork } from '../database/unit-of-work';
 import { MailReadService } from '../llm-parse/mail-read.service';
 import type { WalkedMail } from '../llm-parse/mail-read.types';
 import { MailIngestApplyService } from './mail-ingest-apply.service';
+import { MailIngestNotifyService } from './mail-ingest-notify.service';
 import { implausible, type TripOption } from './mail-ingest-checks';
 import { MailIngestTripsService } from './mail-ingest-trips.service';
 import { IMPORT_SOURCES, UNDO_WINDOW_DAYS } from './mail-ingest.constants';
@@ -57,6 +58,7 @@ export class MailIngestActionsService {
     @InjectRepository(MailIngestCandidates) private readonly candidates: MailIngestCandidatesRepository,
     @InjectRepository(MailIngestMessages) private readonly messages: MailIngestMessagesRepository,
     private readonly apply: MailIngestApplyService,
+    private readonly notify: MailIngestNotifyService,
     private readonly trips: MailIngestTripsService,
     private readonly reader: MailReadService,
     private readonly airports: AirportsService,
@@ -265,9 +267,11 @@ export class MailIngestActionsService {
     return raw ? this.reader.walk(raw) : null;
   }
 
+  /** The entry as it is now; its person's clients hear the inbox changed. */
   private async reload(id: number): Promise<MailIngestCandidateRow> {
     const row = await this.candidates.findById(id);
     if (!row) throw new DomainError(404, 'Entry not found');
+    this.notify.inboxChanged(row.user_id, await this.candidates.countPending(row.user_id));
     return row;
   }
 }

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

@@ -1,6 +1,7 @@
 import type { MailIngestPersonRow } from '../../db/repositories/MailIngestCandidates.repository';
 import { logError } from '../audit/audit-log.logger';
 import { NotificationsService } from '../notifications/notifications.service';
+import { RealtimeService } from '../realtime/realtime.service';
 import { KEEP_ACTION, UNDO_ACTION } from './mail-ingest.constants';
 import { Injectable } from '@nestjs/common';
 
@@ -22,7 +23,15 @@ interface AddedBooking {
  */
 @Injectable()
 export class MailIngestNotifyService {
-  constructor(private readonly notifications: NotificationsService) {}
+  constructor(
+    private readonly notifications: NotificationsService,
+    private readonly realtime: RealtimeService,
+  ) {}
+
+  /** Tells the person's open clients how many entries wait in their inbox now. */
+  inboxChanged(userId: number, pending: number): void {
+    this.realtime.broadcastToUser(userId, { type: 'mailInbox:changed', pending });
+  }
 
   /** @txStandalone a notice reports a write that committed; a failed one undoes nothing. */
   async bookingAdded(forwarder: MailIngestPersonRow, booking: AddedBooking): Promise<void> {
@@ -63,6 +72,7 @@ export class MailIngestNotifyService {
 
   /** @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);
     await this.quietly(
       this.notifications.send({
         event: 'mail_booking_pending',

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

@@ -8,6 +8,7 @@ import { MailIngestActionsService } from '../../../../src/nest/mail-ingest/mail-
 import { MailIngestNotificationActions } from '../../../../src/nest/mail-ingest/mail-ingest-notification-actions';
 import { MailIngestNotifyService } from '../../../../src/nest/mail-ingest/mail-ingest-notify.service';
 import { NotificationsService } from '../../../../src/nest/notifications/notifications.service';
+import { RealtimeService } from '../../../../src/nest/realtime/realtime.service';
 import { Test } from '@nestjs/testing';
 
 import { beforeEach, describe, expect, it, vi } from 'vitest';
@@ -22,6 +23,7 @@ vi.mock('../../../../src/nest/audit/audit-log.logger', () => ({
 
 const notifications = { send: vi.fn(), registerResponseAction: vi.fn() };
 const actions = { undo: vi.fn(async () => ({})) };
+const realtime = { broadcastToUser: vi.fn() };
 let notify: MailIngestNotifyService;
 let buttons: MailIngestNotificationActions;
 
@@ -34,6 +36,7 @@ beforeEach(async () => {
       MailIngestNotificationActions,
       { provide: NotificationsService, useValue: notifications },
       { provide: MailIngestActionsService, useValue: actions },
+      { provide: RealtimeService, useValue: realtime },
     ],
   }).compile();
   notify = moduleRef.get(MailIngestNotifyService);
@@ -67,6 +70,7 @@ describe('MailIngestNotifyService', () => {
     expect(notifications.send).toHaveBeenCalledWith(
       expect.objectContaining({ event: 'mail_booking_pending', params: { count: '2' } }),
     );
+    expect(realtime.broadcastToUser).toHaveBeenCalledWith(3, { type: 'mailInbox:changed', pending: 2 });
   });
 });