From 05a7516f9c3c6c225473a6fe0ef0a4d51a59ff3b Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Fri, 30 Oct 2026 03:01:13 -0700 Subject: [PATCH 1/6] fix(messages): a configured Postgres is the only message write path (TASK-503) Once PG_HOST is set, the HTTP send, the socket send and agent posts write PostgreSQL only. A failed write returns MESSAGE_STORE_UNAVAILABLE (503 or a socket error) instead of silently writing a MongoDB row that pod chat, which reads PG, cannot show. The socket path wrote MongoDB whenever PG was configured but still booting (every rollout until onMounted), with no pgFallback flag, so the reconciler never moved those rows. Its write moves into services/socketMessageWrite.ts so it can be unit-tested, and it gains the same Mongo-only pod heal as the HTTP and agent paths. The reconciler, the read fallbacks and the no-PG-configured Mongo write stay until the existing fallback rows are drained (item C status in docs/architecture/long-term-review-2026-10.md). Co-Authored-By: Claude Opus 5.5 Co-authored-by: Ember Agent-Model: claude/claude-opus-5-5/high --- .../controllers/messageController.test.js | 33 +++----- ...MessageService.threadRootBroadcast.test.js | 20 ++++- .../unit/services/socketMessageWrite.test.js | 79 +++++++++++++++++++ backend/controllers/messageController.ts | 38 +++------ backend/server.ts | 77 +++++------------- backend/services/agentMessageService.ts | 18 ++--- backend/services/socketMessageWrite.ts | 68 ++++++++++++++++ docs/architecture/long-term-review-2026-10.md | 5 ++ 8 files changed, 218 insertions(+), 120 deletions(-) create mode 100644 backend/__tests__/unit/services/socketMessageWrite.test.js create mode 100644 backend/services/socketMessageWrite.ts diff --git a/backend/__tests__/unit/controllers/messageController.test.js b/backend/__tests__/unit/controllers/messageController.test.js index 122b11c8c3..16caf62836 100644 --- a/backend/__tests__/unit/controllers/messageController.test.js +++ b/backend/__tests__/unit/controllers/messageController.test.js @@ -356,8 +356,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 +398,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/services/agentMessageService.threadRootBroadcast.test.js b/backend/__tests__/unit/services/agentMessageService.threadRootBroadcast.test.js index 64e97dbadf..d91722e3fc 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.toThrow('connection terminated'); + + 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..cf5a3783e8 --- /dev/null +++ b/backend/__tests__/unit/services/socketMessageWrite.test.js @@ -0,0 +1,79 @@ +/** + * 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'); + +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({ ok: false, error: MESSAGE_STORE_UNAVAILABLE }); + 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({ ok: false, error: MESSAGE_STORE_UNAVAILABLE }); + 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..d6d550d722 100644 --- a/backend/controllers/messageController.ts +++ b/backend/controllers/messageController.ts @@ -245,11 +245,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) { @@ -324,30 +322,16 @@ exports.createMessage = async (req: AuthRequest, res: Response): Promise = } 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.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 (replyToMessageId) { - res.status(503).json({ - error: 'Replies are temporarily unavailable. Please try again shortly.', - code: 'REPLY_REQUIRES_POSTGRES', - }); - return; - } - const mongoMsg = await MongoMessage.create({ - pgFallback: true, - podId, - userId, - content: messageContent || '', - messageType: 'text', + 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', }); - const populated = await MongoMessage.findById(mongoMsg._id) - .populate('userId', 'username profilePicture'); - message = normalizeMongo(populated || mongoMsg); + return; } // Scout's first-task proposal is a deterministic handoff tied to the pod diff --git a/backend/server.ts b/backend/server.ts index 8731615859..075264e230 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 } = require('./services/socketMessageWrite'); + const written = await writeSocketMessage({ + pgReady: pgAvailable, + podId, + userId: socketUserId, + content, + messageType, + replyToMessageId, + }); + if (!written.ok) { + socket.emit('error', 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..05db553fba 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,18 @@ 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 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..7d192dd8a0 --- /dev/null +++ b/backend/services/socketMessageWrite.ts @@ -0,0 +1,68 @@ +/** + * 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; + +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: typeof MESSAGE_STORE_UNAVAILABLE }; + +export async function writeSocketMessage({ + pgReady, podId, userId, content, messageType, replyToMessageId = null, +}: SocketMessageWriteInput): Promise { + if (!pgReady) return { ok: false, error: MESSAGE_STORE_UNAVAILABLE }; + + 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 { ok: false, error: MESSAGE_STORE_UNAVAILABLE }; + } + + // 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 }; 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. From 804b286b2faf0c4d69a5132f64014ec568fc5a12 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Fri, 30 Oct 2026 03:14:16 -0700 Subject: [PATCH 2/6] fix(messages): map an agent-post store outage to 503; give e2e suites a PG schema (TASK-503) Agent posts that hit a PG write failure now throw MESSAGE_STORE_UNAVAILABLE with statusCode 503, and both runtime post routes return it as a 503 instead of a 500. The two-way-integration and clawdbot e2e service suites ran with PG_HOST set but never created a PG schema, so their agent posts passed only through the silent MongoDB fallback. Each suite now runs setupPgDb. Co-Authored-By: Claude Opus 5.5 Co-authored-by: Ember Agent-Model: claude/claude-opus-5-5/high --- .../__tests__/service/clawdbot-e2e.test.js | 8 +++++- .../service/two-way-integration-e2e.test.js | 8 +++++- .../agentsRuntime.post-message.config.test.js | 26 +++++++++++++++++++ ...MessageService.threadRootBroadcast.test.js | 2 +- backend/routes/agentsRuntime.ts | 8 ++++-- backend/services/agentMessageService.ts | 5 +++- 6 files changed, 51 insertions(+), 6 deletions(-) diff --git a/backend/__tests__/service/clawdbot-e2e.test.js b/backend/__tests__/service/clawdbot-e2e.test.js index d6cd410d5a..dec08d182f 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, 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(); }); 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/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 d91722e3fc..25d6e0499b 100644 --- a/backend/__tests__/unit/services/agentMessageService.threadRootBroadcast.test.js +++ b/backend/__tests__/unit/services/agentMessageService.threadRootBroadcast.test.js @@ -184,7 +184,7 @@ describe('a configured PostgreSQL store is the only write path (TASK-503)', () = instanceId: 'default', podId: POD, content: 'A live agent message', - })).rejects.toThrow('connection terminated'); + })).rejects.toMatchObject({ code: 'MESSAGE_STORE_UNAVAILABLE', statusCode: 503 }); expect(Message).not.toHaveBeenCalled(); expect(emitted.find((e) => e.event === 'newMessage')).toBeUndefined(); 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/services/agentMessageService.ts b/backend/services/agentMessageService.ts index 05db553fba..f412b79767 100644 --- a/backend/services/agentMessageService.ts +++ b/backend/services/agentMessageService.ts @@ -1818,7 +1818,10 @@ class AgentMessageService { // (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 pgError; + throw Object.assign( + new Error('Messages are temporarily unavailable. Please try again shortly.'), + { code: 'MESSAGE_STORE_UNAVAILABLE', statusCode: 503, cause: pgError }, + ); } } From 2ba6ae42faa01fd7040b8361eb05c9f6772f9ffb Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Fri, 30 Oct 2026 03:22:59 -0700 Subject: [PATCH 3/6] test(clawdbot-e2e): clear PG messages between tests (TASK-503) Agent posts now land in PG, so one test's posts counted toward the next test's consecutive-run cap and the bridge-flow post was refused. Co-Authored-By: Claude Opus 5.5 Co-authored-by: Ember Agent-Model: claude/claude-opus-5-5/high --- backend/__tests__/service/clawdbot-e2e.test.js | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/backend/__tests__/service/clawdbot-e2e.test.js b/backend/__tests__/service/clawdbot-e2e.test.js index dec08d182f..a32dbaa197 100644 --- a/backend/__tests__/service/clawdbot-e2e.test.js +++ b/backend/__tests__/service/clawdbot-e2e.test.js @@ -16,7 +16,7 @@ const jwt = require('jsonwebtoken'); const mongoose = require('mongoose'); const { - setupMongoDb, closeMongoDb, clearMongoDb, setupPgDb, closePgDb, + setupMongoDb, closeMongoDb, clearMongoDb, setupPgDb, clearPgDb, closePgDb, } = require('../utils/testUtils'); // This suite exercises the legacy Mongo-backed Clawdbot flow. Freshness has @@ -178,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: [] } }); From c3c5c50943c89e9c4f2ad3acf6542d1c47335a55 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Fri, 30 Oct 2026 03:38:23 -0700 Subject: [PATCH 4/6] fix(messages): the HTTP send follows PG_HOST like the socket and agent paths (TASK-503) Lark's second gate on #2989: createMessage wrote PG unconditionally, so on an instance with no PG_HOST every REST send returned 503 while the socket and agent paths (and the readers) used MongoDB. With no PG_HOST the HTTP send now writes MongoDB, and replies still refuse with REPLY_REQUIRES_POSTGRES. Drops the unused pgAvailable helper. Both new tests fail with the gate removed. Co-Authored-By: Claude Opus 5.5 Co-authored-by: Ember Agent-Model: claude/claude-opus-5-5/high --- .../controllers/messageController.test.js | 54 +++++++++++ backend/controllers/messageController.ts | 94 +++++++++++-------- 2 files changed, 107 insertions(+), 41 deletions(-) diff --git a/backend/__tests__/unit/controllers/messageController.test.js b/backend/__tests__/unit/controllers/messageController.test.js index 16caf62836..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', diff --git a/backend/controllers/messageController.ts b/backend/controllers/messageController.ts index d6d550d722..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 @@ -297,41 +287,63 @@ exports.createMessage = async (req: AuthRequest, res: Response): Promise = throw err; } - try { - const created = await PGMessage.create( + 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.', + code: 'REPLY_REQUIRES_POSTGRES', + }); + return; + } + const mongoMsg = await MongoMessage.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; + content: messageContent || '', + messageType: 'text', + }); + const populated = await MongoMessage.findById(mongoMsg._id) + .populate('userId', 'username profilePicture'); + message = normalizeMongo(populated || mongoMsg); + } else { try { - populated = created?.id ? await PGMessage.findById(created.id) : null; - } catch (readErr) { - console.warn('[messageController] PG post-write read failed:', (readErr as Error).message); + 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; } - 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 From 973009b56bd090e3a07a1b8a6080e76225acae0c Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Fri, 30 Oct 2026 03:45:53 -0700 Subject: [PATCH 5/6] test(messages): the create-route test opts into PG_HOST (TASK-503) The HTTP send now writes PostgreSQL only when PG_HOST is set, and the unit setup unsets it, so the route test that asserts the PG insert sets it for that request. Co-Authored-By: Claude Opus 5.5 Co-authored-by: Ember Agent-Model: claude/claude-opus-5-5/high --- backend/__tests__/service/messages.test.js | 19 ++++++++++++++----- 1 file changed, 14 insertions(+), 5 deletions(-) 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( From b08aee59c3fb2769ed3384886b05c8adb7630ca5 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Fri, 30 Oct 2026 03:55:55 -0700 Subject: [PATCH 6/6] fix(chat): a refused socket send reaches the sender without breaking the socket (TASK-503) Otto's first gate on #2989: the socket path emitted MESSAGE_STORE_UNAVAILABLE on the generic `error` event. SocketContext treats that as a broken connection (setConnected(false) while the socket stays up), so ChatRoom's optimistic message looked delivered and every later send was dropped until a reload. The refusal now goes out on its own `messageSendFailed` event, naming the pod, content and type. ChatRoom drops the matching optimistic copy and shows the error above the composer, without the page-level error that would tear down the room's listeners. The new ChatRoom test fails with the listener removed. Co-Authored-By: Claude Opus 5.5 Co-authored-by: Ember Agent-Model: claude/claude-opus-5-5/high --- .../unit/services/socketMessageWrite.test.js | 12 ++++- backend/server.ts | 4 +- backend/services/socketMessageWrite.ts | 26 ++++++++-- frontend/src/components/ChatRoom.test.tsx | 48 +++++++++++++++++++ frontend/src/components/ChatRoom.tsx | 36 ++++++++++++++ 5 files changed, 118 insertions(+), 8 deletions(-) diff --git a/backend/__tests__/unit/services/socketMessageWrite.test.js b/backend/__tests__/unit/services/socketMessageWrite.test.js index cf5a3783e8..b813f875e7 100644 --- a/backend/__tests__/unit/services/socketMessageWrite.test.js +++ b/backend/__tests__/unit/services/socketMessageWrite.test.js @@ -15,6 +15,14 @@ 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, }); @@ -31,7 +39,7 @@ describe('writeSocketMessage', () => { it('refuses while PG is still booting, without touching either store', async () => { const result = await writeSocketMessage(input({ pgReady: false })); - expect(result).toEqual({ ok: false, error: MESSAGE_STORE_UNAVAILABLE }); + expect(result).toEqual(REFUSED); expect(PGMessage.create).not.toHaveBeenCalled(); expect(MongoMessage).not.toHaveBeenCalled(); }); @@ -41,7 +49,7 @@ describe('writeSocketMessage', () => { const result = await writeSocketMessage(input()); - expect(result).toEqual({ ok: false, error: MESSAGE_STORE_UNAVAILABLE }); + expect(result).toEqual(REFUSED); expect(MongoMessage).not.toHaveBeenCalled(); }); diff --git a/backend/server.ts b/backend/server.ts index 075264e230..7f0305b52e 100644 --- a/backend/server.ts +++ b/backend/server.ts @@ -635,7 +635,7 @@ io.on('connection', (socket: any) => { // (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 } = require('./services/socketMessageWrite'); + const { writeSocketMessage, MESSAGE_SEND_FAILED_EVENT } = require('./services/socketMessageWrite'); const written = await writeSocketMessage({ pgReady: pgAvailable, podId, @@ -645,7 +645,7 @@ io.on('connection', (socket: any) => { replyToMessageId, }); if (!written.ok) { - socket.emit('error', written.error); + socket.emit(MESSAGE_SEND_FAILED_EVENT, written.error); return; } message = written.message; diff --git a/backend/services/socketMessageWrite.ts b/backend/services/socketMessageWrite.ts index 7d192dd8a0..79e803d9bb 100644 --- a/backend/services/socketMessageWrite.ts +++ b/backend/services/socketMessageWrite.ts @@ -21,6 +21,20 @@ export const MESSAGE_STORE_UNAVAILABLE = { 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; @@ -32,12 +46,16 @@ export interface SocketMessageWriteInput { export type SocketMessageWriteResult = | { ok: true; message: Record } - | { ok: false; error: typeof MESSAGE_STORE_UNAVAILABLE }; + | { ok: false; error: MessageSendFailure }; export async function writeSocketMessage({ pgReady, podId, userId, content, messageType, replyToMessageId = null, }: SocketMessageWriteInput): Promise { - if (!pgReady) return { ok: false, error: MESSAGE_STORE_UNAVAILABLE }; + const failed = (): SocketMessageWriteResult => ({ + ok: false, + error: { ...MESSAGE_STORE_UNAVAILABLE, podId, content, messageType }, + }); + if (!pgReady) return failed(); let created: Record; try { @@ -51,7 +69,7 @@ export async function writeSocketMessage({ created = await PGMessage.create(podId, userId, content, messageType, replyToMessageId); } catch (writeErr) { console.error('[socket] PG message write failed:', (writeErr as Error).message); - return { ok: false, error: MESSAGE_STORE_UNAVAILABLE }; + return failed(); } // The joined re-read only adds author fields. Its failure must not fail @@ -65,4 +83,4 @@ export async function writeSocketMessage({ return { ok: true, message: joined || created }; } -export default { writeSocketMessage, MESSAGE_STORE_UNAVAILABLE }; +export default { writeSocketMessage, MESSAGE_STORE_UNAVAILABLE, MESSAGE_SEND_FAILED_EVENT }; 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 && (