import fs from 'fs/promises'; import os from 'os'; import path from 'path'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { ConversationOutcomesRepository } from '../src/database/repositories.js'; import { ChatCache } from '../src/services/chat/cache.js'; import { MatrixClient } from '../src/services/chat/matrix/client.js'; import type { MatrixSyncResponse } 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:example.test', capturedAt: Date.now(), }; const temporaryDirectories: string[] = []; const openDatabases: Array> = []; afterEach(async () => { for (const database of openDatabases.splice(0)) database.close(); await Promise.all(temporaryDirectories.splice(0).map((directory) => fs.rm(directory, { recursive: true, force: true }))); }); describe('ChatService outcome ingestion', () => { it('records delta replies, promotes reciprocal engagement, and avoids duplicate examples', async () => { const fixture = await createFixture(); const firstReplyAt = fixture.now - 2_000; const secondSelfAt = fixture.now - 1_500; const secondReplyAt = fixture.now - 1_000; fixture.matrix.sync = vi.fn() .mockResolvedValueOnce(syncResponse('s2', [ messageEvent('$reply-1', '@alice:example.test', firstReplyAt, 'Thanks!'), ])) .mockResolvedValue(syncResponse('s3', [ messageEvent('$reply-1', '@alice:example.test', firstReplyAt, 'Thanks!'), messageEvent('$self-2', credentials.userId, secondSelfAt, 'What do you think?'), messageEvent('$reply-2', '@alice:example.test', secondReplyAt, 'Tell me more'), ])) as typeof fixture.matrix.sync; await fixture.service.refreshTopChats(); expect(fixture.outcomes.get(fixture.sentMessageId)?.outcome).toBe('unknown'); expect(fixture.repos.successfulMessagesRepo.count()).toBe(0); await fixture.service.refreshTopChats(); expect(fixture.outcomes.get(fixture.sentMessageId)?.outcome).toBe('engaged'); expect(fixture.repos.successfulMessagesRepo.count()).toBe(1); await fixture.service.refreshTopChats(); expect(fixture.repos.successfulMessagesRepo.count()).toBe(1); expect(fixture.repos.messagesRepo.getByChatRoomId(fixture.roomId)).toMatchObject({ gotResponse: true, }); }); it('backfills cached replies with missing outcomes once at startup', async () => { const fixture = await createFixture({ cachedMessages: [ { sender: 'Me', text: 'Hello', timestampOffset: -5_000, isSelf: true }, { sender: 'alice', text: 'Thanks!', timestampOffset: -4_000, isSelf: false }, { sender: 'Me', text: 'What do you think?', timestampOffset: -3_000, isSelf: true }, { sender: 'alice', text: 'Tell me more', timestampOffset: -2_000, isSelf: false }, ], }); fixture.repos.messagesRepo.markGotResponse(fixture.sentMessageId); fixture.matrix.sync = vi.fn(async () => syncResponse('s2', [])) as typeof fixture.matrix.sync; await fixture.service.runStartupSync(); expect(fixture.outcomes.get(fixture.sentMessageId)?.outcome).toBe('engaged'); expect(fixture.repos.successfulMessagesRepo.count()).toBe(1); await fixture.service.runStartupSync(); expect(fixture.repos.successfulMessagesRepo.count()).toBe(1); }); }); async function createFixture(options: { cachedMessages?: Array<{ sender: string; text: string; timestampOffset: number; isSelf: boolean; }>; } = {}) { const logger = createMockLogger(); const directory = await fs.mkdtemp(path.join(os.tmpdir(), 'chat-outcomes-')); temporaryDirectories.push(directory); const cacheFile = path.join(directory, 'chat-cache.json'); const db = createTestDb(); openDatabases.push(db); const repos = createTestRepositories(db); const outcomes = new ConversationOutcomesRepository(db); const now = Date.now(); const roomId = '!alice:example.test'; repos.postsRepo.insert({ id: 'source-post', subreddit: 'example', author: 'alice', title: 'Source', imageUrl: null, processedAt: now - 10_000, }); repos.messagesRepo.insert({ postId: 'source-post', recipient: 'alice', subject: 'alice', message: 'Hello', sentAt: now - 9_000, strategyId: 'direct', chatRoomId: roomId, }); const sentMessageId = repos.messagesRepo.getByChatRoomId(roomId)!.id; const cachedMessages = options.cachedMessages ?? [ { sender: 'Me', text: 'Hello', timestampOffset: -5_000, isSelf: true }, ]; await fs.writeFile(cacheFile, JSON.stringify({ version: 1, savedAt: now - 60_000, chats: [{ username: 'alice', roomName: 'alice', lastMessageTime: now - 5_000, lastMessageSnippet: 'Hello', isFromUs: true, unread: false, roomId, messages: cachedMessages.map(({ timestampOffset, ...message }) => ({ ...message, timestamp: now + timestampOffset, })), }], nextBatch: 's1', lastFullSyncAt: now - 120_000, lastDeltaSyncAt: now - 60_000, cooldown: null, }), 'utf8'); const matrix = new MatrixClient(credentials, logger, { minRequestIntervalMs: 0 }); const cache = new ChatCache(logger, cacheFile); const service = new ChatService( matrix, logger, repos.successfulMessagesRepo, undefined, undefined, undefined, cache, repos.messagesRepo, undefined, authenticatedAccountAuth, outcomes, ); await service.loadCache(); return { db, repos, outcomes, now, roomId, sentMessageId, matrix, service }; } function syncResponse(nextBatch: string, events: ReturnType[]): MatrixSyncResponse { return { next_batch: nextBatch, rooms: { join: { '!alice:example.test': { timeline: { events }, }, }, }, }; } function messageEvent(eventId: string, sender: string, timestamp: number, body: string) { return { event_id: eventId, type: 'm.room.message' as const, sender, origin_server_ts: timestamp, content: { body }, }; }