import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { HealthRegistry } from '../src/health/registry.js'; import { IngestBufferFullError, makeArenaDeltaPoll } from '../src/poller/arenaDeltaPoll.js'; import { DeltaPoller } from '../src/poller/deltaPoller.js'; import { InMemoryEventQueue } from '../src/queue/webhookQueue.js'; describe('makeArenaDeltaPoll', () => { it('fetches changes since the cursor and enqueues each as a poll-sourced event', async () => { const queue = new InMemoryEventQueue({ maxSize: 10 }); const fetchChangesSince = vi.fn().mockResolvedValue([{ id: 'c1' }, { id: 'c2' }]); const warn = vi.fn(); const poll = makeArenaDeltaPoll({ client: { fetchChangesSince }, queue, log: { warn } }); const since = new Date('2026-09-14T08:00:00Z'); await poll(since); expect(fetchChangesSince).toHaveBeenCalledWith(since); expect(queue.peekAll().map((e) => [e.source, e.payload])).toEqual([ ['poll', { id: 'c1' }], ['poll', { id: 'c2' }], ]); expect(warn).not.toHaveBeenCalled(); }); it('fails the tick WITHOUT calling Arena when the buffer is already full (B1)', async () => { const queue = new InMemoryEventQueue({ maxSize: 1 }); queue.enqueue({ source: 'webhook', payload: { n: 1 } }); const fetchChangesSince = vi.fn().mockResolvedValue([{ id: 'c1' }]); const warn = vi.fn(); const poll = makeArenaDeltaPoll({ client: { fetchChangesSince }, queue, log: { warn } }); await expect(poll(new Date(0))).rejects.toBeInstanceOf(IngestBufferFullError); expect(fetchChangesSince).not.toHaveBeenCalled(); expect(queue.size).toBe(1); expect(warn).toHaveBeenCalledTimes(1); expect(warn.mock.calls[0]?.[1]).toMatch(/queue full/); }); it('fails the tick after a partial enqueue when the buffer fills mid-poll (B1)', async () => { const queue = new InMemoryEventQueue({ maxSize: 1 }); const fetchChangesSince = vi.fn().mockResolvedValue([{ id: 'c1' }, { id: 'c2' }, { id: 'c3' }]); const warn = vi.fn(); const poll = makeArenaDeltaPoll({ client: { fetchChangesSince }, queue, log: { warn } }); const failure = await poll(new Date(0)).catch((err: unknown) => err); expect(failure).toBeInstanceOf(IngestBufferFullError); expect(failure).toMatchObject({ fetched: 3, dropped: 2 }); // What fit stays queued; the rest is re-covered by the next tick (same window). expect(queue.size).toBe(1); expect(queue.peekAll()[0]?.payload).toEqual({ id: 'c1' }); expect(warn).toHaveBeenCalledTimes(1); expect(warn.mock.calls[0]?.[0]).toMatchObject({ dropped: 2, fetched: 3 }); }); it('propagates a fetch failure so the poller keeps its cursor', async () => { const queue = new InMemoryEventQueue({ maxSize: 1 }); const fetchChangesSince = vi.fn().mockRejectedValue(new Error('arena unreachable')); const poll = makeArenaDeltaPoll({ client: { fetchChangesSince }, queue, log: { warn: vi.fn() } }); await expect(poll(new Date(0))).rejects.toThrow('arena unreachable'); expect(queue.size).toBe(0); }); }); describe('makeArenaDeltaPoll driven by DeltaPoller (B1 end to end)', () => { beforeEach(() => { vi.useFakeTimers(); }); afterEach(() => { vi.useRealTimers(); }); it('a partial drop holds the cursor, flips health to degraded, and the next tick re-covers the window', async () => { const queue = new InMemoryEventQueue({ maxSize: 1 }); const fetchChangesSince = vi.fn().mockResolvedValue([{ id: 'c1' }, { id: 'c2' }]); const health = new HealthRegistry(); const start = new Date('2026-09-14T08:00:00Z'); let nowMs = start.getTime(); const poller = new DeltaPoller({ intervalMs: 1000, poll: makeArenaDeltaPoll({ client: { fetchChangesSince }, queue, log: { warn: vi.fn() } }), health, now: () => new Date(nowMs), }); const initialSince = poller.nextSince; poller.start(); nowMs += 1000; await vi.advanceTimersByTimeAsync(1000); expect(fetchChangesSince).toHaveBeenCalledTimes(1); expect(fetchChangesSince).toHaveBeenLastCalledWith(initialSince); expect(poller.nextSince).toEqual(initialSince); // cursor did NOT advance expect(health.status()).toBe('degraded'); expect(health.pollerSnapshot()?.lastError).toBe('Ingest buffer full at poll time.'); // Buffer drains; the next tick re-covers the SAME window and succeeds. queue.dequeue(); fetchChangesSince.mockResolvedValue([{ id: 'c2' }]); nowMs += 1000; await vi.advanceTimersByTimeAsync(1000); expect(fetchChangesSince).toHaveBeenCalledTimes(2); expect(fetchChangesSince).toHaveBeenLastCalledWith(initialSince); expect(poller.nextSince.getTime()).toBeGreaterThan(initialSince.getTime()); expect(health.status()).toBe('ok'); await poller.stop(); }); it('a full buffer before the fetch never calls Arena and leaves the cursor untouched', async () => { const queue = new InMemoryEventQueue({ maxSize: 1 }); queue.enqueue({ source: 'webhook', payload: {} }); const fetchChangesSince = vi.fn().mockResolvedValue([]); const health = new HealthRegistry(); const poller = new DeltaPoller({ intervalMs: 1000, poll: makeArenaDeltaPoll({ client: { fetchChangesSince }, queue, log: { warn: vi.fn() } }), health, }); const initialSince = poller.nextSince; poller.start(); await vi.advanceTimersByTimeAsync(2500); expect(fetchChangesSince).not.toHaveBeenCalled(); expect(poller.nextSince).toEqual(initialSince); expect(health.status()).toBe('degraded'); await poller.stop(); }); });