import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { emit, subscribe, type PipelineEvent } from '../../src/lib/server/pipeline/bus.js';
import { GET as serverEvents } from '../../src/routes/api/events/servers/+server.js';
import { GET as pipelineEvents } from '../../src/routes/api/events/pipeline/+server.js';

vi.mock('../../src/lib/server/pipeline/bus.js', async importOriginal => {
    const bus = await importOriginal<typeof import('../../src/lib/server/pipeline/bus.js')>();
    return { ...bus, subscribe: vi.fn((listener: Parameters<typeof bus.subscribe>[0]) => vi.fn(bus.subscribe(listener))) };
});
vi.mock('../../src/lib/server/fleet/status.js', () => ({ observationReader: vi.fn(() => ({})) }));
vi.mock('../../src/lib/server/fleet/player-capacity.js', () => ({ playerCapacityReader: vi.fn(() => () => 4) }));
vi.mock('../../src/lib/server/fleet/servers.js', () => ({ listServers: vi.fn(() => [{ name: 'server-1' }]) }));
vi.mock('../../src/lib/server/pipeline/worker.js', () => ({
    statusEntry: vi.fn((server: { name: string }) => ({ server: server.name, players: 0, health: 'online' }))
}));
vi.mock('../../src/lib/server/fleet/telemetry.js', () => ({
    listTelemetry: vi.fn(() => [{ server: 'server-1', event: 'heartbeat' }])
}));

const decoder = new TextDecoder();
const readers: ReadableStreamDefaultReader<Uint8Array>[] = [];
beforeEach(() => { vi.clearAllMocks(); vi.useFakeTimers(); });
afterEach(async () => {
    for (const reader of readers.splice(0)) await reader.cancel();
    vi.useRealTimers();
});

async function open(get: () => Promise<Response>) {
    const response = await get();
    const reader = response.body!.getReader();
    readers.push(reader);
    return { response, reader, unsubscribe: vi.mocked(subscribe).mock.results.at(-1)!.value as ReturnType<typeof vi.fn> };
}

async function readFrame(reader: ReadableStreamDefaultReader<Uint8Array>) {
    const next = await reader.read();
    expect(next.done).toBe(false);
    return decoder.decode(next.value);
}

async function drain(reader: ReadableStreamDefaultReader<Uint8Array>) {
    const frames: string[] = [];
    for (;;) {
        const next = await reader.read();
        if (next.done) return frames;
        frames.push(decoder.decode(next.value));
    }
}

const status: PipelineEvent = { type: 'server-status', entries: [], delta: true };
const telemetry: PipelineEvent = { type: 'server-telemetry', entries: [] };
const sweep: PipelineEvent = { type: 'sweep-tick', frozen: false };
const routes = [
    { name: 'servers', get: serverEvents, initialCount: 3, event: status, ignored: sweep, accepted: [status, telemetry] },
    { name: 'pipeline', get: pipelineEvents, initialCount: 1, event: sweep, ignored: status, accepted: [
        sweep,
        { type: 'deployment-opened', deploymentId: 'deploy-1', action: 'push', servers: ['server-1'] },
        { type: 'server-changed', deploymentId: 'deploy-1', server: 'server-1', files: 2 },
        { type: 'server-state', deploymentId: 'deploy-1', server: 'server-1', state: 'done', players: 0, error: null },
        { type: 'fleet-invalidated' },
        { type: 'beta-changed', betaId: 1 }
    ] as PipelineEvent[] }
];

