fix(notifications): claim a reminder before sending it, give it back when sending fails

The trip and todo reminder ticks sent with `.catch(() => {})` and then
marked the reminder sent whatever happened. A send that threw (a DB error
resolving the recipients) or whose every channel failed (SMTP down, no
in-app) was recorded as delivered and never retried, without a log line.
And because the mark came after the send, two processes running the same
tick (a rolling update overlap, a stray container on the same volume)
both sent.

Each reminder is now claimed first with one conditional UPDATE through
its repository: reminder_sent_for is set only while it differs from the
start date (and the start date is still the one read), reminded_at only
while it is empty or older than the 20 hour dedup window. Only the
process whose update lands sends. If the send throws, or it attempted
in-app and channels and none went out, the claim is given back, the
failure is logged, and the next tick tries again until the trip starts
or the todo falls out of its window. A send nobody wanted (everyone
opted out) stays marked, since a retry would reach no one.

For that, NotificationsService.send() now reports what it attempted and
what was delivered; a channel that rejects or answers false counts as
not delivered. Existing callers ignore the result.

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

@@ -222,7 +222,7 @@ export class TodoItemsRepository extends TrekRepository<TodoItems> {
     const platform = this.getEntityManager().getPlatform();
     const rows = await this.qb('ti')
       .join('ti.trip', 't')
-      .select(['ti.id', columnRef(platform, 'ti.trip_id'), 'ti.name', 'ti.due_date', columnRef(platform, 'ti.assigned_user_id'), 't.title', 't.user'])
+      .select(['ti.id', columnRef(platform, 'ti.trip_id'), 'ti.name', 'ti.due_date', columnRef(platform, 'ti.assigned_user_id'), 'ti.reminded_at', 't.title', 't.user'])
       .andWhere({ checked: 0 })
       .andWhere({ due_date: { $ne: null } })
       .andWhere({ due_date: { $ne: '' } })
@@ -236,21 +236,32 @@ export class TodoItemsRepository extends TrekRepository<TodoItems> {
       name: row.name,
       due_date: row.due_date,
       assigned_user_id: row.assigned_user_id,
+      reminded_at: row.reminded_at,
       trip_title: row.title,
       trip_owner_id: row.user_id,
     }));
   }
 
   /**
-   * RJ5 (`reminder-jobs.service.ts#todoTick`, looped) — `UPDATE todo_items
-   * SET reminded_at = CURRENT_TIMESTAMP WHERE id = ?`. Stays AFTER the
-   * notification send in call order (plan3f-inputs.md correction #8 — a
-   * documented, not-fixed-by-this-plan at-least-once ordering, unlike
-   * `notifications.service.ts#respond`'s deliberately claim-then-act shape).
+   * RJ5 (`reminder-jobs.service.ts#todoTick`, looped): claims the reminder
+   * before it is sent. `UPDATE todo_items SET reminded_at = CURRENT_TIMESTAMP
+   * WHERE id = ? AND (reminded_at IS NULL OR reminded_at <= now - 20 h)`, the
+   * same dedup bound {@link listDueForReminder} selects by, so of two
+   * processes running the tick only the one whose update lands sends. True
+   * when this caller holds the claim.
    */
-  async markReminded(id: number): Promise<void> {
+  async claimReminder(id: number): Promise<boolean> {
     const platform = this.getEntityManager().getPlatform();
-    await this.nativeUpdate({ id }, { reminded_at: currentTimestamp(platform) });
+    const changed = await this.nativeUpdate(
+      { id, $or: [{ reminded_at: null }, { reminded_at: { $lte: nowMinusHours(platform, 20) } }] },
+      { reminded_at: currentTimestamp(platform) },
+    );
+    return changed === 1;
+  }
+
+  /** Gives a claim back after the send failed, so the next tick tries again: `reminded_at` returns to what it was. */
+  async releaseReminder(id: number, remindedAt: string | null): Promise<void> {
+    await this.nativeUpdate({ id }, { reminded_at: remindedAt });
   }
 }
 
@@ -261,6 +272,7 @@ interface TodoReminderQueryRow {
   name: string;
   due_date: string;
   assigned_user_id: number | null;
+  reminded_at: string | null;
   title: string;
   user_id: number;
 }
@@ -272,6 +284,8 @@ export interface TodoReminderRow {
   name: string;
   due_date: string;
   assigned_user_id: number | null;
+  /** The previous reminder's time, what {@link TodoItemsRepository.releaseReminder} restores. */
+  reminded_at: string | null;
   trip_title: string;
   trip_owner_id: number;
 }

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

@@ -889,9 +889,28 @@ export class TripsRepository extends TrekRepository<Trips> {
       .execute<TripReminderCandidateRow[]>('all', false);
   }
 
