feat(messages): reconcile Mongo fallback messages into Postgres (TASK-313 part 2)

backend/__tests__/service/pgMessageReconcile.test.js

@@ -0,0 +1,321 @@
+/**
+ * TASK-313 part 2 — Mongo fallback messages reconciled into PostgreSQL.
+ *
+ * Runs against real PostgreSQL and real MongoDB (Tier 1). The claims need a
+ * database that enforces them: ON CONFLICT on a unique index, a foreign key
+ * from messages.pod_id to pods(id), and two concurrent passes racing on the
+ * same rows. pg-mem can do none of that faithfully.
+ */
+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-reconcile-313';
+
+let pool;
+let MongoMessage;
+let reconcile;
+let copyOne;
+let ensureLegacyIdIndex;
+let PGMessage;
+let MessageReaction;
+
+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 insertFallback = (content, extra = {}) => MongoMessage.create({
+  pgFallback: true,
+  podId: POD,
+  userId: USER,
+  content: `${MARKER} ${content}`,
+  messageType: 'text',
+  createdAt: new Date(),
+  ...extra,
+});
+
+const pgRowsFor = async (mongoId) => (await pool.query(
+  'SELECT id, content, created_at, payload, author_kind FROM messages WHERE legacy_mongo_id = $1',
+  [String(mongoId)],
+)).rows;
+
+d('TASK-313 part 2: reconcile Mongo fallback messages into PostgreSQL', () => {
+  beforeAll(async () => {
+    pool = connect();
+    await pool.query(fs.readFileSync(path.join(__dirname, '../../config/schema.sql'), 'utf8'));
+    await pool.query(
+      `INSERT INTO pods (id, name, type, created_by) VALUES ($1, $2, 'chat', $3)
+       ON CONFLICT (id) DO NOTHING`,
+      [POD, 'reconcile-313', USER],
+    );
+    await setupMongoDb();
+    // The app's own pool and models: this is the code under test.
+    // eslint-disable-next-line global-require
+    MongoMessage = require('../../models/Message');
+    // eslint-disable-next-line global-require
+    ({ reconcileMongoFallbackMessages: reconcile, copyOne, ensureLegacyIdIndex } = require('../../services/pgMessageReconcileService'));
+    // eslint-disable-next-line global-require
+    PGMessage = require('../../models/pg/Message');
+    // eslint-disable-next-line global-require
+    MessageReaction = require('../../models/pg/MessageReaction').default;
+  });
+
+  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();
+    // The app's pool is module-scoped; close it so Jest does not hold the process.
+    // eslint-disable-next-line global-require
+    const { pool: appPool } = require('../../config/db-pg');
+    if (appPool?.end) await appPool.end();
+  });
+
+  test('moves a fallback message once and keeps its Mongo id as the legacy id', async () => {
+    const doc = await insertFallback('a');
+
+    const pass = await reconcile();
+
+    expect(pass.moved).toBe(1);
+    const rows = await pgRowsFor(doc._id);
+    expect(rows).toHaveLength(1);
+    expect(rows[0].content).toBe(`${MARKER} a`);
+    expect(new Date(rows[0].created_at).getTime()).toBe(doc.createdAt.getTime());
+    const after = await MongoMessage.findById(doc._id).lean();
+    expect(after.reconciledAt).toBeInstanceOf(Date);
+    expect(after.reconciledPgId).toBe(Number(rows[0].id));
+  });
+
+  test('old unmarked history is never imported but an explicitly marked fallback moves', async () => {
+    const old = await insertFallback('old-history', { pgFallback: undefined, createdAt: new Date('2020-01-01') });
+    const fallback = await insertFallback('marked-fallback');
+    expect((await reconcile()).moved).toBe(1);
+    expect(await pgRowsFor(old._id)).toHaveLength(0);
+    expect(await pgRowsFor(fallback._id)).toHaveLength(1);
+    const unchanged = await MongoMessage.findById(old._id).lean();
+    expect(unchanged.reconciledAt).toBeUndefined();
+    expect(unchanged.reconcileFailedAt).toBeUndefined();
+  });
+
+  test('each message appears once in the Postgres read after a reconcile, and a second pass moves nothing', async () => {
+    await insertFallback('b1');
+    await insertFallback('b2');
+
+    await reconcile();
+    const second = await reconcile();
+
+    expect(second.moved).toBe(0);
+    const listed = await PGMessage.findByPodId(POD, 50);
+    const ours = listed.filter((m) => String(m.content).startsWith(`${MARKER} b`));
+    expect(ours).toHaveLength(2);
+    expect(new Set(ours.map((m) => m._id)).size).toBe(2);
+  });
+
+  test('two concurrent passes leave exactly one Postgres row per message', async () => {
+    const doc = await insertFallback('c');
+
+    await ensureLegacyIdIndex();
+    const [x, y] = await Promise.all([copyOne(doc), copyOne(doc)]);
+
+    expect([x, y].filter((outcome) => outcome === 'moved')).toHaveLength(1);
+    expect(await pgRowsFor(doc._id)).toHaveLength(1);
+    const stamped = await MongoMessage.findById(doc._id).lean();
+    expect(stamped.reconciledAt).toBeInstanceOf(Date);
+  });
+
+  test('another replica holding the index maintenance lock defers the pass', async () => {
+    const doc = await insertFallback('index-lock');
+    const client = await pool.connect();
+    try {
+      await client.query('SELECT pg_advisory_lock(313, 2)');
+      expect((await reconcile()).moved).toBe(0);
+      expect(await pgRowsFor(doc._id)).toHaveLength(0);
+    } finally {
+      await client.query('SELECT pg_advisory_unlock(313, 2)');
+      client.release();
+    }
+    expect((await reconcile()).moved).toBe(1);
+  });
+
+  test('a pass is bounded by maxPerPass and the remainder moves on the next pass', async () => {
+    await insertFallback('d1');
+    await insertFallback('d2');
+    await insertFallback('d3');
+
+    const first = await reconcile({ maxPerPass: 2 });
+    const second = await reconcile({ maxPerPass: 2 });
+
+    expect(first.moved).toBe(2);
+    expect(second.moved).toBe(1);
+  });
+
+  test('a missing PG pod is deferred without blocking later rows, then retries after sync', async () => {
+    const orphan = await insertFallback('e-orphan', { podId: new mongoose.Types.ObjectId().toHexString() });
+    const doc = await insertFallback('e-ok');
+
+    const pass = await reconcile({ maxPerPass: 1 });
+    const next = await reconcile({ maxPerPass: 1 });
+
+    expect(pass.skipped).toBe(1);
+    const refused = await MongoMessage.findById(orphan._id).lean();
+    expect(refused.reconcileFailedAt).toBeUndefined();
+    expect(refused.reconcileRetryAt).toBeInstanceOf(Date);
+    await pool.query(`INSERT INTO pods (id, name, type, created_by) VALUES ($1, 'late-pod', 'chat', $2)`, [String(orphan.podId), USER]);
+    await MongoMessage.updateOne({ _id: orphan._id }, { $set: { reconcileRetryAt: new Date(0) } });
+    expect((await reconcile()).moved).toBe(1);
+    expect(await pgRowsFor(orphan._id)).toHaveLength(1);
+    await pool.query('DELETE FROM messages WHERE pod_id = $1', [String(orphan.podId)]);
+    await pool.query('DELETE FROM pods WHERE id = $1', [String(orphan.podId)]);
+    expect(await pgRowsFor(doc._id)).toHaveLength(1);
+    expect(next.moved).toBe(1);
+  });
+
+  test('more than a full batch of poison rows cannot starve a valid message', async () => {
+    const bad = Array.from({ length: 201 }, () => ({
+      pgFallback: true, podId: new mongoose.Types.ObjectId(POD), userId: new mongoose.Types.ObjectId(USER),
+      content: `${MARKER} bad\0`, createdAt: new Date(0),
+    }));
+    await MongoMessage.collection.insertMany([...bad,
+      { pgFallback: true, content: `${MARKER} missing-pod`, userId: new mongoose.Types.ObjectId(USER), createdAt: new Date(0) },
+      { pgFallback: true, content: `${MARKER} missing-user`, podId: new mongoose.Types.ObjectId(POD), createdAt: new Date(0) },
+    ]);
+    const good = await insertFallback('after-poison');
+    expect((await reconcile()).skipped).toBe(200);
+    expect((await reconcile()).moved).toBe(1);
+    expect(await pgRowsFor(good._id)).toHaveLength(1);
+    expect(await MongoMessage.countDocuments({ reconcileFailedAt: { $exists: true } })).toBe(203);
+    expect((await reconcile()).skipped).toBe(0);
+  });
+
+  test('retry-ready orphan rows do not crowd out unattempted messages', async () => {
+    await MongoMessage.collection.insertMany(Array.from({ length: 201 }, () => ({
+      pgFallback: true, podId: new mongoose.Types.ObjectId(), userId: new mongoose.Types.ObjectId(USER),
+      content: `${MARKER} retry-orphan`, createdAt: new Date(0), reconcileRetryAt: new Date(0),
+    })));
+    const good = await insertFallback('new-after-retries');
+    expect((await reconcile({ maxPerPass: 1 })).moved).toBe(1);
+    expect(await pgRowsFor(good._id)).toHaveLength(1);
+  });
+
+  test('a second copy leaves the first reconciledAt and pgId unchanged', async () => {
+    const doc = await insertFallback('set-once');
+    await copyOne(doc);
+    const first = await MongoMessage.findById(doc._id).lean();
+    await new Promise((resolve) => setTimeout(resolve, 5));
+    await copyOne(doc);
+    const second = await MongoMessage.findById(doc._id).lean();
+    expect(second.reconciledAt).toEqual(first.reconciledAt);
+    expect(second.reconciledPgId).toBe(first.reconciledPgId);
+    expect(await pgRowsFor(doc._id)).toHaveLength(1);
+  });
+
+  test('a conflict row deleted before lookup is not stamped with a NaN pgId', async () => {
+    const doc = await insertFallback('deleted-during-copy');
+    const { pool: appPool } = require('../../config/db-pg');
+    const spy = jest.spyOn(appPool, 'query').mockResolvedValueOnce({ rows: [] }).mockResolvedValueOnce({ rows: [] });
+    await expect(copyOne(doc)).rejects.toThrow('disappeared');
+    spy.mockRestore();
+    const after = await MongoMessage.findById(doc._id).lean();
+    expect(after.reconciledAt).toBeUndefined();
+    expect(after.reconciledPgId).toBeUndefined();
+  });
+
+  test('card payload and kernel authorship survive reconciliation', async () => {
+    const payload = { type: 'decision', options: ['continue'] };
+    const doc = await insertFallback('card', { messageType: 'card', payload, authorKind: 'kernel' });
+    await reconcile();
+    expect((await pgRowsFor(doc._id))[0]).toMatchObject({ payload, author_kind: 'kernel' });
+  });
+
+  test('reconciled IDs work as replies and thread roots and reject missing or cross-pod targets', async () => {
+    const doc = await insertFallback('thread-root');
+    await reconcile();
+    const { resolveThreadRoot, assertReplyTargetInPod } = require('../../services/threadRootResolver');
+    const root = await resolveThreadRoot({ podId: POD, threadRootId: String(doc._id) });
+    await expect(assertReplyTargetInPod({ podId: POD, replyToMessageId: String(doc._id) })).resolves.toBeUndefined();
+    const reply = await PGMessage.create(POD, USER, `${MARKER} reply`, 'text', String(doc._id));
+    expect(Number(reply.reply_to_message_id)).toBe(root);
+    expect(Number(reply.thread_root_id)).toBe(root);
+    await expect(resolveThreadRoot({ podId: POD, threadRootId: new mongoose.Types.ObjectId().toHexString() })).rejects.toMatchObject({ code: 'thread_root_invalid' });
+    await expect(assertReplyTargetInPod({ podId: new mongoose.Types.ObjectId().toHexString(), replyToMessageId: String(doc._id) })).rejects.toMatchObject({ code: 'reply_target_unavailable' });
+  });
+
+  test('all-digit ObjectIds resolve as Mongo IDs rather than PG integers', async () => {
+    const doc = await insertFallback('digit-id', { _id: new mongoose.Types.ObjectId('123456789012345678901234') });
+    await reconcile();
+    expect(await PGMessage.resolveId(String(doc._id))).toBe(String((await pgRowsFor(doc._id))[0].id));
+  });
+
+  test('a pod that exists only in Mongo is backfilled and its message then moves', async () => {
+    const orphanPod = new mongoose.Types.ObjectId().toHexString();
+    await mongoose.connection.db.collection('pods').insertOne({
+      _id: new mongoose.Types.ObjectId(orphanPod),
+      name: 'reconcile-313-backfill',
+      type: 'chat',
+      createdBy: new mongoose.Types.ObjectId(USER),
+      members: [new mongoose.Types.ObjectId(USER)],
+    });
+    const doc = await insertFallback('i', { podId: orphanPod });
+
+    const pass = await reconcile();
+
+    expect(pass.moved).toBe(1);
+    expect(await pgRowsFor(doc._id)).toHaveLength(1);
+    expect((await MongoMessage.findById(doc._id).lean()).reconcileFailedAt).toBeUndefined();
+    await pool.query('DELETE FROM pods WHERE id = $1', [orphanPod]);
+  });
+
+  test('a moved message resolves its Mongo id for delete', async () => {
+    const doc = await insertFallback('f');
+    await reconcile();
+    const pgId = await PGMessage.resolveId(String(doc._id));
+    expect(pgId).toBe(String((await pgRowsFor(doc._id))[0].id));
+
+    await PGMessage.delete(String(doc._id));
+
+    expect(await pgRowsFor(doc._id)).toHaveLength(0);
+  });
+
+  test.each(['mongo', 'pg'])('delete by %s ID removes the unstamped availability copy before another pass', async (idKind) => {
+    const doc = await insertFallback('crashed-stamp-delete');
+    await copyOne(doc);
+    const pgId = await PGMessage.resolveId(String(doc._id));
+    await MongoMessage.updateOne({ _id: doc._id }, { $unset: { reconciledAt: 1, reconciledPgId: 1 } });
+    const controller = require('../../controllers/messageController');
+    const res = { status: jest.fn().mockReturnThis(), json: jest.fn() };
+    await controller.deleteMessage({ params: { id: idKind === 'mongo' ? String(doc._id) : pgId }, userId: USER }, res);
+    expect(res.json).toHaveBeenCalledWith({ msg: 'Message deleted' });
+    expect(await MongoMessage.findById(doc._id).lean()).toBeNull();
+    expect(await pgRowsFor(doc._id)).toHaveLength(0);
+    expect((await reconcile()).moved).toBe(0);
+    expect(await pgRowsFor(doc._id)).toHaveLength(0);
+  });
+
+  test('a moved message resolves its Mongo id for reactions', async () => {
+    const doc = await insertFallback('g');
+    await reconcile();
+    const pgId = await PGMessage.resolveId(String(doc._id));
+
+    await MessageReaction.add(pgId, USER, '👍');
+
+    const summaries = await MessageReaction.listForMessage(pgId, USER);
+    expect(summaries.map((s) => s.emoji)).toEqual(['👍']);
+  });
+});

