diff --git a/backend/__tests__/service/backfillLocalMessagesToPg.test.js b/backend/__tests__/service/backfillLocalMessagesToPg.test.js index 8abca942f3..d8955f1156 100644 --- a/backend/__tests__/service/backfillLocalMessagesToPg.test.js +++ b/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')); diff --git a/backend/__tests__/unit/scripts/backfillLocalMessagesToPg.test.js b/backend/__tests__/unit/scripts/backfillLocalMessagesToPg.test.js index dd03db8a91..66c15a0736 100644 --- a/backend/__tests__/unit/scripts/backfillLocalMessagesToPg.test.js +++ b/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(); + }); +}); diff --git a/backend/scripts/backfill-local-messages-to-pg.ts b/backend/scripts/backfill-local-messages-to-pg.ts index 5fb82a9200..993875dbdb 100644 --- a/backend/scripts/backfill-local-messages-to-pg.ts +++ b/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 ` (or `--since=`) 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; } -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 Promise } /** Connects both stores for a real run. main() takes it as a parameter so a test can swap it. */ -export const openStores = async (): Promise => { +export const openStores = async (scope: BackfillScope = {}): Promise => { 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 => { 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 `); + } + return args; }; /** Runs the backfill, sets process.exitCode (see exitCodeFor) and returns it. */ export async function main( argv: string[] = process.argv, - open: () => Promise = openStores, + open: (scope: BackfillScope) => Promise = openStores, ): Promise { - 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.'