Fase 1b: cableado del matcher (ingesta polling, runner shadow/full, comparador)

- 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.

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<NotificationDoc, 'userId' | 'geo' | 'when'>): 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<string, NotificationDoc>();
+  for (const s of shadow) shadowByKey.set(notifKey(s), s);
+  const realByKey = new Map<string, NotificationDoc>();
+  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 };
+}

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(),
+  };
+}

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<FireDoc[]> {
+  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<void> {
+  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;
+}

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 };
+}