import { expect, test, vi, type Mock } from 'vitest';

import { SSESession } from '../../source/sse-session/sse-session.ts';
import { ExponentialBackoff } from '../../source/exponential-backoff.ts';
import type { SSESessionOptions, SSEvent } from '../../source/sse-session/types.ts';

import { SseTestStream, type SseTestStreamOptions } from './helpers/sse-stream.ts';
import { priceOracleEvents, storageEvents } from './fixtures/events.fixtures.ts';
import { ExponentialBackoffMaxRetriesHitError, HTTPError, ResponseBodyNullError } from '../../source/errors.ts';

/** URL passed to every session under test. */
const EVENTS_URL = '/events';

/** Headers required for a valid SSE response in these tests. */
const SSE_HEADERS = { 'Content-Type': 'text/event-stream' };

type FetchFn = SSESessionOptions['fetch'];

/**
 * Builds a minimal ExponentialBackoff so reconnect and retry paths finish quickly in tests.
 * Real production delays would make vi.waitFor-based assertions time out.
 *
 * @param maxAttempts - Maximum retry attempts; defaults to 1.
 */
const testRetry = (maxAttempts = 1): ExponentialBackoff => {
    return new ExponentialBackoff({
        baseDelay: 1,
        maxDelay: 1,
        maxAttempts,
        growthRate: 1,
        jitter: 0,
    });
};

/**
 * Wraps fixture chunks in a Response backed by {@link SseTestStream}.
 * SseTestStream simulates a real HTTP body: chunks arrive over time and the stream
 * can optionally close itself when all chunks are sent.
 *
 * @param chunks - Raw SSE payload strings to stream.
 * @param options - Optional stream timing and close behavior.
 */
const sseFetchResponse = (chunks: string[], options: SseTestStreamOptions = {}): Response => {
    return new Response(new SseTestStream(chunks, options).stream, {
        status: 200,
        headers: SSE_HEADERS,
    });
};

/**
 * Creates a vitest mock fetch that delegates to the given responder.
 * SSESession requires fetch injection so tests never hit the network.
 *
 * @param responder - Function that returns the Response for each fetch call.
 */
const createFetchMock = (responder: (url: string, init: RequestInit) => Response | Promise<Response>): Mock<FetchFn> => {
    return vi.fn(async (url: string, init: RequestInit) => responder(url, init));
};

/**
 * Session defaults that disable reconnect noise unless a test opts in.
 * attemptReconnect: false — transport errors should not auto-retry by default.
 * persistent: false — server closing the stream should close the message iterator.
 */
const defaultSessionOptions: Partial<SSESessionOptions> = {
    attemptReconnect: false,
    persistent: false,
    retry: testRetry(),
};

/**
 * Creates an SSESession wired to the injected fetch mock.
 * SSESession.create immediately calls connect(), so fetch is invoked during creation.
 *
 * @param fetch - Mock fetch implementation.
 * @param options - Per-test session overrides merged on top of defaults.
 */
const createSession = async (fetch: FetchFn, options: Partial<SSESessionOptions> = {}): Promise<SSESession> => {
    return SSESession.create(EVENTS_URL, {
        ...defaultSessionOptions,
        onError: vi.fn(),
        ...options,
        fetch,
    });
};

/**
 * Runs a callback against a session and always disconnects afterward.
 * Most tests use this so session lifecycle (connect on create, disconnect on exit)
 * is consistent and resources are not leaked between cases.
 *
 * @param fetch - Mock fetch for the session.
 * @param options - Session options.
 * @param run - Test body receiving the connected session.
 */
const withSession = async <T>(fetch: FetchFn, options: Partial<SSESessionOptions>, run: (session: SSESession) => Promise<T>): Promise<T> => {
    const session = await createSession(fetch, options);

    try {
        return await run(session);
    } finally {
        await session.disconnect();
    }
};

