From a1ed206985673d7d1daf401296f91159b2fccdc5 Mon Sep 17 00:00:00 2001 From: vjrj Date: Mon, 13 Jul 2026 18:07:06 +0200 Subject: [PATCH] Fase 1b: cableado del matcher (ingesta polling, runner shadow/full, comparador) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - config: bloque matcher (MATCHER_MODE off/shadow/full, IRON_PASSWORD, audiencia, poll, batch) + validacion - db: colecciones subscriptions/activefires/notifications_shadow/matcher_state + ensureMatcherIndexes (2dsphere en notifications_shadow para el dedupe 500m) - matcher/context: MatchContext sobre Mongo real (candidatas geo $near 1000km + audiencia; dedupe geo $near 500m; lang del owner) - matcher/ingest: polling de activefires por createdAt>checkpoint (Mongo 3.2 sin change streams), checkpoint persistido en matcher_state - matcher/runner: loop shadow(->notifications_shadow)/full(->notifications), idempotente por dedupe+checkpoint - matcher/compare: comparador de shadow (shadowOnly=spam, realOnly=faltantes, contentMismatches) — el gate antes del cutover - index: arranca el matcher; fake-repo ampliado (upsert, sort, $gt/$lt) - docs/env/README actualizados 73 tests en verde, typecheck y build OK. --- .env.example | 7 +++ README.md | 22 ++++--- docs/rollout.md | 23 ++++++++ src/config.ts | 30 ++++++++++ src/db.ts | 26 +++++++++ src/index.ts | 9 ++- src/matcher/compare.ts | 55 ++++++++++++++++++ src/matcher/context.ts | 57 ++++++++++++++++++ src/matcher/ingest.ts | 39 +++++++++++++ src/matcher/runner.ts | 77 +++++++++++++++++++++++++ test/compare.test.ts | 52 +++++++++++++++++ test/fake-repo.ts | 42 +++++++++++--- test/matcher-runner.test.ts | 111 ++++++++++++++++++++++++++++++++++++ 13 files changed, 531 insertions(+), 19 deletions(-) create mode 100644 src/matcher/compare.ts create mode 100644 src/matcher/context.ts create mode 100644 src/matcher/ingest.ts create mode 100644 src/matcher/runner.ts create mode 100644 test/compare.test.ts create mode 100644 test/matcher-runner.test.ts diff --git a/.env.example b/.env.example index cb89ea7..bec2925 100644 --- a/.env.example +++ b/.env.example @@ -40,6 +40,13 @@ MAIL_URL=smtps://USER:PASS@smtp.host:465 MAIL_FROM=Todos contra el Fuego # MAIL_REDIRECT_TO=buzon-de-test@example.com # canary/staging: redirige todo el email aquí +# --- Matcher geoespacial (fase 1b: genera notifications a partir de activefires) --- +MATCHER_MODE=off # off | shadow | full +IRON_PASSWORD= # SECRETO: Meteor.settings.private.ironPassword (para sellar notification.sealed) +MATCHER_AUDIENCE=web,mobile # tipos de suscripción que generan notifs aquí (telegram es fase 1c) +MATCHER_POLL_MS=30000 # cada cuánto sondea activefires (Mongo 3.2, sin change streams) +MATCHER_BATCH=2000 # máx fuegos por ciclo + # --- Enlaces / mapas --- ROOT_URL=https://fuegos.comunes.org/ GMAPS_KEY= # Meteor.settings.gmaps.key diff --git a/README.md b/README.md index fe412de..9a1fcde 100644 --- a/README.md +++ b/README.md @@ -5,10 +5,10 @@ Microservicio de **envío de notificaciones** para *Todos contra el Fuego* hacía la web Meteor con la API legacy de GCM (apagada por Google en jun-2024) y saca el envío del observer de Meteor a una cola controlada. -Fase **1a** del [plan de modernización](../plan-modernizacion/fase-1a-notif-workers.md): -solo **workers de envío**. Consume la colección `notifications` que hoy genera -node-red; el *matching* geoespacial (fase 1b) y Telegram (fase 1c) vienen -después. +Fases **1a** (workers de envío) y **1b** (matching geoespacial) del +[plan de modernización](../plan-modernizacion/README.md). La 1a consume la +colección `notifications`; la 1b la **genera** (fuego → subs 2dsphere), +sustituyendo el matching O(fuegos×subs) de node-red. Telegram (1c) viene después. ## Qué hace @@ -68,8 +68,12 @@ sistema) + Redis local. En fase 3 pasa a Docker Compose. ## Estado -- [x] Workers FCM v1 + email, poller, idempotencia, salvaguardas, tests. -- [x] Flag de cutover en Meteor (`notifDisabledChannels`). -- [ ] **Falta**: service account de Firebase (pedir al usuario) para poder - enviar push de verdad. -- [ ] Rollout en shiva (dry-run → shadow → canary → full), con el usuario. +- [x] **1a**: workers FCM v1 + email, poller, idempotencia, salvaguardas, tests. +- [x] **1a**: flag de cutover en Meteor (`notifDisabledChannels`). +- [x] **1a**: canary FCM verificado (push real entregada a dispositivos propios). +- [x] **1b**: matcher geoespacial (content i18n, geolib, `sealed` vía Iron — + compatible con la web —, dedupe 500 m), ingesta por polling, modos + `shadow`/`full`, comparador de shadow. Tests en verde. +- [ ] **1a full**: envío masivo FCM (~8176 users) — con el usuario, vigilando. +- [ ] **1b**: rollout `shadow` ≥3 días → comparar → cutover node-red → `full`. +- [ ] **1c**: envío Telegram. diff --git a/docs/rollout.md b/docs/rollout.md index 7eec22a..e0d643a 100644 --- a/docs/rollout.md +++ b/docs/rollout.md @@ -108,6 +108,29 @@ MODE=full OWNED_CHANNELS=fcm 3. Nada se pierde: los docs pendientes siguen en `notifications` con `notified`/`emailNotified` sin marcar. +## Fase 1b — matcher geoespacial (rollout) + +El matcher **genera** los docs `notifications` (fuego → subs 2dsphere) que hoy +crea node-red. Config: `MATCHER_MODE` (`off`/`shadow`/`full`), `IRON_PASSWORD` +(sella `sealed`; = `settings.private.ironPassword`), `MATCHER_AUDIENCE` +(`web,mobile`). Ingesta por **polling** de `activefires` (Mongo 3.2 no tiene +change streams). Ver `docs/legacy-matching.md`. + +Secuencia (empezar solo con la fase 1a estable): + +1. **`MATCHER_MODE=shadow`** ≥3 días / varios ciclos NASA reales. Escribe en + `notifications_shadow` (no toca `notifications`, node-red sigue generando). +2. **Comparar** con `src/matcher/compare.ts` (`compareShadow`): correr el + comparador sobre la ventana y exigir **`shadowOnly` vacío** (0 notifs de más = + 0 spam) — cada `realOnly` (faltante) se analiza uno a uno; `contentMismatches` + revisados. +3. **Cutover** (con confirmación explícita): desactivar el subflow de matching en + node-red (exportar el flow antes como respaldo; documentar qué nodos) y poner + `MATCHER_MODE=full`. node-red deja de insertar en `notifications`; el matcher + pasa a ser el único origen. Los workers de 1a consumen igual. +4. **Rollback**: reimportar el flow de node-red y `MATCHER_MODE=shadow`/`off`. + Los workers de 1a siguen consumiendo `notifications` venga de donde venga. + ## Checklist de verificación (por paso) - [ ] Tests en verde (`npm test`). diff --git a/src/config.ts b/src/config.ts index 3b75af1..8c78ace 100644 --- a/src/config.ts +++ b/src/config.ts @@ -50,6 +50,18 @@ export interface Config { redirectTo?: string; }; + /** Fase 1b — geospatial matcher (fire → subscriptions → notifications). */ + matcher: { + /** off = disabled; shadow = write to notifications_shadow + compare; full = write to notifications. */ + mode: 'off' | 'shadow' | 'full'; + /** Iron password to seal `notification.sealed`, same as Meteor settings.private.ironPassword. */ + ironPassword?: string; + /** Subscription types this produces notifications for (workers 1a consume web/mobile). */ + audienceTypes: string[]; + pollIntervalMs: number; + batchSize: number; + }; + /** Base URL for building fire/subscription links (Meteor ROOT_URL). */ rootUrl: string; gmaps: { key?: string; fireIconUrl?: string }; @@ -67,6 +79,12 @@ const DEFAULTS = { limits: { perUserPerDay: 20, maxPerBatch: 2000 }, canary: { userIds: [] as string[], subsIds: [] as string[] }, retries: { attempts: 3, backoffMs: 5_000 }, + matcher: { + mode: 'off' as 'off' | 'shadow' | 'full', + audienceTypes: ['web', 'mobile'], + pollIntervalMs: 30_000, + batchSize: 2000, + }, rootUrl: 'https://fuegos.comunes.org/', logLevel: 'info', serviceName: 'tcef-notifications', @@ -109,6 +127,11 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): Config { const mongoUrl = env.MONGO_URL ?? file.mongoUrl; if (!mongoUrl) throw new Error('MONGO_URL is required'); + const matcherMode = (env.MATCHER_MODE as string) ?? file.matcher?.mode ?? DEFAULTS.matcher.mode; + if (!['off', 'shadow', 'full'].includes(matcherMode)) { + throw new Error(`Invalid MATCHER_MODE: ${matcherMode}`); + } + const cfg: Config = { mode, killSwitch: parseBool(env.KILL_SWITCH) ?? file.killSwitch ?? DEFAULTS.killSwitch, @@ -146,6 +169,13 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): Config { from: env.MAIL_FROM ?? file.email?.from ?? 'Todos contra el Fuego ', redirectTo: env.MAIL_REDIRECT_TO ?? file.email?.redirectTo, }, + matcher: { + mode: matcherMode as 'off' | 'shadow' | 'full', + ironPassword: env.IRON_PASSWORD ?? file.matcher?.ironPassword, + audienceTypes: parseList(env.MATCHER_AUDIENCE) ?? file.matcher?.audienceTypes ?? DEFAULTS.matcher.audienceTypes, + pollIntervalMs: Number(env.MATCHER_POLL_MS ?? file.matcher?.pollIntervalMs ?? DEFAULTS.matcher.pollIntervalMs), + batchSize: Number(env.MATCHER_BATCH ?? file.matcher?.batchSize ?? DEFAULTS.matcher.batchSize), + }, rootUrl: env.ROOT_URL ?? file.rootUrl ?? DEFAULTS.rootUrl, gmaps: { key: env.GMAPS_KEY ?? file.gmaps?.key, diff --git a/src/db.ts b/src/db.ts index 717b793..c13535c 100644 --- a/src/db.ts +++ b/src/db.ts @@ -5,6 +5,14 @@ import type { Config } from './config.js'; // Node's cjs-lexer, so pull runtime values off the default (module.exports). const { MongoClient } = mongodb; import type { NotificationDoc, SendRecord, UserDoc } from './types.js'; +import type { FireDoc, SubscriptionDoc } from './matcher/types.js'; + +/** matcher_state: persisted polling checkpoint for fire ingestion. */ +export interface MatcherState { + _id: string; // e.g. 'fires' + checkpoint: Date; + updatedAt: Date; +} export interface Repos { client: mongodb.MongoClient; @@ -12,6 +20,11 @@ export interface Repos { notifications: Collection; users: Collection; sends: Collection; + // Fase 1b — matcher collections + subscriptions: Collection; + activefires: Collection; + notificationsShadow: Collection; + matcherState: Collection; close(): Promise; } @@ -36,6 +49,10 @@ export async function connect(cfg: Config): Promise { notifications, users, sends, + subscriptions: db.collection('subscriptions'), + activefires: db.collection('activefires'), + notificationsShadow: db.collection('notifications_shadow'), + matcherState: db.collection('matcher_state'), close: () => client.close(), }; } @@ -52,3 +69,12 @@ export async function ensureIndexes(repos: Repos): Promise { // For per-user/day rate limiting queries. await repos.sends.createIndex({ createdAt: 1 }, { name: 'created_at' }); } + +/** + * Fase 1b matcher indexes: notifications_shadow needs a 2dsphere index on `geo` + * for the 500 m dedupe $near query (the real `notifications` already has one). + */ +export async function ensureMatcherIndexes(repos: Repos): Promise { + await repos.notificationsShadow.createIndex({ geo: '2dsphere' }, { name: 'geo_2dsphere' }); + await repos.notificationsShadow.createIndex({ userId: 1 }, { name: 'userId' }); +} diff --git a/src/index.ts b/src/index.ts index db6f9aa..00c974e 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,7 +1,8 @@ import { Worker } from 'bullmq'; import { loadConfig } from './config.js'; import { createLogger } from './logger.js'; -import { connect, ensureIndexes } from './db.js'; +import { connect, ensureIndexes, ensureMatcherIndexes } from './db.js'; +import { createMatcherRunner } from './matcher/runner.js'; import { createQueues, redisConnection, QUEUE_NAMES } from './queue.js'; import { createFcmClient, type FcmClient } from './fcm.js'; import { createMailer, type Mailer } from './mailer.js'; @@ -55,9 +56,15 @@ async function main(): Promise { const poller = createPoller(cfg, repos, queues, log); poller.start(); + // Fase 1b — geospatial matcher (fire → subscriptions → notifications). + if (cfg.matcher.mode !== 'off') await ensureMatcherIndexes(repos); + const matcher = createMatcherRunner(cfg, repos, log); + matcher.start(); + const shutdown = async (signal: string): Promise => { log.info({ signal }, 'shutting down'); poller.stop(); + matcher.stop(); await Promise.all(workers.map((w) => w.close())); await Promise.all(Object.values(queues).map((q) => q.close())); await repos.close(); diff --git a/src/matcher/compare.ts b/src/matcher/compare.ts new file mode 100644 index 0000000..d8f428d --- /dev/null +++ b/src/matcher/compare.ts @@ -0,0 +1,55 @@ +import type { NotificationDoc } from '../types.js'; + +/** + * Shadow comparison (docs/rollout.md — fase 1b). Compares the notifications our + * matcher WOULD generate (`notifications_shadow`) against the ones node-red + * actually created (`notifications`) for the same period, so we can prove + * equivalence before cutover. + * + * Match key = userId + fire coordinates + fire time. Content/sealed are compared + * separately for matched pairs (sealed is decrypted-equivalence, not byte-equal, + * so it is NOT part of the key). + * + * The critical bucket is `shadowOnly`: notifications we would send that node-red + * did NOT — i.e. potential over-notification (spam). It must be empty (or every + * case explained) before going to `full`. + */ +export interface ShadowDiff { + shadowOnly: string[]; // keys we generate but node-red didn't (SPAM RISK) + realOnly: string[]; // keys node-red generated but we didn't (MISSING) + matched: number; + contentMismatches: Array<{ key: string; shadow: string; real: string }>; +} + +export function notifKey(n: Pick): string { + const [lon, lat] = n.geo.coordinates; + const when = n.when instanceof Date ? n.when.toISOString() : String(n.when); + return `${n.userId}|${lon},${lat}|${when}`; +} + +export function compareShadow(shadow: NotificationDoc[], real: NotificationDoc[]): ShadowDiff { + const shadowByKey = new Map(); + for (const s of shadow) shadowByKey.set(notifKey(s), s); + const realByKey = new Map(); + for (const r of real) realByKey.set(notifKey(r), r); + + const shadowOnly: string[] = []; + const contentMismatches: ShadowDiff['contentMismatches'] = []; + let matched = 0; + + for (const [key, s] of shadowByKey) { + const r = realByKey.get(key); + if (!r) { + shadowOnly.push(key); + } else { + matched += 1; + if (s.content !== r.content) contentMismatches.push({ key, shadow: s.content, real: r.content }); + } + } + const realOnly: string[] = []; + for (const key of realByKey.keys()) { + if (!shadowByKey.has(key)) realOnly.push(key); + } + + return { shadowOnly, realOnly, matched, contentMismatches }; +} diff --git a/src/matcher/context.ts b/src/matcher/context.ts new file mode 100644 index 0000000..00bff04 --- /dev/null +++ b/src/matcher/context.ts @@ -0,0 +1,57 @@ +import type { Config } from '../config.js'; +import type { Repos } from '../db.js'; +import type { MatchContext } from './matcher.js'; +import type { FireDoc } from './types.js'; + +/** + * Build a MatchContext backed by the real MongoDB (3.2, driver v3.7). + * All geo queries rely on the 2dsphere index on `.geo` (subscriptions already + * has one; notifications and notifications_shadow are ensured, see db.ts). + */ +export function createMatchContext(cfg: Config, repos: Repos): MatchContext { + const shadow = cfg.matcher.mode === 'shadow'; + // In shadow we dedupe against our own shadow output; in full against the real one. + const dedupeColl = shadow ? repos.notificationsShadow : repos.notifications; + const audienceTypes = new Set(cfg.matcher.audienceTypes); + if (!cfg.matcher.ironPassword && cfg.matcher.mode !== 'off') { + throw new Error('IRON_PASSWORD required for the matcher (to seal notification.sealed)'); + } + + return { + async candidateSubs(fire: FireDoc) { + // Coarse geo prefilter (1000 km) + audience type, like the old "near?" node. + return repos.subscriptions + .find({ + type: { $in: cfg.matcher.audienceTypes }, + geo: { + $near: { + $geometry: { type: 'Point', coordinates: [fire.lon, fire.lat] }, + $minDistance: 0, + $maxDistance: 1_000_000, + }, + }, + }) + .toArray(); + }, + async existsNotifNear(userId, lon, lat, meters) { + const doc = await dedupeColl.findOne({ + userId, + geo: { + $near: { + $geometry: { type: 'Point', coordinates: [lon, lat] }, + $minDistance: 0, + $maxDistance: meters, + }, + }, + }); + return doc !== null; + }, + async userLang(userId) { + const u = await repos.users.findOne({ _id: userId }); + return u?.lang; + }, + ironPassword: cfg.matcher.ironPassword ?? '', + audienceTypes, + now: () => new Date(), + }; +} diff --git a/src/matcher/ingest.ts b/src/matcher/ingest.ts new file mode 100644 index 0000000..d0f052f --- /dev/null +++ b/src/matcher/ingest.ts @@ -0,0 +1,39 @@ +import type { Config } from '../config.js'; +import type { Repos } from '../db.js'; +import type { FireDoc } from './types.js'; + +const STATE_ID = 'fires'; + +/** + * Poll `activefires` for fires created since the last checkpoint. MongoDB 3.2 + * has no change streams (docs/legacy-matching.md §0), so ingestion is by + * `createdAt > checkpoint`, ordered ascending, capped at batchSize. + */ +export async function pollNewFires(cfg: Config, repos: Repos): Promise { + const state = await repos.matcherState.findOne({ _id: STATE_ID }); + // First run: only look a little way back to avoid reprocessing history. + const since = state?.checkpoint ?? new Date(Date.now() - cfg.matcher.pollIntervalMs); + return repos.activefires + .find({ createdAt: { $gt: since } }) + .sort({ createdAt: 1 }) + .limit(cfg.matcher.batchSize) + .toArray(); +} + +/** Advance the checkpoint to the newest processed fire's createdAt. */ +export async function advanceCheckpoint(repos: Repos, checkpoint: Date): Promise { + await repos.matcherState.updateOne( + { _id: STATE_ID }, + { $set: { checkpoint, updatedAt: new Date() } }, + { upsert: true }, + ); +} + +/** Newest createdAt among fires (fires must be non-empty). */ +export function newestCreatedAt(fires: FireDoc[]): Date | undefined { + let max: Date | undefined; + for (const f of fires) { + if (f.createdAt && (!max || f.createdAt > max)) max = f.createdAt; + } + return max; +} diff --git a/src/matcher/runner.ts b/src/matcher/runner.ts new file mode 100644 index 0000000..3d0b92c --- /dev/null +++ b/src/matcher/runner.ts @@ -0,0 +1,77 @@ +import type { Config } from '../config.js'; +import type { Repos } from '../db.js'; +import type { Logger } from '../logger.js'; +import { createMatchContext } from './context.js'; +import { advanceCheckpoint, newestCreatedAt, pollNewFires } from './ingest.js'; +import { matchFire, type MatchContext } from './matcher.js'; + +export interface MatcherRunner { + runOnce(): Promise<{ fires: number; generated: number }>; + start(): void; + stop(): void; +} + +/** + * Fase 1b matcher loop: poll new fires, match each against subscriptions, and + * write the resulting notification docs — to `notifications_shadow` (shadow) or + * `notifications` (full). Idempotent across restarts: docs are written before + * the checkpoint advances, and the 500 m dedupe (against the target collection) + * means reprocessing the same fires produces no duplicates. + */ +export function createMatcherRunner( + cfg: Config, + repos: Repos, + log: Logger, + ctx: MatchContext = createMatchContext(cfg, repos), +): MatcherRunner { + let timer: NodeJS.Timeout | null = null; + let running = false; + const target = cfg.matcher.mode === 'shadow' ? repos.notificationsShadow : repos.notifications; + + async function runOnce(): Promise<{ fires: number; generated: number }> { + if (cfg.matcher.mode === 'off' || running) return { fires: 0, generated: 0 }; + running = true; + try { + const fires = await pollNewFires(cfg, repos); + if (fires.length === 0) return { fires: 0, generated: 0 }; + + let generated = 0; + for (const fire of fires) { + const notifs = await matchFire(fire, ctx); + if (notifs.length > 0) { + await target.insertMany(notifs as never[]); + generated += notifs.length; + } + } + + const checkpoint = newestCreatedAt(fires); + if (checkpoint) await advanceCheckpoint(repos, checkpoint); + + log.info({ fires: fires.length, generated, mode: cfg.matcher.mode, target: target.collectionName }, 'matcher: cycle'); + return { fires: fires.length, generated }; + } finally { + running = false; + } + } + + function start(): void { + if (cfg.matcher.mode === 'off') { + log.info('matcher: disabled (MATCHER_MODE=off)'); + return; + } + log.info({ mode: cfg.matcher.mode, intervalMs: cfg.matcher.pollIntervalMs, audience: cfg.matcher.audienceTypes }, 'matcher: starting'); + timer = setInterval(() => { + void runOnce().catch((err) => log.error({ err: String(err) }, 'matcher: cycle error')); + }, cfg.matcher.pollIntervalMs); + void runOnce().catch((err) => log.error({ err: String(err) }, 'matcher: initial cycle error')); + } + + function stop(): void { + if (timer) { + clearInterval(timer); + timer = null; + } + } + + return { runOnce, start, stop }; +} diff --git a/test/compare.test.ts b/test/compare.test.ts new file mode 100644 index 0000000..760f324 --- /dev/null +++ b/test/compare.test.ts @@ -0,0 +1,52 @@ +import { describe, expect, it } from 'vitest'; +import { ObjectId } from 'mongodb'; +import { compareShadow, notifKey } from '../src/matcher/compare.js'; +import type { NotificationDoc } from '../src/types.js'; + +function n(userId: string, lon: number, lat: number, content = 'c'): NotificationDoc { + return { + _id: new ObjectId(), + userId, + content, + geo: { type: 'Point', coordinates: [lon, lat] }, + type: 'mobile', + when: new Date('2026-07-13T12:00:00Z'), + sealed: 's', + }; +} + +describe('compareShadow', () => { + it('matches identical sets', () => { + const a = [n('u1', -3.7, 40.4), n('u2', -4, 41)]; + const b = [n('u1', -3.7, 40.4), n('u2', -4, 41)]; + const diff = compareShadow(a, b); + expect(diff.matched).toBe(2); + expect(diff.shadowOnly).toHaveLength(0); + expect(diff.realOnly).toHaveLength(0); + }); + + it('flags shadowOnly (spam risk) when we generate an extra notification', () => { + const shadow = [n('u1', -3.7, 40.4), n('u3', -5, 42)]; + const real = [n('u1', -3.7, 40.4)]; + const diff = compareShadow(shadow, real); + expect(diff.shadowOnly).toEqual([notifKey(n('u3', -5, 42))]); + expect(diff.realOnly).toHaveLength(0); + }); + + it('flags realOnly (missing) when node-red had one we do not', () => { + const shadow = [n('u1', -3.7, 40.4)]; + const real = [n('u1', -3.7, 40.4), n('u2', -4, 41)]; + const diff = compareShadow(shadow, real); + expect(diff.realOnly).toEqual([notifKey(n('u2', -4, 41))]); + expect(diff.shadowOnly).toHaveLength(0); + }); + + it('reports content mismatches on matched keys', () => { + const shadow = [n('u1', -3.7, 40.4, 'nuevo texto')]; + const real = [n('u1', -3.7, 40.4, 'viejo texto')]; + const diff = compareShadow(shadow, real); + expect(diff.matched).toBe(1); + expect(diff.contentMismatches).toHaveLength(1); + expect(diff.contentMismatches[0]).toMatchObject({ shadow: 'nuevo texto', real: 'viejo texto' }); + }); +}); diff --git a/test/fake-repo.ts b/test/fake-repo.ts index 5d77ac1..afd4fde 100644 --- a/test/fake-repo.ts +++ b/test/fake-repo.ts @@ -47,8 +47,12 @@ function matchField(docVal: any, cond: any): boolean { return Array.isArray(v) && v.some((x) => valEq(docVal, x)); case '$gte': return docVal != null && docVal >= (v as any); + case '$gt': + return docVal != null && docVal > (v as any); case '$lte': return docVal != null && docVal <= (v as any); + case '$lt': + return docVal != null && docVal < (v as any); case '$exists': return (docVal !== undefined) === v; default: @@ -70,6 +74,7 @@ function applyUpdate(doc: Doc, update: Doc): void { class FakeCollection { docs: Doc[] = []; + collectionName = 'fake'; private uniqueFields: string[] | null = null; async createIndex(spec: Doc, opts?: { unique?: boolean }): Promise { @@ -102,11 +107,19 @@ class FakeCollection { return this.docs.find((d) => matches(d, filter)) ?? null; } - async updateOne(filter: Doc, update: Doc): Promise<{ modifiedCount: number }> { + async updateOne(filter: Doc, update: Doc, opts?: { upsert?: boolean }): Promise<{ modifiedCount: number; upsertedCount: number }> { const d = this.docs.find((x) => matches(x, filter)); - if (!d) return { modifiedCount: 0 }; + if (!d) { + if (opts?.upsert) { + const created: Doc = { ...filter }; + applyUpdate(created, update); + this.docs.push(created); + return { modifiedCount: 0, upsertedCount: 1 }; + } + return { modifiedCount: 0, upsertedCount: 0 }; + } applyUpdate(d, update); - return { modifiedCount: 1 }; + return { modifiedCount: 1, upsertedCount: 0 }; } async updateMany(filter: Doc, update: Doc): Promise<{ modifiedCount: number }> { @@ -138,6 +151,16 @@ class FakeCollection { project() { return cursor; }, + sort(spec: Doc) { + const [field, dir] = Object.entries(spec)[0] as [string, number]; + result = [...result].sort((a, b) => { + const av = a[field]; + const bv = b[field]; + if (av === bv) return 0; + return (av < bv ? -1 : 1) * (dir < 0 ? -1 : 1); + }); + return cursor; + }, limit(n: number) { result = result.slice(0, n); return cursor; @@ -151,15 +174,16 @@ class FakeCollection { } export function makeRepos(): Repos { - const notifications = new FakeCollection(); - const users = new FakeCollection(); - const sends = new FakeCollection(); return { client: {} as any, db: {} as any, - notifications: notifications as any, - users: users as any, - sends: sends as any, + notifications: new FakeCollection() as any, + users: new FakeCollection() as any, + sends: new FakeCollection() as any, + subscriptions: new FakeCollection() as any, + activefires: new FakeCollection() as any, + notificationsShadow: new FakeCollection() as any, + matcherState: new FakeCollection() as any, close: async () => {}, }; } diff --git a/test/matcher-runner.test.ts b/test/matcher-runner.test.ts new file mode 100644 index 0000000..7b576f6 --- /dev/null +++ b/test/matcher-runner.test.ts @@ -0,0 +1,111 @@ +import { beforeEach, describe, expect, it } from 'vitest'; +import { ObjectId } from 'mongodb'; +import type { Repos } from '../src/db.js'; +import type { Config } from '../src/config.js'; +import { createMatcherRunner } from '../src/matcher/runner.js'; +import { pollNewFires, advanceCheckpoint, newestCreatedAt } from '../src/matcher/ingest.js'; +import type { MatchContext } from '../src/matcher/matcher.js'; +import type { FireDoc } from '../src/matcher/types.js'; +import { makeRepos } from './fake-repo.js'; +import { silentLog, testCfg } from './helpers.js'; + +let repos: Repos; +beforeEach(() => { + repos = makeRepos(); +}); + +function matcherCfg(mode: 'off' | 'shadow' | 'full', over: Partial = {}): Config { + return testCfg({ matcher: { mode, audienceTypes: ['web', 'mobile'], pollIntervalMs: 1000, batchSize: 2000, ironPassword: 'x'.repeat(32), ...over } }); +} + +function fireDoc(over: Partial = {}): FireDoc { + return { _id: new ObjectId(), lat: 40.4, lon: -3.7, when: new Date('2026-07-13T12:00:00Z'), type: 'viirs', createdAt: new Date('2026-07-13T12:05:00Z'), ...over }; +} + +describe('ingest', () => { + it('polls fires created after the checkpoint, ascending', async () => { + await repos.activefires.insertMany([ + fireDoc({ createdAt: new Date('2026-07-13T10:00:00Z') }), + fireDoc({ createdAt: new Date('2026-07-13T12:00:00Z') }), + fireDoc({ createdAt: new Date('2026-07-13T13:00:00Z') }), + ] as never); + await advanceCheckpoint(repos, new Date('2026-07-13T11:00:00Z')); + const cfg = matcherCfg('shadow'); + + const fires = await pollNewFires(cfg, repos); + expect(fires).toHaveLength(2); // only the 12:00 and 13:00 ones + expect(fires[0]!.createdAt! < fires[1]!.createdAt!).toBe(true); + }); + + it('newestCreatedAt returns the max', () => { + const max = newestCreatedAt([ + fireDoc({ createdAt: new Date('2026-07-13T12:00:00Z') }), + fireDoc({ createdAt: new Date('2026-07-13T14:00:00Z') }), + ]); + expect(max).toEqual(new Date('2026-07-13T14:00:00Z')); + }); +}); + +describe('matcher runner', () => { + // Deterministic context so we don't need geo indexes in the fake. + function fakeCtx(over: Partial = {}): MatchContext { + return { + candidateSubs: async () => [ + { _id: new ObjectId(), location: { lat: 40.4, lon: -3.7 }, geo: { type: 'Point', coordinates: [-3.7, 40.4] }, distance: 50, owner: 'u1', type: 'mobile' }, + ], + existsNotifNear: async () => false, + userLang: async () => 'es', + ironPassword: 'x'.repeat(32), + audienceTypes: new Set(['web', 'mobile']), + now: () => new Date('2026-07-13T12:06:00Z'), + ...over, + }; + } + + it('does nothing when mode is off', async () => { + const cfg = matcherCfg('off'); + const r = createMatcherRunner(cfg, repos, silentLog, fakeCtx()); + const res = await r.runOnce(); + expect(res).toEqual({ fires: 0, generated: 0 }); + }); + + it('shadow mode writes to notifications_shadow, not notifications', async () => { + await repos.activefires.insertOne(fireDoc() as never); + await advanceCheckpoint(repos, new Date('2026-07-13T00:00:00Z')); + const cfg = matcherCfg('shadow'); + const r = createMatcherRunner(cfg, repos, silentLog, fakeCtx()); + + const res = await r.runOnce(); + + expect(res.generated).toBe(1); + expect(await repos.notificationsShadow.countDocuments({})).toBe(1); + expect(await repos.notifications.countDocuments({})).toBe(0); + // checkpoint advanced + const state = await repos.matcherState.findOne({ _id: 'fires' }); + expect(state?.checkpoint).toEqual(fireDoc().createdAt); + }); + + it('full mode writes to notifications', async () => { + await repos.activefires.insertOne(fireDoc() as never); + await advanceCheckpoint(repos, new Date('2026-07-13T00:00:00Z')); + const cfg = matcherCfg('full'); + const r = createMatcherRunner(cfg, repos, silentLog, fakeCtx()); + + const res = await r.runOnce(); + + expect(res.generated).toBe(1); + expect(await repos.notifications.countDocuments({})).toBe(1); + const n = await repos.notifications.findOne({}); + expect(n?.type).toBe('mobile'); + expect(n?.sealed).toContain('Fe26.2'); + }); + + it('generates nothing for fires with no matching subs', async () => { + await repos.activefires.insertOne(fireDoc() as never); + await advanceCheckpoint(repos, new Date('2026-07-13T00:00:00Z')); + const cfg = matcherCfg('shadow'); + const r = createMatcherRunner(cfg, repos, silentLog, fakeCtx({ candidateSubs: async () => [] })); + const res = await r.runOnce(); + expect(res.generated).toBe(0); + }); +});