RedditBot/tests/chat-service-progressive-backfill.test.ts

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<MatrixMessagesResponse>((resolve) => {
      finishHistory = resolve;
    });
    let finishUsernameResolution!: (usernames: Record<string, string>) => void;
    const pendingUsernames = new Promise<Record<string, string>>((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('derives isFromUs from history when /sync returns no timeline for the room', async () => {
    const logger = createMockLogger();
    const tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), 'chat-authorship-backfill-'));
    try {
      const matrix = new MatrixClient(credentials, logger, { minRequestIntervalMs: 0 });
      // Reddit omits timeline events for idle rooms, so the seeded summary
      // carries no authorship signal at all.
      matrix.sync = vi.fn(async () => ({ next_batch: 's1', rooms: { join: {} } })) as typeof matrix.sync;
      matrix.getJoinedRooms = vi.fn(async () => ['!unanswered:reddit.com', '!answered:reddit.com']);
      matrix.getRoomMessages = vi.fn(async (roomId: string) => ({
        start: 'start',
        chunk: roomId === '!unanswered:reddit.com'
          ? [{
              event_id: '$opener',
              type: 'm.room.message',
              sender: '@me:reddit.com',
              origin_server_ts: 1_000,
              content: { body: 'Hello there' },
            }]
          : [{
              event_id: '$reply',
              type: 'm.room.message',
              sender: '@t2_bob:reddit.com',
              origin_server_ts: 2_000,
              content: { body: 'Hi back' },
            }, {
              event_id: '$opener',
              type: 'm.room.message',
              sender: '@me:reddit.com',
              origin_server_ts: 1_000,
              content: { body: 'Hello there' },
            }],
      }));

      const service = new ChatService(
        matrix, logger, undefined, undefined, undefined, undefined,
        new ChatCache(logger, path.join(tmpDir, 'chat-cache.json')),
        undefined, undefined, authenticatedAccountAuth,
      );

      const chats = await service.fetchAllMessages({ requireComplete: true });
      const unanswered = chats.find((c) => c.roomId === '!unanswered:reddit.com');
      const answered = chats.find((c) => c.roomId === '!answered:reddit.com');
      expect(unanswered?.isFromUs).toBe(true);
      expect(answered?.isFromUs).toBe(false);
    } 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 });
    }
  });
});