/**
 * Reads up to `count` messages from the session's async iterator.
 * Breaking out of the for-await loop early leaves the underlying stream open,
 * which is intentional for tests that inspect post-read session state.
 *
 * @param session - Connected SSE session.
 * @param count - Number of events to collect before stopping.
 */
const readMessages = async (session: SSESession, count: number): Promise<SSEvent[]> => {
    const events: SSEvent[] = [];

    for await (const event of session.messages) {
        events.push(event);

        if (events.length >= count) {
            break;
        }
    }

    return events;
};

/**
 * Builds a ReadableStream that emits one chunk then errors after a delay.
 * Simulates a mid-stream network failure: the client receives partial data,
 * then the connection drops before the server finishes sending.
 *
 * @param raw - SSE payload to enqueue before the error.
 * @param delayMs - Milliseconds to wait before erroring (gives the parser time to process the chunk).
 */
const failingStreamAfter = (raw: string, delayMs = 50): ReadableStream<Uint8Array> => {
    const encoder = new TextEncoder();

    return new ReadableStream({
        async start(controller): Promise<void> {
            controller.enqueue(encoder.encode(raw));
            await new Promise((resolve) => setTimeout(resolve, delayMs));
            controller.error(new Error('network failure'));
        },
    });
};

/**
 * Scenario: a client opens a standard GET SSE connection.
 * Verifies the session passes the correct fetch options for a spec-compliant SSE request.
 */
const testSseSessionConnectCallsFetchWithExpectedOptions = async (): Promise<void> => {
    // Return one valid storage fixture event so connect succeeds and the stream stays open briefly.
    const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ]));

    try {
        // withSession → createSession → SSESession.create → connect → fetchMock is called once.
        await withSession(fetchMock, {}, async () => {
            expect(fetchMock).toHaveBeenCalledWith(
                EVENTS_URL,
                expect.objectContaining({
                    method: 'GET',
                    cache: 'no-store',
                    signal: expect.any(AbortSignal),
                    headers: expect.objectContaining({
                        Accept: 'text/event-stream',
                        'Cache-Control': 'no-cache',
                    }),
                }),
            );

            // While the stream is active the abort signal must not yet be triggered.
            expect(fetchMock.mock.calls[0]?.[1]?.signal?.aborted).toBe(false);
        });
    } finally {
        vi.restoreAllMocks();
    }
};

/**
 * Scenario: a client connects and the application wants a lifecycle callback when the stream is ready.
 * Verifies onConnected fires after the transport is established.
 */
const testSseSessionConnectInvokesOnConnected = async (): Promise<void> => {
    const onConnected = vi.fn();
    const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ]));

    try {
        await withSession(fetchMock, { onConnected }, async () => {
            expect(onConnected).toHaveBeenCalledOnce();
        });
    } finally {
        vi.restoreAllMocks();
    }
};

/**
 * Scenario: a client is already connected and something calls connect() again (e.g. a duplicate init).
 * Verifies the session does not open a second HTTP transport.
 */
const testSseSessionConnectDoesNotOpenSecondTransport = async (): Promise<void> => {
    const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ]));

    try {
        await withSession(fetchMock, {}, async (session) => {
            // Pull one event so the first transport is fully established and reading.
            await readMessages(session, 1);

            // Idempotent connect — should be a no-op at the fetch layer.
            await session.connect();

            expect(fetchMock).toHaveBeenCalledOnce();
        });
    } finally {
        vi.restoreAllMocks();
    }
};

/**
 * Scenario: the application attaches auth or other headers via onRequest before each fetch.
 * Verifies mutations from onRequest reach the actual fetch call.
 */
