import { describe, expect, it } from 'vitest'; import { redis } from '../src/lib/redis'; import { claimDue, ensureScheduled, markWatched, pollSoon, schedulePoll } from '../src/services/scheduler'; import { useTestApp } from './helpers'; useTestApp(); const score = async (id: string) => Number(await redis.zscore('poll:due', id)); describe('poll scheduler', () => { it('claims only due connections, and concurrent claimers never get the same one', async () => { for (let i = 0; i < 50; i++) await schedulePoll(`due-${i}`); await schedulePoll('later', 60_000); const batches = await Promise.all(Array.from({ length: 5 }, () => claimDue(20))); const claimed = batches.flat(); expect(claimed).toHaveLength(50); expect(new Set(claimed).size).toBe(50); expect(claimed).not.toContain('later'); }); it('parks claimed items in the future so a crashed worker gets retried, not lost', async () => { await schedulePoll('c1'); await claimDue(); expect(await score('c1')).toBeGreaterThan(Date.now() + 4 * 60_000); }); it('pollSoon only moves polls earlier', async () => { await schedulePoll('soon', 1_000); await schedulePoll('late', 600_000); const before = await score('soon'); await pollSoon(['soon', 'late']); expect(await score('late')).toBeLessThanOrEqual(Date.now()); expect(await score('soon')).toBeLessThanOrEqual(before); }); it('ensureScheduled never overrides an existing schedule', async () => { await schedulePoll('existing', 600_000); const before = await score('existing'); await ensureScheduled(['existing', 'new']); expect(await score('existing')).toBe(before); expect(await score('new')).toBeLessThanOrEqual(Date.now()); }); it('markWatched reports only newly watched users', async () => { expect(await markWatched(['a', 'b'])).toEqual(['a', 'b']); expect(await markWatched(['a', 'b', 'c'])).toEqual(['c']); expect(await redis.ttl('watched:a')).toBeGreaterThan(80); }); });