refactor(messages): one pod-message read behind a repository (TASK-503)

backend/__tests__/service/messages.test.js

@@ -69,7 +69,7 @@ describe('Message Routes Integration Tests', () => {
     expect(PGMessage.findByPodId).toHaveBeenCalledWith(
       pod._id.toString(),
       50,
-      undefined,
+      null,
     );
   });
 

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

@@ -139,7 +139,7 @@ describe('messageController', () => {
       // The Mongo fallback serves through the same gate.
       PGMessage.findByPodId.mockRejectedValue(new Error('pg down'));
       MongoMessage.find.mockReturnValue({
-        populate: () => ({ sort: () => ({ limit: async () => [] }) }),
+        populate: () => ({ sort: () => ({ limit: () => ({ lean: async () => [] }) }) }),
       });
       gateWelcomeHostedActions.mockResolvedValueOnce(gated);
       const fallback = { status: jest.fn().mockReturnThis(), json: jest.fn() };

backend/__tests__/unit/repositories/messageRepository.test.js

@@ -0,0 +1,116 @@
+/**
+ * TASK-503: the one pod-message read. PG answers when it can; the MongoDB
+ * fallback lives only here and returns the PG row shape, oldest first.
+ */
+jest.mock('../../../models/pg/Message', () => ({ findByPodId: jest.fn() }));
+jest.mock('../../../models/Message', () => ({ find: jest.fn() }));
+
+const PGMessage = require('../../../models/pg/Message');
+const MongoMessage = require('../../../models/Message');
+const { listPodMessages, toPodMessage } = require('../../../repositories/messageRepository');
+
+const mongoChain = (docs) => {
+  const calls = {};
+  const chain = {
+    populate: (...args) => { calls.populate = args; return chain; },
+    sort: (arg) => { calls.sort = arg; return chain; },
+    limit: (arg) => { calls.limit = arg; return chain; },
+    lean: async () => docs,
+  };
+  MongoMessage.find.mockImplementation((query) => { calls.query = query; return chain; });
+  return calls;
+};
+
+describe('listPodMessages', () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+    jest.spyOn(console, 'warn').mockImplementation(() => {});
+  });
+  afterEach(() => console.warn.mockRestore());
+
+  it('returns the PG rows as they are and never reads MongoDB', async () => {
+    const rows = [{ id: '1' }, { id: '2' }];
+    PGMessage.findByPodId.mockResolvedValue(rows);
+    await expect(listPodMessages('p1', { limit: 20, before: '2026-10-01T00:00:00.000Z' })).resolves.toBe(rows);
+    expect(PGMessage.findByPodId).toHaveBeenCalledWith('p1', 20, '2026-10-01T00:00:00.000Z');
+    expect(MongoMessage.find).not.toHaveBeenCalled();
+  });
+
+  it('clamps the limit to 1..200 for both stores', async () => {
+    PGMessage.findByPodId.mockResolvedValue([]);
+    await listPodMessages('p1', { limit: 5000 });
+    expect(PGMessage.findByPodId).toHaveBeenLastCalledWith('p1', 200, null);
+
+    PGMessage.findByPodId.mockRejectedValue(new Error('pg down'));
+    const calls = mongoChain([]);
+    await listPodMessages('p1', { limit: 5000 });
+    expect(calls.limit).toBe(200);
+  });
+
+  it('on a PG failure reads MongoDB, newest page first, and returns it oldest first', async () => {
+    PGMessage.findByPodId.mockRejectedValue(new Error('pg down'));
+    const calls = mongoChain([
+      { _id: 'm2', podId: 'p1', userId: 'u2', content: 'second', createdAt: new Date('2026-01-02') },
+      { _id: 'm1', podId: 'p1', userId: 'u1', content: 'first', createdAt: new Date('2026-01-01') },
+    ]);
+    const rows = await listPodMessages('p1', { limit: 2, before: '2026-02-01T00:00:00.000Z' });
+    expect(rows.map((r) => r.id)).toEqual(['m1', 'm2']);
+    expect(calls.query).toEqual({ podId: 'p1', createdAt: { $lt: new Date('2026-02-01T00:00:00.000Z') } });
+    expect(calls.sort).toEqual({ createdAt: -1 });
+    expect(calls.populate).toEqual(['userId', 'username profilePicture isBot']);
+  });
+});
+
+describe('toPodMessage', () => {
+  it('maps a populated MongoDB message onto the PG row shape', () => {
+    const at = new Date('2026-01-01');
+    const row = toPodMessage({
+      _id: 'm1',
+      podId: 'p1',
+      userId: { _id: 'u1', username: 'sam', profilePicture: 'pic', isBot: false },
+      content: 'hi',
+      messageType: 'text',
+      authorKind: 'kernel',
+      payload: { kind: 'card' },
+      createdAt: at,
+      updatedAt: at,
+    });
+    expect(row).toEqual({
+      id: 'm1',
+      _id: 'm1',
+      pgId: '',
+      pod_id: 'p1',
+      user_id: 'u1',
+      content: 'hi',
+      text: 'hi',
+      message_type: 'text',
+      messageType: 'text',
+      author_kind: 'kernel',
+      payload: { kind: 'card' },
+      created_at: at,
+      createdAt: at,
+      updated_at: at,
+      username: 'sam',
+      profile_picture: 'pic',
+      is_bot: false,
+      userId: { _id: 'u1', username: 'sam', profilePicture: 'pic', isBot: false },
+      replyTo: null,
+    });
+  });
+
+  it('keeps an unpopulated author as a plain id, and an unknown bot flag absent', () => {
+    const row = toPodMessage({ _id: 'm1', podId: 'p1', userId: 'u1', content: 'hi' });
+    expect(row.userId).toBe('u1');
+    expect(row.user_id).toBe('u1');
+    expect(row).not.toHaveProperty('is_bot');
+    expect(row.message_type).toBe('text');
+    expect(row.author_kind).toBeNull();
+    expect(row.pgId).toBe('');
+  });
+
+  it('carries a reconciled row\'s PG id, so its reactions still resolve', () => {
+    const row = toPodMessage({ _id: 'm1', podId: 'p1', userId: 'u1', content: 'hi', reconciledPgId: 42 });
+    expect(row.pgId).toBe('42');
+    expect(row.id).toBe('m1');
+  });
+});

