fix(local): a refused backfill exits non-zero; overlap SQL tested on real PG (TASK-503)

backend/__tests__/service/backfillLocalMessagesToPg.test.js

@@ -0,0 +1,152 @@
+/**
+ * TASK-503 — the local-upgrade backfill against real PostgreSQL and MongoDB.
+ *
+ * The unit suite fakes every dependency, so it cannot see the overlap SQL or
+ * copyOne's ON CONFLICT. Here both run for real. Overlap is asserted as a
+ * delta, since other suites may leave unmarked rows in a shared database.
+ */
+const fs = require('fs');
+const path = require('path');
+const mongoose = require('mongoose');
+const { Pool } = require('pg');
+const { setupMongoDb, closeMongoDb, clearMongoDb } = require('../utils/testUtils');
+
+const RUN = process.env.INTEGRATION_TEST === 'true';
+const d = RUN ? describe : describe.skip;
+const MARKER = 'tier1-local-backfill-503';
+
+let pool;
+let MongoMessage;
+let backfillLocalMessages;
+let productionDeps;
+let ensureLegacyIdIndex;
+
+const POD = new mongoose.Types.ObjectId().toHexString();
+const USER = new mongoose.Types.ObjectId().toHexString();
+
+const connect = () => new Pool({
+  host: process.env.PG_HOST,
+  port: Number(process.env.PG_PORT || 5432),
+  database: process.env.PG_DATABASE,
+  user: process.env.PG_USER,
+  password: process.env.PG_PASSWORD,
+  ssl: false,
+});
+
+const mongoMessage = (content, createdAt = new Date()) => MongoMessage.create({
+  podId: POD, userId: USER, content: `${MARKER} ${content}`, messageType: 'text', createdAt,
+});
+
+const pgUnmarked = (content, createdAt) => pool.query(
+  'INSERT INTO messages (pod_id, user_id, content, created_at) VALUES ($1, $2, $3, $4)',
+  [POD, USER, `${MARKER} ${content}`, createdAt],
+);
+
+const pgCopies = async () => (await pool.query(
+  'SELECT legacy_mongo_id FROM messages WHERE content LIKE $1 AND legacy_mongo_id IS NOT NULL',
+  [`${MARKER}%`],
+)).rows.map((r) => r.legacy_mongo_id).sort();
+
+d('TASK-503: local-upgrade backfill into PostgreSQL', () => {
+  beforeAll(async () => {
+    pool = connect();
+    await pool.query(fs.readFileSync(path.join(__dirname, '../../config/schema.sql'), 'utf8'));
+    // Added by migration 0001, which schema.sql alone does not carry.
+    await pool.query('ALTER TABLE messages ADD COLUMN IF NOT EXISTS legacy_mongo_id TEXT');
+    await pool.query(
+      `INSERT INTO pods (id, name, type, created_by) VALUES ($1, $2, 'chat', $3)
+       ON CONFLICT (id) DO NOTHING`,
+      [POD, 'local-backfill-503', USER],
+    );
+    await setupMongoDb();
+    // eslint-disable-next-line global-require
+    MongoMessage = require('../../models/Message');
+    // eslint-disable-next-line global-require
+    ({ backfillLocalMessages, productionDeps } = require('../../scripts/backfill-local-messages-to-pg'));
+    // eslint-disable-next-line global-require
+    ({ ensureLegacyIdIndex } = require('../../services/pgMessageReconcileService'));
+    expect(await ensureLegacyIdIndex()).toBeTruthy();
+  });
+
+  beforeEach(async () => {
+    await pool.query('DELETE FROM messages WHERE content LIKE $1', [`${MARKER}%`]);
+    await clearMongoDb();
+  });
+
+  afterAll(async () => {
+    await pool.query('DELETE FROM messages WHERE content LIKE $1', [`${MARKER}%`]);
+    await pool.query('DELETE FROM pods WHERE id = $1', [POD]);
+    await pool.end();
+    await closeMongoDb();
+    // eslint-disable-next-line global-require
+    const { pool: appPool } = require('../../config/db-pg');
+    if (appPool?.end) await appPool.end();
+  });
+
+  test('overlap counts only unmarked PG rows written at or before the given time', async () => {
+    const deps = productionDeps(pool);
+    const at = new Date('2026-01-10T00:00:00Z');
+    const before = await deps.countOverlap(at);
+    await pgUnmarked('older', new Date('2026-01-01T00:00:00Z'));
+    await pgUnmarked('same instant', at);
+    await pgUnmarked('newer', new Date('2026-02-01T00:00:00Z'));
+    await pool.query(
+      `INSERT INTO messages (pod_id, user_id, content, created_at, legacy_mongo_id)
+       VALUES ($1, $2, $3, $4, $5)`,
+      [POD, USER, `${MARKER} marked`, new Date('2026-01-01T00:00:00Z'), new mongoose.Types.ObjectId().toHexString()],
+    );
+    expect(await deps.countOverlap(at)).toBe(before + 2);
+  });
+
+  test('a dry run writes nothing, --apply copies once, and a rerun copies nothing', async () => {
+    const docs = [await mongoMessage('a'), await mongoMessage('b'), await mongoMessage('c')];
+    const ids = docs.map((doc) => doc._id.toString()).sort();
+    const deps = productionDeps(pool);
+    // force: other suites may leave unmarked rows; the refusal has its own test.
+    const dry = await backfillLocalMessages(deps, { force: true });
+    expect(dry).toMatchObject({ dryRun: true, pending: 3, moved: 0 });
+    expect(await pgCopies()).toEqual([]);
+
+    const applied = await backfillLocalMessages(deps, { apply: true, force: true, batchSize: 2 });
+    expect(applied).toMatchObject({ pending: 3, moved: 3, failed: 0 });
+    expect(await pgCopies()).toEqual(ids);
+
+    const rerun = await backfillLocalMessages(deps, { apply: true, force: true });
+    expect(rerun).toMatchObject({ pending: 0, moved: 0 });
+    expect(await pgCopies()).toEqual(ids);
+  });
+
+  test('a pod that exists only in MongoDB is mirrored into PG, then its message copies', async () => {
+    const mongoOnlyPod = new mongoose.Types.ObjectId();
+    await mongoose.connection.db.collection('pods').insertOne({
+      _id: mongoOnlyPod,
+      name: 'mongo-only-backfill-503',
+      type: 'chat',
+      createdBy: new mongoose.Types.ObjectId(USER),
+      members: [new mongoose.Types.ObjectId(USER)],
+    });
+    const doc = await MongoMessage.create({
+      podId: mongoOnlyPod, userId: USER, content: `${MARKER} mongo-only pod`, messageType: 'text',
+    });
+    try {
+      const report = await backfillLocalMessages(productionDeps(pool), { apply: true, force: true });
+      expect(report).toMatchObject({ pending: 1, moved: 1, failed: 0 });
+      expect(await pgCopies()).toEqual([doc._id.toString()]);
+      const { rows } = await pool.query('SELECT id FROM pods WHERE id = $1', [mongoOnlyPod.toHexString()]);
+      expect(rows).toHaveLength(1);
+    } finally {
+      await pool.query('DELETE FROM messages WHERE pod_id = $1', [mongoOnlyPod.toHexString()]);
+      await pool.query('DELETE FROM pods WHERE id = $1', [mongoOnlyPod.toHexString()]);
+    }
+  });
+
+  test('refuses when PG holds unmarked history older than the newest Mongo message', async () => {
+    await pgUnmarked('copied by an older migration', new Date('2026-01-01T00:00:00Z'));
+    await mongoMessage('same history', new Date('2026-01-02T00:00:00Z'));
+    const report = await backfillLocalMessages(productionDeps(pool), { apply: true });
+    expect(report.refused).toBe(true);
+    expect(report.overlap).toBeGreaterThanOrEqual(1);
+    expect(report.moved).toBe(0);
+    expect(await pgCopies()).toEqual([]);
+  });
+});

