diff --git a/backend/__tests__/service/clawdbot-e2e.test.js b/backend/__tests__/service/clawdbot-e2e.test.js index d6cd410d5a..a32dbaa197 100644 --- a/backend/__tests__/service/clawdbot-e2e.test.js +++ b/backend/__tests__/service/clawdbot-e2e.test.js @@ -15,7 +15,9 @@ const request = require('supertest'); const jwt = require('jsonwebtoken'); const mongoose = require('mongoose'); -const { setupMongoDb, closeMongoDb, clearMongoDb } = require('../utils/testUtils'); +const { + setupMongoDb, closeMongoDb, clearMongoDb, setupPgDb, clearPgDb, closePgDb, +} = require('../utils/testUtils'); // This suite exercises the legacy Mongo-backed Clawdbot flow. Freshness has // its own real-database service coverage; Tier 1's PostgreSQL host is present @@ -66,6 +68,9 @@ describe('Clawdbot E2E Integration Tests', () => { beforeAll(async () => { process.env.JWT_SECRET = JWT_SECRET; await setupMongoDb(); + // Agent posts write PostgreSQL only (TASK-503); before it, these suites + // passed through the silent MongoDB fallback without a PG schema. + await setupPgDb(); // Create Express app with routes app = express(); @@ -163,6 +168,7 @@ describe('Clawdbot E2E Integration Tests', () => { }); afterAll(async () => { + await closePgDb(); await closeMongoDb(); }); @@ -172,6 +178,9 @@ describe('Clawdbot E2E Integration Tests', () => { await AgentEvent.deleteMany({}); await AgentProfile.deleteMany({}); await Message.deleteMany({}); + // Agent posts now land in PG (TASK-503). Clear them too, or one test's + // posts count toward the next test's consecutive-run cap. + await clearPgDb(); await Summary.deleteMany({}); // Reset agent runtime tokens so each beforeEach gets a fresh token await User.updateMany({ isBot: true }, { $set: { agentRuntimeTokens: [] } }); diff --git a/backend/__tests__/service/messages.test.js b/backend/__tests__/service/messages.test.js index 4f0c7c0648..9e27d58dea 100644 --- a/backend/__tests__/service/messages.test.js +++ b/backend/__tests__/service/messages.test.js @@ -102,11 +102,20 @@ describe('Message Routes Integration Tests', () => { PGMessage.create.mockResolvedValue(mockMessage); const token = generateTestToken(user._id); - const response = await request(app) - .post(`/api/messages/${pod._id}`) - .set('Authorization', `Bearer ${token}`) - .send({ text: 'Hi there' }) - .expect(200); + // The send writes PostgreSQL only when PG_HOST is configured (TASK-503). + const oldPgHost = process.env.PG_HOST; + process.env.PG_HOST = 'localhost'; + let response; + try { + response = await request(app) + .post(`/api/messages/${pod._id}`) + .set('Authorization', `Bearer ${token}`) + .send({ text: 'Hi there' }) + .expect(200); + } finally { + if (oldPgHost === undefined) delete process.env.PG_HOST; + else process.env.PG_HOST = oldPgHost; + } expect(response.body.content).toBe('Hi there'); expect(PGMessage.create).toHaveBeenCalledWith( diff --git a/backend/__tests__/service/two-way-integration-e2e.test.js b/backend/__tests__/service/two-way-integration-e2e.test.js index 475f50d186..35b8ee5a7e 100644 --- a/backend/__tests__/service/two-way-integration-e2e.test.js +++ b/backend/__tests__/service/two-way-integration-e2e.test.js @@ -21,7 +21,9 @@ const request = require('supertest'); const express = require('express'); const jwt = require('jsonwebtoken'); -const { setupMongoDb, closeMongoDb } = require('../utils/testUtils'); +const { + setupMongoDb, closeMongoDb, setupPgDb, closePgDb, +} = require('../utils/testUtils'); // Models const User = require('../../models/User'); @@ -103,6 +105,9 @@ describe('Two-Way Integration E2E Tests', () => { beforeAll(async () => { process.env.JWT_SECRET = JWT_SECRET; await setupMongoDb(); + // Agent posts write PostgreSQL only (TASK-503); before it, these suites + // passed through the silent MongoDB fallback without a PG schema. + await setupPgDb(); // Create Express app with routes app = express(); @@ -128,6 +133,7 @@ describe('Two-Way Integration E2E Tests', () => { }); afterAll(async () => { + await closePgDb(); await closeMongoDb(); }); diff --git a/backend/__tests__/unit/controllers/messageController.test.js b/backend/__tests__/unit/controllers/messageController.test.js index 122b11c8c3..8c53cb73bd 100644 --- a/backend/__tests__/unit/controllers/messageController.test.js +++ b/backend/__tests__/unit/controllers/messageController.test.js @@ -172,6 +172,60 @@ describe('messageController', () => { }); describe('createMessage', () => { + // The write follows PG_HOST like the socket and agent paths; the unit + // setup unsets it, so the PostgreSQL cases opt in here. + const OLD_PG_HOST = process.env.PG_HOST; + beforeEach(() => { process.env.PG_HOST = 'localhost'; }); + afterEach(() => { + if (OLD_PG_HOST === undefined) delete process.env.PG_HOST; + else process.env.PG_HOST = OLD_PG_HOST; + }); + + describe('with no PostgreSQL configured (no PG_HOST)', () => { + beforeEach(() => { delete process.env.PG_HOST; }); + + it('writes the send to MongoDB, the only store, instead of failing it with 503', async () => { + const row = { + _id: { toString: () => 'mongo-1' }, + podId: { toString: () => 'p1' }, + userId: { toString: () => 'u1' }, + content: 'hello', + messageType: 'text', + }; + Pod.findById.mockResolvedValue({ members: ['u1'], type: 'chat' }); + MongoMessage.create.mockResolvedValueOnce(row); + MongoMessage.findById.mockReturnValueOnce({ populate: jest.fn().mockResolvedValue(row) }); + const req = { params: { podId: 'p1' }, body: { content: 'hello' }, user: { id: 'u1' } }; + const res = { status: jest.fn().mockReturnThis(), json: jest.fn() }; + + await messageController.createMessage(req, res); + + expect(PGMessage.create).not.toHaveBeenCalled(); + expect(MongoMessage.create).toHaveBeenCalledWith({ + podId: 'p1', userId: 'u1', content: 'hello', messageType: 'text', + }); + expect(res.status).not.toHaveBeenCalledWith(503); + expect(res.json).toHaveBeenCalledWith(expect.objectContaining({ id: 'mongo-1', content: 'hello' })); + }); + + it('refuses a reply, which a MongoDB row cannot carry', async () => { + Pod.findById.mockResolvedValue({ members: ['u1'], type: 'chat' }); + const req = { + params: { podId: 'p1' }, + body: { content: 'a reply', replyToMessageId: 'parent-1' }, + user: { id: 'u1' }, + }; + const res = { status: jest.fn().mockReturnThis(), json: jest.fn() }; + + await messageController.createMessage(req, res); + + expect(res.status).toHaveBeenCalledWith(503); + expect(res.json).toHaveBeenCalledWith(expect.objectContaining({ code: 'REPLY_REQUIRES_POSTGRES' })); + expect(MongoMessage.create).not.toHaveBeenCalled(); + expect(PGMessage.create).not.toHaveBeenCalled(); + }); + }); + it('turns the creator\'s first answer into one proposal while preserving explicit delivery', async () => { Pod.findById.mockResolvedValue({ members: ['u1'], type: 'chat', createdBy: 'u1', @@ -356,8 +410,8 @@ describe('messageController', () => { expect(res.status).toHaveBeenCalledWith(503); expect(res.json).toHaveBeenCalledWith({ - error: 'Replies are temporarily unavailable. Please try again shortly.', - code: 'REPLY_REQUIRES_POSTGRES', + error: 'Messages are temporarily unavailable. Please try again shortly.', + code: 'MESSAGE_STORE_UNAVAILABLE', }); expect(MongoMessage.create).not.toHaveBeenCalled(); expect(AgentMentionService.enqueueMentions).not.toHaveBeenCalled(); @@ -398,38 +452,25 @@ describe('messageController', () => { })); }); - it('keeps the Mongo availability fallback for a non-reply when PostgreSQL is unavailable', async () => { - const mongoFallbackRow = { - _id: { toString: () => 'mongo-message' }, - podId: { toString: () => 'p1' }, - userId: { - _id: { toString: () => 'u1' }, username: 'alice', profilePicture: 'default', - }, - content: 'ordinary message', - messageType: 'text', - }; + it('fails a non-reply with 503 when PostgreSQL is unavailable, writing nothing to MongoDB (TASK-503)', async () => { Pod.findById.mockResolvedValue({ members: ['u1'], type: 'chat' }); PGMessage.create.mockRejectedValueOnce(new Error('PostgreSQL unavailable')); - MongoMessage.create.mockResolvedValueOnce(mongoFallbackRow); - MongoMessage.findById.mockReturnValueOnce({ - populate: jest.fn().mockResolvedValue(mongoFallbackRow), - }); const req = { params: { podId: 'p1' }, - body: { content: mongoFallbackRow.content }, + body: { content: 'ordinary message' }, user: { id: 'u1' }, }; const res = { status: jest.fn().mockReturnThis(), json: jest.fn() }; await messageController.createMessage(req, res); - expect(MongoMessage.create).toHaveBeenCalledWith({ - pgFallback: true, podId: 'p1', userId: 'u1', content: mongoFallbackRow.content, messageType: 'text', + expect(res.status).toHaveBeenCalledWith(503); + expect(res.json).toHaveBeenCalledWith({ + error: 'Messages are temporarily unavailable. Please try again shortly.', + code: 'MESSAGE_STORE_UNAVAILABLE', }); - expect(res.status).not.toHaveBeenCalled(); - expect(res.json).toHaveBeenCalledWith(expect.objectContaining({ - id: 'mongo-message', content: mongoFallbackRow.content, - })); + expect(MongoMessage.create).not.toHaveBeenCalled(); + expect(AgentMentionService.enqueueMentions).not.toHaveBeenCalled(); }); it('uses the joined PG author for a JWT-posted chat wake', async () => { diff --git a/backend/__tests__/unit/routes/agentsRuntime.post-message.config.test.js b/backend/__tests__/unit/routes/agentsRuntime.post-message.config.test.js index 8a5c909780..8dae066cad 100644 --- a/backend/__tests__/unit/routes/agentsRuntime.post-message.config.test.js +++ b/backend/__tests__/unit/routes/agentsRuntime.post-message.config.test.js @@ -170,6 +170,32 @@ describe('agentsRuntime post message installation config forwarding', () => { } }); + it('returns a retryable 503 when the message store is unavailable (TASK-503)', async () => { + const handler = getRouteHandler('/pods/:podId/messages', 'post'); + AgentMessageService.postMessage.mockRejectedValueOnce(Object.assign( + new Error('Messages are temporarily unavailable. Please try again shortly.'), + { code: 'MESSAGE_STORE_UNAVAILABLE', statusCode: 503 }, + )); + const installation = { podId: 'pod-1', agentName: 'openclaw', instanceId: 'tarik', config: {} }; + const req = { + params: { podId: 'pod-1' }, body: { content: 'hello' }, + agentInstallation: installation, agentInstallations: [installation], + }; + const res = { status: jest.fn().mockReturnThis(), json: jest.fn() }; + const warn = jest.spyOn(console, 'warn').mockImplementation(() => {}); + + try { + await handler(req, res); + expect(res.status).toHaveBeenCalledWith(503); + expect(res.json).toHaveBeenCalledWith({ + message: 'Messages are temporarily unavailable. Please try again shortly.', + code: 'MESSAGE_STORE_UNAVAILABLE', + }); + } finally { + warn.mockRestore(); + } + }); + // answeringMessageId tells the answer rule which message a post answers // (agentAnswerService): it completes "Give your agent its first task" and // suppresses a needs-you row. Only the native runtime sets it, from its run's diff --git a/backend/__tests__/unit/services/agentMessageService.threadRootBroadcast.test.js b/backend/__tests__/unit/services/agentMessageService.threadRootBroadcast.test.js index 64e97dbadf..25d6e0499b 100644 --- a/backend/__tests__/unit/services/agentMessageService.threadRootBroadcast.test.js +++ b/backend/__tests__/unit/services/agentMessageService.threadRootBroadcast.test.js @@ -175,6 +175,22 @@ describe('agent broadcast carries thread_root_id', () => { }); }); +describe('a configured PostgreSQL store is the only write path (TASK-503)', () => { + it('fails the post when the PG write throws, writing nothing to MongoDB', async () => { + PGMessage.create.mockRejectedValue(new Error('connection terminated')); + + await expect(AgentMessageService.postMessage({ + agentName: 'sprint-review', + instanceId: 'default', + podId: POD, + content: 'A live agent message', + })).rejects.toMatchObject({ code: 'MESSAGE_STORE_UNAVAILABLE', statusCode: 503 }); + + expect(Message).not.toHaveBeenCalled(); + expect(emitted.find((e) => e.event === 'newMessage')).toBeUndefined(); + }); +}); + describe('agent broadcast carries author identity', () => { it('preserves the bot marker through Mongo population and the socket whitelist', async () => { delete process.env.PG_HOST; @@ -201,7 +217,9 @@ describe('agent broadcast carries author identity', () => { }); const msg = emitted.find((e) => e.event === 'newMessage'); - expect(Message).toHaveBeenCalledWith(expect.objectContaining({ pgFallback: true })); + // No PG configured: a plain MongoDB row, not a pgFallback one (TASK-503). + expect(Message).toHaveBeenCalledWith(expect.objectContaining({ content: 'A live agent message' })); + expect(Message.mock.calls[0][0]).not.toHaveProperty('pgFallback'); expect(msg.payload.userId).toEqual(expect.objectContaining({ _id: 'agent-user-1', isBot: true, diff --git a/backend/__tests__/unit/services/socketMessageWrite.test.js b/backend/__tests__/unit/services/socketMessageWrite.test.js new file mode 100644 index 0000000000..b813f875e7 --- /dev/null +++ b/backend/__tests__/unit/services/socketMessageWrite.test.js @@ -0,0 +1,87 @@ +/** + * TASK-503: with PostgreSQL configured, the socket send writes PG only. A PG + * that is still booting or a failed insert is a retryable error, never a + * MongoDB row (those rows carried no pgFallback flag and were never moved). + */ + +jest.mock('../../../models/pg/Message', () => ({ create: jest.fn(), findById: jest.fn() })); +jest.mock('../../../models/pg/Pod', () => ({ findById: jest.fn() })); +jest.mock('../../../services/pgPodSyncService', () => ({ syncPodFromMongo: jest.fn() })); +jest.mock('../../../models/Message', () => jest.fn()); + +const PGMessage = require('../../../models/pg/Message'); +const PGPod = require('../../../models/pg/Pod'); +const { syncPodFromMongo } = require('../../../services/pgPodSyncService'); +const MongoMessage = require('../../../models/Message'); +const { writeSocketMessage, MESSAGE_STORE_UNAVAILABLE } = require('../../../services/socketMessageWrite'); + +// The refusal names the send so the client can drop its optimistic copy. +const REFUSED = { + ok: false, + error: { + ...MESSAGE_STORE_UNAVAILABLE, podId: 'p1', content: 'hi', messageType: 'text', + }, +}; + +const input = (over = {}) => ({ + pgReady: true, podId: 'p1', userId: 'u1', content: 'hi', messageType: 'text', replyToMessageId: null, ...over, +}); + +describe('writeSocketMessage', () => { + beforeEach(() => { + jest.clearAllMocks(); + jest.spyOn(console, 'error').mockImplementation(() => {}); + jest.spyOn(console, 'warn').mockImplementation(() => {}); + PGPod.findById.mockResolvedValue({ id: 'p1' }); + }); + afterEach(() => jest.restoreAllMocks()); + + it('refuses while PG is still booting, without touching either store', async () => { + const result = await writeSocketMessage(input({ pgReady: false })); + + expect(result).toEqual(REFUSED); + expect(PGMessage.create).not.toHaveBeenCalled(); + expect(MongoMessage).not.toHaveBeenCalled(); + }); + + it('returns a retryable error when the PG insert throws, writing nothing to MongoDB', async () => { + PGMessage.create.mockRejectedValue(new Error('connection terminated')); + + const result = await writeSocketMessage(input()); + + expect(result).toEqual(REFUSED); + expect(MongoMessage).not.toHaveBeenCalled(); + }); + + it('returns the joined row on success, passing the reply edge through', async () => { + PGMessage.create.mockResolvedValue({ id: 7, content: 'hi' }); + PGMessage.findById.mockResolvedValue({ id: 7, content: 'hi', username: 'sam' }); + + const result = await writeSocketMessage(input({ replyToMessageId: '5' })); + + expect(PGMessage.create).toHaveBeenCalledWith('p1', 'u1', 'hi', 'text', '5'); + expect(result).toEqual({ ok: true, message: { id: 7, content: 'hi', username: 'sam' } }); + }); + + it('keeps an already-stored message when only the joined re-read fails', async () => { + PGMessage.create.mockResolvedValue({ id: 8, content: 'hi' }); + PGMessage.findById.mockRejectedValue(new Error('read timeout')); + + const result = await writeSocketMessage(input()); + + expect(result).toEqual({ ok: true, message: { id: 8, content: 'hi' } }); + expect(PGMessage.create).toHaveBeenCalledTimes(1); + }); + + it('heals a MongoDB-only pod into PG before the insert', async () => { + PGPod.findById.mockResolvedValue(null); + PGMessage.create.mockResolvedValue({ id: 9 }); + PGMessage.findById.mockResolvedValue({ id: 9 }); + + await writeSocketMessage(input()); + + expect(syncPodFromMongo).toHaveBeenCalledWith('p1', 'u1'); + expect(syncPodFromMongo.mock.invocationCallOrder[0]) + .toBeLessThan(PGMessage.create.mock.invocationCallOrder[0]); + }); +}); diff --git a/backend/controllers/messageController.ts b/backend/controllers/messageController.ts index 24e9b8e0d3..d80b7bdcd8 100644 --- a/backend/controllers/messageController.ts +++ b/backend/controllers/messageController.ts @@ -45,16 +45,6 @@ interface NormalizedMessage { user?: { username: string; profile_picture?: string }; } -const pgAvailable = (): boolean => { - try { - // eslint-disable-next-line global-require - const { pool } = require('../config/db-pg'); - return !!pool; - } catch { - return false; - } -}; - const normalizeMongo = (m: Record): NormalizedMessage => { // When .populate('userId') ran, m.userId is a populated User document. Calling // .toString() on a Mongoose document returns util.inspect output @@ -245,11 +235,9 @@ exports.createMessage = async (req: AuthRequest, res: Response): Promise = // Best-effort backfill: pods created via the Mongo-only POST /api/pods // path (podController) have no row in the PG `pods` table, so the // subsequent PGMessage.create fails the messages.pod_id foreign key - // and the message lands in the Mongo fallback indefinitely. - // pgMessageController already does this; mirror it here so the dual-DB - // path stays consistent with the PG-primary path. Swallow errors — - // if PG is unreachable, PGMessage.create below will throw and the - // existing Mongo fallback handles it. + // and the send fails. pgMessageController already does this; mirror it + // here. Swallow errors: if PG is unreachable, PGMessage.create below + // throws and the send returns 503. try { const pgPodExists = await PGPod.findById(podId); if (!pgPodExists) { @@ -299,38 +287,10 @@ exports.createMessage = async (req: AuthRequest, res: Response): Promise = throw err; } - try { - const created = await PGMessage.create( - podId, - userId, - messageContent || '', - 'text', - replyToMessageId || null, - null, - resolvedThreadRootId, - ); - // create() returns the raw INSERT row with no users JOIN, so the - // response would lack username/profile_picture and the v2 chat would - // render the author as "Unknown" until a refresh pulled the joined - // row. findById re-fetches with the JOIN so the optimistic render - // already has the right author identity. A JOIN failure is not a write - // failure: falling into the Mongo fallback after INSERT would duplicate - // the message (or falsely reject an already-persisted reply). - let populated = null; - try { - populated = created?.id ? await PGMessage.findById(created.id) : null; - } catch (readErr) { - console.warn('[messageController] PG post-write read failed:', (readErr as Error).message); - } - message = (populated || created) as NormalizedMessage; - } catch (pgErr) { - const e = pgErr as { message?: string }; - console.warn('PG unavailable for createMessage, falling back to MongoDB:', e.message); - // Mongo rows have no reply_to_message_id column, and the reconciler - // (services/pgMessageReconcileService, TASK-313 part 2) copies them into - // PG without a reply edge. A reply written here would look successful - // while losing its parent edge, so replies still wait for Postgres. - // Non-replies take this fallback and are moved once PG is back. + if (!process.env.PG_HOST) { + // No PostgreSQL configured: MongoDB is this instance's only message + // store, as on the socket and agent paths and in the readers. Mongo + // rows have no reply edge, so replies still need PostgreSQL. if (replyToMessageId) { res.status(503).json({ error: 'Replies are temporarily unavailable. Please try again shortly.', @@ -339,7 +299,6 @@ exports.createMessage = async (req: AuthRequest, res: Response): Promise = return; } const mongoMsg = await MongoMessage.create({ - pgFallback: true, podId, userId, content: messageContent || '', @@ -348,6 +307,43 @@ exports.createMessage = async (req: AuthRequest, res: Response): Promise = const populated = await MongoMessage.findById(mongoMsg._id) .populate('userId', 'username profilePicture'); message = normalizeMongo(populated || mongoMsg); + } else { + try { + const created = await PGMessage.create( + podId, + userId, + messageContent || '', + 'text', + replyToMessageId || null, + null, + resolvedThreadRootId, + ); + // create() returns the raw INSERT row with no users JOIN, so the + // response would lack username/profile_picture and the v2 chat would + // render the author as "Unknown" until a refresh pulled the joined + // row. findById re-fetches with the JOIN so the optimistic render + // already has the right author identity. A JOIN failure is not a write + // failure: falling into the Mongo fallback after INSERT would duplicate + // the message (or falsely reject an already-persisted reply). + let populated = null; + try { + populated = created?.id ? await PGMessage.findById(created.id) : null; + } catch (readErr) { + console.warn('[messageController] PG post-write read failed:', (readErr as Error).message); + } + message = (populated || created) as NormalizedMessage; + } catch (pgErr) { + // PostgreSQL is the only message store (TASK-503). A failed write is + // a failed send the client can retry, never a silent MongoDB row that + // pod chat (which reads PG) cannot show. + const e = pgErr as { message?: string }; + console.error('[messageController] PG message write failed:', e.message); + res.status(503).json({ + error: 'Messages are temporarily unavailable. Please try again shortly.', + code: 'MESSAGE_STORE_UNAVAILABLE', + }); + return; + } } // Scout's first-task proposal is a deterministic handoff tied to the pod diff --git a/backend/routes/agentsRuntime.ts b/backend/routes/agentsRuntime.ts index 4f7e0501a4..7666ff8485 100644 --- a/backend/routes/agentsRuntime.ts +++ b/backend/routes/agentsRuntime.ts @@ -1621,7 +1621,9 @@ router.post( return res.json(result); } catch (error: any) { const err = error as Error & { code?: string; statusCode?: number }; - if (err.statusCode && err.statusCode >= 400 && err.statusCode < 500) { + // A message-store outage is retryable (TASK-503): 503, not 500. + if (err.statusCode && ((err.statusCode >= 400 && err.statusCode < 500) + || err.code === 'MESSAGE_STORE_UNAVAILABLE')) { console.warn('Bot message refused:', { code: err.code, status: err.statusCode, message: err.message }); return res.status(err.statusCode).json({ message: err.message, @@ -2382,7 +2384,9 @@ router.post('/pods/:podId/messages', phase4RateLimit, agentRuntimeAuth, async (r // from "pod truly missing" (404) and unexpected failures (500). Without // this, the post-message route 500'd on every legitimate guard refusal, // and the gateway logged "Failed to post message: 500" floods. - if (err.statusCode && err.statusCode >= 400 && err.statusCode < 500) { + // A message-store outage is retryable (TASK-503): 503, not 500. + if (err.statusCode && ((err.statusCode >= 400 && err.statusCode < 500) + || err.code === 'MESSAGE_STORE_UNAVAILABLE')) { console.warn('Agent message refused:', { code: err.code, status: err.statusCode, message: err.message }); return res.status(err.statusCode).json({ message: err.message, diff --git a/backend/server.ts b/backend/server.ts index 8731615859..7f0305b52e 100644 --- a/backend/server.ts +++ b/backend/server.ts @@ -631,67 +631,26 @@ io.on('connection', (socket: any) => { } } - // Use PostgreSQL for messages if available, otherwise fallback to MongoDB - if (pgAvailable) { - try { - // Create message in PostgreSQL - console.log('Creating message in PostgreSQL:', { - podId, - userId: socketUserId, - content, - messageType, - }); - const newMessage = await PGMessage.create( - podId, - socketUserId, - content, - messageType, - replyToMessageId, - ); - console.log('Message created successfully:', newMessage); - - message = await PGMessage.findById(newMessage.id); - console.log('Message retrieved for broadcast:', message); - } catch (dbError) { - console.error( - 'Database error with PostgreSQL, falling back to MongoDB:', - dbError, - ); - try { - // Create message in MongoDB - const user = await User.findById(socketUserId); - const newMessage = new Message({ - podId, - userId: socketUserId, - content, - messageType, - }); - - await newMessage.save(); - console.log( - 'Message saved to MongoDB after PG fallback:', - newMessage._id, - ); - - // Populate user info - message = { - ...newMessage.toObject(), - username: user.username, - profilePicture: user.profilePicture, - }; - } catch (mongoDbError) { - console.error( - 'Failed to save message in MongoDB fallback:', - mongoDbError, - ); - socket.emit('error', { - message: 'Failed to save message to any database', - }); - return; - } + // PostgreSQL is the only message store once it is configured + // (TASK-503); MongoDB is written only when no PG is configured. + if (process.env.PG_HOST) { + // eslint-disable-next-line global-require, @typescript-eslint/no-require-imports + const { writeSocketMessage, MESSAGE_SEND_FAILED_EVENT } = require('./services/socketMessageWrite'); + const written = await writeSocketMessage({ + pgReady: pgAvailable, + podId, + userId: socketUserId, + content, + messageType, + replyToMessageId, + }); + if (!written.ok) { + socket.emit(MESSAGE_SEND_FAILED_EVENT, written.error); + return; } + message = written.message; } else { - console.log('Using MongoDB for messages (PostgreSQL not available)'); + console.log('Using MongoDB for messages (PostgreSQL not configured)'); try { // Create message in MongoDB const user = await User.findById(socketUserId); diff --git a/backend/services/agentMessageService.ts b/backend/services/agentMessageService.ts index cab7151c87..f412b79767 100644 --- a/backend/services/agentMessageService.ts +++ b/backend/services/agentMessageService.ts @@ -1743,7 +1743,7 @@ class AgentMessageService { // to pod chat, which reads PG. Found in the 2026-07-03 live smoke: // the BYO hero path ("install agent → talk to it") broke at the // agent's first reply. Swallow errors: if PG is truly down the - // create below throws and the existing Mongo fallback handles it. + // create below throws and the post fails. try { // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const PGPod = require('../models/pg/Pod'); @@ -1814,20 +1814,21 @@ class AgentMessageService { thread_root_id: (newMessage.thread_root_id as number | string | null) ?? null, }; } catch (pgError) { - console.error('PostgreSQL message creation failed, falling back to MongoDB:', pgError); + // PostgreSQL is the only message store once it is configured + // (TASK-503). A failed write fails the post so the caller can retry; + // a MongoDB row here would be invisible to pod chat, which reads PG. + console.error('[agent-msg] PostgreSQL message write failed:', (pgError as Error).message); + throw Object.assign( + new Error('Messages are temporarily unavailable. Please try again shortly.'), + { code: 'MESSAGE_STORE_UNAVAILABLE', statusCode: 503, cause: pgError }, + ); } } - if (!message && resolvedThreadRootId != null) { - // PG accepted the resolve but the INSERT failed. Falling to Mongo here - // would strand an explicitly-threaded post as an unthreaded row — - // fail the post instead; the caller can retry. - throw new Error('thread-scoped post failed at the PostgreSQL store'); - } - + // Reached only when PostgreSQL is not configured at all (no PG_HOST): a + // configured store that fails has already thrown above. if (!message) { const mongoMessage = new Message({ - pgFallback: true, content, userId: agentUser._id, podId, diff --git a/backend/services/socketMessageWrite.ts b/backend/services/socketMessageWrite.ts new file mode 100644 index 0000000000..79e803d9bb --- /dev/null +++ b/backend/services/socketMessageWrite.ts @@ -0,0 +1,86 @@ +/** + * The socket `sendMessage` store write when PostgreSQL is configured + * (TASK-503). PostgreSQL is the only message store then: a failed write is a + * failed send the client can retry, never a MongoDB row that pod chat (which + * reads PG) cannot show. + * + * Before this, server.ts wrote MongoDB whenever PG had not finished booting + * (every rollout until onMounted) and whenever the PG insert threw. Neither + * row carried `pgFallback`, so the reconciler never moved them into PG. + */ + +// 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 PGPod = require('../models/pg/Pod'); +// eslint-disable-next-line @typescript-eslint/no-require-imports +const { syncPodFromMongo } = require('./pgPodSyncService'); + +export const MESSAGE_STORE_UNAVAILABLE = { + message: 'Messages are temporarily unavailable. Please try again shortly.', + code: 'MESSAGE_STORE_UNAVAILABLE', +} as const; + +// A refused send goes to the sender on its own event, not the generic socket +// `error`: the client treats `error` as a broken connection and stops sending +// until a reload. The payload names the send so the client can drop its +// optimistic copy and tell the person. +export const MESSAGE_SEND_FAILED_EVENT = 'messageSendFailed'; + +export interface MessageSendFailure { + podId: string; + content: string; + messageType: string; + message: string; + code: string; +} + +export interface SocketMessageWriteInput { + pgReady: boolean; + podId: string; + userId: string; + content: string; + messageType: string; + replyToMessageId?: string | null; +} + +export type SocketMessageWriteResult = + | { ok: true; message: Record } + | { ok: false; error: MessageSendFailure }; + +export async function writeSocketMessage({ + pgReady, podId, userId, content, messageType, replyToMessageId = null, +}: SocketMessageWriteInput): Promise { + const failed = (): SocketMessageWriteResult => ({ + ok: false, + error: { ...MESSAGE_STORE_UNAVAILABLE, podId, content, messageType }, + }); + if (!pgReady) return failed(); + + let created: Record; + try { + // Best-effort heal for a pod that exists only in MongoDB, as the HTTP + // and agent paths do; without it the insert FK-fails. + try { + if (!await PGPod.findById(podId)) await syncPodFromMongo(podId, userId); + } catch (syncErr) { + console.warn('[socket] PG pod backfill skipped:', (syncErr as Error).message); + } + created = await PGMessage.create(podId, userId, content, messageType, replyToMessageId); + } catch (writeErr) { + console.error('[socket] PG message write failed:', (writeErr as Error).message); + return failed(); + } + + // The joined re-read only adds author fields. Its failure must not fail + // (or duplicate) a message that is already stored. + let joined: Record | null = null; + try { + joined = await PGMessage.findById(created.id); + } catch (readErr) { + console.warn('[socket] PG post-write read failed:', (readErr as Error).message); + } + return { ok: true, message: joined || created }; +} + +export default { writeSocketMessage, MESSAGE_STORE_UNAVAILABLE, MESSAGE_SEND_FAILED_EVENT }; diff --git a/docs/architecture/long-term-review-2026-10.md b/docs/architecture/long-term-review-2026-10.md index 3e9fccf2e6..6eccdb79b7 100644 --- a/docs/architecture/long-term-review-2026-10.md +++ b/docs/architecture/long-term-review-2026-10.md @@ -387,6 +387,11 @@ Items are ranked by measured cost removed per day of effort. **Moves:** message stores, from two to one, and files with a MongoDB message fallback, from 7 to 0. +**Status (2026-10-30, TASK-503):** the trigger fired. SRE counted 13 MongoDB message rows created in the 30 days to 10-30, so this moved to Do now. +- *Writes, done in the TASK-503 write PR:* once `PG_HOST` is set, the HTTP send, the socket send (`services/socketMessageWrite.ts`) and agent posts write PostgreSQL only. A failed write returns `MESSAGE_STORE_UNAVAILABLE` (503, or a socket `error`) instead of writing MongoDB. +- *Where the rows came from (not yet confirmed per row):* there are three MongoDB writers. The socket wrote MongoDB whenever PostgreSQL was configured but had not finished booting, which happens on every rollout until `onMounted`. It set no `pgFallback` flag, so the reconciler never moved those rows and pod chat, which reads PostgreSQL, never showed them. The controller and agent fallbacks set `pgFallback: true`, and agent rows also carry `metadata`. Grouping the 13 rows by `pgFallback`, `metadata` and `reconciledAt` will tell the writers apart. +- *Still open:* removing the read, get-by-id and delete fallbacks, `pgMessageReconcileService`, and the MongoDB write used when no PostgreSQL is configured (the unit harness runs that way). That waits until the existing fallback rows are copied into PostgreSQL. + ### D. Remove OpenClaw, per ADR-021 Phases 3–4 (3–5 days, plus 1 for docs) **Removes:** code and agent context that a parked runtime still costs. diff --git a/frontend/src/components/ChatRoom.test.tsx b/frontend/src/components/ChatRoom.test.tsx index a4e6626827..9c44b8e2f8 100644 --- a/frontend/src/components/ChatRoom.test.tsx +++ b/frontend/src/components/ChatRoom.test.tsx @@ -273,3 +273,51 @@ describe('uploadImageSrc', () => { expect(uploadImageSrc('data:image/png;base64,AAAA')).toBe('data:image/png;base64,AAAA'); }); }); + +test('a send the server refuses is removed and shown in the composer (TASK-503)', async () => { + axios.get + .mockResolvedValueOnce({ data: { _id: '1', name: 'Room', members: [{ _id: 'u' }], createdBy: { _id: 'u', username: 'me', profilePicture: null } } }) + .mockResolvedValueOnce({ data: [] }) + .mockResolvedValueOnce({ data: [] }) + .mockResolvedValueOnce({ data: [] }) + .mockResolvedValueOnce({ data: [] }) + .mockResolvedValueOnce({ data: { entries: [] } }); + const socket = { on: jest.fn(), off: jest.fn() }; + const sendMessage = jest.fn(); + useSocket.mockReturnValue({ + socket, connected: true, pgAvailable: false, joinPod: jest.fn(), leavePod: jest.fn(), sendMessage, + }); + + await TestUtils.act(async () => { root.render(); }); + await TestUtils.act(async () => Promise.resolve()); + + const input = container.querySelector('[placeholder="Message #Room"]'); + const setValue = Object.getOwnPropertyDescriptor(Object.getPrototypeOf(input), 'value').set; + await TestUtils.act(async () => { + setValue.call(input, 'store is down'); + input.dispatchEvent(new Event('input', { bubbles: true })); + }); + await TestUtils.act(async () => { + container.querySelector('form.message-input-container') + .dispatchEvent(new Event('submit', { bubbles: true, cancelable: true })); + }); + expect(sendMessage).toHaveBeenCalledWith('1', 'store is down', 'text', null); + expect(container.textContent).toContain('store is down'); + + const [, onFailed] = socket.on.mock.calls.find(([event]) => event === 'messageSendFailed'); + await TestUtils.act(async () => { + onFailed({ + podId: '1', + content: 'store is down', + messageType: 'text', + code: 'MESSAGE_STORE_UNAVAILABLE', + message: 'Messages are temporarily unavailable. Please try again shortly.', + }); + }); + + expect(container.textContent).not.toContain('store is down'); + expect(container.querySelector('[role="alert"]').textContent) + .toContain('Messages are temporarily unavailable. Please try again shortly.'); + // The room stays mounted and listening: the refusal is not a page error. + expect(container.textContent).toContain('Room'); +}); diff --git a/frontend/src/components/ChatRoom.tsx b/frontend/src/components/ChatRoom.tsx index ab9b6675f5..3e6e9749db 100644 --- a/frontend/src/components/ChatRoom.tsx +++ b/frontend/src/components/ChatRoom.tsx @@ -518,6 +518,9 @@ const ChatRoom = () => { const [message, setMessage] = useState(''); const [loading, setLoading] = useState(true); const [error, setError] = useState(''); + // A send the server refused (TASK-503). Shown in the composer, not through + // `error`, which replaces the whole room and stops listening to it. + const [sendError, setSendError] = useState(''); const [showEmojiPicker, setShowEmojiPicker] = useState(false); const [showMembers, setShowMembers] = useState(false); // Default to collapsed sidebar const messagesEndRef = useRef(null); @@ -1918,10 +1921,26 @@ const ChatRoom = () => { }); }); + // The server refused a send (message store unavailable): drop the + // optimistic copy so it doesn't look delivered, and say so. + const handleSendFailed = (failure) => { + if (!failure || String(failure.podId) !== String(roomId)) return; + setMessages(prevMessages => { + const i = prevMessages.findIndex(msg => + !msg._id && msg.content === failure.content && msg.messageType === failure.messageType + ); + if (i < 0) return prevMessages; + return [...prevMessages.slice(0, i), ...prevMessages.slice(i + 1)]; + }); + setSendError(failure.message || 'Failed to send message. Please try again.'); + }; + socket.on('messageSendFailed', handleSendFailed); + // Clean up return () => { leavePod(roomId); socket.off('newMessage'); + socket.off('messageSendFailed', handleSendFailed); }; } }, [connected, roomId, socket, joinPod, leavePod, error]); @@ -2227,6 +2246,7 @@ const ChatRoom = () => { }]); // Send text message via socket + setSendError(''); sendMessage(roomId, message, 'text', replyingTo?._id || replyingTo?.id || null); // Clear the message input and reply state @@ -4787,6 +4807,22 @@ const ChatRoom = () => { onSubmit={handleSendMessage} className={`message-input-container ${showMembers ? 'sidebar-visible' : 'sidebar-hidden'}`} > + {sendError && ( + + {sendError} + setSendError('')} + > + + + + )} {replyingTo && (