diff --git a/backend/__tests__/service/messages.test.js b/backend/__tests__/service/messages.test.js index 11078df641..4f0c7c0648 100644 --- a/backend/__tests__/service/messages.test.js +++ b/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, ); }); diff --git a/backend/__tests__/unit/controllers/messageController.test.js b/backend/__tests__/unit/controllers/messageController.test.js index d17a8fd63c..77188fcc23 100644 --- a/backend/__tests__/unit/controllers/messageController.test.js +++ b/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() }; diff --git a/backend/__tests__/unit/repositories/messageRepository.test.js b/backend/__tests__/unit/repositories/messageRepository.test.js new file mode 100644 index 0000000000..0c77c9a9a8 --- /dev/null +++ b/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'); + }); +}); diff --git a/backend/__tests__/unit/services/nativeRuntimeService.recentMessages.test.js b/backend/__tests__/unit/services/nativeRuntimeService.recentMessages.test.js index 64a65bb3a9..01ac9c4ba0 100644 --- a/backend/__tests__/unit/services/nativeRuntimeService.recentMessages.test.js +++ b/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); }); diff --git a/backend/controllers/messageController.ts b/backend/controllers/messageController.ts index 397f174e0d..32dde515c3 100644 --- a/backend/controllers/messageController.ts +++ b/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 => // 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>, 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 = { 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); diff --git a/backend/repositories/messageRepository.ts b/backend/repositories/messageRepository.ts new file mode 100644 index 0000000000..2cac872e32 --- /dev/null +++ b/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): 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 { + 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 = { 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>; + return docs.slice().reverse().map(toPodMessage); +} diff --git a/backend/routes/showcase.ts b/backend/routes/showcase.ts index 429b5fce6d..a3cb7f69af 100644 --- a/backend/routes/showcase.ts +++ b/backend/routes/showcase.ts @@ -23,6 +23,7 @@ // recognises the limiter on the router — same pattern as routes/uploads.ts. import rateLimit from 'express-rate-limit'; import { cloudflareIpRateLimitKeyGenerator } from '../middleware/ipRateLimit'; +import { listPodMessages, type PodMessage } from '../repositories/messageRepository'; // eslint-disable-next-line @typescript-eslint/no-require-imports const express = require('express'); @@ -33,10 +34,6 @@ const Pod = require('../models/Pod'); // eslint-disable-next-line @typescript-eslint/no-require-imports const User = require('../models/User'); // eslint-disable-next-line @typescript-eslint/no-require-imports -const MongoMessage = require('../models/Message'); -// 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 AgentMessageService = require('../services/agentMessageService'); // eslint-disable-next-line @typescript-eslint/no-require-imports const { resolveAgentDisplayLabel } = require('../services/agentIdentityService'); @@ -204,8 +201,8 @@ interface RawMessage { } // Read newest-`limit` messages before the cursor, ascending (chronological), -// PG-first with a Mongo fallback (mirrors messageController.getMessages). The -// raw read is store-agnostic; display identity + the noise-filter + the +// through the message repository (PG first; the store fallback lives there). +// The raw read is store-agnostic; display identity + the noise-filter + the // whitelist serializer are applied uniformly afterwards so neither store can // leak a non-whitelisted field. const readRawMessages = async ( @@ -213,43 +210,15 @@ const readRawMessages = async ( limit: number, before?: string, ): Promise<{ raw: RawMessage[]; hasMore: boolean }> => { - try { - const rows = await PGMessage.findByPodId(podId, limit, before || null); - const raw = (rows as Array>).map((r) => { - const uid = r.user_id - || (r.userId && typeof r.userId === 'object' - ? (r.userId as { _id?: { toString(): string } })._id - : r.userId); - return { - id: String(r.id ?? r._id ?? ''), - content: String(r.content ?? ''), - messageType: String(r.messageType ?? r.message_type ?? 'text'), - createdAt: r.createdAt ?? r.created_at, - userId: uid ? String(uid) : '', - }; - }); - return { raw, hasMore: rows.length === limit }; - } catch (pgErr) { - console.warn('[showcase] PG unavailable, falling back to Mongo:', (pgErr as Error).message); - } - - const query: Record = { podId }; - if (before) query.createdAt = { $lt: new Date(before) }; - const docs = await MongoMessage.find(query) - .sort({ createdAt: -1 }) - .limit(limit) - .lean(); - const raw = (docs as Array>) - .slice() - .reverse() - .map((m) => ({ - id: String(m._id), - content: String(m.content ?? ''), - messageType: String(m.messageType ?? 'text'), - createdAt: m.createdAt, - userId: m.userId ? String(m.userId) : '', - })); - return { raw, hasMore: docs.length === limit }; + const rows = await listPodMessages(podId, { limit, before: before || null }); + const raw = rows.map((r: PodMessage) => ({ + id: String(r.id ?? r._id ?? ''), + content: String(r.content ?? ''), + messageType: String(r.messageType ?? r.message_type ?? 'text'), + createdAt: r.createdAt ?? r.created_at, + userId: r.user_id ? String(r.user_id) : '', + })); + return { raw, hasMore: rows.length === limit }; }; // GET /api/showcase/:podId/messages — filtered, whitelisted message feed. diff --git a/backend/services/nativeRuntimeService.ts b/backend/services/nativeRuntimeService.ts index 4ec7e856d8..5be155ad2c 100644 --- a/backend/services/nativeRuntimeService.ts +++ b/backend/services/nativeRuntimeService.ts @@ -1502,34 +1502,15 @@ export async function fetchRecentPodMessages( id: string; author: string; authorId: string | null; authorIsBot: boolean | null; content: string; createdAt: unknown; }>> { // eslint-disable-next-line global-require, @typescript-eslint/no-require-imports - const Message = require('../models/pg/Message'); - let rows: Array> = []; - try { - rows = (await Message.findByPodId(String(podId), limit)) as Array>; - } catch (pgError) { - console.warn( - '[native-runtime] recent pod messages: PG fetch failed, falling back to Mongo:', - (pgError as Error).message, - ); - // eslint-disable-next-line global-require, @typescript-eslint/no-require-imports - const MongoMessage = require('../models/Message'); - const docs = await MongoMessage.find({ podId }) - .sort({ createdAt: -1 }) - .limit(limit) - .populate('userId', 'username isBot') - .lean(); - rows = (docs as Array>).reverse(); - } + const { listPodMessages } = require('../repositories/messageRepository'); + const rows = await listPodMessages(String(podId), { limit }) as Array>; return rows.map((row) => { const user = (row.userId as Record | undefined) || {}; const authorId = user._id || row.user_id || row.user_db_id; - // A PG row (it has `user_id`) carries the users mirror's `is_bot` as - // stored; its mapper's `isBot` reads null or missing as false, so only the - // raw column is trusted, and a PG row without it is unknown. A Mongo row - // carries the populated `isBot`. Unknown is null, which gives no address. - let authorIsBot: boolean | null = null; - if ('user_id' in row) authorIsBot = typeof row.is_bot === 'boolean' ? row.is_bot : null; - else if (typeof user.isBot === 'boolean') authorIsBot = user.isBot; + // Only the raw `is_bot` is trusted: the PG mapper's `userId.isBot` reads + // null or missing as false. The repository carries a Mongo row's populated + // `isBot` into `is_bot` the same way. Unknown is null, which gives no address. + const authorIsBot = typeof row.is_bot === 'boolean' ? row.is_bot : null; return { id: String(row.id || row._id || ''), author: (user.username as string) || (row.username as string) || 'unknown',