backend/__tests__/unit/scripts/backfillLocalMessagesToPg.test.js

@@ -1,6 +1,6 @@
 jest.mock('../../../services/pgMessageReconcileService', () => ({ copyOne: jest.fn(), ensureLegacyIdIndex: jest.fn() }));
 
-const { backfillLocalMessages } = require('../../../scripts/backfill-local-messages-to-pg');
+const { backfillLocalMessages, main } = require('../../../scripts/backfill-local-messages-to-pg');
 
 const doc = (id) => ({ _id: { toString: () => id }, podId: 'p', userId: 'u', content: id, createdAt: new Date('2026-01-01') });
 
@@ -59,3 +59,34 @@ describe('backfillLocalMessages', () => {
     expect(report).toMatchObject({ moved: 2, failed: 1 });
   });
 });
+
+describe('main() exit code', () => {
+  const run = async (deps, argv = ['node', 'script', '--apply']) => {
+    const close = jest.fn(async () => {});
+    const log = jest.spyOn(console, 'log').mockImplementation(() => {});
+    const before = process.exitCode;
+    try {
+      const code = await main(argv, async () => ({ deps, close }));
+      return { code, exitCode: process.exitCode, close };
+    } finally {
+      process.exitCode = before;
+      log.mockRestore();
+    }
+  };
+
+  it('a refused run sets process.exitCode to 2 and closes the stores', async () => {
+    const deps = fakeDeps({ overlap: 4 });
+    const { code, exitCode, close } = await run(deps);
+    expect(exitCode).toBe(2);
+    expect(code).toBe(2);
+    expect(deps.copy).not.toHaveBeenCalled();
+    expect(close).toHaveBeenCalledTimes(1);
+  });
+
+  it('failed rows set 1; a clean run and a clean dry run set 0; a refused dry run sets 2', async () => {
+    expect((await run(fakeDeps({ copy: async () => { throw new Error('boom'); } }))).exitCode).toBe(1);
+    expect((await run(fakeDeps())).exitCode).toBe(0);
+    expect((await run(fakeDeps(), ['node', 'script'])).exitCode).toBe(0);
+    expect((await run(fakeDeps({ overlap: 4 }), ['node', 'script'])).exitCode).toBe(2);
+  });
+});

