arena-plm-fiori/worker/test/arenaDeltaPoll.test.ts

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