backend/__tests__/unit/services/nativeRuntimeService.recentMessages.test.js

@@ -119,7 +119,7 @@ describe('native runtime: recent pod messages inline (recentMessages)', () => {
     ]);
     await runAgent(installationWith({ recentMessages: 3 }), trigger);
 
-    expect(PgMessage.findByPodId).toHaveBeenCalledWith('pod-1', 4);
+    expect(PgMessage.findByPodId).toHaveBeenCalledWith('pod-1', 4, null);
     const sent = sentUserMessage();
     expect(sent.startsWith('Recent messages in this pod, oldest first')).toBe(true);
     const block = sent.split('\n\n')[0];
@@ -144,7 +144,7 @@ describe('native runtime: recent pod messages inline (recentMessages)', () => {
   test('one long message is clipped, and the count is capped', async () => {
     PgMessage.findByPodId.mockResolvedValue([row('m1', 'hamza', 'x'.repeat(RECENT_MESSAGE_CHARS + 500)), row('m4', 'hamza', 'claude code')]);
     await runAgent(installationWith({ recentMessages: 999 }), trigger);
-    expect(PgMessage.findByPodId).toHaveBeenCalledWith('pod-1', RECENT_MESSAGES_MAX + 1);
+    expect(PgMessage.findByPodId).toHaveBeenCalledWith('pod-1', RECENT_MESSAGES_MAX + 1, null);
     const line = sentUserMessage().split('\n').find((l) => l.startsWith('- hamza: x'));
     expect(line.length).toBeLessThanOrEqual('- hamza: '.length + RECENT_MESSAGE_CHARS + 1);
   });

backend/controllers/messageController.ts

@@ -1,4 +1,5 @@
 import type { Request, Response } from 'express';
+import { listPodMessages } from '../repositories/messageRepository';
 
 // eslint-disable-next-line global-require
 const Pod = require('../models/Pod');
@@ -146,48 +147,31 @@ exports.getMessages = async (req: AuthRequest, res: Response): Promise<void> =>
     // workspace. Write-once, fire-and-forget (services/onboardingMarks.ts).
     if (!before) markPodOpened(userId, pod);
 
+    const messages = await listPodMessages(podId, { limit, before });
+    // Sprint B5: attach reactions per message in one batched query
+    // (MessageReaction.listForMessages aggregates by GROUP BY).
+    // Falls through to `messages` unchanged on any reaction lookup error.
     try {
-      const messages = await PGMessage.findByPodId(podId, limit, before);
-      // Sprint B5: attach reactions per message in one batched query
-      // (MessageReaction.listForMessages aggregates by GROUP BY).
-      // Falls through to `messages` unchanged on any reaction lookup error.
-      try {
+      // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require
+      const MessageReaction = require('../models/pg/MessageReaction').default;
+      // Keyed by the PG row id: a reconciled message's client `id` is its
+      // Mongo id, which reaction rows do not know.
+      const ids = messages.map((m) => m.pgId).filter(Boolean);
+      if (ids.length > 0) {
+        const reactionsMap = await MessageReaction.listForMessages(ids, String(userId));
         // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require
-        const MessageReaction = require('../models/pg/MessageReaction').default;
-        // Keyed by the PG row id: a reconciled message's client `id` is its
-        // Mongo id, which reaction rows do not know.
-        const ids = (messages as Array<{ pgId?: string }>)
-          .map((m) => m.pgId)
-          .filter(Boolean);
-        if (ids.length > 0) {
-          const reactionsMap = await MessageReaction.listForMessages(ids, String(userId));
-          // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require
-          const { decorateReactionMap } = require('../services/reactionAttributionService');
-          const decorated = await decorateReactionMap(reactionsMap);
-          for (const m of messages as Array<{ pgId?: string; reactions?: unknown }>) {
-            const key = String(m.pgId || '');
-            m.reactions = decorated.get(key) || [];
-          }
+        const { decorateReactionMap } = require('../services/reactionAttributionService');
+        const decorated = await decorateReactionMap(reactionsMap);
+        for (const m of messages) {
+          m.reactions = decorated.get(String(m.pgId || '')) || [];
         }
-      } catch (rxnErr) {
-        // Non-fatal: reactions are a render enhancement.
-        // eslint-disable-next-line no-console
-        console.warn('[messages] reaction enrichment skipped:', (rxnErr as Error).message);
       }
-      res.json(await gateWelcomeHostedActions(messages as Array<Record<string, unknown>>, podId, userId));
-      return;
-    } catch (pgErr) {
-      const e = pgErr as { message?: string };
-      console.warn('PG unavailable for getMessages, falling back to MongoDB:', e.message);
+    } catch (rxnErr) {
+      // Non-fatal: reactions are a render enhancement.
+      // eslint-disable-next-line no-console
+      console.warn('[messages] reaction enrichment skipped:', (rxnErr as Error).message);
     }
-
-    const query: Record<string, unknown> = { podId };
-    if (before) query.createdAt = { $lt: new Date(before) };
-    const messages = await MongoMessage.find(query)
-      .populate('userId', 'username profilePicture')
-      .sort({ createdAt: -1 })
-      .limit(limit);
-    res.json(await gateWelcomeHostedActions(messages.map(normalizeMongo), podId, userId));
+    res.json(await gateWelcomeHostedActions(messages, podId, userId));
   } catch (err) {
     const e = err as { message?: string; kind?: string };
     console.error('Error in getMessages:', e.message);

backend/repositories/messageRepository.ts

@@ -0,0 +1,120 @@
+/**
+ * Pod message reads (TASK-503). PostgreSQL is the message store; MongoDB holds
+ * only legacy rows and the `pgFallback` rows written while PG was down.
+ *
+ * Every read that pages a pod's messages goes through `listPodMessages`, so the
+ * Mongo fallback lives here and nowhere else. Removing it once the 30-day
+ * trigger check clears is a change to this file only; a PG failure then
+ * surfaces as an error instead of a silent read from the other store.
+ *
+ * Callers get one row shape whichever store answered: the shape of
+ * `models/pg/Message` `findByPodId`, oldest first.
+ */
+
+// eslint-disable-next-line @typescript-eslint/no-require-imports
+const PGMessage = require('../models/pg/Message');
+// eslint-disable-next-line @typescript-eslint/no-require-imports
+const MongoMessage = require('../models/Message');
+
+export type PodMessageAuthor = { _id: string; username: string; profilePicture?: string; isBot?: boolean };
+
+export interface PodMessage {
+  id: string;
+  _id: string;
+  /**
+   * The PG row id. Reactions are keyed by it. A MongoDB row carries its
+   * reconciled PG id, or '' if it was never copied to PG.
+   */
+  pgId: string;
+  pod_id: string;
+  user_id: string;
+  content: string;
+  text: string;
+  message_type: string;
+  messageType: string;
+  author_kind: 'kernel' | null;
+  payload?: unknown;
+  created_at: unknown;
+  createdAt: unknown;
+  updated_at?: unknown;
+  username?: string;
+  profile_picture?: string;
+  is_bot?: boolean;
+  userId: PodMessageAuthor | string;
+  replyTo: { id: string; content: string; username: string; userId: string } | null;
+  [key: string]: unknown;
+}
+
+export interface ListPodMessagesOptions {
+  limit?: number;
+  /** ISO timestamp; only messages created strictly before it. */
+  before?: string | null;
+}
+
+/** Same clamp as the PG read, so the fallback can never page more rows. */
+const clampLimit = (limit: unknown): number => Math.min(200, Math.max(1, Number.parseInt(String(limit), 10) || 50));
+
+type MongoAuthor = { _id?: { toString(): string }; username?: string; profilePicture?: string; isBot?: boolean };
+
+/** A MongoDB message document in the PG read's row shape. */
+export const toPodMessage = (doc: Record<string, unknown>): PodMessage => {
+  const raw = doc.userId as MongoAuthor | { toString(): string } | null | undefined;
+  const populated = raw && typeof raw === 'object' && '_id' in raw && (raw as MongoAuthor)._id
+    ? raw as MongoAuthor
+    : null;
+  const userIdString = populated ? String(populated._id) : (raw ? String(raw) : '');
+  const id = String(doc._id);
+  const content = String(doc.content ?? '');
+  const messageType = String(doc.messageType || 'text');
+  return {
+    id,
+    _id: id,
+    pgId: doc.reconciledPgId != null ? String(doc.reconciledPgId) : '',
+    pod_id: String(doc.podId ?? ''),
+    user_id: userIdString,
+    content,
+    text: content,
+    message_type: messageType,
+    messageType,
+    author_kind: doc.authorKind === 'kernel' || doc.author_kind === 'kernel' ? 'kernel' : null,
+    ...(doc.payload != null ? { payload: doc.payload } : {}),
+    created_at: doc.createdAt,
+    createdAt: doc.createdAt,
+    updated_at: doc.updatedAt,
+    ...(populated ? {
+      username: populated.username,
+      profile_picture: populated.profilePicture,
+      ...(typeof populated.isBot === 'boolean' ? { is_bot: populated.isBot } : {}),
+    } : {}),
+    userId: populated?.username
+      ? {
+        _id: userIdString,
+        username: populated.username,
+        profilePicture: populated.profilePicture,
+        isBot: populated.isBot === true,
+      }
+      : userIdString,
+    replyTo: null,
+  };
+};
+
+/** The pod's newest `limit` messages before `before`, oldest first. */
+export async function listPodMessages(
+  podId: string,
+  { limit = 50, before = null }: ListPodMessagesOptions = {},
+): Promise<PodMessage[]> {
+  const safeLimit = clampLimit(limit);
+  try {
+    return await PGMessage.findByPodId(podId, safeLimit, before || null) as PodMessage[];
+  } catch (pgError) {
+    console.warn('[messages] PG read failed, falling back to MongoDB:', (pgError as Error).message);
+  }
+  const query: Record<string, unknown> = { podId };
+  if (before) query.createdAt = { $lt: new Date(before) };
+  const docs = await MongoMessage.find(query)
+    .populate('userId', 'username profilePicture isBot')
+    .sort({ createdAt: -1 })
+    .limit(safeLimit)
+    .lean() as Array<Record<string, unknown>>;
+  return docs.slice().reverse().map(toPodMessage);
+}