-  /** `UPDATE trips SET reminder_sent_for = ? WHERE id = ?`: the start date the reminder went out for. */
-  async markReminderSent(id: number, startDate: string): Promise<void> {
-    await this.nativeUpdate({ id }, { reminder_sent_for: startDate });
+  /**
+   * Claims a trip reminder before it is sent: `UPDATE trips SET
+   * reminder_sent_for = :start WHERE id = ? AND start_date = :start AND
+   * (reminder_sent_for IS NULL OR reminder_sent_for <> :start)`. Of two
+   * processes running the tick only the one whose update lands sends, and a
+   * trip moved since it was read is left for the next tick. True when this
+   * caller holds the claim.
+   */
+  async claimReminder(id: number, startDate: string): Promise<boolean> {
+    const changed = await this.nativeUpdate(
+      { id, start_date: startDate, $or: [{ reminder_sent_for: null }, { reminder_sent_for: { $ne: startDate } }] },
+      { reminder_sent_for: startDate },
+    );
+    return changed === 1;
+  }
+
+  /**
+   * Gives a claim back after the send failed, so the next tick tries again:
+   * `reminder_sent_for` returns to what it was, unless the row moved on since.
+   */
+  async releaseReminder(id: number, startDate: string, previous: string | null): Promise<void> {
+    await this.nativeUpdate({ id, reminder_sent_for: startDate }, { reminder_sent_for: previous });
   }
 
   // ---------------------------------------------------------------------------

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

@@ -257,6 +257,17 @@ export interface NotificationPayload {
   };
 }
 
