Fase 1a: microservicio tcef-notifications (workers FCM v1 + email)
Sustituye el envío de push por la API legacy de GCM (muerta desde jun-2024) y saca el envío del observer de Meteor a una cola BullMQ controlada. - poller: consume 'notifications' pendientes (campos correctos; el cron viejo tenía un typo nofitied/emailNofitied que nunca casaba) - fcm-worker: firebase-admin (FCM HTTP v1), payload compatible con la app Flutter (FLUTTER_NOTIFICATION_CLICK, collapseKey=_id, data); purga tokens muertos sin reintentar - email-worker: nodemailer + plantillas new-fire portadas verbatim - idempotencia persistente: notification_sends, indice unico (notificationId, channel) -> sobrevive reinicios y replays - salvaguardas anti-spam: modos dry-run/shadow/canary/full, exclusion mutua por canal (OWNED_CHANNELS + notifDisabledChannels en Meteor), kill switch, limites por usuario/dia y circuit breaker por batch - docs/legacy-behavior.md (caracterizacion) + docs/rollout.md - 47 tests (vitest + mongodb-memory-server), typecheck y build OK - deploy: systemd + pm2 (Node 22 standalone) Falta el service account de Firebase para envio real de push (pedir al usuario).
This commit is contained in:
commit
d20e168c90
39 changed files with 8378 additions and 0 deletions
218
test/handlers.test.ts
Normal file
218
test/handlers.test.ts
Normal file
|
|
@ -0,0 +1,218 @@
|
|||
import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest';
|
||||
import type { MongoMemoryServer } from 'mongodb-memory-server';
|
||||
import type { Repos } from '../src/db.js';
|
||||
import type { Config } from '../src/config.js';
|
||||
import { processEmailJob, processFcmJob, type Deps } from '../src/handlers.js';
|
||||
import { aNotif, aUser, fakeFcm, fakeMailer, silentLog, startMongo } from './helpers.js';
|
||||
|
||||
let server: MongoMemoryServer;
|
||||
let repos: Repos;
|
||||
let baseCfg: Config;
|
||||
|
||||
beforeAll(async () => {
|
||||
const s = await startMongo();
|
||||
server = s.server;
|
||||
repos = s.repos;
|
||||
baseCfg = s.cfg;
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await repos.close();
|
||||
await server.stop();
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
await repos.notifications.deleteMany({});
|
||||
await repos.users.deleteMany({});
|
||||
await repos.sends.deleteMany({});
|
||||
});
|
||||
|
||||
function deps(over: Partial<Deps> = {}): Deps {
|
||||
return { cfg: baseCfg, repos, log: silentLog, ...over };
|
||||
}
|
||||
|
||||
describe('FCM handler', () => {
|
||||
it('sends push and marks the doc notified (full mode)', async () => {
|
||||
const notif = aNotif();
|
||||
await repos.notifications.insertOne(notif);
|
||||
await repos.users.insertOne(aUser());
|
||||
const fcm = fakeFcm({ ok: true, messageId: 'mid-1' });
|
||||
|
||||
await processFcmJob(deps({ fcm }), notif._id.toHexString());
|
||||
|
||||
expect(fcm.calls).toHaveLength(1);
|
||||
expect(fcm.calls[0]!.token).toBe('tok-abc');
|
||||
const after = await repos.notifications.findOne({ _id: notif._id });
|
||||
expect(after?.notified).toBe(true);
|
||||
expect(after?.notifiedAt).toBeInstanceOf(Date);
|
||||
const send = await repos.sends.findOne({ notificationId: notif._id, channel: 'fcm' });
|
||||
expect(send?.status).toBe('sent');
|
||||
});
|
||||
|
||||
it('idempotent across a simulated restart: never double-sends', async () => {
|
||||
const notif = aNotif();
|
||||
await repos.notifications.insertOne(notif);
|
||||
await repos.users.insertOne(aUser());
|
||||
const fcm = fakeFcm({ ok: true, messageId: 'mid-1' });
|
||||
|
||||
// First run sends and marks notified.
|
||||
await processFcmJob(deps({ fcm }), notif._id.toHexString());
|
||||
// Simulate a replay/restart: the doc is already notified -> guard skips.
|
||||
await processFcmJob(deps({ fcm }), notif._id.toHexString());
|
||||
// Even if the notified flag were somehow cleared, the send record blocks it.
|
||||
await repos.notifications.updateOne({ _id: notif._id }, { $unset: { notified: '' } });
|
||||
await processFcmJob(deps({ fcm }), notif._id.toHexString());
|
||||
|
||||
expect(fcm.calls).toHaveLength(1); // only ONE real send, ever
|
||||
});
|
||||
|
||||
it('dry-run does not send and does not mark the doc', async () => {
|
||||
const notif = aNotif();
|
||||
await repos.notifications.insertOne(notif);
|
||||
await repos.users.insertOne(aUser());
|
||||
const fcm = fakeFcm({ ok: true, messageId: 'x' });
|
||||
|
||||
await processFcmJob(deps({ cfg: { ...baseCfg, mode: 'dry-run' }, fcm }), notif._id.toHexString());
|
||||
|
||||
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
|
||||
});
|
||||
|
||||
it('kill switch stops sending', async () => {
|
||||
const notif = aNotif();
|
||||
await repos.notifications.insertOne(notif);
|
||||
await repos.users.insertOne(aUser());
|
||||
const fcm = fakeFcm({ ok: true, messageId: 'x' });
|
||||
|
||||
await processFcmJob(deps({ cfg: { ...baseCfg, killSwitch: true }, fcm }), notif._id.toHexString());
|
||||
|
||||
expect(fcm.calls).toHaveLength(0);
|
||||
});
|
||||
|
||||
it('canary sends only to the cohort', async () => {
|
||||
const inCohort = aNotif({ userId: 'u1' });
|
||||
const outCohort = aNotif({ userId: 'u2', subsId: undefined });
|
||||
await repos.notifications.insertMany([inCohort, outCohort]);
|
||||
await repos.users.insertMany([aUser({ _id: 'u1' }), aUser({ _id: 'u2' })]);
|
||||
const fcm = fakeFcm({ ok: true, messageId: 'x' });
|
||||
const cfg = { ...baseCfg, mode: 'canary' as const, canary: { userIds: ['u1'], subsIds: [] } };
|
||||
|
||||
await processFcmJob(deps({ cfg, fcm }), inCohort._id.toHexString());
|
||||
await processFcmJob(deps({ cfg, fcm }), outCohort._id.toHexString());
|
||||
|
||||
expect(fcm.calls).toHaveLength(1);
|
||||
expect(fcm.calls[0]!.token).toBe('tok-abc');
|
||||
});
|
||||
|
||||
it('purges dead FCM tokens and does not retry', async () => {
|
||||
const notif = aNotif();
|
||||
await repos.notifications.insertOne(notif);
|
||||
await repos.users.insertOne(aUser());
|
||||
const fcm = fakeFcm({ ok: false, deadToken: true, error: 'messaging/registration-token-not-registered' });
|
||||
|
||||
await processFcmJob(deps({ fcm }), notif._id.toHexString());
|
||||
|
||||
const user = await repos.users.findOne({ _id: 'u1' });
|
||||
expect(user?.fireBaseToken).toBeNull();
|
||||
expect(user?.fireBaseTokenDeadAt).toBeInstanceOf(Date);
|
||||
const send = await repos.sends.findOne({ notificationId: notif._id, channel: 'fcm' });
|
||||
expect(send?.status).toBe('dead-token'); // recorded -> won't retry
|
||||
const after = await repos.notifications.findOne({ _id: notif._id });
|
||||
expect(after?.notified).toBeUndefined(); // not marked as notified
|
||||
});
|
||||
|
||||
it('transient error releases the reservation so it can retry', async () => {
|
||||
const notif = aNotif();
|
||||
await repos.notifications.insertOne(notif);
|
||||
await repos.users.insertOne(aUser());
|
||||
const fcm = fakeFcm({ ok: false, deadToken: false, error: 'UNAVAILABLE' });
|
||||
|
||||
await expect(processFcmJob(deps({ fcm }), notif._id.toHexString())).rejects.toThrow();
|
||||
// reservation released -> a later retry can proceed
|
||||
expect(await repos.sends.countDocuments({ notificationId: notif._id })).toBe(0);
|
||||
|
||||
const fcm2 = fakeFcm({ ok: true, messageId: 'ok' });
|
||||
await processFcmJob(deps({ fcm: fcm2 }), notif._id.toHexString());
|
||||
expect(fcm2.calls).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('skips users without a fireBaseToken', async () => {
|
||||
const notif = aNotif();
|
||||
await repos.notifications.insertOne(notif);
|
||||
await repos.users.insertOne(aUser({ fireBaseToken: null }));
|
||||
const fcm = fakeFcm({ ok: true, messageId: 'x' });
|
||||
|
||||
await processFcmJob(deps({ fcm }), notif._id.toHexString());
|
||||
|
||||
expect(fcm.calls).toHaveLength(0);
|
||||
const after = await repos.notifications.findOne({ _id: notif._id });
|
||||
expect(after?.notified).toBeUndefined();
|
||||
});
|
||||
|
||||
it('enforces the per-user/day limit', async () => {
|
||||
await repos.users.insertOne(aUser());
|
||||
const cfg = { ...baseCfg, limits: { ...baseCfg.limits, perUserPerDay: 2 } };
|
||||
const fcm = fakeFcm({ ok: true, messageId: 'x' });
|
||||
|
||||
// Three pending notifs for the same user.
|
||||
const notifs = [aNotif(), aNotif(), aNotif()];
|
||||
await repos.notifications.insertMany(notifs);
|
||||
for (const n of notifs) {
|
||||
await processFcmJob(deps({ cfg, fcm }), n._id.toHexString());
|
||||
}
|
||||
|
||||
expect(fcm.calls).toHaveLength(2); // third blocked by limit
|
||||
});
|
||||
});
|
||||
|
||||
describe('Email handler', () => {
|
||||
it('sends email and marks emailNotified', async () => {
|
||||
const notif = aNotif({ type: 'web' });
|
||||
await repos.notifications.insertOne(notif);
|
||||
await repos.users.insertOne(aUser());
|
||||
const mailer = fakeMailer();
|
||||
|
||||
await processEmailJob(deps({ mailer }), notif._id.toHexString());
|
||||
|
||||
expect(mailer.calls).toHaveLength(1);
|
||||
expect(mailer.calls[0]!.to).toBe('ana@example.com');
|
||||
expect(mailer.calls[0]!.content.subject.length).toBeGreaterThan(0);
|
||||
const after = await repos.notifications.findOne({ _id: notif._id });
|
||||
expect(after?.emailNotified).toBe(true);
|
||||
});
|
||||
|
||||
it('removes the doc when there is no verified recipient (full mode)', async () => {
|
||||
const notif = aNotif({ type: 'web' });
|
||||
await repos.notifications.insertOne(notif);
|
||||
await repos.users.insertOne(aUser({ emails: [{ address: 'ana@example.com', verified: false }] }));
|
||||
const mailer = fakeMailer();
|
||||
|
||||
await processEmailJob(deps({ mailer }), notif._id.toHexString());
|
||||
|
||||
expect(mailer.calls).toHaveLength(0);
|
||||
expect(await repos.notifications.findOne({ _id: notif._id })).toBeNull(); // removed
|
||||
});
|
||||
|
||||
it('does not remove the doc for missing recipient in shadow mode', async () => {
|
||||
const notif = aNotif({ type: 'web' });
|
||||
await repos.notifications.insertOne(notif);
|
||||
await repos.users.insertOne(aUser({ emails: [{ address: 'x@y', verified: false }] }));
|
||||
const mailer = fakeMailer();
|
||||
|
||||
await processEmailJob(deps({ cfg: { ...baseCfg, mode: 'shadow' }, mailer }), notif._id.toHexString());
|
||||
|
||||
expect(await repos.notifications.findOne({ _id: notif._id })).not.toBeNull();
|
||||
});
|
||||
|
||||
it('transient SMTP error releases reservation and rethrows', async () => {
|
||||
const notif = aNotif({ type: 'web' });
|
||||
await repos.notifications.insertOne(notif);
|
||||
await repos.users.insertOne(aUser());
|
||||
const mailer = fakeMailer({ fail: true });
|
||||
|
||||
await expect(processEmailJob(deps({ mailer }), notif._id.toHexString())).rejects.toThrow();
|
||||
expect(await repos.sends.countDocuments({ notificationId: notif._id })).toBe(0);
|
||||
});
|
||||
});
|
||||
Loading…
Add table
Add a link
Reference in a new issue