describe.each(routes)('$name live events', route => {
    it('preserves headers, initial frames, event filtering and heartbeats', async () => {
        const { response, reader, unsubscribe } = await open(route.get);
        expect(response.headers.get('content-type')).toBe('text/event-stream');
        expect(response.headers.get('cache-control')).toBe('no-cache, no-transform');
        expect(response.headers.get('connection')).toBe('keep-alive');
        expect(await readFrame(reader)).toBe(': hello\n\n');
        if (route.name === 'servers') {
            expect(await readFrame(reader)).toBe(`event: server-status\ndata: ${JSON.stringify({
                entries: [{ server: 'server-1', players: 0, health: 'online' }], snapshot: true, serverNow: Date.now()
            })}\n\n`);
            expect(await readFrame(reader)).toBe(`event: server-telemetry\ndata: ${JSON.stringify({
                entries: [{ server: 'server-1', event: 'heartbeat' }], snapshot: true, serverNow: Date.now()
            })}\n\n`);
        }
        emit(route.ignored);
        for (const event of route.accepted) {
            emit(event);
            const data = route.name === 'servers' ? { ...event, serverNow: Date.now() } : event;
            expect(await readFrame(reader)).toBe(`event: ${event.type}\ndata: ${JSON.stringify(data)}\n\n`);
        }
        vi.advanceTimersByTime(20_000);
        expect(await readFrame(reader)).toBe(': ping\n\n');
        expect(unsubscribe).not.toHaveBeenCalled();
    });

    it('bounds an unread connection and releases its subscription and heartbeat on overflow', async () => {
        const { reader, unsubscribe } = await open(route.get);
        for (let i = route.initialCount; i < 64; i++) emit({ ...route.event });
        // An unrelated event must not close a full connection or use queue space.
        for (let i = 0; i < 100; i++) emit(route.ignored);
        expect(unsubscribe).not.toHaveBeenCalled();
        for (let i = 0; i < 1000; i++) emit({ ...route.event });
        expect(unsubscribe).toHaveBeenCalledOnce();
        expect(vi.getTimerCount()).toBe(0);
        const frames = await drain(reader);
        expect(frames).toHaveLength(64);
        expect(frames[0]).toBe(': hello\n\n');
        expect(frames.at(-1)).toContain(`event: ${route.event.type}\n`);
        await reader.cancel();
        expect(unsubscribe).toHaveBeenCalledOnce();

        // A reconnect gets the greeting and current snapshots again.
        vi.advanceTimersByTime(1000);
        const reconnect = await open(route.get);
        expect(await readFrame(reconnect.reader)).toBe(': hello\n\n');
        for (let i = 1; i < route.initialCount; i++) {
            expect(await readFrame(reconnect.reader)).toContain(`"snapshot":true,"serverNow":${Date.now()}`);
        }
        expect(reconnect.unsubscribe).not.toHaveBeenCalled();
    });

    it('keeps a reading client live when another client stalls', async () => {
        const slow = await open(route.get);
        const fast = await open(route.get);
        for (let i = 0; i < route.initialCount; i++) await readFrame(fast.reader);
        for (let i = 0; i < 200; i++) {
            const next = fast.reader.read();
            emit({ ...route.event });
            const frame = await next;
            expect(frame.done).toBe(false);
            expect(decoder.decode(frame.value)).toContain(`event: ${route.event.type}\n`);
        }
        expect(slow.unsubscribe).toHaveBeenCalledOnce();
        expect(fast.unsubscribe).not.toHaveBeenCalled();
        expect(vi.getTimerCount()).toBe(1);
        expect(await drain(slow.reader)).toHaveLength(64);
        await fast.reader.cancel();
        expect(fast.unsubscribe).toHaveBeenCalledOnce();
        expect(vi.getTimerCount()).toBe(0);
    });

    it('also bounds heartbeats when no events arrive', async () => {
        const { reader, unsubscribe } = await open(route.get);
        vi.advanceTimersByTime(20_000 * 65);
        expect(unsubscribe).toHaveBeenCalledOnce();
        expect(vi.getTimerCount()).toBe(0);
        const frames = await drain(reader);
        expect(frames).toHaveLength(64);
        expect(frames.slice(route.initialCount)).toEqual(Array(64 - route.initialCount).fill(': ping\n\n'));
    });

    it.each([false, true])('cleans up cancellation with initial frames consumed: %s', async consumed => {
        const { reader, unsubscribe } = await open(route.get);
        const listener = vi.mocked(subscribe).mock.calls.at(-1)![0];
        if (consumed) for (let i = 0; i < route.initialCount; i++) await readFrame(reader);
        await reader.cancel();
        await reader.cancel();
        expect(unsubscribe).toHaveBeenCalledOnce();
        expect(vi.getTimerCount()).toBe(0);
        // A callback already captured by the bus must be harmless after cleanup.
        expect(() => { listener(route.event); emit(route.event); vi.advanceTimersByTime(60_000); }).not.toThrow();
        expect(await reader.read()).toEqual({ done: true, value: undefined });
    });
});