backend/__tests__/unit/controllers/messageController.test.js

@@ -8,6 +8,8 @@ const { AgentInstallation } = require('../../../models/AgentRegistry');
 jest.mock('../../../models/Pod');
 jest.mock('../../../models/Message', () => ({
   create: jest.fn(),
+  deleteOne: jest.fn(),
+  findByIdAndDelete: jest.fn(),
   findById: jest.fn(),
   find: jest.fn(),
 }));
@@ -395,7 +397,7 @@ describe('messageController', () => {
       await messageController.createMessage(req, res);
 
       expect(MongoMessage.create).toHaveBeenCalledWith({
-        podId: 'p1', userId: 'u1', content: mongoFallbackRow.content, messageType: 'text',
+        pgFallback: true, podId: 'p1', userId: 'u1', content: mongoFallbackRow.content, messageType: 'text',
       });
       expect(res.status).not.toHaveBeenCalled();
       expect(res.json).toHaveBeenCalledWith(expect.objectContaining({
@@ -599,18 +601,53 @@ describe('messageController', () => {
 
   describe('deleteMessage', () => {
     it('returns 404 when message not found', async () => {
+      PGMessage.resolveId.mockResolvedValue(null);
       PGMessage.findById.mockResolvedValue(null);
       const req = { params: { id: 'm1' }, userId: 'u1', user: { id: 'u1' } };
       const res = { status: jest.fn().mockReturnThis(), json: jest.fn() };
       await messageController.deleteMessage(req, res);
       expect(res.status).toHaveBeenCalledWith(404);
     });
 
+    it('deletes the resolved PG row and availability copy for a reconciled Mongo id', async () => {
+      PGMessage.resolveId.mockResolvedValue('41');
+      PGMessage.findById.mockResolvedValue({ user_id: 'u1', pod_id: 'p1' });
+      const res = { status: jest.fn().mockReturnThis(), json: jest.fn() };
+      await messageController.deleteMessage({ params: { id: '123456789012345678901234' }, userId: 'u1' }, res);
+      expect(PGMessage.delete).toHaveBeenCalledWith('41');
+      expect(MongoMessage.deleteOne).toHaveBeenCalledWith({ $or: [
+        { reconciledPgId: 41 }, { _id: '123456789012345678901234' },
+      ] });
+      expect(MongoMessage.findByIdAndDelete).not.toHaveBeenCalled();
+      expect(res.json).toHaveBeenCalledWith({ msg: 'Message deleted' });
+    });
+
+    it('refuses deletion of a reconciled row while PG is down without deleting either copy', async () => {
+      PGMessage.resolveId.mockRejectedValue(new Error('PG unavailable'));
+      MongoMessage.findById.mockResolvedValue({ reconciledAt: new Date(), userId: 'u1', podId: 'p1' });
+      const res = { status: jest.fn().mockReturnThis(), json: jest.fn() };
+      await messageController.deleteMessage({ params: { id: '123456789012345678901234' }, userId: 'u1' }, res);
+      expect(res.status).toHaveBeenCalledWith(503);
+      expect(PGMessage.delete).not.toHaveBeenCalled();
+      expect(MongoMessage.deleteOne).not.toHaveBeenCalled();
+      expect(MongoMessage.findByIdAndDelete).not.toHaveBeenCalled();
+    });
+
+    it('deletes an unreconciled Mongo-only row', async () => {
+      PGMessage.resolveId.mockResolvedValue(null);
+      MongoMessage.findById.mockResolvedValue({ _id: 'mongo-only', userId: 'u1', podId: 'p1' });
+      const res = { status: jest.fn().mockReturnThis(), json: jest.fn() };
+      await messageController.deleteMessage({ params: { id: 'mongo-only' }, userId: 'u1' }, res);
+      expect(MongoMessage.findByIdAndDelete).toHaveBeenCalledWith('mongo-only');
+      expect(PGMessage.delete).not.toHaveBeenCalled();
+    });
+
     it('deletes message successfully for message owner', async () => {
       const mockMessage = { user_id: 'u1', pod_id: 'p1' };
+      PGMessage.resolveId.mockResolvedValue('1');
       PGMessage.findById.mockResolvedValue(mockMessage);
       PGMessage.delete.mockResolvedValue(true);
-      const req = { params: { id: 'm1' }, userId: 'u1', user: { id: 'u1' } };
+      const req = { params: { id: '1' }, userId: 'u1', user: { id: 'u1' } };
       const res = { status: jest.fn().mockReturnThis(), json: jest.fn() };
       await messageController.deleteMessage(req, res);
       expect(res.json).toHaveBeenCalledWith({ msg: 'Message deleted' });