Fix chat visibility and conversation sync

This commit is contained in:
jjbrunton committed 2026-08-08 19:29:40 +01:00
1 parent f4e8fdc42f
commit 34ae532cdd
10 files changed
+408 -15

No files matched your search

@@ -35,12 +35,19 @@ function MessageHistory({ conversation }: { conversation: ConversationDetailResp
{message.media?.length ? (
<div className="conversation-message-images">
{message.media.map((media) => (
<a
aria-label={`View original image ${message.isSelf ? 'sent by you' : `received from ${redditUsernameLabel(conversation.username)}`}`}
href={mediaPath(media.localPath)}
key={`${message.timestamp}-${media.localPath}`}
rel="noreferrer"
target="_blank"
>
<img
alt={`Image ${message.isSelf ? 'sent by you' : `received from ${redditUsernameLabel(conversation.username)}`}`}
key={`${message.timestamp}-${media.localPath}`}
loading="lazy"
src={mediaPath(media.localPath)}
/>
</a>
))}
</div>
) : null}
@@ -60,6 +60,8 @@
.conversation-messages time { color: var(--muted); font-size: .7rem; }
.conversation-messages p { margin: 0; line-height: 1.45; white-space: pre-wrap; }
.conversation-message-images { display: grid; grid-template-columns: repeat(auto-fit, minmax(min(12rem, 100%), 1fr)); gap: .5rem; margin-top: .65rem; }
.conversation-message-images a { display: block; border-radius: .6rem; }
.conversation-message-images a:focus-visible { outline: 2px solid var(--accent); outline-offset: 2px; }
.conversation-message-images img { display: block; width: 100%; max-width: 28rem; max-height: 32rem; object-fit: contain; background: #08100c; border: 1px solid var(--border); border-radius: .6rem; }
.conversation-empty-detail { padding: 2rem; color: var(--muted); text-align: center; }
.conversation-composer { display: grid; gap: .65rem; padding: 1rem; border-top: 1px solid var(--border); }
+55 -2
View File
@@ -54,6 +54,11 @@ export interface RoomSummary {
recentEvents: MatrixTimelineEvent[];
}
export interface DirectRoomMapping {
recipientMxid: string;
roomId: string;
}
/**
* Cap on how long we'll sit and sleep waiting out a 429. Anything longer
* is surfaced as a `MatrixRateLimitError` so the main loop can back off
@@ -405,7 +410,7 @@ export class MatrixClient {
* "Start chat" in the UI:
*
* POST /_matrix/client/v3/createRoom
* { "preset": "reddit_dm", "invite": [mxid] }
* { "preset": "reddit_dm", "invite": [mxid], "is_direct": true }
*
* `reddit_dm` is a Reddit-specific preset that the homeserver expands
* server-side into the right power levels, history visibility, and
@@ -413,12 +418,53 @@ export class MatrixClient {
*/
async createDirectRoom(recipientMxid: string): Promise<string> {
const res = await this.request<{ room_id: string }>('POST', '/_matrix/client/v3/createRoom', {
body: { preset: 'reddit_dm', invite: [recipientMxid] },
body: { preset: 'reddit_dm', invite: [recipientMxid], is_direct: true },
});
this.logger.debug(`[matrix] created DM room ${res.room_id} with ${recipientMxid}`);
return res.room_id;
}
/**
* Add missing rooms to the account-level `m.direct` map used by Matrix
* clients to classify joined rooms as direct chats. Existing mappings are
* preserved verbatim; the write is skipped when there is nothing to repair.
*/
async repairDirectRoomMappings(mappings: DirectRoomMapping[]): Promise<number> {
if (mappings.length === 0) return 0;
let directRooms: Record<string, string[]>;
try {
directRooms = await this.request<Record<string, string[]>>(
'GET',
`/_matrix/client/v3/user/${encodeURIComponent(this.creds.userId)}/account_data/m.direct`,
);
} catch (error) {
if (error instanceof Error && error.message.includes('→ 404')) {
directRooms = {};
} else {
throw error;
}
}
let repaired = 0;
for (const { recipientMxid, roomId } of mappings) {
const existing = Array.isArray(directRooms[recipientMxid])
? directRooms[recipientMxid]
: [];
if (existing.includes(roomId)) continue;
directRooms[recipientMxid] = [...existing, roomId];
repaired++;
}
if (repaired === 0) return 0;
await this.request<Record<string, never>>(
'PUT',
`/_matrix/client/v3/user/${encodeURIComponent(this.creds.userId)}/account_data/m.direct`,
{ body: directRooms },
);
return repaired;
}
/**
* One-shot: create a DM room with the recipient and send the opening
* message. Returns the new room_id and the event_id of the message.
@@ -428,6 +474,13 @@ export class MatrixClient {
body: string,
): Promise<{ roomId: string; eventId: string }> {
const roomId = await this.createDirectRoom(recipientMxid);
try {
await this.repairDirectRoomMappings([{ recipientMxid, roomId }]);
} catch (error) {
// Room creation has already succeeded. Do not turn a metadata repair
// failure into an orphan-and-retry cycle that creates duplicate rooms.
this.logger.warn(`[matrix] failed to register ${roomId} as a direct chat:`, error);
}
const eventId = await this.sendTextMessage(roomId, body);
return { roomId, eventId };
}
+52 -6
View File
@@ -393,6 +393,7 @@ export class ChatService {
if (!chat.roomName || chat.roomName.startsWith('t2_')) chat.roomName = resolved;
}
await this.cache.save(chats);
await this.repairDirectRoomMappings(chats);
let done = 0;
for (const chat of chats) {
@@ -727,17 +728,21 @@ export class ChatService {
existing.set(roomId, next);
}
if (this.mediaService) {
for (const [roomId, room] of Object.entries(joined)) {
const scraped = this.scrapeMessageEvents(room?.timeline?.events ?? []);
const mediaMessages = scraped.filter((m) => m.mediaUrls.length > 0);
if (mediaMessages.length === 0) continue;
if (scraped.length === 0) continue;
const chat = existing.get(roomId);
if (!chat) continue;
let newMessages: ChatMessage[] = scraped.map(
({ mediaUrls: _ignored, ...message }) => message,
);
if (this.mediaService && scraped.some((message) => message.mediaUrls.length > 0)) {
try {
const withMedia = await this.mediaService.attachMedia(chat, mediaMessages);
this.mergeNewMessages(chat, withMedia);
if (withMedia.some((message) => !message.isSelf && (message.media?.length ?? 0) > 0)) {
newMessages = await this.mediaService.attachMedia(chat, scraped);
if (newMessages.some(
(message) => !message.isSelf && (message.media?.length ?? 0) > 0,
)) {
this.recordReceivedImage(chat);
}
} catch (err) {
@@ -746,6 +751,7 @@ export class ChatService {
);
}
}
this.mergeNewMessages(chat, newMessages);
}
const merged = Array.from(existing.values()).sort(
@@ -757,9 +763,49 @@ export class ChatService {
// Resolve any t2_-prefixed usernames the delta just introduced, so the
// dashboard doesn't show raw Matrix-style IDs for new rooms.
await this.enrichUsernames();
await this.repairDirectRoomMappings(merged);
return merged;
}
/**
* Reconcile cached one-to-one chats with Matrix account data. The durable
* username repository lets delta sync repair older rooms even when Reddit
* omits their member state from the latest /sync response.
*/
private async repairDirectRoomMappings(chats: ActiveChat[]): Promise<void> {
const mappings: Array<{ recipientMxid: string; roomId: string }> = [];
const seen = new Set<string>();
for (const chat of chats) {
if (!chat.roomId) continue;
const rawT2 = this.extractT2(chat);
const persistedT2 = rawT2
? `t2_${rawT2}`
: this.redditUsersRepo?.getByUsername(chat.username)?.t2Id;
if (!persistedT2) continue;
const recipientMxid = MatrixClient.t2ToMxid(persistedT2);
const key = `${recipientMxid}|${chat.roomId}`;
if (seen.has(key)) continue;
seen.add(key);
mappings.push({ recipientMxid, roomId: chat.roomId });
}
if (mappings.length === 0) return;
try {
const repaired = await this.withAuth(() =>
this.matrix.repairDirectRoomMappings(mappings),
);
if (repaired > 0) {
this.logger.info(`[chat] repaired ${repaired} missing direct-room mapping(s)`);
}
} catch (error) {
// Sync remains useful even if Reddit temporarily rejects account-data
// maintenance. A later delta/full sync will try the additive repair again.
this.logger.warn('[chat] direct-room mapping repair failed:', error);
}
}
private scrapeMessageEvents(events: MatrixTimelineEvent[]): ScrapedChatMessage[] {
return events
.filter((e) => e.type === 'm.room.message')
+11 -4
View File
@@ -273,11 +273,18 @@ export function createConversationsRouter(ctx: WebContext): Router {
}
ctx.activity.push('sent', `Sent chat reply to u/${conversation.username}`, conversation.username);
ctx.broadcaster.push('info', `Chat reply sent to u/${conversation.username}`);
try {
// Wait for Matrix /sync to merge the sent event into the cached
// message history. The dashboard invalidates its queries when this
// request completes, so returning earlier can immediately re-read
// the stale pre-send conversation.
await ctx.services.chat.refreshTopChats(5);
} catch (err) {
// Delivery is already confirmed by Matrix's send response. A sync
// failure must not turn a successful send into a reported failure.
ctx.logger.error('Failed to refresh chats after send-reply:', err);
}
res.json({ success: true });
ctx.services.chat.refreshTopChats(5).catch((err) =>
ctx.logger.error('Failed to refresh chats after send-reply:', err),
);
} else {
res.status(500).json({ error: 'Failed to send chat message' });
}
@@ -0,0 +1,67 @@
import { describe, expect, it, vi } from 'vitest';
import os from 'os';
import path from 'path';
import type { RedditUsersRepository } from '../src/database/repositories.js';
import { ChatCache } from '../src/services/chat/cache.js';
import { ChatService } from '../src/services/chat/service.js';
import { MatrixClient } from '../src/services/chat/matrix/client.js';
import { createMockLogger } from './mockLogger.js';
import { authenticatedAccountAuth } from './testUtils.js';
const creds = {
homeserver: 'https://matrix.example.test',
accessToken: 'tok',
userId: '@t2_me:reddit.com',
capturedAt: Date.now(),
};
describe('ChatService direct-room repair', () => {
it('repairs cached outbound rooms during delta sync', async () => {
const logger = createMockLogger();
const matrix = new MatrixClient(creds, logger, {
minRequestIntervalMs: 0,
fetchImpl: vi.fn(),
});
matrix.sync = vi.fn().mockResolvedValue({ next_batch: 'next', rooms: { join: {} } });
matrix.repairDirectRoomMappings = vi.fn().mockResolvedValue(1);
const users = {
getAll: () => [{ t2Id: 't2_alice', username: 'alice' }],
getByUsername: () => ({ t2Id: 't2_alice', username: 'alice' }),
upsert: vi.fn(),
} as unknown as RedditUsersRepository;
const cache = new ChatCache(
logger,
path.join(os.tmpdir(), `chat-direct-repair-${Date.now()}-${Math.random()}.json`),
);
cache.set([{
username: 'alice',
roomName: 'alice',
lastMessageTime: Date.now(),
lastMessageSnippet: 'hello',
isFromUs: true,
unread: false,
url: 'https://chat.reddit.com/room/room',
roomId: '!room:reddit.com',
messages: [],
}]);
const service = new ChatService(
matrix,
logger,
undefined, undefined, undefined, undefined,
cache,
undefined,
users,
authenticatedAccountAuth,
);
await service.refreshTopChats();
expect(matrix.repairDirectRoomMappings).toHaveBeenCalledWith([{
recipientMxid: '@t2_alice:reddit.com',
roomId: '!room:reddit.com',
}]);
});
});
+96
View File
@@ -0,0 +1,96 @@
import fs from 'fs/promises';
import os from 'os';
import path from 'path';
import { afterEach, describe, expect, it, vi } from 'vitest';
import { ChatCache } from '../src/services/chat/cache.js';
import { MatrixClient } from '../src/services/chat/matrix/client.js';
import { ChatService } from '../src/services/chat/service.js';
import { createMockLogger } from './mockLogger.js';
import { authenticatedAccountAuth } from './testUtils.js';
const temporaryDirectories: string[] = [];
afterEach(async () => {
await Promise.all(temporaryDirectories.splice(0).map((directory) =>
fs.rm(directory, { recursive: true, force: true }),
));
});
describe('ChatService text delta sync', () => {
it('merges a sent text event into conversation history without a media service', async () => {
const directory = await fs.mkdtemp(path.join(os.tmpdir(), 'chat-text-sync-'));
temporaryDirectories.push(directory);
const cacheFile = path.join(directory, 'chat-cache.json');
const now = Date.now();
await fs.writeFile(cacheFile, JSON.stringify({
version: 1,
savedAt: now,
chats: [{
username: 'alice',
roomName: 'alice',
roomId: '!alice:example.test',
lastMessageTime: now - 1_000,
lastMessageSnippet: 'hello',
isFromUs: false,
unread: false,
messages: [{ sender: 'alice', text: 'hello', timestamp: now - 1_000, isSelf: false }],
}],
nextBatch: 's1',
lastDeltaSyncAt: now,
}), 'utf-8');
const logger = createMockLogger();
const matrix = new MatrixClient({
homeserver: 'https://matrix.example.test',
accessToken: 'token',
userId: '@me:example.test',
capturedAt: now,
}, logger, { minRequestIntervalMs: 0 });
matrix.sync = vi.fn(async () => ({
next_batch: 's2',
rooms: {
join: {
'!alice:example.test': {
timeline: {
events: [{
event_id: '$sent',
type: 'm.room.message',
sender: '@me:example.test',
origin_server_ts: now,
content: { msgtype: 'm.text', body: 'AI-assisted reply' },
}],
},
},
},
},
})) as typeof matrix.sync;
const cache = new ChatCache(logger, cacheFile);
const service = new ChatService(
matrix,
logger,
undefined,
undefined,
undefined,
undefined,
cache,
undefined,
undefined,
authenticatedAccountAuth,
);
await service.loadCache();
await service.refreshTopChats();
const [conversation] = await service.getCachedConversations();
expect(conversation.messages).toEqual([
{ sender: 'alice', text: 'hello', timestamp: now - 1_000, isSelf: false },
{ sender: 'Me', text: 'AI-assisted reply', timestamp: now, isSelf: true },
]);
const persisted = JSON.parse(await fs.readFile(cacheFile, 'utf-8')) as {
chats: Array<{ messages: unknown[] }>;
};
expect(persisted.chats[0].messages).toHaveLength(2);
});
});
+8
View File
@@ -179,6 +179,14 @@ describe('ConversationsPage', () => {
'src',
'/media/received%20image.jpg',
);
expect(screen.getByRole('link', { name: 'View original image received from u/alice' })).toHaveAttribute(
'href',
'/media/received%20image.jpg',
);
expect(screen.getByRole('link', { name: 'View original image received from u/alice' })).toHaveAttribute(
'target',
'_blank',
);
});
it('turns a suggestion into an editable draft and confirms sending without losing failed input', async () => {
+59
View File
@@ -93,6 +93,65 @@ describe('MatrixClient HTTP timeout', () => {
}, 10_000);
});
describe('MatrixClient direct room registration', () => {
it('creates rooms with the direct-message flag', async () => {
let requestBody: unknown;
const client = new MatrixClient(creds, createMockLogger(), {
minRequestIntervalMs: 0,
fetchImpl: vi.fn(async (_url, init) => {
requestBody = JSON.parse(String(init?.body));
return new Response(JSON.stringify({ room_id: '!new:example.test' }), {
status: 200,
headers: { 'content-type': 'application/json' },
});
}),
});
await expect(client.createDirectRoom('@them:example.test'))
.resolves.toBe('!new:example.test');
expect(requestBody).toEqual({
preset: 'reddit_dm',
invite: ['@them:example.test'],
is_direct: true,
});
});
it('additively repairs m.direct account data without removing existing rooms', async () => {
const requests: Array<{ method: string; url: string; body?: unknown }> = [];
const client = new MatrixClient(creds, createMockLogger(), {
minRequestIntervalMs: 0,
fetchImpl: vi.fn(async (url, init) => {
requests.push({
method: init?.method ?? 'GET',
url: String(url),
...(init?.body ? { body: JSON.parse(String(init.body)) } : {}),
});
if (init?.method === 'GET') {
return new Response(JSON.stringify({
'@existing:example.test': ['!old:example.test'],
'@them:example.test': ['!already:example.test'],
}), { status: 200, headers: { 'content-type': 'application/json' } });
}
return new Response('{}', { status: 200, headers: { 'content-type': 'application/json' } });
}),
});
await expect(client.repairDirectRoomMappings([
{ recipientMxid: '@them:example.test', roomId: '!already:example.test' },
{ recipientMxid: '@them:example.test', roomId: '!new:example.test' },
])).resolves.toBe(1);
expect(requests).toHaveLength(2);
expect(requests[1]).toMatchObject({
method: 'PUT',
body: {
'@existing:example.test': ['!old:example.test'],
'@them:example.test': ['!already:example.test', '!new:example.test'],
},
});
});
});
describe('MatrixClient authentication classification', () => {
it('does not classify a room-level 403 as an expired account token', async () => {
const client = new MatrixClient(creds, createMockLogger(), {
+50 -2
View File
@@ -6,12 +6,16 @@ import { createConversationsRouter } from '../../src/web/routes/conversations.js
const servers: Array<ReturnType<express.Express['listen']>> = [];
async function request(app: express.Express, path: string): Promise<Response> {
async function request(
app: express.Express,
path: string,
init?: RequestInit,
): Promise<Response> {
const server = app.listen(0);
servers.push(server);
await new Promise<void>((resolve) => server.once('listening', resolve));
const { port } = server.address() as AddressInfo;
return fetch(`http://127.0.0.1:${port}${path}`);
return fetch(`http://127.0.0.1:${port}${path}`, init);
}
afterEach(async () => {
@@ -177,4 +181,48 @@ describe('conversations route', () => {
});
expect(context.repos.messages.getLatestByRecipient).not.toHaveBeenCalled();
});
it('waits for the conversation delta sync after sending a reply', async () => {
const conversation = {
username: 'alice', roomName: 'alice', roomId: 'room-alice',
url: 'https://chat.reddit.com/room/room-alice',
lastMessageTime: 10, lastMessageSnippet: 'Hello', isFromUs: false, unread: false,
messages: [{ sender: 'alice', text: 'Hello', timestamp: 10, isSelf: false }],
};
let completeSync: (() => void) | undefined;
const refreshTopChats = vi.fn(() => new Promise<void>((resolve) => {
completeSync = resolve;
}));
const context = {
logger: { error: vi.fn() },
activity: { push: vi.fn() },
broadcaster: { push: vi.fn() },
services: { chat: {
getCachedConversations: vi.fn().mockResolvedValue([conversation]),
sendChatMessage: vi.fn().mockResolvedValue(true),
refreshTopChats,
} },
} as unknown as WebContext;
const app = express()
.use(express.json())
.use('/api', createConversationsRouter(context));
let responseSettled = false;
const responsePromise = request(app, '/api/conversations/room-alice/send-reply', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ message: 'AI-assisted reply' }),
}).then((response) => {
responseSettled = true;
return response;
});
await vi.waitFor(() => expect(refreshTopChats).toHaveBeenCalledWith(5));
expect(responseSettled).toBe(false);
completeSync?.();
const response = await responsePromise;
expect(response.status).toBe(200);
expect(await response.json()).toEqual({ success: true });
});
});