RedditBot/tests/chat-service-delta-media.test.ts

import { describe, it, expect, vi } from 'vitest';
import fs from 'fs/promises';
import os from 'os';
import path from 'path';
import { ChatService } from '../src/services/chat/service.js';
import { ChatCache } from '../src/services/chat/cache.js';
import { MatrixClient } from '../src/services/chat/matrix/client.js';
import { MediaService } from '../src/services/media.js';
import type { MediaDownloader } from '../src/services/media.js';
import { createMockLogger } from './mockLogger.js';
import { authenticatedAccountAuth, createTestDb, createTestRepositories } from './testUtils.js';

const creds = {
  homeserver: 'https://matrix.example.test',
  accessToken: 'tok',
  userId: '@me:example.test',
  capturedAt: Date.now(),
};

function buildDownloader(): MediaDownloader {
  return {
    fetch: vi.fn(async () => ({
      buffer: Buffer.from('image-bytes'),
      contentType: 'image/jpeg',
    })),
  };
}

describe('ChatService delta sync — media attachment', () => {
  it('downloads media for image events arriving via /sync', async () => {
    const logger = createMockLogger();
    const tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), 'chat-delta-media-'));
    try {
      const matrix = new MatrixClient(creds, logger, { minRequestIntervalMs: 0 });
      matrix.sync = vi.fn(async () => ({
        next_batch: 's2',
        rooms: {
          join: {
            '!room1:example.test': {
              timeline: {
                events: [
                  {
                    type: 'm.room.message',
                    sender: '@t2_abc:example.test',
                    origin_server_ts: 1000,
                    content: { msgtype: 'm.image', url: 'mxc://example.test/img1' },
                  },
                  {
                    type: 'm.room.message',
                    sender: '@t2_abc:example.test',
                    origin_server_ts: 1100,
                    content: { body: 'plain text follow-up' },
                  },
                ],
              },
            },
          },
        },
      })) as typeof matrix.sync;

      const downloader = buildDownloader();
      const media = new MediaService(downloader, logger, tmpDir);
      const cache = new ChatCache(logger, path.join(tmpDir, 'chat-cache.json'));
      const svc = new ChatService(matrix, logger, undefined, media, undefined, undefined, cache, undefined, undefined, authenticatedAccountAuth);

      await svc.getActiveChats();

      expect(downloader.fetch).toHaveBeenCalledTimes(1);
      const items = await media.getMediaItems();
      expect(items).toHaveLength(1);
      expect(items[0].originalUrl).toBe('mxc://example.test/img1');
      expect(items[0].chatRoomId).toBe('!room1:example.test');
    } finally {
      await fs.rm(tmpDir, { recursive: true, force: true });
    }
  });

  it('appends new media messages to chats that already have loaded history', async () => {
    const logger = createMockLogger();
    const tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), 'chat-delta-media-merge-'));
    const cacheFile = path.join(tmpDir, 'chat-cache.json');
    const db = createTestDb();
    const repos = createTestRepositories(db);
    try {
      const now = Date.now();
      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: '!room1:example.test',
      });
      const existing = {
        version: 1,
        savedAt: now - 60_000,
        chats: [
          {
            username: 'alice',
            roomName: 'alice',
            lastMessageTime: now - 5_000,
            lastMessageSnippet: 'hi',
            isFromUs: false,
            unread: false,
            roomId: '!room1:example.test',
            messages: [
              { sender: 't2_abc', text: 'hi', timestamp: now - 5_000, isSelf: false },
            ],
          },
        ],
        nextBatch: 's1',
        lastFullSyncAt: now - 120_000,
        lastDeltaSyncAt: now - 60_000,
        cooldown: null,
      };
      const imageTs = now - 1_000;
      await fs.writeFile(cacheFile, JSON.stringify(existing), 'utf-8');

      const matrix = new MatrixClient(creds, logger, { minRequestIntervalMs: 0 });
      matrix.sync = vi.fn(async () => ({
        next_batch: 's2',
        rooms: {
          join: {
            '!room1:example.test': {
              timeline: {
                events: [
                  {
                    type: 'm.room.message',
                    sender: '@t2_abc:example.test',
                    origin_server_ts: imageTs,
                    content: { msgtype: 'm.image', url: 'mxc://example.test/img1' },
                  },
                ],
              },
            },
          },
        },
      })) as typeof matrix.sync;

      const media = new MediaService(buildDownloader(), logger, tmpDir);
      const cache = new ChatCache(logger, cacheFile);
      const svc = new ChatService(matrix, logger, undefined, media, undefined, undefined, cache, repos.messagesRepo, undefined, authenticatedAccountAuth);
      await svc.loadCache();

      // refreshTopChats always runs a /sync (unlike getActiveChats, which can
      // short-circuit on a fresh cache).
      await svc.refreshTopChats();
      const chats = await svc.getCachedConversations();
      const alice = chats.find((c) => c.roomId === '!room1:example.test');
      expect(alice).toBeDefined();
      expect(alice!.messages).toBeDefined();
      // Original message + the new image-bearing message.
      expect(alice!.messages!.length).toBe(2);
      const imageMsg = alice!.messages!.find((m) => m.timestamp === imageTs);
      expect(imageMsg).toBeDefined();
      expect(imageMsg!.media).toBeDefined();
      expect(imageMsg!.media!.length).toBe(1);
      const performanceMessage = repos.messagesRepo.getByChatRoomId('!room1:example.test');
      expect(performanceMessage).toMatchObject({ gotResponse: true, gotImage: true });
      expect(repos.messagesRepo.getStrategyStats()).toEqual([
        expect.objectContaining({
          strategyId: 'direct', totalResponses: 1, totalImages: 1, responseRate: 100, imageRate: 100,
        }),
      ]);
    } finally {
      db.close();
      await fs.rm(tmpDir, { recursive: true, force: true });
    }
  });

  it('is a no-op when no MediaService is configured', async () => {
    const logger = createMockLogger();
    const matrix = new MatrixClient(creds, logger, { minRequestIntervalMs: 0 });
    matrix.sync = vi.fn(async () => ({
      next_batch: 's2',
      rooms: {
        join: {
          '!room1:example.test': {
            timeline: {
              events: [
                {
                  type: 'm.room.message',
                  sender: '@t2_abc:example.test',
                  origin_server_ts: 1000,
                  content: { url: 'mxc://example.test/img1' },
                },
              ],
            },
          },
        },
      },
    })) as typeof matrix.sync;

    const cache = new ChatCache(
      logger,
      path.join(os.tmpdir(), `chat-cache-${Date.now()}-${Math.random()}.json`),
    );
    const svc = new ChatService(matrix, logger, undefined, undefined, undefined, undefined, cache, undefined, undefined, authenticatedAccountAuth);
    await expect(svc.getActiveChats()).resolves.toBeDefined();
  });

  it('backfills the room when Matrix marks a delta timeline as limited', async () => {
    const logger = createMockLogger();
    const tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), 'chat-limited-delta-media-'));
    try {
      const matrix = new MatrixClient(creds, logger, { minRequestIntervalMs: 0 });
      matrix.sync = vi.fn(async () => ({
        next_batch: 's2',
        rooms: { join: { '!room1:example.test': {
          timeline: {
            limited: true,
            prev_batch: 'gap-token',
            events: [],
          },
        } } },
      })) as typeof matrix.sync;
      matrix.getRoomMessages = vi.fn(async () => ({
        start: 'latest',
        chunk: [{
          event_id: '$missed-image', type: 'm.room.message', sender: '@t2_abc:example.test',
          origin_server_ts: 1_000,
          content: { msgtype: 'm.image', info: { thumbnail_url: 'mxc://example.test/missed' } },
        }],
      }));

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

      await expect(svc.refreshTopChats()).resolves.toBe(1);
      expect(matrix.getRoomMessages).toHaveBeenCalledWith(
        '!room1:example.test', { from: undefined, limit: 100, dir: 'b' },
      );
      expect(await media.getMediaItems()).toEqual([
        expect.objectContaining({ originalUrl: 'mxc://example.test/missed' }),
      ]);
    } finally {
      await fs.rm(tmpDir, { recursive: true, force: true });
    }
  });

  it('does not persist the next delta cursor when a media download fails', async () => {
    const logger = createMockLogger();
    const tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), 'chat-media-retry-cursor-'));
    const cacheFile = path.join(tmpDir, 'chat-cache.json');
    try {
      await fs.writeFile(cacheFile, JSON.stringify({
        version: 1, savedAt: 1, chats: [], nextBatch: 's1',
        lastFullSyncAt: 1, lastDeltaSyncAt: 1, cooldown: null,
      }));
      const matrix = new MatrixClient(creds, logger, { minRequestIntervalMs: 0 });
      matrix.sync = vi.fn(async () => ({
        next_batch: 's2',
        rooms: { join: { '!room1:example.test': { timeline: { events: [{
          event_id: '$image', type: 'm.room.message', sender: '@t2_abc:example.test',
          origin_server_ts: 1_000, content: { msgtype: 'm.image', url: 'mxc://example.test/fails' },
        }] } } } },
      })) as typeof matrix.sync;
      const media = new MediaService({ fetch: vi.fn().mockResolvedValue(null) }, logger, tmpDir);
      const cache = new ChatCache(logger, cacheFile);
      const svc = new ChatService(
        matrix, logger, undefined, media, undefined, undefined, cache,
        undefined, undefined, authenticatedAccountAuth,
      );
      await svc.loadCache();

      await expect(svc.refreshTopChats()).rejects.toThrow('Failed to retrieve');

      const persisted = JSON.parse(await fs.readFile(cacheFile, 'utf-8')) as { nextBatch: string };
      expect(persisted.nextBatch).toBe('s1');
    } finally {
      await fs.rm(tmpDir, { recursive: true, force: true });
    }
  });
});