const testSseSessionConnectPassesOnRequestMutations = async (): Promise<void> => {
    const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ]));

    try {
        await withSession(
            fetchMock,
            {
                // onRequest runs during connect and can rewrite headers/body before fetch sees them.
                onRequest: async (request) => ({
                    ...request,
                    headers: { ...request.headers, Authorization: 'Bearer test-token' },
                }),
            },
            async () => {
                expect(fetchMock).toHaveBeenCalledWith(
                    EVENTS_URL,
                    expect.objectContaining({
                        headers: expect.objectContaining({
                            Authorization: 'Bearer test-token',
                        }),
                    }),
                );
            },
        );
    } finally {
        vi.restoreAllMocks();
    }
};

/**
 * Scenario: some SSE endpoints require POST with a form body instead of a plain GET.
 * Verifies method and body are forwarded to fetch.
 */
const testSseSessionConnectSendsPostBody = async (): Promise<void> => {
    const body = new FormData();
    body.set('topic', 'prices');

    const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ]));

    try {
        await withSession(
            fetchMock,
            {
                method: 'POST',
                body,
            },
            async () => {
                expect(fetchMock).toHaveBeenCalledWith(
                    EVENTS_URL,
                    expect.objectContaining({
                        method: 'POST',
                        body,
                    }),
                );
            },
        );
    } finally {
        vi.restoreAllMocks();
    }
};

/**
 * Scenario: a single long-lived connection delivers many events from different domains (LLM, oracle, storage).
 * Verifies the parser and session deliver every fixture event in order through one stream.
 */
const testSseSessionDeliversMultipleFixtureEvents = async (): Promise<void> => {
    const fixtures = [ ...priceOracleEvents, ...storageEvents ];
    const expected = fixtures.flatMap(({ parsed }) => parsed ?? []);

    // All raw payloads are concatenated into one SseTestStream response.
    const fetchMock = createFetchMock(() => sseFetchResponse(fixtures.map(({ raw }) => raw)));

    try {
        await withSession(fetchMock, {}, async (session) => {
            const received = await readMessages(session, expected.length);

            expect(received).toEqual(expected);
        });
    } finally {
        vi.restoreAllMocks();
    }
};

/**
 * Scenario: application code listens via session.on('message') instead of the async iterator.
 * Verifies the EventEmitter path receives the same parsed events as the iterator.
 */
const testSseSessionEmitsMessageEvents = async (): Promise<void> => {
    const { raw, parsed } = storageEvents[0]!;
    const emitted: SSEvent[] = [];

    // ':\n\n' is an SSE comment/heartbeat; chunkDelayMs forces it to arrive as a separate chunk
    // so incremental parsing is exercised before the real data frame.
    const fetchMock = createFetchMock(() => sseFetchResponse([ ':\n\n', raw ], { chunkDelayMs: 50 }));

    // Use createSession directly (not withSession) so we control disconnect timing in finally.
    const session = await createSession(fetchMock, {});

    try {
        session.on('message', (event) => emitted.push(event));

        await readMessages(session, 1);

        expect(emitted).toEqual([ parsed![0] ]);
    } finally {
        await session.disconnect();
        vi.restoreAllMocks();
    }
};

/**
 * Scenario: network chunks split an SSE frame at an arbitrary byte boundary.
 * Verifies the internal parser buffers partial data and still emits a complete event.
 */
const testSseSessionParsesEventSplitAcrossChunks = async (): Promise<void> => {
    const { raw, parsed } = storageEvents[0]!;
    const mid = Math.floor(raw.length / 2);

    // SseTestStream sends each array element as a separate chunk — the event is cut in half.
    const fetchMock = createFetchMock(() => sseFetchResponse([ raw.slice(0, mid), raw.slice(mid) ]));

    try {
        await withSession(fetchMock, {}, async (session) => {
            const [ event ] = await readMessages(session, 1);

            expect(event).toEqual(parsed![0]);
        });
    } finally {
        vi.restoreAllMocks();
    }
};

/**
 * Scenario: server closes the stream after one event and the client is not in persistent mode.
 * Verifies the message iterator closes and onDisconnected fires.
 */
