diff --git a/apps/dashboard/src/features/conversations/conversation-detail.tsx b/apps/dashboard/src/features/conversations/conversation-detail.tsx index b172991..b09f40a 100644 --- a/apps/dashboard/src/features/conversations/conversation-detail.tsx +++ b/apps/dashboard/src/features/conversations/conversation-detail.tsx @@ -35,12 +35,19 @@ function MessageHistory({ conversation }: { conversation: ConversationDetailResp {message.media?.length ? (
{message.media.map((media) => ( - {`Image + rel="noreferrer" + target="_blank" + > + {`Image + ))}
) : null} diff --git a/apps/dashboard/src/features/conversations/conversations.css b/apps/dashboard/src/features/conversations/conversations.css index cdc1834..493323b 100644 --- a/apps/dashboard/src/features/conversations/conversations.css +++ b/apps/dashboard/src/features/conversations/conversations.css @@ -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); } diff --git a/src/services/chat/matrix/client.ts b/src/services/chat/matrix/client.ts index 961a13f..131bf3e 100644 --- a/src/services/chat/matrix/client.ts +++ b/src/services/chat/matrix/client.ts @@ -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 { 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 { + if (mappings.length === 0) return 0; + + let directRooms: Record; + try { + directRooms = await this.request>( + '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>( + '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 }; } diff --git a/src/services/chat/service.ts b/src/services/chat/service.ts index eb07c8c..2759abd 100644 --- a/src/services/chat/service.ts +++ b/src/services/chat/service.ts @@ -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; - const chat = existing.get(roomId); - if (!chat) continue; + for (const [roomId, room] of Object.entries(joined)) { + const scraped = this.scrapeMessageEvents(room?.timeline?.events ?? []); + 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 { + const mappings: Array<{ recipientMxid: string; roomId: string }> = []; + const seen = new Set(); + + 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') diff --git a/src/web/routes/conversations.ts b/src/web/routes/conversations.ts index abb5b7b..1c7883d 100644 --- a/src/web/routes/conversations.ts +++ b/src/web/routes/conversations.ts @@ -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' }); } diff --git a/tests/chat-service-direct-room-repair.test.ts b/tests/chat-service-direct-room-repair.test.ts new file mode 100644 index 0000000..f045d93 --- /dev/null +++ b/tests/chat-service-direct-room-repair.test.ts @@ -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', + }]); + }); +}); diff --git a/tests/chat-service-text-sync.test.ts b/tests/chat-service-text-sync.test.ts new file mode 100644 index 0000000..ae4bd56 --- /dev/null +++ b/tests/chat-service-text-sync.test.ts @@ -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); + }); +}); diff --git a/tests/dashboard/conversations.test.tsx b/tests/dashboard/conversations.test.tsx index 0d9a55a..75fc8a9 100644 --- a/tests/dashboard/conversations.test.tsx +++ b/tests/dashboard/conversations.test.tsx @@ -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 () => { diff --git a/tests/matrix-client.test.ts b/tests/matrix-client.test.ts index b36457b..fa1ca6e 100644 --- a/tests/matrix-client.test.ts +++ b/tests/matrix-client.test.ts @@ -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(), { diff --git a/tests/web/conversations-route.test.ts b/tests/web/conversations-route.test.ts index cf047e5..71666c8 100644 --- a/tests/web/conversations-route.test.ts +++ b/tests/web/conversations-route.test.ts @@ -6,12 +6,16 @@ import { createConversationsRouter } from '../../src/web/routes/conversations.js const servers: Array> = []; -async function request(app: express.Express, path: string): Promise { +async function request( + app: express.Express, + path: string, + init?: RequestInit, +): Promise { const server = app.listen(0); servers.push(server); await new Promise((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((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 }); + }); });