diff --git a/server/src/db/repositories/TodoItems.repository.ts b/server/src/db/repositories/TodoItems.repository.ts index 3c6d023c8..3feae7474 100644 --- a/server/src/db/repositories/TodoItems.repository.ts +++ b/server/src/db/repositories/TodoItems.repository.ts @@ -222,7 +222,7 @@ export class TodoItemsRepository extends TrekRepository { 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 { 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 { + async claimReminder(id: number): Promise { 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 { + 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; } diff --git a/server/src/db/repositories/Trips.repository.ts b/server/src/db/repositories/Trips.repository.ts index a5317d224..344c60712 100644 --- a/server/src/db/repositories/Trips.repository.ts +++ b/server/src/db/repositories/Trips.repository.ts @@ -889,9 +889,28 @@ export class TripsRepository extends TrekRepository { .execute('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 { - 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 { + 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 { + await this.nativeUpdate({ id, reminder_sent_for: startDate }, { reminder_sent_for: previous }); } // --------------------------------------------------------------------------- diff --git a/server/src/nest/notifications/notifications.service.ts b/server/src/nest/notifications/notifications.service.ts index f89342f33..7f90d38e4 100644 --- a/server/src/nest/notifications/notifications.service.ts +++ b/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 { + async send(payload: NotificationPayload): Promise { 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; } } diff --git a/server/src/nest/notifications/reminder-jobs.service.ts b/server/src/nest/notifications/reminder-jobs.service.ts index 946562cbe..f176f99fe 100644 --- a/server/src/nest/notifications/reminder-jobs.service.ts +++ b/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 { 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 { + 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 { 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}`); diff --git a/server/tests/helpers/notifications.ts b/server/tests/helpers/notifications.ts index 1b57a757e..02c9a26f1 100644 --- a/server/tests/helpers/notifications.ts +++ b/server/tests/helpers/notifications.ts @@ -95,8 +95,10 @@ export async function makeNotificationPreferencesService(db: Database.Database): * `tsconfig`'s `include`, so leaving it off would not fail to compile. It would * land as `undefined` and throw inside the send. */ -export function notificationsStub(send: NotificationSend = async () => {}): NotificationsService { +export function notificationsStub( + send: NotificationSend = async () => ({ attempted: 0, delivered: 0 }), +): NotificationsService { return { send } as unknown as NotificationsService; } -type NotificationSend = (payload: Parameters[0]) => Promise; +type NotificationSend = (payload: Parameters[0]) => Promise; diff --git a/server/tests/unit/nest/notifications.service.test.ts b/server/tests/unit/nest/notifications.service.test.ts index bc3bfe055..4abf56430 100644 --- a/server/tests/unit/nest/notifications.service.test.ts +++ b/server/tests/unit/nest/notifications.service.test.ts @@ -584,6 +584,49 @@ describe('send() — channel failure resilience', () => { }); +describe('send() reports what it delivered', () => { + it('NSVC-022: counts in-app and every channel that went out', async () => { + const { user } = createUser(testDb); + setSmtp(); + setUserWebhookUrl(user.id); + setNotificationChannels(testDb, 'email,webhook'); + testDb.prepare('UPDATE users SET email = ? WHERE id = ?').run('recipient@test.com', user.id); + const tripId = (testDb.prepare('INSERT INTO trips (title, user_id) VALUES (?, ?)').run('Trip', user.id)).lastInsertRowid as number; + + await expect( + send({ event: 'trip_invite', actorId: null, scope: 'user', targetId: user.id, params: { trip: 'Trip', actor: 'Alice', invitee: 'Bob', tripId: String(tripId) } }), + ).resolves.toEqual({ attempted: 3, delivered: 3 }); + }); + + it('NSVC-023: a channel that fails is attempted but not delivered', async () => { + const { user } = createUser(testDb); + setSmtp(); + setNotificationChannels(testDb, 'email'); + testDb.prepare('UPDATE users SET email = ? WHERE id = ?').run('recipient@test.com', user.id); + disableNotificationPref(testDb, user.id, 'trip_reminder', 'inapp'); + sendMailMock.mockRejectedValueOnce(new Error('SMTP connection refused')); + const tripId = (testDb.prepare('INSERT INTO trips (title, user_id) VALUES (?, ?)').run('Trip', user.id)).lastInsertRowid as number; + + await expect( + send({ event: 'trip_reminder', actorId: null, scope: 'user', targetId: user.id, params: { trip: 'Trip', tripId: String(tripId) } }), + ).resolves.toEqual({ attempted: 1, delivered: 0 }); + }); + + it('NSVC-024: nothing attempted when there is nobody to tell or nobody wants it', async () => { + const { user } = createUser(testDb); + setNotificationChannels(testDb, 'none'); + disableNotificationPref(testDb, user.id, 'collab_message', 'inapp'); + const tripId = (testDb.prepare('INSERT INTO trips (title, user_id) VALUES (?, ?)').run('Trip', user.id)).lastInsertRowid as number; + + await expect( + send({ event: 'collab_message', actorId: null, scope: 'user', targetId: user.id, params: { trip: 'Trip', actor: 'Alice', tripId: String(tripId) } }), + ).resolves.toEqual({ attempted: 0, delivered: 0 }); + await expect( + send({ event: 'booking_change', actorId: user.id, scope: 'trip', targetId: tripId, params: { trip: 'Trip', actor: 'a', booking: 'Hotel', type: 'hotel', tripId: String(tripId) } }), + ).resolves.toEqual({ attempted: 0, delivered: 0 }); + }); +}); + // ── Ntfy dispatch ───────────────────────────────────────────────────────────── function setUserNtfyTopic(userId: number, topic = 'my-trek-topic'): void { diff --git a/server/tests/unit/nest/reminder-jobs.service.test.ts b/server/tests/unit/nest/reminder-jobs.service.test.ts index 584c99186..c6c8f60dd 100644 --- a/server/tests/unit/nest/reminder-jobs.service.test.ts +++ b/server/tests/unit/nest/reminder-jobs.service.test.ts @@ -2,8 +2,9 @@ * ReminderJobsService — the trip + todo reminder crons in their owning domain * (moved from src/scheduler.ts). Proves the bootstrap registration + banners, * the per-tick enable gates, the reminder selection windows, the todo 20h - * dedup via reminded_at, the user-vs-trip scope routing, and that a failing - * tick is contained to a log line. + * dedup via reminded_at, the user-vs-trip scope routing, that a failing + * tick is contained to a log line, and the claim before each send: a second + * process skips a claimed reminder, a failed send gives the claim back. */ import { describe, it, expect, vi, beforeAll, beforeEach } from 'vitest'; @@ -62,7 +63,7 @@ function makeJobs(overrides: { notifications?: NotificationsService } = {}) { // exactly, so every existing assertion below is unaffected. runOnBoot: vi.fn(async (_name: string, fn: () => void | Promise) => { await fn(); }), }; - const send = vi.fn().mockResolvedValue(undefined); + const send = vi.fn().mockResolvedValue({ attempted: 1, delivered: 1 }); const svc = new ReminderJobsService( appSettingsRepo, tripsRepo, @@ -203,7 +204,7 @@ describe('trip reminder tick', () => { const svc = new ReminderJobsService( { getValue: throwGone } as unknown as AppSettingsRepository, { listReminderCandidates: throwGone, countActiveWithReminders: throwGone } as unknown as TripsRepository, - { listDueForReminder: throwGone, markReminded: throwGone } as unknown as TodoItemsRepository, + { listDueForReminder: throwGone, claimReminder: throwGone } as unknown as TodoItemsRepository, broken, { isEnabled: () => true, register: () => true, unregister: () => {} } as unknown as CronRegistrarService, ); @@ -214,6 +215,57 @@ describe('trip reminder tick', () => { }); }); +describe('trip reminder claims', () => { + const sentFor = (id: number) => + (testDb.prepare('SELECT reminder_sent_for FROM trips WHERE id = ?').get(id) as { reminder_sent_for: string | null }).reminder_sent_for; + + it('RJOB-013: a send that throws gives the claim back, logs it, and the next run sends again', async () => { + const { user } = createUser(testDb); + const tripId = tripWithReminder(user.id, 2); + const { svc, send } = makeJobs(); + send.mockRejectedValueOnce(new Error('db gone mid-send')); + + await svc.tripTick(); + expect(sentFor(tripId)).toBeNull(); + expect(logMock.logError).toHaveBeenCalledWith(expect.stringContaining('db gone mid-send')); + expect(logMock.logWarn).toHaveBeenCalledWith('Trip reminder for "Lisbon" was not delivered; the next run tries again'); + expect(logMock.logInfo).not.toHaveBeenCalledWith(expect.stringMatching(/^Trip reminders sent/)); + + await svc.tripTick(); + expect(send).toHaveBeenCalledTimes(2); + expect(sentFor(tripId)).not.toBeNull(); + }); + + it('RJOB-014: every channel failing is retried, a send nobody wanted is not', async () => { + const { user } = createUser(testDb); + const failing = tripWithReminder(user.id, 2, 'Failing'); + const unwanted = tripWithReminder(user.id, 3, 'Unwanted'); + const { svc, send } = makeJobs(); + send.mockImplementation(async (payload: { targetId: number }) => + payload.targetId === failing ? { attempted: 2, delivered: 0 } : { attempted: 0, delivered: 0 }, + ); + + await svc.tripTick(); + expect(sentFor(failing)).toBeNull(); + expect(sentFor(unwanted)).not.toBeNull(); + }); + + it('RJOB-015: a reminder another process claimed first is not sent again', async () => { + const { user } = createUser(testDb); + const tripId = tripWithReminder(user.id, 2); + // Both processes read the candidates; the other one claims first. + const stale = await tripsRepo.listReminderCandidates(); + const start = stale.find((t) => t.id === tripId)!.start_date; + await expect(tripsRepo.claimReminder(tripId, start)).resolves.toBe(true); + vi.spyOn(tripsRepo, 'listReminderCandidates').mockResolvedValueOnce(stale); + + const { svc, send } = makeJobs(); + await svc.tripTick(); + expect(send).not.toHaveBeenCalled(); + await expect(tripsRepo.claimReminder(tripId, start)).resolves.toBe(false); + }); +}); + describe('todo reminder tick', () => { it('RJOB-008 — sends for a due todo, routes trip-scope, stamps reminded_at, and dedups within 20h', async () => { const { user } = createUser(testDb); @@ -276,3 +328,42 @@ describe('todo reminder tick', () => { expect(send).not.toHaveBeenCalled(); }); }); + +describe('todo reminder claims', () => { + function dueTodo(): number { + const { user } = createUser(testDb); + const trip = createTrip(testDb, user.id, { title: 'Lisbon' }); + const todo = createTodoItem(testDb, trip.id, { name: 'Pack bags' }); + testDb.prepare("UPDATE todo_items SET due_date = date('now', '+1 day') WHERE id = ?").run(todo.id); + return todo.id; + } + const remindedAt = (id: number) => + (testDb.prepare('SELECT reminded_at FROM todo_items WHERE id = ?').get(id) as { reminded_at: string | null }).reminded_at; + + it('RJOB-016: a failed send gives the claim back, so the next run sends again', async () => { + const todoId = dueTodo(); + const { svc, send } = makeJobs(); + send.mockResolvedValueOnce({ attempted: 1, delivered: 0 }); + + await svc.todoTick(); + expect(remindedAt(todoId)).toBeNull(); + expect(logMock.logWarn).toHaveBeenCalledWith('Todo reminder for "Pack bags" was not delivered; the next run tries again'); + expect(logMock.logInfo).not.toHaveBeenCalledWith(expect.stringMatching(/^Todo reminders sent/)); + + await svc.todoTick(); + expect(send).toHaveBeenCalledTimes(2); + expect(remindedAt(todoId)).not.toBeNull(); + }); + + it('RJOB-017: a todo another process claimed first is not sent again', async () => { + const todoId = dueTodo(); + const stale = await todoItemsRepo.listDueForReminder(new Date().toISOString().slice(0, 10), '2999-12-31'); + await expect(todoItemsRepo.claimReminder(todoId)).resolves.toBe(true); + vi.spyOn(todoItemsRepo, 'listDueForReminder').mockResolvedValueOnce(stale); + + const { svc, send } = makeJobs(); + await svc.todoTick(); + expect(send).not.toHaveBeenCalled(); + await expect(todoItemsRepo.claimReminder(todoId)).resolves.toBe(false); + }); +});