const testSseSessionClosesOnServerCloseWhenNotPersistent = async (): Promise<void> => {
    // closeWhenDone: true makes SseTestStream end the body after sending the chunk.
    const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ], { closeWhenDone: true }));
    const disconnected = vi.fn();

    try {
        // persistent defaults to false — terminal server close should shut down messages.
        await withSession(fetchMock, { onDisconnected: disconnected }, async (session) => {
            await readMessages(session, 1);

            await vi.waitFor(() => expect(session.messages.closed).toBe(true));
            expect(disconnected).toHaveBeenCalled();
        });
    } finally {
        vi.restoreAllMocks();
    }
};

/**
 * Scenario: server closes the stream, the message iterator closes, then the app manually reconnects.
 * Verifies a second fetch opens and events flow again (non-persistent manual reconnect path).
 */
const testSseSessionReconnectsAfterTerminalClose = async (): Promise<void> => {
    const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ], { closeWhenDone: true }));

    // Override the default responder: first fetch closes after one event, second stays open.
    fetchMock
        .mockResolvedValueOnce(sseFetchResponse([ storageEvents[0]!.raw ], { closeWhenDone: true }))
        .mockResolvedValueOnce(sseFetchResponse([ storageEvents[0]!.raw ]));

    try {
        await withSession(fetchMock, {}, async (session) => {
            await readMessages(session, 1);

            await vi.waitFor(() => expect(session.messages.closed).toBe(true));

            // App-initiated reconnect after the first stream ended.
            await session.connect();

            const [ event ] = await readMessages(session, 1);
            expect(event).toEqual(storageEvents[0]!.parsed![0]);
            expect(fetchMock).toHaveBeenCalledTimes(2);
        });
    } finally {
        vi.restoreAllMocks();
    }
};

/**
 * Scenario: persistent client — server closes the stream but the session should auto-reconnect.
 * Verifies two fetch calls and events from both streams arrive on the same message iterator.
 */
const testSseSessionPersistentReconnectsOnServerClose = async (): Promise<void> => {
    const fetchMock = createFetchMock(() => sseFetchResponse([ priceOracleEvents[0]!.raw ], { closeWhenDone: true }));

    fetchMock
        .mockResolvedValueOnce(sseFetchResponse([ priceOracleEvents[0]!.raw ], { closeWhenDone: true }))
        .mockResolvedValueOnce(sseFetchResponse([ priceOracleEvents[1]!.raw ]));

    try {
        await withSession(
            fetchMock,
            {
                persistent: true,
                retry: testRetry(2),
            },
            async (session) => {
                const [ first, second ] = await readMessages(session, 2);

                expect(first).toEqual(priceOracleEvents[0]!.parsed![0]);
                expect(second).toEqual(priceOracleEvents[1]!.parsed![0]);
                await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(2));
            },
        );
    } finally {
        vi.restoreAllMocks();
    }
};

/**
 * Scenario: mid-stream network error with attemptReconnect enabled.
 * Verifies the session fetches again and the iterator continues delivering events.
 */
const testSseSessionReconnectsOnTransportError = async (): Promise<void> => {
    const fetchMock = createFetchMock(() => sseFetchResponse([ priceOracleEvents[0]!.raw ]));

    fetchMock
        // First connection: delivers one event then the stream errors (failingStreamAfter).
        .mockResolvedValueOnce(new Response(failingStreamAfter(priceOracleEvents[0]!.raw), { status: 200, headers: SSE_HEADERS }))
        // Second connection: clean stream with the next fixture event.
        .mockResolvedValueOnce(sseFetchResponse([ priceOracleEvents[1]!.raw ]));

    try {
        await withSession(
            fetchMock,
            {
                attemptReconnect: true,
                retry: testRetry(2),
            },
            async (session) => {
                const [ beforeError, recovered ] = await readMessages(session, 2);

                expect(beforeError).toEqual(priceOracleEvents[0]!.parsed![0]);
                expect(recovered).toEqual(priceOracleEvents[1]!.parsed![0]);
                await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(2));
            },
        );
    } finally {
        vi.restoreAllMocks();
    }
};