+/**
+ * What one {@link NotificationsService.send} reached: every in-app notification
+ * and channel message it tried, and how many of those went out. A channel that
+ * rejects or answers `false` (SMTP down, a webhook 5xx) did not deliver.
+ * `attempted` is 0 when nobody wanted the event at all.
+ */
+export interface NotificationDelivery {
+  attempted: number;
+  delivered: number;
+}
+
 /**
  * Should this channel deliver this event to this recipient?
  *
@@ -644,12 +655,13 @@ export class NotificationsService {
 
   // ── Unified dispatcher (from services/notificationService.ts) ─────────────
 
-  async send(payload: NotificationPayload): Promise<void> {
+  async send(payload: NotificationPayload): Promise<NotificationDelivery> {
     const { event, actorId, params, scope, targetId, inApp } = payload;
+    const delivery: NotificationDelivery = { attempted: 0, delivered: 0 };
 
     // Resolve recipients based on scope
     const recipients = await this.resolveRecipients(scope, targetId, actorId);
-    if (recipients.length === 0) return;
+    if (recipients.length === 0) return delivery;
 
     const configEntry = EVENT_NOTIFICATION_CONFIG[event];
     if (!configEntry) {
@@ -770,9 +782,12 @@ export class NotificationsService {
       }
 
       const results = await Promise.allSettled(promises);
+      delivery.attempted += results.length;
       for (const result of results) {
         if (result.status === 'rejected') {
           logError(`notificationService.send channel dispatch failed event=${event} recipient=${recipientId}: ${result.reason instanceof Error ? result.reason.message : result.reason}`);
+        } else if (result.value !== false) {
+          delivery.delivered += 1;
         }
       }
     }));
@@ -792,15 +807,19 @@ export class NotificationsService {
       if (globalChannels.length > 0) {
         const { title, body } = getEventText('en', event, params);
         const msg: ChannelMessage = { event, title, body, navigateTarget: navigateTarget ?? undefined, url: fullLink };
-        await Promise.all(
+        const sent = await Promise.all(
           globalChannels.map(ch =>
             ch.sendGlobal!(msg).catch((err: unknown) => {
               logError(`notificationService.send admin ${ch.id} failed event=${event}: ${err instanceof Error ? err.message : err}`);
+              return false;
             }),
           ),
         );
+        delivery.attempted += sent.length;
+        delivery.delivered += sent.filter((result) => result !== false).length;
       }
     }
+    return delivery;
   }
 }
 

server/src/nest/notifications/reminder-jobs.service.ts

@@ -1,13 +1,13 @@
 import { Injectable, type OnApplicationBootstrap } from '@nestjs/common';
 import { InjectRepository } from '@mikro-orm/nestjs';
-import { logInfo, logError } from '../audit/audit-log.logger';
+import { logInfo, logError, logWarn } from '../audit/audit-log.logger';
 import { AppSettingsRepository } from '../../db/repositories/AppSettings.repository';
 import { TripsRepository } from '../../db/repositories/Trips.repository';
 import { TodoItemsRepository } from '../../db/repositories/TodoItems.repository';
 import { AppSettings } from '../../db/entities/AppSettings.entity';
 import { Trips } from '../../db/entities/Trips.entity';
 import { TodoItems } from '../../db/entities/TodoItems.entity';
-import { NotificationsService } from './notifications.service';
+import { NotificationsService, type NotificationDelivery, type NotificationPayload } from './notifications.service';
 import { CronRegistrarService } from '../scheduling/cron-registrar.service';
 import { addIsoDays } from '@trek/shared';
 import { appClock } from '../common/timezoneService';
@@ -34,6 +34,15 @@ import { appClock } from '../common/timezoneService';
 // leading up to the deadline.
 const TODO_REMINDER_LEAD_DAYS = 3;
 
+/**
+ * Not delivered: the send threw (a DB error resolving the recipients), or it
+ * tried in-app and channels and none of them went out. A send nobody wanted
+ * (every recipient opted out) counts as done, since a retry would reach no one.
+ */
+function undelivered(delivery: NotificationDelivery | null): boolean {
+  return delivery === null || (delivery.attempted > 0 && delivery.delivered === 0);
+}
+
 @Injectable()
 export class ReminderJobsService implements OnApplicationBootstrap {
   constructor(
@@ -83,6 +92,10 @@ export class ReminderJobsService implements OnApplicationBootstrap {
    * date. A day the job did not run is caught up on the next one, and a trip
    * moved to a new start date is reminded again. "Today" is the date in the
    * TZ the job is scheduled in, not UTC.
+   *
+   * Each reminder is claimed before it is sent, so a second process running
+   * the same tick skips it, and given back when the send fails, so the next
+   * tick retries it until the trip starts.
    */
   async tripTick(): Promise<void> {
     try {
@@ -95,20 +108,40 @@ export class ReminderJobsService implements OnApplicationBootstrap {
         && t.start_date >= today
         && addIsoDays(t.start_date, -t.reminder_days) <= today);
 
+      const sent: typeof trips = [];
       for (const trip of trips) {
-        await this.notifications.send({ event: 'trip_reminder', actorId: null, scope: 'trip', targetId: trip.id, params: { trip: trip.title, tripId: String(trip.id) } }).catch(() => {});
-        await this.trips.markReminderSent(trip.id, trip.start_date);
+        if (!(await this.trips.claimReminder(trip.id, trip.start_date))) continue;
+        const delivery = await this.deliver({ event: 'trip_reminder', actorId: null, scope: 'trip', targetId: trip.id, params: { trip: trip.title, tripId: String(trip.id) } });
+        if (undelivered(delivery)) {
+          await this.trips.releaseReminder(trip.id, trip.start_date, trip.reminder_sent_for);
+          logWarn(`Trip reminder for "${trip.title}" was not delivered; the next run tries again`);
+          continue;
+        }
+        sent.push(trip);
       }
 
-      if (trips.length > 0) {
-        logInfo(`Trip reminders sent for ${trips.length} trip(s): ${trips.map(t => `"${t.title}" (${t.reminder_days}d)`).join(', ')}`);
+      if (sent.length > 0) {
+        logInfo(`Trip reminders sent for ${sent.length} trip(s): ${sent.map(t => `"${t.title}" (${t.reminder_days}d)`).join(', ')}`);
       }
     } catch (err: unknown) {
       logError(`Trip reminder check failed: ${err instanceof Error ? err.message : err}`);
     }
   }
 
-  /** Daily check for unchecked todos due inside the lead window. */
+  /** One send, with a failure logged and reported as null instead of ending the tick. */
+  private async deliver(payload: NotificationPayload): Promise<NotificationDelivery | null> {
+    try {
+      return await this.notifications.send(payload);
+    } catch (err: unknown) {
+      logError(`Reminder ${payload.event} for ${payload.scope} ${payload.targetId} failed: ${err instanceof Error ? err.message : err}`);
+      return null;
+    }
+  }
+
+  /**
+   * Daily check for unchecked todos due inside the lead window. Claimed before
+   * the send and given back when it fails, like the trip reminders.
+   */
   async todoTick(): Promise<void> {
     try {
       if ((await this.appSettings.getValue('notify_todo_due')) === 'false') return;
@@ -120,10 +153,12 @@ export class ReminderJobsService implements OnApplicationBootstrap {
 
       const todos = await this.todoItems.listDueForReminder(todayDate, cutoffDate);
 
+      let sent = 0;
       for (const todo of todos) {
+        if (!(await this.todoItems.claimReminder(todo.id))) continue;
         const targetScope: 'user' | 'trip' = todo.assigned_user_id ? 'user' : 'trip';
         const targetId = todo.assigned_user_id ?? todo.trip_id;
-        await this.notifications.send({
+        const delivery = await this.deliver({
           event: 'todo_due',
           actorId: null,
           scope: targetScope,
@@ -134,13 +169,17 @@ export class ReminderJobsService implements OnApplicationBootstrap {
             tripId: String(todo.trip_id),
             due: todo.due_date,
           },
-        }).catch(() => {});
-        // RJ5 stays AFTER the send, unchanged (plan3f-inputs.md correction #8).
-        await this.todoItems.markReminded(todo.id);
+        });
+        if (undelivered(delivery)) {
+          await this.todoItems.releaseReminder(todo.id, todo.reminded_at);
+          logWarn(`Todo reminder for "${todo.name}" was not delivered; the next run tries again`);
+          continue;
+        }
+        sent += 1;
       }
 
-      if (todos.length > 0) {
-        logInfo(`Todo reminders sent for ${todos.length} item(s)`);
+      if (sent > 0) {
+        logInfo(`Todo reminders sent for ${sent} item(s)`);
       }
     } catch (err: unknown) {
       logError(`Todo reminder check failed: ${err instanceof Error ? err.message : err}`);