backend/scripts/backfill-local-messages-to-pg.ts

@@ -25,6 +25,9 @@
  *
  *   npm run backfill:local-messages            # report only
  *   npm run backfill:local-messages -- --apply
+ *
+ * Exit codes: 0 done, or a dry run that would proceed; 1 some rows failed;
+ * 2 refused (a dry run that finds the overlap refuses too, so it also exits 2).
  */
 /* eslint-disable no-console, no-await-in-loop */
 
@@ -96,7 +99,7 @@ interface PgPool {
   end: () => Promise<void>;
 }
 
-const productionDeps = (pool: PgPool): LocalBackfillDeps => {
+export const productionDeps = (pool: PgPool): LocalBackfillDeps => {
   // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require
   const Message = require('../models/Message');
   // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require
@@ -127,18 +130,46 @@ const productionDeps = (pool: PgPool): LocalBackfillDeps => {
   };
 };
 
-export async function main(): Promise<void> {
-  const apply = process.argv.includes('--apply');
-  const force = process.argv.includes('--force');
+/**
+ * The process exit code for a finished run: 1 when rows failed, 2 when the run
+ * refused, so an upgrade script that runs this stops instead of carrying on
+ * without history; 0 otherwise (including a dry run).
+ */
+export const exitCodeFor = (report: Pick<LocalBackfillReport, 'refused' | 'failed'>): number => {
+  if (report.failed) return 1;
+  if (report.refused) return 2;
+  return 0;
+};
+
+export interface OpenedStores { deps: LocalBackfillDeps; close: () => Promise<void> }
+
+/** Connects both stores for a real run. main() takes it as a parameter so a test can swap it. */
+export const openStores = async (): Promise<OpenedStores> => {
   if (!process.env.MONGO_URI) throw new Error('MONGO_URI is required');
   // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require
   const { pool } = require('../config/db-pg') as { pool: PgPool | null };
   if (!pool) throw new Error('PostgreSQL is not configured (PG_HOST unset)');
-
   await mongoose.connect(process.env.MONGO_URI);
+  const close = async () => { await mongoose.disconnect(); await pool.end(); };
   try {
     if (!await ensureLegacyIdIndex()) throw new Error('could not ensure the legacy_mongo_id index');
-    const report = await backfillLocalMessages(productionDeps(pool), { apply, force });
+  } catch (err) {
+    await close();
+    throw err;
+  }
+  return { deps: productionDeps(pool), close };
+};
+
+/** Runs the backfill, sets process.exitCode (see exitCodeFor) and returns it. */
+export async function main(
+  argv: string[] = process.argv,
+  open: () => Promise<OpenedStores> = openStores,
+): Promise<number> {
+  const apply = argv.includes('--apply');
+  const force = argv.includes('--force');
+  const { deps, close } = await open();
+  try {
+    const report = await backfillLocalMessages(deps, { apply, force });
     console.log(`[backfill-local-messages] ${JSON.stringify(report)}`);
     if (report.refused) {
       console.log(`REFUSED — PostgreSQL already holds ${report.overlap} messages with no Mongo id`
@@ -147,10 +178,10 @@ export async function main(): Promise<void> {
     } else if (!apply) {
       console.log('DRY RUN — nothing written. Re-run with --apply.');
     }
-    if (report.failed) process.exitCode = 1;
+    process.exitCode = exitCodeFor(report);
+    return process.exitCode;
   } finally {
-    await mongoose.disconnect();
-    await pool.end();
+    await close();
   }
 }