SnowLuma/packages/core/tests/webui-connection-diff-loop.test.ts

import { getLogLevel, setLogLevel, subscribeLogs, type LogEntry } from '@snowluma/common/logger';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { StateBus } from '../src/webui/state-bus';
import {
    createConnectionPoller,
    type ConnectionDiffLoopOptions,
} from '../src/webui/connection-diff-loop';
import { comparableConnectionSnapshot } from '../src/webui/connection-snapshot';

const previousLogLevel = getLogLevel();

function startConnectionDiffLoop(options: ConnectionDiffLoopOptions) {
    return { dispose: createConnectionPoller(options)() };
}

describe('connection polling', () => {
    it('shares one timer and releases it with the last subscriber', () => {
        const bus = new StateBus();
        const getSnapshot = vi.fn(() => []);
        const subscribe = createConnectionPoller({ bus, getSnapshot });
        vi.advanceTimersByTime(1000);
        expect(getSnapshot).not.toHaveBeenCalled();
        expect(vi.getTimerCount()).toBe(0);
        const releaseFirst = subscribe();
        const releaseSecond = subscribe();
        expect(getSnapshot).toHaveBeenCalledOnce();
        expect(vi.getTimerCount()).toBe(1);
        releaseFirst();
        releaseFirst();
        vi.advanceTimersByTime(500);
        expect(getSnapshot).toHaveBeenCalledTimes(2);
        releaseSecond();
        expect(vi.getTimerCount()).toBe(0);
        vi.advanceTimersByTime(1000);
        expect(getSnapshot).toHaveBeenCalledTimes(2);
    });

    it('takes a fresh baseline on resubscription and detects changes from the first interval', () => {
        const bus = new StateBus();
        const publish = vi.spyOn(bus, 'publish');
        let snapshot = 'initial';
        const subscribe = createConnectionPoller({ bus, getSnapshot: () => snapshot });
        const release = subscribe();
        snapshot = 'connected';
        vi.advanceTimersByTime(500);
        expect(publish).toHaveBeenCalledOnce();
        release();
        snapshot = 'offline';
        const releaseNext = subscribe();
        vi.advanceTimersByTime(500);
        expect(publish).toHaveBeenCalledOnce();
        releaseNext();
    });

    it('treats an undefined snapshot as an established baseline', () => {
        vi.useFakeTimers();
        const bus = new StateBus();
        const publish = vi.spyOn(bus, 'publish');
        let snapshot: unknown = undefined;
        const handle = startConnectionDiffLoop({ bus, getSnapshot: () => snapshot });
        vi.advanceTimersByTime(1000);
        expect(publish).not.toHaveBeenCalled();
        snapshot = null;
        vi.advanceTimersByTime(500);
        expect(publish).toHaveBeenCalledOnce();
        handle.dispose();
    });

    beforeEach(() => {
        vi.useFakeTimers();
    });
    afterEach(() => {
        setLogLevel(previousLogLevel);
        vi.clearAllTimers();
        vi.useRealTimers();
    });

    it('does NOT publish when the snapshot is stable tick after tick', () => {
        const bus = new StateBus();
        const seen: string[] = [];
        bus.subscribe((r) => seen.push(r));
        const handle = startConnectionDiffLoop({
            bus,
            getSnapshot: () => [{ uin: '12345', nickname: 'me', adapters: [] }],
            intervalMs: 500,
        });

        vi.advanceTimersByTime(2_000); // 4 ticks
        expect(seen).toEqual([]);
        handle.dispose();
    });

    it('traces only changed snapshot publication under one semantic context', () => {
        const bus = new StateBus();
        const entries: LogEntry[] = [];
        let snap: unknown = [{ uin: '1', adapters: [] }];
        setLogLevel('trace');
        const unsubscribe = subscribeLogs((entry) => entries.push(entry));
        const handle = startConnectionDiffLoop({
            bus,
            getSnapshot: () => snap,
            intervalMs: 500,
        });

        try {
            vi.advanceTimersByTime(500);
            vi.advanceTimersByTime(1_500);
            expect(entries.filter((entry) => entry.scope === 'WebUI.Connections')).toEqual([]);

            snap = [{ uin: '1', adapters: [{ kind: 'ws', status: 'connected' }] }];
            vi.advanceTimersByTime(500);
            vi.advanceTimersByTime(1_000);

            const trace = entries.filter(
                (entry) => entry.level === 'trace' && entry.scope === 'WebUI.Connections',
            );
            expect(trace.map((entry) => entry.message)).toEqual([
                'connection_diff_start snapshot=[{uin:"1",adapters:[{kind:"ws",status:"connected"}]}]',
                'connection_diff_branch branch=publish resource=connections',
                expect.stringMatching(
                    /^connection_diff_terminal outcome=completed reason=published elapsedMs=\d+$/,
                ),
            ]);
            expect(trace.map((entry) => entry.req)).toEqual([
                expect.any(Number),
                trace[0]!.req,
                trace[0]!.req,
            ]);
        } finally {
            handle.dispose();
            unsubscribe();
        }
    });

    it('routes snapshot and projector failures through the common logger without repeated noise', () => {
        const bus = new StateBus();
        const entries: LogEntry[] = [];
        const getSnapshot = vi.fn(() => {
            throw new Error('snapshot boom');
        });
        setLogLevel('trace');
        const unsubscribe = subscribeLogs((entry) => entries.push(entry));
        const snapshotHandle = startConnectionDiffLoop({
            bus,
            getSnapshot,
            intervalMs: 500,
        });

        try {
            vi.advanceTimersByTime(2_000);
            expect(
                entries.filter(
                    (entry) =>
                        entry.scope === 'WebUI.Connections' &&
                        entry.message.includes('reason=snapshot_failed'),
                ),
            ).toHaveLength(1);
        } finally {
            snapshotHandle.dispose();
        }

        const projectorHandle = startConnectionDiffLoop({
            bus,
            getSnapshot: () => [{ uin: '1' }],
            pickComparable: () => {
                throw new Error('projector boom');
            },
            intervalMs: 500,
        });

        try {
            vi.advanceTimersByTime(2_000);
            const failures = entries.filter(
                (entry) =>
                    entry.scope === 'WebUI.Connections' &&
                    entry.message.includes('reason=projector_failed'),
            );
            expect(failures).toHaveLength(1);
            expect(failures[0]!.message).toMatch(
                /^connection_diff_terminal outcome=failed reason=projector_failed error="projector boom" elapsedMs=\d+$/,
            );
        } finally {
            projectorHandle.dispose();
            unsubscribe();
        }
    });

    it('keeps one-shot projector failures visible when TRACE is disabled', () => {
        const bus = new StateBus();
        const entries: LogEntry[] = [];
        setLogLevel('info');
        const unsubscribe = subscribeLogs((entry) => entries.push(entry));
        const handle = startConnectionDiffLoop({
            bus,
            getSnapshot: () => [{ uin: '1' }],
            pickComparable: () => {
                throw new Error('projector boom');
            },
            intervalMs: 500,
        });

        try {
            vi.advanceTimersByTime(2_000);
            expect(
                entries.filter(
                    (entry) =>
                        entry.level === 'error' &&
                        entry.scope === 'WebUI.Connections' &&
                        entry.message.includes('projector boom'),
                ),
            ).toHaveLength(1);
        } finally {
            handle.dispose();
            unsubscribe();
        }
    });

    it('publishes only when the current snapshot differs from the subscription baseline', () => {
        const bus = new StateBus();
        const seen: string[] = [];
        bus.subscribe((r) => seen.push(r));
        let snap = [
            { uin: '12345', nickname: 'me', adapters: [{ kind: 'http', status: 'listening' }] },
        ];
        const handle = startConnectionDiffLoop({
            bus,
            getSnapshot: () => snap,
            intervalMs: 500,
        });
        vi.advanceTimersByTime(500);
        expect(seen).toEqual([]);
        snap = [
            { uin: '12345', nickname: 'me', adapters: [{ kind: 'http', status: 'connected' }] },
        ];
        vi.advanceTimersByTime(500);
        expect(seen).toEqual(['connections']);
        // Next tick: stable again → no extra publish.
        vi.advanceTimersByTime(500);
        expect(seen).toEqual(['connections']);
        handle.dispose();
    });

    it('publishes once per CHANGE, not once per tick the snapshot differs from initial', () => {
        const bus = new StateBus();
        const seen: string[] = [];
        bus.subscribe((r) => seen.push(r));
        const versions = [
            [{ uin: '1', nickname: 'a', adapters: [] }], // baseline (no publish on first observation)
            [{ uin: '1', nickname: 'a', adapters: [{ kind: 'http', status: 'listening' }] }],
            [{ uin: '1', nickname: 'a', adapters: [{ kind: 'http', status: 'listening' }] }],
            [{ uin: '1', nickname: 'a', adapters: [{ kind: 'http', status: 'connected' }] }],
        ];
        let i = 0;
        const handle = startConnectionDiffLoop({
            bus,
            getSnapshot: () => versions[Math.min(i, versions.length - 1)] ?? [],
            intervalMs: 500,
        });

        vi.advanceTimersByTime(500); // i=0 baseline
        i = 1;
        vi.advanceTimersByTime(500); // change #1
        i = 2;
        vi.advanceTimersByTime(500); // stable vs i=1
        i = 3;
        vi.advanceTimersByTime(500); // change #2

        expect(seen).toEqual(['connections', 'connections']);
        handle.dispose();
    });

    it('the connection projector ignores adapter detail changes', () => {
        const baseline = [
            {
                uin: '1',
                nickname: 'account',
                adapters: [
                    {
                        name: 'webhook',
                        kind: 'httpClient',
                        status: 'ok',
                        detail: '上次推送 14:00:00',
                    },
                ],
            },
        ];
        const changed = [
            {
                uin: '1',
                nickname: 'account',
                adapters: [
                    {
                        name: 'webhook',
                        kind: 'httpClient',
                        status: 'ok',
                        detail: '上次推送 14:00:01',
                    },
                ],
            },
        ];

        expect(comparableConnectionSnapshot(changed)).toEqual(
            comparableConnectionSnapshot(baseline),
        );
    });

    it('publishes when database migration progress changes', () => {
        const bus = new StateBus();
        const seen: string[] = [];
        bus.subscribe((resource) => seen.push(resource));
        let processed = 100;
        const handle = startConnectionDiffLoop({
            bus,
            getSnapshot: () => [
                {
                    uin: '1',
                    nickname: 'account',
                    adapters: [],
                    databaseMigration: {
                        phase: 'migrating',
                        usable: true,
                        processed,
                        total: 1000,
                        progress: processed / 1000,
                        estimatedRemainingSeconds: 90,
                    },
                },
            ],
            pickComparable: comparableConnectionSnapshot,
            intervalMs: 500,
        });

        vi.advanceTimersByTime(500);
        processed = 200;
        vi.advanceTimersByTime(500);

        expect(seen).toEqual(['connections']);
        handle.dispose();
    });

    it('dispose() stops further ticks, no publish even after a change', () => {
        const bus = new StateBus();
        const seen: string[] = [];
        bus.subscribe((r) => seen.push(r));
        let snap: object[] = [];
        const handle = startConnectionDiffLoop({
            bus,
            getSnapshot: () => snap,
            intervalMs: 500,
        });
        handle.dispose();
        snap = [{ uin: '1', nickname: 'a', adapters: [] }];
        vi.advanceTimersByTime(5_000);
        expect(seen).toEqual([]);
    });

    it('pickComparable() strips volatile fields so detail-only changes do NOT publish', () => {
        const bus = new StateBus();
        const seen: string[] = [];
        bus.subscribe((r) => seen.push(r));
        // Reproduces the real-world bug: HttpPostAdapter.describeStatus()
        // embeds HH:MM:SS in `detail`; every webhook delivery rotates that
        // string every second. WITHOUT pickComparable, the loop would publish
        // every tick under any active webhook. WITH it (stripping `detail`),
        // only level changes publish.
        let snap = [
            {
                uin: '1',
                adapters: [{ kind: 'httpClient', status: 'ok', detail: '上次推送 14:00:00' }],
            },
        ];
        const handle = startConnectionDiffLoop({
            bus,
            getSnapshot: () => snap,
            pickComparable: (s) => {
                if (!Array.isArray(s)) {
                    return s;
                }

                return (s as Array<{ uin: string; adapters?: unknown[] }>).map((acc) => ({
                    uin: acc.uin,
                    adapters: Array.isArray(acc.adapters)
                        ? acc.adapters.map((a: unknown) => {
                              const o = a as { name?: string; kind?: string; status?: string };
                              return { name: o.name, kind: o.kind, status: o.status };
                          })
                        : acc.adapters,
                }));
            },
            intervalMs: 500,
        });
        vi.advanceTimersByTime(500); // baseline
        expect(seen).toEqual([]);

        // detail changes every second; raw JSON differs each tick.
        for (let t = 1; t <= 10; t++) {
            snap = [
                {
                    uin: '1',
                    adapters: [
                        {
                            kind: 'httpClient',
                            status: 'ok',
                            detail: `上次推送 14:00:${String(t).padStart(2, '0')}`,
                        },
                    ],
                },
            ];
            vi.advanceTimersByTime(500);
        }

        expect(seen).toEqual([]); // ZERO publishes — detail is not comparable
        // Now a REAL state change (status level) → publish exactly once.
        snap = [
            { uin: '1', adapters: [{ kind: 'httpClient', status: 'down', detail: '连接失败' }] },
        ];
        vi.advanceTimersByTime(500);
        expect(seen).toEqual(['connections']);
        handle.dispose();
    });

    it('a throwing pickComparable() is isolated — loop keeps ticking, no publish', () => {
        const bus = new StateBus();
        const seen: string[] = [];
        bus.subscribe((r) => seen.push(r));
        const handle = startConnectionDiffLoop({
            bus,
            getSnapshot: () => [{ uin: '1' }],
            pickComparable: () => {
                throw new Error('projector boom');
            },
            intervalMs: 500,
        });
        vi.advanceTimersByTime(2_000);
        expect(seen).toEqual([]);
        handle.dispose();
    });

    it('survives a throwing snapshot() — keeps trying on the next tick', () => {
        const bus = new StateBus();
        const seen: string[] = [];
        bus.subscribe((r) => seen.push(r));
        let mode: 'throw' | 'stable' | 'changed' = 'throw';
        const handle = startConnectionDiffLoop({
            bus,
            getSnapshot: () => {
                if (mode === 'throw') {
                    throw new Error('getConnectionStatuses failed');
                }

                if (mode === 'stable') {
                    return [{ uin: '1', adapters: [] }];
                }

                return [{ uin: '1', adapters: [{ kind: 'ws', status: 'connected' }] }];
            },
            intervalMs: 500,
        });
        vi.advanceTimersByTime(500); // throws — skipped, no publish
        expect(seen).toEqual([]);
        mode = 'stable';
        vi.advanceTimersByTime(500); // first SUCCESSFUL read — baseline, no publish
        expect(seen).toEqual([]);
        mode = 'changed';
        vi.advanceTimersByTime(500); // real change → publish
        expect(seen).toEqual(['connections']);
        handle.dispose();
    });
});