/**
 * Scenario: mid-stream network error with attemptReconnect disabled (default).
 * Verifies the session terminates: messages close, onError runs, and an error event is emitted.
 */
const testSseSessionClosesOnTransportErrorWhenNotReconnecting = async (): Promise<void> => {
    const onError = vi.fn();
    const response = new Response(failingStreamAfter(priceOracleEvents[0]!.raw), { status: 200, headers: SSE_HEADERS });
    const fetchMock = createFetchMock(() => response);

    try {
        await withSession(fetchMock, { onError }, async (session) => {
            const errorEvent = new Promise<Error>((resolve) => session.once('error', resolve));

            await vi.waitFor(() => expect(session.messages.closed).toBe(true));
            expect(onError).toHaveBeenCalled();
            await expect(errorEvent).resolves.toBeInstanceOf(Error);
        });
    } finally {
        vi.restoreAllMocks();
    }
};

/**
 * Scenario: the user aborts an active connection (e.g. navigation away) but may reconnect later.
 * Verifies onDisconnected fires, fetch is aborted, but the message iterator stays open.
 */
const testSseSessionAbortEmitsDisconnectedAndKeepsMessagesOpen = async (): Promise<void> => {
    const onDisconnected = vi.fn();
    const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ]));

    try {
        await withSession(fetchMock, { onDisconnected }, async (session) => {
            await readMessages(session, 1);
            await session.abort();

            expect(onDisconnected).toHaveBeenCalled();
            expect(session.messages.closed).toBe(false);
        });
    } finally {
        vi.restoreAllMocks();
    }
};

/**
 * Scenario: the application fully tears down the session (logout, component unmount, etc.).
 * Verifies disconnect closes the message iterator and emits the closed event.
 */
const testSseSessionDisconnectClosesMessagesAndEmitsClosed = async (): Promise<void> => {
    const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ]));

    const session = await createSession(fetchMock, {});
    const closed = new Promise<void>((resolve) => session.once('closed', () => resolve()));

    try {
        await session.disconnect();

        expect(session.messages.closed).toBe(true);
        await expect(closed).resolves.toBeUndefined();
    } finally {
        vi.restoreAllMocks();
    }
};

/**
 * Scenario: the server returns a non-2xx HTTP status (500).
 * Verifies create rejects, onError is invoked, and the error is a real Error instance.
 */
const testSseSessionHttpErrorCallsOnErrorAndThrows = async (): Promise<void> => {
    const onError = vi.fn();
    const fetchMock = createFetchMock(() =>
            new Response('nope', {
                status: 500,
                statusText: 'Internal Server Error',
            }));

    const expectedError = new ExponentialBackoffMaxRetriesHitError([ new HTTPError(500, 'nope') ]);

    try {
        await expect(createSession(fetchMock, { onError })).rejects.toThrow(expectedError);

        expect(onError).toHaveBeenCalled();
        expect(onError.mock.calls[0]?.[0]).toBeInstanceOf(Error);
    } finally {
        vi.restoreAllMocks();
    }
};

/**
 * Scenario: fetch returns 200 but with a null body (misconfigured proxy or server bug).
 * Verifies create rejects because SSE requires a readable stream body.
 */
const testSseSessionRejectsWhenResponseBodyIsNull = async (): Promise<void> => {
    const onError = vi.fn();
    const fetchMock = createFetchMock(() => new Response(null, { status: 200, headers: SSE_HEADERS }));

    const expectedError = new ExponentialBackoffMaxRetriesHitError([ new ResponseBodyNullError() ]);

    try {
        await expect(createSession(fetchMock, { onError, retry: testRetry() })).rejects.toThrow(expectedError);

        expect(onError).toHaveBeenCalled();
    } finally {
        vi.restoreAllMocks();
    }
};

