feat(backfill): bound the Mongo-to-PG message backfill with --since (TASK-503)

backend/__tests__/service/backfillLocalMessagesToPg.test.js

@@ -140,6 +140,21 @@ d('TASK-503: local-upgrade backfill into PostgreSQL', () => {
     }
   });
 
+  test('since copies only Mongo rows created at or after it', async () => {
+    const since = new Date('2026-09-30T00:00:00Z');
+    await mongoMessage('older history', new Date('2026-09-29T23:59:59Z'));
+    const atSince = await mongoMessage('at since', since);
+    const after = await mongoMessage('recent tail', new Date('2026-10-20T00:00:00Z'));
+    const deps = productionDeps(pool, { since });
+
+    const dry = await backfillLocalMessages(deps, { force: true });
+    expect(dry).toMatchObject({ dryRun: true, pending: 2, moved: 0 });
+
+    const applied = await backfillLocalMessages(deps, { apply: true, force: true });
+    expect(applied).toMatchObject({ pending: 2, moved: 2, failed: 0 });
+    expect(await pgCopies()).toEqual([atSince._id.toString(), after._id.toString()].sort());
+  });
+
   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'));

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

@@ -1,6 +1,6 @@
 jest.mock('../../../services/pgMessageReconcileService', () => ({ copyOne: jest.fn(), ensureLegacyIdIndex: jest.fn() }));
 
-const { backfillLocalMessages, main } = require('../../../scripts/backfill-local-messages-to-pg');
+const { backfillLocalMessages, main, parseArgs } = 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') });
 
@@ -90,3 +90,49 @@ describe('main() exit code', () => {
     expect((await run(fakeDeps({ overlap: 4 }), ['node', 'script'])).exitCode).toBe(2);
   });
 });
+
+describe('--since', () => {
+  it('passes the parsed time to the stores, and no flag passes none', async () => {
+    const open = jest.fn(async () => ({ deps: fakeDeps(), close: async () => {} }));
+    const log = jest.spyOn(console, 'log').mockImplementation(() => {});
+    const before = process.exitCode;
+    try {
+      await main(['node', 'script', '--since', '2026-09-30T00:00:00Z'], open);
+      await main(['node', 'script'], open);
+    } finally {
+      process.exitCode = before;
+      log.mockRestore();
+    }
+    expect(open.mock.calls[0][0].since.toISOString()).toBe('2026-09-30T00:00:00.000Z');
+    expect(open.mock.calls[1][0].since).toBeUndefined();
+  });
+
+  it('an unparseable or missing date throws before any store opens', async () => {
+    const open = jest.fn();
+    await expect(main(['node', 'script', '--since', 'last-month'], open)).rejects.toThrow('--since needs an ISO date');
+    await expect(main(['node', 'script', '--since'], open)).rejects.toThrow('got nothing');
+    expect(open).not.toHaveBeenCalled();
+    expect(parseArgs(['node', 'script']).since).toBeUndefined();
+  });
+
+  it('reads the = form the same as the space form', () => {
+    const spaced = parseArgs(['node', 'script', '--since', '2026-09-30T00:00:00Z', '--apply', '--force']);
+    const joined = parseArgs(['node', 'script', '--since=2026-09-30T00:00:00Z', '--apply', '--force']);
+    expect(joined).toEqual(spaced);
+    expect(joined).toMatchObject({ apply: true, force: true });
+    expect(joined.since.toISOString()).toBe('2026-09-30T00:00:00.000Z');
+    expect(() => parseArgs(['node', 'script', '--since='])).toThrow('--since needs an ISO date');
+  });
+
+  it('a misspelt or unknown flag throws before any store opens, so the run cannot go unbounded', async () => {
+    const open = jest.fn();
+    await Promise.all([
+      ['node', 'script', '--snce', '2026-09-30T00:00:00Z', '--apply', '--force'],
+      ['node', 'script', '--since-date=2026-09-30T00:00:00Z', '--apply', '--force'],
+      ['node', 'script', '--apply', '--force', '2026-09-30T00:00:00Z'],
+    ].map((argv) => expect(main(argv, open)).rejects.toThrow('unknown argument')));
+    await expect(main(['node', 'script', '--since', '2026-09-30', '--since=2026-10-01'], open))
+      .rejects.toThrow('more than once');
+    expect(open).not.toHaveBeenCalled();
+  });
+});

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

@@ -23,8 +23,18 @@
  * Dry run by default; `--apply` writes. Replies keep their text but not their
  * reply edge, same as the reconciler.
  *
