diff --git a/src/handlers.ts b/src/handlers.ts index fe454d3..a300f14 100644 --- a/src/handlers.ts +++ b/src/handlers.ts @@ -14,7 +14,7 @@ import { getEmailOf } from './recipient.js'; import { decideSend } from './mode.js'; import { underPerUserLimit } from './limits.js'; import { markResult, release, reserve } from './sends.js'; -import type { NotificationDoc, UserDoc } from './types.js'; +import type { Channel, NotificationDoc, UserDoc } from './types.js'; export interface Deps { cfg: Config; @@ -43,6 +43,25 @@ async function loadPair( return { notif, user }; } +/** + * Mark a notification as terminally processed for a channel WITHOUT sending it. + * Sets a service-owned marker (`processedBy.`) that the poller excludes, + * so a doc we will never send (shadow/dry-run/kill-switch, not in the canary + * cohort, no token/recipient/chat) is not re-enqueued every cycle. It never + * touches the shared notified/emailNotified/telegramNotified flags, so the old + * system is unaffected. + */ +async function markProcessed( + repos: Repos, + id: NotificationDoc['_id'], + channel: Channel, +): Promise { + await repos.notifications.updateOne( + { _id: id }, + { $set: { [`processedBy.${channel}`]: new Date() } }, + ); +} + /** * FCM push handler. Mirrors the old processNotif mobile branch, with all * safeguards. Throws RetryableError on transient failures so BullMQ retries. @@ -60,7 +79,8 @@ export async function processFcmJob(deps: Deps, notificationId: string): Promise if (!user?.fireBaseToken) { log.warn({ notificationId, userId: notif.userId }, 'fcm: user has no fireBaseToken'); - return; // matches old behavior: warn, no send, no mark + await markProcessed(repos, notif._id, 'fcm'); // terminal: no token, don't re-enqueue + return; } const decision = decideSend(cfg, notif); @@ -69,6 +89,7 @@ export async function processFcmJob(deps: Deps, notificationId: string): Promise { notificationId, userId: notif.userId, mode: cfg.mode, reason: decision.reason, title: payload.title, body: payload.body }, `fcm: not sending (${decision.reason})`, ); + await markProcessed(repos, notif._id, 'fcm'); // won't send in this mode; don't re-enqueue return; } @@ -104,6 +125,7 @@ export async function processFcmJob(deps: Deps, notificationId: string): Promise { _id: notif.userId }, { $set: { fireBaseToken: null, fireBaseTokenDeadAt: new Date() } }, ); + await markProcessed(repos, notif._id, 'fcm'); // dead token: terminal, don't re-enqueue log.warn({ notificationId, userId: notif.userId, error: result.error }, 'fcm: dead token purged'); return; } @@ -126,6 +148,7 @@ export async function processEmailJob(deps: Deps, notificationId: string): Promi if (notif.type !== 'web' || notif.emailNotified === true) return; if (!user) { log.warn({ notificationId, userId: notif.userId }, 'email: user not found'); + await markProcessed(repos, notif._id, 'email'); // terminal: no user, don't re-enqueue return; } @@ -135,6 +158,8 @@ export async function processEmailJob(deps: Deps, notificationId: string): Promi if (cfg.mode === 'full' || cfg.mode === 'canary') { await repos.notifications.deleteOne({ _id: notif._id }); await markResult(repos, notif._id, 'email', 'no-recipient'); + } else { + await markProcessed(repos, notif._id, 'email'); // shadow/dry-run: keep doc but don't re-enqueue } log.info({ notificationId, userId: notif.userId }, 'email: no verified recipient'); return; @@ -153,6 +178,7 @@ export async function processEmailJob(deps: Deps, notificationId: string): Promi { notificationId, userId: notif.userId, mode: cfg.mode, reason: decision.reason, to: emailAddress, subject: content.subject }, `email: not sending (${decision.reason})`, ); + await markProcessed(repos, notif._id, 'email'); // won't send in this mode; don't re-enqueue return; } @@ -197,12 +223,14 @@ export async function processTelegramJob(deps: Deps, notificationId: string): Pr if (notif.type !== 'telegram' || notif.telegramNotified === true) return; if (!notif.subsId) { log.warn({ notificationId }, 'telegram: notification has no subsId'); + await markProcessed(repos, notif._id, 'telegram'); // terminal: no subsId, don't re-enqueue return; } const sub = await repos.subscriptions.findOne({ _id: notif.subsId }); if (!sub || typeof sub.chatId !== 'number') { log.warn({ notificationId, subsId: String(notif.subsId) }, 'telegram: subscription/chatId not found'); + await markProcessed(repos, notif._id, 'telegram'); // terminal: no chatId, don't re-enqueue return; } const chatId = sub.chatId; @@ -212,6 +240,7 @@ export async function processTelegramJob(deps: Deps, notificationId: string): Pr const decision = decideSend(cfg, notif); if (!decision.send) { log.info({ notificationId, chatId, mode: cfg.mode, reason: decision.reason }, `telegram: not sending (${decision.reason})`); + await markProcessed(repos, notif._id, 'telegram'); // won't send in this mode; don't re-enqueue return; } @@ -256,6 +285,7 @@ async function handleTgFailure( { _id: subId }, { $set: { active: false, blockedAt: new Date() } } as never, ); + await markProcessed(repos, notif._id, 'telegram'); // blocked: terminal, don't re-enqueue log.warn({ chatId, error: res.error }, 'telegram: bot blocked, sub marked inactive'); return; } diff --git a/src/poller.ts b/src/poller.ts index 25ece1d..c6c1c73 100644 --- a/src/poller.ts +++ b/src/poller.ts @@ -11,11 +11,11 @@ import type { Channel, NotificationDoc, SendJob } from './types.js'; export function pendingFilter(channel: Channel): FilterQuery { switch (channel) { case 'fcm': - return { type: 'mobile', notified: { $ne: true } }; + return { type: 'mobile', notified: { $ne: true }, 'processedBy.fcm': { $exists: false } }; case 'email': - return { type: 'web', emailNotified: { $ne: true } }; + return { type: 'web', emailNotified: { $ne: true }, 'processedBy.email': { $exists: false } }; case 'telegram': - return { type: 'telegram', telegramNotified: { $ne: true } }; + return { type: 'telegram', telegramNotified: { $ne: true }, 'processedBy.telegram': { $exists: false } }; } } diff --git a/src/types.ts b/src/types.ts index 2dbdf4b..1656a41 100644 --- a/src/types.ts +++ b/src/types.ts @@ -21,6 +21,14 @@ export interface NotificationDoc { // provenance set by our poller (invisible to the old system) claimedBy?: string; claimedAt?: Date; + /** + * Per-channel terminal marker set by our sender when a notification will NOT + * be sent and must not be re-enqueued (shadow/dry-run/kill-switch, not in the + * canary cohort, no token/recipient/chat). Service-owned and invisible to the + * old system: unlike `notified`, setting it never makes the old sender skip a + * real send. Stops the poller from re-selecting the same batch forever. + */ + processedBy?: Partial>; } export interface UserDoc { diff --git a/test/fake-repo.ts b/test/fake-repo.ts index afd4fde..87357fc 100644 --- a/test/fake-repo.ts +++ b/test/fake-repo.ts @@ -3,14 +3,51 @@ * MongoDB 3.2, for which no mongod binary is downloadable/runnable on modern * dev/CI hosts (glibc/libssl). So the DB integration tests run against this * fake, which faithfully emulates the subset of behaviour the code relies on: - * the query operators used ($ne/$in/$gte/$exists/equality), $set/$unset updates, - * projection/limit, and — crucially — unique-index duplicate-key errors (11000) - * that back the idempotency guarantee. + * the query operators used ($ne/$in/$gte/$exists/equality), $set/$unset updates + * (including dotted paths like `processedBy.fcm`), projection/limit, and — + * crucially — unique-index duplicate-key errors (11000) that back the + * idempotency guarantee. */ import type { Repos } from '../src/db.js'; type Doc = Record; +/** Read a possibly-dotted path (`a.b.c`) from a doc, like MongoDB does. */ +function getPath(doc: Doc, key: string): any { + if (!key.includes('.')) return doc[key]; + return key.split('.').reduce((cur, part) => (cur == null ? undefined : cur[part]), doc); +} + +/** Set a possibly-dotted path, creating intermediate objects as MongoDB would. */ +function setPath(doc: Doc, key: string, value: any): void { + if (!key.includes('.')) { + doc[key] = value; + return; + } + const parts = key.split('.'); + let cur = doc; + for (const part of parts.slice(0, -1)) { + if (cur[part] == null || typeof cur[part] !== 'object') cur[part] = {}; + cur = cur[part]; + } + cur[parts[parts.length - 1]!] = value; +} + +/** Delete a possibly-dotted path. */ +function delPath(doc: Doc, key: string): void { + if (!key.includes('.')) { + delete doc[key]; + return; + } + const parts = key.split('.'); + let cur = doc; + for (const part of parts.slice(0, -1)) { + if (cur[part] == null) return; + cur = cur[part]; + } + delete cur[parts[parts.length - 1]!]; +} + function isObjectId(v: any): boolean { return v != null && typeof v === 'object' && typeof v.equals === 'function' && v._bsontype === 'ObjectID'; } @@ -64,12 +101,12 @@ function matchField(docVal: any, cond: any): boolean { } function matches(doc: Doc, filter: Doc): boolean { - return Object.entries(filter).every(([k, cond]) => matchField(doc[k], cond)); + return Object.entries(filter).every(([k, cond]) => matchField(getPath(doc, k), cond)); } function applyUpdate(doc: Doc, update: Doc): void { - if (update.$set) Object.assign(doc, update.$set); - if (update.$unset) for (const k of Object.keys(update.$unset)) delete doc[k]; + if (update.$set) for (const [k, v] of Object.entries(update.$set)) setPath(doc, k, v); + if (update.$unset) for (const k of Object.keys(update.$unset)) delPath(doc, k); } class FakeCollection { diff --git a/test/handlers.test.ts b/test/handlers.test.ts index 5220415..4e6e49c 100644 --- a/test/handlers.test.ts +++ b/test/handlers.test.ts @@ -2,6 +2,7 @@ import { beforeEach, describe, expect, it } from 'vitest'; import type { Repos } from '../src/db.js'; import type { Config } from '../src/config.js'; import { processEmailJob, processFcmJob, type Deps } from '../src/handlers.js'; +import { pendingFilter } from '../src/poller.js'; import { aNotif, aUser, fakeFcm, fakeMailer, freshRepos, silentLog, testCfg } from './helpers.js'; let repos: Repos; @@ -60,8 +61,31 @@ describe('FCM handler', () => { expect(fcm.calls).toHaveLength(0); const after = await repos.notifications.findOne({ _id: notif._id }); - expect(after?.notified).toBeUndefined(); - expect(await repos.sends.countDocuments({})).toBe(0); // no reservation in dry-run + expect(after?.notified).toBeUndefined(); // never touches the old system's flag + expect(after?.processedBy?.fcm).toBeInstanceOf(Date); // but marks it processed… + expect(await repos.sends.countDocuments({})).toBe(0); // …with no reservation in dry-run + }); + + it('does not re-enqueue a doc already processed in shadow mode', async () => { + // Regression for the infinite fcm:500/cycle re-enqueue: in shadow the handler + // marks processedBy so the poller filter no longer selects the doc. + const notif = aNotif(); + await repos.notifications.insertOne(notif); + await repos.users.insertOne(aUser()); + const fcm = fakeFcm({ ok: true, messageId: 'x' }); + const shadow = { ...baseCfg, mode: 'shadow' as const }; + + // Before: the poller would select this pending mobile doc. + expect(await repos.notifications.countDocuments(pendingFilter('fcm'))).toBe(1); + + await processFcmJob(deps({ cfg: shadow, fcm }), notif._id.toHexString()); + + expect(fcm.calls).toHaveLength(0); // nothing sent in shadow + const after = await repos.notifications.findOne({ _id: notif._id }); + expect(after?.notified).toBeUndefined(); // old system's flag untouched + expect(after?.processedBy?.fcm).toBeInstanceOf(Date); + // After: the poller no longer selects it -> no re-enqueue. + expect(await repos.notifications.countDocuments(pendingFilter('fcm'))).toBe(0); }); it('kill switch stops sending', async () => { @@ -133,6 +157,7 @@ describe('FCM handler', () => { expect(fcm.calls).toHaveLength(0); const after = await repos.notifications.findOne({ _id: notif._id }); expect(after?.notified).toBeUndefined(); + expect(after?.processedBy?.fcm).toBeInstanceOf(Date); // terminal: won't be re-enqueued }); it('enforces the per-user/day limit', async () => { diff --git a/test/poller.test.ts b/test/poller.test.ts index dd6a5ca..362f050 100644 --- a/test/poller.test.ts +++ b/test/poller.test.ts @@ -28,8 +28,8 @@ function fakeQueues() { describe('pendingFilter', () => { it('uses the CORRECT field names (old cron had a typo)', () => { - expect(pendingFilter('fcm')).toEqual({ type: 'mobile', notified: { $ne: true } }); - expect(pendingFilter('email')).toEqual({ type: 'web', emailNotified: { $ne: true } }); + expect(pendingFilter('fcm')).toEqual({ type: 'mobile', notified: { $ne: true }, 'processedBy.fcm': { $exists: false } }); + expect(pendingFilter('email')).toEqual({ type: 'web', emailNotified: { $ne: true }, 'processedBy.email': { $exists: false } }); }); });