/**
 * Scenario: SSE spec resume — after receiving an event with an id, reconnect should send Last-Event-ID.
 * Verifies the second fetch includes the id from the first event (1234 in storageEvents[0]).
 */
const testSseSessionSendsLastEventIdOnReconnect = async (): Promise<void> => {
    const { raw } = storageEvents[0]!;
    let reconnectHeaders: Record<string, string> | undefined;
    const fetchMock = createFetchMock(() => sseFetchResponse([ raw ]));

    fetchMock
        // First connection: heartbeat chunk, then the event (with id: 1234), then server closes.
        .mockResolvedValueOnce(sseFetchResponse([ ':\n\n', raw ], { chunkDelayMs: 10, closeWhenDone: true }))
        // Second connection: capture whatever headers the reconnect logic attached.
        .mockImplementationOnce(async (_url, init) => {
            reconnectHeaders = init.headers as Record<string, string>;

            return sseFetchResponse([]);
        });

    const session = await createSession(fetchMock, { persistent: true, retry: testRetry(2) });

    try {
        // Registers an onRequest hook that copies the last seen event id into reconnect headers.
        await SSESession.addLastEventIdReconnect(session);

        await readMessages(session, 1);

        await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(2));
        expect(reconnectHeaders?.['Last-Event-ID']).toBe('1234');
    } finally {
        await session.disconnect();
        vi.restoreAllMocks();
    }
};

const runTests = async (): Promise<void> => {
    test('SSESession.connect: calls injected fetch with method, headers, and abort signal', testSseSessionConnectCallsFetchWithExpectedOptions);
    test('SSESession.connect: invokes onConnected when the stream is established', testSseSessionConnectInvokesOnConnected);
    test('SSESession.connect: does not open a second transport when connect is called again', testSseSessionConnectDoesNotOpenSecondTransport);
    test('SSESession.connect: passes request mutations from onRequest to fetch', testSseSessionConnectPassesOnRequestMutations);
    test('SSESession.connect: sends POST bodies for POST-based SSE endpoints', testSseSessionConnectSendsPostBody);
    test('SSESession: delivers multiple fixture events through a single session', testSseSessionDeliversMultipleFixtureEvents);
    test('SSESession: emits message events for incoming SSE frames', testSseSessionEmitsMessageEvents);
    test('SSESession: parses a fixture event split across chunk boundaries', testSseSessionParsesEventSplitAcrossChunks);
    test('SSESession: closes messages and emits disconnected when persistent is false', testSseSessionClosesOnServerCloseWhenNotPersistent);
    test('SSESession: opens a new message stream after reconnecting following a terminal close', testSseSessionReconnectsAfterTerminalClose);
    test('SSESession: reconnects when the server closes the stream and persistent is true', testSseSessionPersistentReconnectsOnServerClose);
    test('SSESession: reconnects when attemptReconnect is true', testSseSessionReconnectsOnTransportError);
    test('SSESession: closes messages and emits error when attemptReconnect is false', testSseSessionClosesOnTransportErrorWhenNotReconnecting);
    test('SSESession.abort: aborts fetch, emits disconnected, and keeps messages open', testSseSessionAbortEmitsDisconnectedAndKeepsMessagesOpen);
    test('SSESession.disconnect: closes messages and emits closed', testSseSessionDisconnectClosesMessagesAndEmitsClosed);
    test('SSESession: calls onError, emits error, and throws from create on HTTP error', testSseSessionHttpErrorCallsOnErrorAndThrows);
    test('SSESession: rejects when the response body is null', testSseSessionRejectsWhenResponseBodyIsNull);
    test('SSESession: sends Last-Event-ID on reconnect after receiving an event with an id', testSseSessionSendsLastEventIdOnReconnect);
};

await runTests();