+ * `--since <ISO date>` (or `--since=<ISO date>`) copies only Mongo rows created
+ * at or after that time; any other argument throws, so a typo cannot widen it.
+ * It is for a deployment that already moved its history into PG and has a
+ * bounded tail of Mongo-only rows left (TASK-503, hosted: socket sends during
+ * PG boot were written to Mongo without the `pgFallback` flag, so the
+ * reconciler never moves them). The overlap check still runs on that tail; a
+ * live deployment has unmarked PG rows in any window, so it needs `--force`
+ * after the dry run shows the expected count.
+ *
  *   npm run backfill:local-messages            # report only
  *   npm run backfill:local-messages -- --apply
+ *   npm run backfill:local-messages -- --since 2026-09-30T00:00:00Z --apply --force
  *
  * 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).
@@ -99,12 +109,16 @@ interface PgPool {
   end: () => Promise<void>;
 }
 
-export const productionDeps = (pool: PgPool): LocalBackfillDeps => {
+export interface BackfillScope { since?: Date }
+
+export const productionDeps = (pool: PgPool, { since }: BackfillScope = {}): 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
   const { syncPodFromMongo } = require('../services/pgPodSyncService');
-  const unreconciled = { reconciledAt: { $exists: false } };
+  const unreconciled = since
+    ? { reconciledAt: { $exists: false }, createdAt: { $gte: since } }
+    : { reconciledAt: { $exists: false } };
   return {
     countPending: () => Message.countDocuments(unreconciled),
     newestPendingAt: async () => {
@@ -144,7 +158,7 @@ export const exitCodeFor = (report: Pick<LocalBackfillReport, 'refused' | 'faile
 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> => {
+export const openStores = async (scope: BackfillScope = {}): 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 };
@@ -157,20 +171,57 @@ export const openStores = async (): Promise<OpenedStores> => {
     await close();
     throw err;
   }
-  return { deps: productionDeps(pool), close };
+  return { deps: productionDeps(pool, scope), close };
+};
+
+export interface BackfillArgs { apply: boolean; force: boolean; since?: Date }
+
+const toSince = (raw: string | undefined): Date => {
+  const since = raw ? new Date(raw) : null;
+  if (!since || Number.isNaN(since.getTime())) {
+    throw new Error(`--since needs an ISO date, got ${raw === undefined ? 'nothing' : JSON.stringify(raw)}`);
+  }
+  return since;
+};
+
+/**
+ * Reads the flags after `node script`. `--since` takes its date as the next
+ * argument or after `=`. Any other argument throws: a misspelt or misread
+ * bound would otherwise run unbounded, and with `--apply --force` that copies
+ * the whole pre-PG history again.
+ */
+export const parseArgs = (argv: string[]): BackfillArgs => {
+  const args: BackfillArgs = { apply: false, force: false };
+  let sinceSeen = false;
+  const setSince = (raw: string | undefined) => {
+    if (sinceSeen) throw new Error('--since given more than once');
+    sinceSeen = true;
+    args.since = toSince(raw);
+  };
+  const rest = argv.slice(2);
+  for (let i = 0; i < rest.length; i += 1) {
+    const arg = rest[i];
+    if (arg === '--apply') args.apply = true;
+    else if (arg === '--force') args.force = true;
+    else if (arg === '--since') {
+      setSince(rest[i + 1]);
+      i += 1;
+    } else if (arg.startsWith('--since=')) setSince(arg.slice('--since='.length));
+    else throw new Error(`unknown argument ${JSON.stringify(arg)}; expected --apply, --force, --since <ISO date>`);
+  }
+  return args;
 };
 
 /** Runs the backfill, sets process.exitCode (see exitCodeFor) and returns it. */
 export async function main(
   argv: string[] = process.argv,
-  open: () => Promise<OpenedStores> = openStores,
+  open: (scope: BackfillScope) => Promise<OpenedStores> = openStores,
 ): Promise<number> {
-  const apply = argv.includes('--apply');
-  const force = argv.includes('--force');
-  const { deps, close } = await open();
+  const { apply, force, since } = parseArgs(argv);
+  const { deps, close } = await open({ since });
   try {
     const report = await backfillLocalMessages(deps, { apply, force });
-    console.log(`[backfill-local-messages] ${JSON.stringify(report)}`);
+    console.log(`[backfill-local-messages] ${JSON.stringify({ ...report, since: since?.toISOString() ?? null })}`);
     if (report.refused) {
       console.log(`REFUSED — PostgreSQL already holds ${report.overlap} messages with no Mongo id`
         + ' that predate the newest Mongo message. They may be the same history; copying would duplicate it.'