import fs from 'fs/promises'; import os from 'os'; import path from 'path'; import { describe, expect, it, vi } from 'vitest'; import { ChatCache } from '../src/services/chat/cache.js'; import { MatrixClient } from '../src/services/chat/matrix/client.js'; import type { MatrixMessagesResponse } from '../src/services/chat/matrix/types.js'; import { ChatService } from '../src/services/chat/service.js'; import { createMockLogger } from './mockLogger.js'; import { authenticatedAccountAuth, createTestDb, createTestRepositories } from './testUtils.js'; const credentials = { homeserver: 'https://matrix.example.test', accessToken: 'token', userId: '@me:reddit.com', capturedAt: Date.now(), }; describe('ChatService progressive full backfill', () => { it('records replies for performance only when the room has source context', async () => { const logger = createMockLogger(); const tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), 'chat-performance-context-')); const db = createTestDb(); const repos = createTestRepositories(db); try { const sentAt = Date.now(); repos.postsRepo.insert({ id: 'post-alice', subreddit: 'example', author: 'alice', title: 'Source post', imageUrl: null, processedAt: sentAt - 1_000, }); repos.messagesRepo.insert({ postId: 'post-alice', recipient: 'alice', subject: 'alice', message: 'Hello Alice', sentAt, strategyId: 'direct', chatRoomId: '!alice:reddit.com', }); const matrix = new MatrixClient(credentials, logger, { minRequestIntervalMs: 0 }); matrix.sync = vi.fn(async () => ({ next_batch: 's1', rooms: { join: {} } })) as typeof matrix.sync; matrix.getJoinedRooms = vi.fn(async () => ['!alice:reddit.com', '!bob:reddit.com']); matrix.getRoomMessages = vi.fn(async (roomId: string) => ({ chunk: [ { event_id: `$reply-${roomId}`, type: 'm.room.message', sender: roomId === '!alice:reddit.com' ? '@t2_alice:reddit.com' : '@t2_bob:reddit.com', origin_server_ts: sentAt + 1_000, content: { body: 'Hi there' }, }, { event_id: `$first-${roomId}`, type: 'm.room.message', sender: '@me:reddit.com', origin_server_ts: sentAt, content: { body: roomId === '!alice:reddit.com' ? 'Hello Alice' : 'Hello Bob' }, }, ], start: 'start', })); const service = new ChatService( matrix, logger, repos.successfulMessagesRepo, undefined, undefined, { resolveT2Ids: vi.fn(async () => ({ t2_alice: 'alice', t2_bob: 'bob' })) }, new ChatCache(logger, path.join(tmpDir, 'chat-cache.json')), repos.messagesRepo, undefined, authenticatedAccountAuth, ); await service.fetchAllMessages(); expect(repos.messagesRepo.getByChatRoomId('!alice:reddit.com')?.gotResponse).toBe(true); expect(repos.messagesRepo.getByChatRoomId('!bob:reddit.com')).toBeUndefined(); expect(repos.messagesRepo.getResponseRateTrend()).toEqual([ expect.objectContaining({ totalSent: 1, totalResponses: 1, responseRate: 100 }), ]); // A single reply counts toward response performance, but the // reciprocal-engagement policy excludes it from learning examples. expect(repos.successfulMessagesRepo.getRecent()).toEqual([]); } finally { db.close(); await fs.rm(tmpDir, { recursive: true, force: true }); } }); it('publishes room summaries and recent timeline messages before deep history finishes', async () => { const logger = createMockLogger(); const tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), 'chat-progressive-backfill-')); let finishHistory!: (response: MatrixMessagesResponse) => void; const pendingHistory = new Promise((resolve) => { finishHistory = resolve; }); let finishUsernameResolution!: (usernames: Record) => void; const pendingUsernames = new Promise>((resolve) => { finishUsernameResolution = resolve; }); try { const matrix = new MatrixClient(credentials, logger, { minRequestIntervalMs: 0 }); matrix.sync = vi.fn(async () => ({ next_batch: 's1', rooms: { join: { '!alice:reddit.com': { summary: { 'm.heroes': ['@t2_alice:reddit.com'] }, timeline: { events: [ { event_id: '$first', type: 'm.room.message', sender: '@me:reddit.com', origin_server_ts: 1_000, content: { body: 'Hello Alice' }, }, { event_id: '$reply', type: 'm.room.message', sender: '@t2_alice:reddit.com', origin_server_ts: 2_000, content: { body: 'Hi there' }, }, ], }, }, }, }, })) as typeof matrix.sync; matrix.getJoinedRooms = vi.fn(async () => ['!alice:reddit.com']); matrix.getRoomMessages = vi.fn(() => pendingHistory); const usernameResolver = { resolveT2Ids: vi.fn(() => pendingUsernames), }; const cachePath = path.join(tmpDir, 'chat-cache.json'); const service = new ChatService( matrix, logger, undefined, undefined, undefined, usernameResolver, new ChatCache(logger, cachePath), undefined, undefined, authenticatedAccountAuth, ); const backfill = service.fetchAllMessages(); await vi.waitFor(() => expect(usernameResolver.resolveT2Ids).toHaveBeenCalledOnce()); const visibleChats = await service.getCachedConversations(); expect(visibleChats).toHaveLength(1); expect(visibleChats[0]).toMatchObject({ roomId: '!alice:reddit.com', username: 't2_alice', lastMessageSnippet: 'Hi there', messages: [ { text: 'Hello Alice', isSelf: true }, { text: 'Hi there', isSelf: false }, ], }); expect(usernameResolver.resolveT2Ids).toHaveBeenCalledWith(['t2_alice']); const persisted = JSON.parse(await fs.readFile(cachePath, 'utf-8')) as { chats: unknown[] }; expect(persisted.chats).toHaveLength(1); finishUsernameResolution({ t2_alice: 'alice' }); await vi.waitFor(() => expect(matrix.getRoomMessages).toHaveBeenCalledOnce()); await vi.waitFor(async () => { const enrichedChats = await service.getCachedConversations(); expect(enrichedChats[0].username).toBe('alice'); }); finishHistory({ chunk: [], start: 'start' }); await backfill; } finally { finishUsernameResolution({}); finishHistory({ chunk: [], start: 'start' }); await fs.rm(tmpDir, { recursive: true, force: true }); } }); it('rejects a required-complete backfill and leaves the full-sync marker unset when a room fails', async () => { const logger = createMockLogger(); const tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), 'chat-incomplete-backfill-')); try { const matrix = new MatrixClient(credentials, logger, { minRequestIntervalMs: 0 }); matrix.sync = vi.fn(async () => ({ next_batch: 's1', rooms: { join: {} } })) as typeof matrix.sync; matrix.getJoinedRooms = vi.fn(async () => ['!alice:reddit.com']); matrix.getRoomMessages = vi.fn().mockRejectedValue(new Error('temporary history failure')); const cache = new ChatCache(logger, path.join(tmpDir, 'chat-cache.json')); const service = new ChatService( matrix, logger, undefined, undefined, undefined, undefined, cache, undefined, undefined, authenticatedAccountAuth, ); await expect(service.fetchAllMessages({ requireComplete: true })) .rejects.toThrow('1 of 1 conversations failed'); expect(cache.getMarkers().lastFullSyncAt).toBeNull(); } finally { await fs.rm(tmpDir, { recursive: true, force: true }); } }); it('paginates beyond the former 50-page history ceiling', async () => { const logger = createMockLogger(); const tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), 'chat-long-backfill-')); try { const matrix = new MatrixClient(credentials, logger, { minRequestIntervalMs: 0 }); matrix.sync = vi.fn(async () => ({ next_batch: 's1', rooms: { join: {} } })) as typeof matrix.sync; matrix.getJoinedRooms = vi.fn(async () => ['!long:reddit.com']); let page = 0; matrix.getRoomMessages = vi.fn(async () => { page++; return { start: `start-${page}`, end: page < 51 ? `page-${page}` : undefined, chunk: [{ event_id: `$event-${page}`, type: 'm.room.message', sender: '@t2_long:reddit.com', origin_server_ts: page, content: { msgtype: 'm.text', body: `message ${page}` }, }], }; }); const service = new ChatService( matrix, logger, undefined, undefined, undefined, undefined, new ChatCache(logger, path.join(tmpDir, 'chat-cache.json')), undefined, undefined, authenticatedAccountAuth, ); const [chat] = await service.fetchAllMessages({ requireComplete: true }); expect(matrix.getRoomMessages).toHaveBeenCalledTimes(51); expect(chat.messages).toHaveLength(51); } finally { await fs.rm(tmpDir, { recursive: true, force: true }); } }); });