|
| 1 | +import { copilotChats } from '@sim/db/schema' |
| 2 | +import { queueTableRows, resetDbChainMock } from '@sim/testing/mocks/database.mock' |
| 3 | +import { beforeEach, describe, expect, it, vi } from 'vitest' |
| 4 | + |
| 5 | +const { mockSweepStoragePrefix, deadline } = vi.hoisted(() => ({ |
| 6 | + mockSweepStoragePrefix: vi.fn(), |
| 7 | + deadline: { near: false }, |
| 8 | +})) |
| 9 | + |
| 10 | +vi.mock('@/lib/uploads/core/prefix-sweep', () => ({ |
| 11 | + sweepStoragePrefix: mockSweepStoragePrefix, |
| 12 | + outboxDeadlineNear: () => deadline.near, |
| 13 | +})) |
| 14 | + |
| 15 | +import { CHAT_INLINE_IMAGE_CLEANUP_EVENT } from '@/lib/cleanup/chat-backend-cleanup-event' |
| 16 | +import { chatBackendCleanupOutboxHandlers } from '@/lib/cleanup/chat-backend-cleanup-outbox' |
| 17 | +import type { OutboxEventContext } from '@/lib/core/outbox/service' |
| 18 | + |
| 19 | +const sweepImages = chatBackendCleanupOutboxHandlers[CHAT_INLINE_IMAGE_CLEANUP_EVENT] |
| 20 | +const chatIds = Array.from({ length: 25 }, (_, index) => `chat-${String(index).padStart(2, '0')}`) |
| 21 | +const prefixOf = (chatId: string) => `chat-images/${chatId}/` |
| 22 | + |
| 23 | +function outboxContext() { |
| 24 | + return { |
| 25 | + eventId: 'event-1', |
| 26 | + eventType: CHAT_INLINE_IMAGE_CLEANUP_EVENT, |
| 27 | + attempts: 0, |
| 28 | + maxAttempts: 10, |
| 29 | + signal: new AbortController().signal, |
| 30 | + checkpointPayload: vi.fn(async (_patch: Record<string, unknown>) => {}), |
| 31 | + } satisfies OutboxEventContext |
| 32 | +} |
| 33 | + |
| 34 | +const sweptPrefixes = () => mockSweepStoragePrefix.mock.calls.map(([prefix]) => prefix) |
| 35 | + |
| 36 | +describe('chat inline image cleanup', () => { |
| 37 | + beforeEach(() => { |
| 38 | + resetDbChainMock() |
| 39 | + deadline.near = false |
| 40 | + mockSweepStoragePrefix.mockReset().mockResolvedValue('done') |
| 41 | + }) |
| 42 | + |
| 43 | + it('sweeps several chats at once, never more than ten', async () => { |
| 44 | + queueTableRows(copilotChats, []) |
| 45 | + let inFlight = 0 |
| 46 | + let peak = 0 |
| 47 | + mockSweepStoragePrefix.mockImplementation(async () => { |
| 48 | + peak = Math.max(peak, ++inFlight) |
| 49 | + await new Promise((resolve) => setImmediate(resolve)) |
| 50 | + inFlight-- |
| 51 | + return 'done' |
| 52 | + }) |
| 53 | + |
| 54 | + await sweepImages({ chatIds }, outboxContext()) |
| 55 | + |
| 56 | + expect(peak).toBe(10) |
| 57 | + expect(new Set(sweptPrefixes())).toEqual(new Set(chatIds.map(prefixOf))) |
| 58 | + }) |
| 59 | + |
| 60 | + it('checkpoints the chats finished before the lowest unfinished one and resumes after them', async () => { |
| 61 | + queueTableRows(copilotChats, []) |
| 62 | + const slow = chatIds[3] |
| 63 | + mockSweepStoragePrefix.mockImplementation(async (prefix: string) => { |
| 64 | + if (prefix !== prefixOf(slow)) return 'done' |
| 65 | + // Finishes after the chats behind it, so only "lowest unfinished" can yield 3 here. |
| 66 | + await new Promise((resolve) => setImmediate(resolve)) |
| 67 | + deadline.near = true |
| 68 | + return 'continue' |
| 69 | + }) |
| 70 | + const context = outboxContext() |
| 71 | + |
| 72 | + const result = await sweepImages({ chatIds: chatIds.slice(0, 6) }, context) |
| 73 | + |
| 74 | + expect(result).toMatchObject({ outcome: 'deferred', consumeAttempt: false }) |
| 75 | + expect(context.checkpointPayload).toHaveBeenCalledWith({ swept: 3 }) |
| 76 | + |
| 77 | + queueTableRows(copilotChats, []) |
| 78 | + deadline.near = false |
| 79 | + mockSweepStoragePrefix.mockReset().mockResolvedValue('done') |
| 80 | + await expect( |
| 81 | + sweepImages({ chatIds: chatIds.slice(0, 6), swept: 3 }, outboxContext()) |
| 82 | + ).resolves.toBe(undefined) |
| 83 | + expect(sweptPrefixes().sort()).toEqual(chatIds.slice(3, 6).map(prefixOf)) |
| 84 | + }) |
| 85 | + |
| 86 | + it('skips a chat id no image key can hold instead of failing the event on every retry', async () => { |
| 87 | + queueTableRows(copilotChats, []) |
| 88 | + |
| 89 | + await expect( |
| 90 | + sweepImages({ chatIds: ['../escape', chatIds[0]] }, outboxContext()) |
| 91 | + ).resolves.toBe(undefined) |
| 92 | + expect(sweptPrefixes()).toEqual([prefixOf(chatIds[0])]) |
| 93 | + }) |
| 94 | + |
| 95 | + it('spends an attempt when a chat’s listing stops advancing', async () => { |
| 96 | + queueTableRows(copilotChats, []) |
| 97 | + mockSweepStoragePrefix.mockImplementation(async (prefix: string) => |
| 98 | + prefix === prefixOf(chatIds[0]) ? 'stalled' : 'done' |
| 99 | + ) |
| 100 | + const context = outboxContext() |
| 101 | + |
| 102 | + const result = await sweepImages({ chatIds: chatIds.slice(0, 2) }, context) |
| 103 | + |
| 104 | + expect(result).toMatchObject({ outcome: 'deferred' }) |
| 105 | + expect(result).not.toMatchObject({ consumeAttempt: false }) |
| 106 | + expect(context.checkpointPayload).not.toHaveBeenCalled() |
| 107 | + }) |
| 108 | + |
| 109 | + it('spends an attempt when the lowest unfinished chat stalls, though a later one continues', async () => { |
| 110 | + queueTableRows(copilotChats, []) |
| 111 | + mockSweepStoragePrefix.mockImplementation(async (prefix: string) => { |
| 112 | + if (prefix === prefixOf(chatIds[0])) return 'stalled' |
| 113 | + deadline.near = true |
| 114 | + return 'continue' |
| 115 | + }) |
| 116 | + const context = outboxContext() |
| 117 | + |
| 118 | + const result = await sweepImages({ chatIds: chatIds.slice(0, 2) }, context) |
| 119 | + |
| 120 | + expect(result).toMatchObject({ outcome: 'deferred' }) |
| 121 | + expect(result).not.toMatchObject({ consumeAttempt: false }) |
| 122 | + expect(context.checkpointPayload).not.toHaveBeenCalled() |
| 123 | + }) |
| 124 | +}) |
0 commit comments