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

import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { UNCLASSIFIED_ERROR_PHRASE } from '../src/errors/publicPhrase.js';
import { HealthRegistry } from '../src/health/registry.js';
import { DeltaPoller } from '../src/poller/deltaPoller.js';

describe('DeltaPoller', () => {
  beforeEach(() => {
    vi.useFakeTimers();
  });

  afterEach(() => {
    vi.useRealTimers();
  });

  it('polls on the configured interval after start and stops cleanly', async () => {
    const poll = vi.fn().mockResolvedValue(undefined);
    const poller = new DeltaPoller({ intervalMs: 1000, poll });

    expect(poller.running).toBe(false);
    poller.start();
    poller.start(); // idempotent
    expect(poller.running).toBe(true);

    await vi.advanceTimersByTimeAsync(3000);
    expect(poll).toHaveBeenCalledTimes(3);

    await poller.stop();
    await poller.stop(); // idempotent
    expect(poller.running).toBe(false);
    await vi.advanceTimersByTimeAsync(3000);
    expect(poll).toHaveBeenCalledTimes(3);
  });

  it('skips a tick while the previous poll is still in flight', async () => {
    let release!: () => void;
    const poll = vi.fn(
      () =>
        new Promise<void>((resolve) => {
          release = resolve;
        }),
    );
    const poller = new DeltaPoller({ intervalMs: 1000, poll });

    poller.start();
    await vi.advanceTimersByTimeAsync(2500); // ticks at 1000 and 2000 — second skipped
    expect(poll).toHaveBeenCalledTimes(1);

    release();
    await vi.advanceTimersByTimeAsync(1000);
    expect(poll).toHaveBeenCalledTimes(2);

    // stop() now awaits the in-flight (second) tick, so release it first.
    const stopped = poller.stop();
    release();
    await stopped;
  });

  it('reports poll errors via onError and keeps polling', async () => {
    const onError = vi.fn();
    const poll = vi.fn().mockRejectedValue(new Error('arena unreachable'));
    const poller = new DeltaPoller({ intervalMs: 1000, poll, onError });

    poller.start();
    await vi.advanceTimersByTimeAsync(2000);

    expect(poll).toHaveBeenCalledTimes(2);
    expect(onError).toHaveBeenCalledTimes(2);
    expect(onError.mock.calls[0]?.[0]).toBeInstanceOf(Error);

    await poller.stop();
  });

  describe('cursor ownership', () => {
    it('starts the cursor at construction time and advances it to the tick time only after a successful poll', async () => {
      let nowMs = 1_000;
      const poll = vi.fn().mockResolvedValue(undefined);
      const poller = new DeltaPoller({ intervalMs: 1000, poll, now: () => new Date(nowMs) });
      expect(poller.nextSince.getTime()).toBe(1_000);

      poller.start();
      nowMs = 2_000;
      await vi.advanceTimersByTimeAsync(1000); // tick 1 at t=2000: since=1000
      nowMs = 3_000;
      await vi.advanceTimersByTimeAsync(1000); // tick 2 at t=3000: since=2000

      expect(poll.mock.calls.map((call) => (call[0] as Date).getTime())).toEqual([1_000, 2_000]);
      expect(poller.nextSince.getTime()).toBe(3_000);
      await poller.stop();
    });

    it('keeps the same `since` across consecutive ticks while the poll keeps rejecting', async () => {
      let nowMs = 1_000;
      const poll = vi.fn().mockRejectedValue(new Error('arena unreachable'));
      const poller = new DeltaPoller({
        intervalMs: 1000,
        poll,
        onError: () => undefined,
        now: () => new Date(nowMs),
      });

      poller.start();
      nowMs = 2_000;
      await vi.advanceTimersByTimeAsync(1000);
      nowMs = 3_000;
      await vi.advanceTimersByTimeAsync(1000);
      nowMs = 4_000;
      await vi.advanceTimersByTimeAsync(1000);

      // A failed poll must not lose its window: every tick re-covers from 1000.
      expect(poll.mock.calls.map((call) => (call[0] as Date).getTime())).toEqual([1_000, 1_000, 1_000]);
      expect(poller.nextSince.getTime()).toBe(1_000);

      // First success advances to that tick's `asOf`, and the next tick uses it.
      poll.mockResolvedValue(undefined);
      nowMs = 5_000;
      await vi.advanceTimersByTimeAsync(1000);
      nowMs = 6_000;
      await vi.advanceTimersByTimeAsync(1000);
      expect(poll.mock.calls.slice(3).map((call) => (call[0] as Date).getTime())).toEqual([1_000, 5_000]);
      await poller.stop();
    });

    it('captures asOf BEFORE the fetch so changes arriving mid-poll are not skipped', async () => {
      let nowMs = 1_000;
      let release!: () => void;
      const poll = vi.fn(
        () =>
          new Promise<void>((resolve) => {
            release = resolve;
          }),
      );
      const poller = new DeltaPoller({ intervalMs: 1000, poll, now: () => new Date(nowMs) });

      poller.start();
      nowMs = 2_000;
      await vi.advanceTimersByTimeAsync(1000); // tick starts at t=2000
      nowMs = 2_900; // time passes while the poll is in flight
      release();
      await vi.advanceTimersByTimeAsync(0);

      // Cursor is the tick's start time, not its completion time.
      expect(poller.nextSince.getTime()).toBe(2_000);
      await poller.stop();
    });
  });

  describe('stop()', () => {
    it('resolves immediately when no tick is in flight', async () => {
      const poller = new DeltaPoller({ intervalMs: 1000, poll: vi.fn() });
      poller.start();
      await expect(poller.stop()).resolves.toBeUndefined();
      expect(poller.running).toBe(false);
    });

    it('returns the in-flight tick promise so shutdown can await it', async () => {
      let release!: () => void;
      const poll = vi.fn(
        () =>
          new Promise<void>((resolve) => {
            release = resolve;
          }),
      );
      const poller = new DeltaPoller({ intervalMs: 1000, poll });
      poller.start();
      await vi.advanceTimersByTimeAsync(1000);
      expect(poll).toHaveBeenCalledTimes(1);

      let settled = false;
      const stopped = poller.stop().then(() => {
        settled = true;
      });
      await vi.advanceTimersByTimeAsync(5000);
      expect(poller.running).toBe(false);
      expect(settled).toBe(false); // still waiting on the tick
      expect(poll).toHaveBeenCalledTimes(1); // and no new ticks were scheduled

      release();
      await stopped;
      expect(settled).toBe(true);
    });
  });

  describe('health reporting', () => {
    it('records success and failure with the tick start time', async () => {
      let nowMs = 10_000;
      const health = new HealthRegistry();
      const poll = vi.fn().mockRejectedValueOnce(new Error('boom')).mockResolvedValueOnce(undefined);
      const poller = new DeltaPoller({
        intervalMs: 1000,
        poll,
        health,
        onError: () => undefined,
        now: () => new Date(nowMs),
      });

      poller.start();
      nowMs = 11_000;
      await vi.advanceTimersByTimeAsync(1000);
      expect(health.status()).toBe('degraded');
      // B5: the registry stores a fixed public phrase, never the raw message.
      expect(health.pollerSnapshot()).toEqual({
        lastRunAt: new Date(11_000).toISOString(),
        lastError: UNCLASSIFIED_ERROR_PHRASE,
      });

      nowMs = 12_000;
      await vi.advanceTimersByTimeAsync(1000);
      expect(health.status()).toBe('ok');
      expect(health.pollerSnapshot()).toEqual({
        lastRunAt: new Date(12_000).toISOString(),
        lastSuccessAt: new Date(12_000).toISOString(),
      });
      await poller.stop();
    });
  });
});