SnowLuma/packages/core/tests/webui-state-stream.test.ts

import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { StateBus, type StateResource } from '../src/webui/state-bus';
import { bindStateStream } from '../src/webui/state-stream';

interface CapturedFrame {
    resource: StateResource;
    data: unknown;
}

function makeSnapshotFn(table: Partial<Record<StateResource, unknown>>) {
    return vi.fn(async (resource: StateResource): Promise<unknown> => table[resource]);
}

describe('bindStateStream', () => {
    beforeEach(() => {
        vi.useFakeTimers();
    });
    afterEach(() => {
        vi.useRealTimers();
    });

    it('joins concurrent initial snapshots for the same resource', async () => {
        const bus = new StateBus();
        const sent: CapturedFrame[] = [];
        let resolve!: (data: unknown) => void;
        const slow = new Promise((done) => {
            resolve = done;
        });
        const snapshot = vi.fn((resource: StateResource) =>
            resource === 'processes' ? slow : resource,
        );
        const handle = bindStateStream({ bus, snapshot, send: (frame) => sent.push(frame) });
        const first = handle.sendAllInitial();
        const second = handle.sendAllInitial();
        await vi.advanceTimersByTimeAsync(0);
        expect(snapshot).toHaveBeenCalledTimes(3);
        resolve('processes');
        await Promise.all([first, second]);
        expect(sent).toHaveLength(3);
        handle.dispose();
    });

    it('absorbs a scheduled publish into an initial snapshot without starting another request', async () => {
        const bus = new StateBus();
        const sent: CapturedFrame[] = [];
        let resolve!: (data: unknown) => void;
        const slow = new Promise((done) => {
            resolve = done;
        });
        const snapshot = vi.fn((resource: StateResource) =>
            resource === 'processes' ? slow : resource,
        );
        const handle = bindStateStream({ bus, snapshot, send: (frame) => sent.push(frame) });
        bus.publish('processes');
        const initial = handle.sendAllInitial();
        await vi.advanceTimersByTimeAsync(100);
        expect(snapshot).toHaveBeenCalledTimes(3);
        resolve('fresh');
        await initial;
        expect(sent.filter((frame) => frame.resource === 'processes')).toEqual([
            { resource: 'processes', data: 'fresh' },
        ]);
        handle.dispose();
    });

    it('sendAllInitial emits one frame per resource with the snapshot data', async () => {
        const bus = new StateBus();
        const sent: CapturedFrame[] = [];
        const snapshot = makeSnapshotFn({
            processes: [{ pid: 4242, name: 'qq', status: 'available' }],
            'qq-list': [{ uin: '12345', nickname: 'me' }],
            connections: [{ uin: '12345', nickname: 'me', adapters: [] }],
        });
        const handle = bindStateStream({
            bus,
            snapshot,
            send: (f) => sent.push(f),
            debounceMs: 50,
        });
        await handle.sendAllInitial();

        expect(snapshot).toHaveBeenCalledTimes(3);
        expect(new Set(sent.map((f) => f.resource))).toEqual(
            new Set(['processes', 'qq-list', 'connections']),
        );
        const procFrame = sent.find((f) => f.resource === 'processes');
        expect(procFrame?.data).toEqual([{ pid: 4242, name: 'qq', status: 'available' }]);

        handle.dispose();
    });

    it('a single bus publish reaches send() exactly once after debounceMs', async () => {
        const bus = new StateBus();
        const sent: CapturedFrame[] = [];
        const snapshot = makeSnapshotFn({ processes: [{ pid: 9999 }] });
        const handle = bindStateStream({
            bus,
            snapshot,
            send: (f) => sent.push(f),
            debounceMs: 50,
        });

        bus.publish('processes');
        expect(sent).toHaveLength(0); // still debounced
        await vi.advanceTimersByTimeAsync(49);
        expect(sent).toHaveLength(0);
        await vi.advanceTimersByTimeAsync(2);
        expect(sent).toEqual([{ resource: 'processes', data: [{ pid: 9999 }] }]);

        handle.dispose();
    });

    it('coalesces a burst of publishes for the SAME resource into one send', async () => {
        const bus = new StateBus();
        const sent: CapturedFrame[] = [];
        const snapshot = makeSnapshotFn({ processes: 'fresh' });
        const handle = bindStateStream({
            bus,
            snapshot,
            send: (f) => sent.push(f),
            debounceMs: 50,
        });

        for (let i = 0; i < 20; i++) {
            bus.publish('processes');
        }

        await vi.advanceTimersByTimeAsync(51);
        expect(sent).toEqual([{ resource: 'processes', data: 'fresh' }]);
        // Snapshot was only invoked once for the whole burst.
        expect(snapshot).toHaveBeenCalledExactlyOnceWith('processes');

        handle.dispose();
    });

    it('each resource has its own independent debounce timer', async () => {
        const bus = new StateBus();
        const sent: CapturedFrame[] = [];
        const snapshot = makeSnapshotFn({
            processes: 'p',
            'qq-list': 'q',
            connections: 'c',
        });
        const handle = bindStateStream({
            bus,
            snapshot,
            send: (f) => sent.push(f),
            debounceMs: 50,
        });

        bus.publish('processes');
        await vi.advanceTimersByTimeAsync(30);
        bus.publish('qq-list');
        await vi.advanceTimersByTimeAsync(25); // processes fires at t=50 (5ms after this), qq-list pending till t=80
        expect(sent.map((f) => f.resource)).toEqual(['processes']);

        await vi.advanceTimersByTimeAsync(30);
        expect(sent.map((f) => f.resource)).toEqual(['processes', 'qq-list']);

        bus.publish('connections');
        await vi.advanceTimersByTimeAsync(51);
        expect(sent.map((f) => f.resource)).toEqual(['processes', 'qq-list', 'connections']);

        handle.dispose();
    });

    it('dispose() stops further sends, both pending and future', async () => {
        const bus = new StateBus();
        const sent: CapturedFrame[] = [];
        const snapshot = makeSnapshotFn({ processes: 'x' });
        const handle = bindStateStream({
            bus,
            snapshot,
            send: (f) => sent.push(f),
            debounceMs: 50,
        });

        bus.publish('processes');
        handle.dispose();
        await vi.advanceTimersByTimeAsync(200);
        expect(sent).toEqual([]);

        // Subsequent publishes are dead too.
        bus.publish('processes');
        await vi.advanceTimersByTimeAsync(200);
        expect(sent).toEqual([]);
    });

    it('snapshot() rejection is isolated — the failed resource is skipped and others keep flowing', async () => {
        const bus = new StateBus();
        const sent: CapturedFrame[] = [];
        const snapshot = vi.fn(async (resource: StateResource): Promise<unknown> => {
            if (resource === 'processes') {
                throw new Error('listProcesses failed');
            }

            if (resource === 'qq-list') {
                return [{ uin: '1' }];
            }

            return [];
        });
        const handle = bindStateStream({
            bus,
            snapshot,
            send: (f) => sent.push(f),
            debounceMs: 50,
        });

        bus.publish('processes');
        bus.publish('qq-list');
        await vi.advanceTimersByTimeAsync(60);

        // qq-list went through; processes didn't (snapshot threw).
        expect(sent.find((f) => f.resource === 'qq-list')?.data).toEqual([{ uin: '1' }]);
        expect(sent.find((f) => f.resource === 'processes')).toBeUndefined();

        handle.dispose();
    });

    it('serialises overlapping flushes for the same resource — the LAST sent frame reflects the LATEST publish (no stale overwrite)', async () => {
        const bus = new StateBus();
        const sent: CapturedFrame[] = [];
        // Variable-latency snapshot: 1st call sleeps 200ms (slow), 2nd call
        // sleeps 10ms (fast). Without single-flight, the second publish would
        // race-overlap and send BEFORE the first, leaving the client's last
        // observed frame as the older snapshot.
        let call = 0;
        const versions = ['snap-T0', 'snap-T60'];
        const snapshot = vi.fn(async (_resource: StateResource): Promise<unknown> => {
            const i = call++;
            const sleep = i === 0 ? 200 : 10;
            await new Promise<void>((r) => setTimeout(r, sleep));
            return versions[i] ?? `snap-#${i}`;
        });
        const handle = bindStateStream({
            bus,
            snapshot,
            send: (f) => sent.push(f),
            debounceMs: 50,
        });

        bus.publish('processes'); // schedules timer1 → fires at 50, snapshot A (200ms)
        await vi.advanceTimersByTimeAsync(60);
        bus.publish('processes'); // arrives WHILE flush A is mid-snapshot
        await vi.advanceTimersByTimeAsync(60); // hits 120: timer2 should have fired, flush B observed_after A completes
        await vi.advanceTimersByTimeAsync(300); // settle everything

        // Exactly two frames; the LAST one must reflect the most recent snapshot.
        expect(sent).toHaveLength(2);
        expect(sent[sent.length - 1]).toEqual({ resource: 'processes', data: 'snap-T60' });
        // For single-flight: snapshot is called once per flush (no overlap),
        // so the SECOND snapshot read sees the latest state.
        expect(call).toBe(2);

        handle.dispose();
    });

    it('multiple publishes during an in-flight flush coalesce into ONE follow-up flush, not N', async () => {
        const bus = new StateBus();
        const sent: CapturedFrame[] = [];
        let call = 0;
        const snapshot = vi.fn(async (_resource: StateResource): Promise<unknown> => {
            call++;
            await new Promise<void>((r) => setTimeout(r, 100));
            return `snap-#${call}`;
        });
        const handle = bindStateStream({
            bus,
            snapshot,
            send: (f) => sent.push(f),
            debounceMs: 50,
        });

        bus.publish('processes');
        await vi.advanceTimersByTimeAsync(55);

        // flush A in flight (100ms more). Send 5 publishes during the in-flight window.
        for (let i = 0; i < 5; i++) {
            bus.publish('processes');
        }

        await vi.advanceTimersByTimeAsync(200);
        await vi.advanceTimersByTimeAsync(200);

        // 1 initial flush + 1 coalesced follow-up = 2 frames, NOT 6.
        expect(sent).toHaveLength(2);
        expect(call).toBe(2);

        handle.dispose();
    });

    it('send() throw is isolated — the next publish still reaches a healthy send()', async () => {
        const bus = new StateBus();
        const sent: CapturedFrame[] = [];
        let mode: 'throw' | 'ok' = 'throw';
        const snapshot = makeSnapshotFn({ processes: 'x' });
        const handle = bindStateStream({
            bus,
            snapshot,
            send: (f) => {
                if (mode === 'throw') {
                    throw new Error('controller dead');
                }

                sent.push(f);
            },
            debounceMs: 50,
        });

        bus.publish('processes');
        await vi.advanceTimersByTimeAsync(60);
        expect(sent).toEqual([]);

        mode = 'ok';
        bus.publish('processes');
        await vi.advanceTimersByTimeAsync(60);
        expect(sent).toEqual([{ resource: 'processes', data: 'x' }]);

        handle.dispose();
    });
});