RedditBot/tests/chat-service-outcome-ingestion.test.ts

import fs from 'fs/promises';
import os from 'os';
import path from 'path';
import { afterEach, describe, expect, it, vi } from 'vitest';
import { ConversationOutcomesRepository, ConversationStateRepository } from '../src/database/repositories.js';
import { buildConversationState } from '../src/services/conversation-state.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<ReturnType<typeof createTestDb>> = [];

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);

    fixture.markConversationComplete();
    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.markConversationComplete();
    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 conversationStates = new ConversationStateRepository(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,
    conversationStates,
  );
  await service.loadCache();

  return {
    db, repos, outcomes, now, roomId, sentMessageId, matrix, service,
    markConversationComplete() {
      const state = buildConversationState([]);
      state.automation.status = 'human-required';
      state.automation.handoffReason = 'Conversation complete.';
      conversationStates.save(roomId, state);
    },
  };
}

function syncResponse(nextBatch: string, events: ReturnType<typeof messageEvent>[]): 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 },
  };
}