Fase 1a: driver mongodb v3.7 (produccion es MongoDB 3.2)

El smoke-test contra prod revelo MongoDB 3.2.11 (EOL): el driver v6 exige >=4.2
y no conecta. Cambios:

- mongodb ^3.7.4 + @types/mongodb; db.ts con useUnifiedTopology, sin retryWrites
- sends.ts: deteccion de duplicate-key (11000) sin depender de MongoServerError
- poller.ts: FilterQuery en vez de Filter
- tests de integracion migrados de mongodb-memory-server a un fake en memoria
  (test/fake-repo.ts) porque no hay mongod 3.x ejecutable en hosts modernos;
  emula operadores ($ne/$in/$gte/$exists), $set/$unset e indice unico (11000)
- docs: legacy-behavior.md §0 (entorno MongoDB 3.2, sin change streams -> fase
  1b usara polling; recomendacion de upgrade de Mongo como fase propia)

47 tests en verde, typecheck y build OK. Datos reales del import: 413 pendientes
FCM, 0 email, 8176 users con fireBaseToken.

README.md

@@ -41,11 +41,17 @@ Detalle del despliegue y la secuencia de cutover: [`docs/rollout.md`](docs/rollo
 
 ```bash
 npm ci
-npm test            # 47 tests (vitest + mongodb-memory-server, sin infra externa)
+npm test            # 47 tests (vitest + fake-repo en memoria, sin infra externa)
 npm run typecheck
 npm run build       # -> dist/ (+ plantillas)
 ```
 
+> **Nota MongoDB**: producción corre **MongoDB 3.2** (EOL), así que el servicio
+> usa el driver `mongodb` **v3.7** (el moderno no conecta a 3.2) y no hay change
+> streams (fase 1b usará polling). Los tests van contra un fake en memoria
+> porque no hay mongod tan antiguo ejecutable en hosts modernos. Detalle en
+> [`docs/legacy-behavior.md`](docs/legacy-behavior.md) §0.
+
 ## Ejecución
 
 Config por env (ver [`.env.example`](.env.example)) o `config.json` (ver

docs/legacy-behavior.md

@@ -13,6 +13,24 @@
 > **campo a campo**. Los tests de caracterización (`test/legacy-behavior.test.ts`)
 > lo verifican.
 
+## 0. Entorno de datos (crítico)
+
+- **Producción corre MongoDB 3.2.11** (verificado en shiva vía `buildInfo`,
+  jul-2026). Es EOL desde 2018. Implica:
+  - El servicio usa el **driver `mongodb` v3.7** (el v6 exige servidor ≥ 4.2 y
+    **no conecta** a 3.2). No subir el driver sin subir antes el servidor.
+  - **No hay change streams** (requieren 3.6+) → la fase 1b debe usar el
+    **polling** de respaldo, no change streams.
+  - Los tests de integración corren contra un **fake en memoria**
+    (`test/fake-repo.ts`), porque en hosts modernos no se puede
+    descargar/ejecutar un mongod tan antiguo (glibc/libssl).
+- Cifras reales del import (jul-2026): ~413 docs `notifications` pendientes de
+  FCM, 0 de email, **8176 users con `fireBaseToken`** (muchos llevarán ~2 años
+  muertos → alta tasa de token inválido en el primer envío real).
+- Recomendación de fondo (fuera de fase 1a): planificar un **upgrade de MongoDB**
+  como fase propia; 3.2 bloquea drivers modernos, change streams y es un riesgo
+  de seguridad. Afecta a Meteor 1.6 y node-red, que dependen de él.
+
 ## 1. Esquema del doc `notifications`
 
 Colección `notifications` (db `fuegos`, `idGeneration: 'MONGO'` → `_id` es

package.json

@@ -22,16 +22,16 @@
     "ioredis": "^5.4.2",
     "juice": "^11.0.0",
     "luxon": "^3.5.0",
-    "mongodb": "^6.12.0",
+    "mongodb": "^3.7.4",
     "nodemailer": "^6.9.16",
     "pino": "^9.5.0",
     "pino-pretty": "^13.0.0"
   },
   "devDependencies": {
     "@types/luxon": "^3.4.2",
+    "@types/mongodb": "^3.6.20",
     "@types/node": "^22.10.2",
     "@types/nodemailer": "^6.4.17",
-    "mongodb-memory-server": "^11.2.0",
     "typescript": "^5.7.2",
     "vitest": "^2.1.8"
   }

src/db.ts

@@ -12,9 +12,12 @@ export interface Repos {
 }
 
 export async function connect(cfg: Config): Promise<Repos> {
+  // Production MongoDB is 3.2 (see docs/legacy-behavior.md). We pin the v3.x
+  // driver; useUnifiedTopology is required, and retryWrites (3.6+) is omitted.
   const client = new MongoClient(cfg.mongoUrl, {
-    // The service only writes provenance/idempotency; prefer safety.
+    useUnifiedTopology: true,
-    retryWrites: true,
+    useNewUrlParser: true,
+    serverSelectionTimeoutMS: 10_000,
   });
   await client.connect();
   const db = client.db(cfg.dbName);

src/poller.ts

@@ -1,5 +1,5 @@
 import type { Queue } from 'bullmq';
-import type { Filter } from 'mongodb';
+import type { FilterQuery } from 'mongodb';
 import type { Config } from './config.js';
 import type { Repos } from './db.js';
 import type { Logger } from './logger.js';
@@ -8,7 +8,7 @@ import { jobId } from './queue.js';
 import type { Channel, NotificationDoc, SendJob } from './types.js';
 
 /** Pending-doc filter for a channel — the CORRECT fields (old cron had a typo). */
-export function pendingFilter(channel: Channel): Filter<NotificationDoc> {
+export function pendingFilter(channel: Channel): FilterQuery<NotificationDoc> {
   return channel === 'fcm'
     ? { type: 'mobile', notified: { $ne: true } }
     : { type: 'web', emailNotified: { $ne: true } };

src/sends.ts

@@ -1,7 +1,12 @@
-import { MongoServerError, type ObjectId } from 'mongodb';
+import type { ObjectId } from 'mongodb';
 import type { Repos } from './db.js';
 import type { Channel, OperatingMode, SendRecord } from './types.js';
 
+/** Duplicate-key error (unique index violated). Works across driver versions. */
+function isDuplicateKey(err: unknown): boolean {
+  return typeof err === 'object' && err !== null && (err as { code?: number }).code === 11000;
+}
+
 /**
  * Persistent idempotency. `reserve()` inserts a placeholder into
  * `notification_sends`; the unique index (notificationId, channel) makes a
@@ -28,7 +33,7 @@ export async function reserve(
     await repos.sends.insertOne(rec);
     return true;
   } catch (err) {
-    if (err instanceof MongoServerError && err.code === 11000) {
+    if (isDuplicateKey(err)) {
       return false; // already reserved/sent — idempotent skip
     }
     throw err;

test/fake-repo.ts

@@ -0,0 +1,165 @@
+/**
+ * In-memory fake of the Mongo collections used by the service. Production runs
+ * 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.
+ */
+import type { Repos } from '../src/db.js';
+
+type Doc = Record<string, any>;
+
+function isObjectId(v: any): boolean {
+  return v != null && typeof v === 'object' && typeof v.equals === 'function' && v._bsontype === 'ObjectID';
+}
+
+function valEq(a: any, b: any): boolean {
+  if (a === b) return true;
+  if (a == null || b == null) return a === b;
+  if (isObjectId(a)) {
+    try {
+      return a.equals(b);
+    } catch {
+      return false;
+    }
+  }
+  if (isObjectId(b)) return b.equals(a);
+  if (a instanceof Date && b instanceof Date) return a.getTime() === b.getTime();
+  return false;
+}
+
+function isOperatorObject(cond: any): boolean {
+  if (cond == null || typeof cond !== 'object') return false;
+  if (cond instanceof Date || isObjectId(cond)) return false;
+  const keys = Object.keys(cond);
+  return keys.length > 0 && keys.every((k) => k.startsWith('$'));
+}
+
+function matchField(docVal: any, cond: any): boolean {
+  if (isOperatorObject(cond)) {
+    return Object.entries(cond).every(([op, v]) => {
+      switch (op) {
+        case '$ne':
+          return !valEq(docVal, v);
+        case '$in':
+          return Array.isArray(v) && v.some((x) => valEq(docVal, x));
+        case '$gte':
+          return docVal != null && docVal >= (v as any);
+        case '$lte':
+          return docVal != null && docVal <= (v as any);
+        case '$exists':
+          return (docVal !== undefined) === v;
+        default:
+          throw new Error(`fake-repo: unsupported operator ${op}`);
+      }
+    });
+  }
+  return valEq(docVal, cond);
+}
+
+function matches(doc: Doc, filter: Doc): boolean {
+  return Object.entries(filter).every(([k, cond]) => matchField(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];
+}
+
+class FakeCollection {
+  docs: Doc[] = [];
+  private uniqueFields: string[] | null = null;
+
+  async createIndex(spec: Doc, opts?: { unique?: boolean }): Promise<string> {
+    if (opts?.unique) this.uniqueFields = Object.keys(spec);
+    return 'idx';
+  }
+
+  private violatesUnique(doc: Doc): boolean {
+    if (!this.uniqueFields) return false;
+    return this.docs.some((d) => this.uniqueFields!.every((f) => valEq(d[f], doc[f])));
+  }
+
+  async insertOne(doc: Doc): Promise<{ insertedId: any }> {
+    if (this.violatesUnique(doc)) {
+      const err: any = new Error('E11000 duplicate key error');
+      err.code = 11000;
+      throw err;
+    }
+    const copy = { ...doc };
+    this.docs.push(copy);
+    return { insertedId: copy._id };
+  }
+
+  async insertMany(docs: Doc[]): Promise<{ insertedCount: number }> {
+    for (const d of docs) await this.insertOne(d);
+    return { insertedCount: docs.length };
+  }
+
+  async findOne(filter: Doc = {}): Promise<Doc | null> {
+    return this.docs.find((d) => matches(d, filter)) ?? null;
+  }
+
+  async updateOne(filter: Doc, update: Doc): Promise<{ modifiedCount: number }> {
+    const d = this.docs.find((x) => matches(x, filter));
+    if (!d) return { modifiedCount: 0 };
+    applyUpdate(d, update);
+    return { modifiedCount: 1 };
+  }
+
+  async updateMany(filter: Doc, update: Doc): Promise<{ modifiedCount: number }> {
+    const ds = this.docs.filter((x) => matches(x, filter));
+    for (const d of ds) applyUpdate(d, update);
+    return { modifiedCount: ds.length };
+  }
+
+  async deleteOne(filter: Doc): Promise<{ deletedCount: number }> {
+    const i = this.docs.findIndex((d) => matches(d, filter));
+    if (i === -1) return { deletedCount: 0 };
+    this.docs.splice(i, 1);
+    return { deletedCount: 1 };
+  }
+
+  async deleteMany(filter: Doc = {}): Promise<{ deletedCount: number }> {
+    const before = this.docs.length;
+    this.docs = this.docs.filter((d) => !matches(d, filter));
+    return { deletedCount: before - this.docs.length };
+  }
+
+  async countDocuments(filter: Doc = {}): Promise<number> {
+    return this.docs.filter((d) => matches(d, filter)).length;
+  }
+
+  find(filter: Doc = {}, _opts?: Doc) {
+    let result = this.docs.filter((d) => matches(d, filter));
+    const cursor = {
+      project() {
+        return cursor;
+      },
+      limit(n: number) {
+        result = result.slice(0, n);
+        return cursor;
+      },
+      async toArray() {
+        return result.map((d) => ({ ...d }));
+      },
+    };
+    return cursor;
+  }
+}
+
+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,
+    close: async () => {},
+  };
+}

test/handlers.test.ts

@@ -1,30 +1,14 @@
-import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest';
+import { 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';
+import { aNotif, aUser, fakeFcm, fakeMailer, freshRepos, silentLog, testCfg } from './helpers.js';
 
-let server: MongoMemoryServer;
 let repos: Repos;
-let baseCfg: Config;
+const baseCfg: Config = testCfg();
-
-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({});
+  repos = await freshRepos();
-  await repos.users.deleteMany({});
-  await repos.sends.deleteMany({});
 });
 
 function deps(over: Partial<Deps> = {}): Deps {

test/helpers.ts

@@ -1,21 +1,20 @@
-import { MongoMemoryServer } from 'mongodb-memory-server';
 import { pino } from 'pino';
 import { ObjectId } from 'mongodb';
-import { connect, ensureIndexes, type Repos } from '../src/db.js';
+import type { Repos } from '../src/db.js';
 import type { Config } from '../src/config.js';
 import type { FcmClient, FcmSendResult } from '../src/fcm.js';
 import type { Mailer } from '../src/mailer.js';
 import type { EmailContent, PushPayload } from '../src/messages.js';
 import type { NotificationDoc, UserDoc } from '../src/types.js';
+import { makeRepos } from './fake-repo.js';
 
 export const silentLog = pino({ level: 'silent' });
 
-export async function startMongo(): Promise<{ server: MongoMemoryServer; repos: Repos; cfg: Config }> {
+/** Fresh in-memory repos with the idempotency unique index applied. */
-  const server = await MongoMemoryServer.create();
+export async function freshRepos(): Promise<Repos> {
-  const cfg = testCfg({ mongoUrl: server.getUri(), dbName: 'fuegos' });
+  const repos = makeRepos();
-  const repos = await connect(cfg);
+  await repos.sends.createIndex({ notificationId: 1, channel: 1 }, { unique: true, name: 'uniq' });
-  await ensureIndexes(repos);
+  return repos;
-  return { server, repos, cfg };
 }
 
 export function testCfg(over: Partial<Config> = {}): Config {

test/poller.test.ts

@@ -1,29 +1,17 @@
-import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest';
+import { beforeEach, describe, expect, it } from 'vitest';
 import type { Queue } from 'bullmq';
-import type { MongoMemoryServer } from 'mongodb-memory-server';
 import type { Repos } from '../src/db.js';
 import type { Config } from '../src/config.js';
 import { createPoller } from '../src/poller.js';
 import { pendingFilter } from '../src/poller.js';
 import type { Channel, SendJob } from '../src/types.js';
-import { aNotif, silentLog, startMongo } from './helpers.js';
+import { aNotif, freshRepos, silentLog, testCfg } from './helpers.js';
 
-let server: MongoMemoryServer;
 let repos: Repos;
-let baseCfg: Config;
+const baseCfg: Config = testCfg();
 
-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({});
+  repos = await freshRepos();
 });
 
 /** Fake queue recording add() calls. */