import type { SSESessionOptions, SSESessionEventMap, SSEvent, SSESessionEventCallbacks, SSESessionOperationContext } from './types.ts';
import { SSE_SESSION_EXPONENTIAL_BACKOFF_DEFAULTS, SSE_SESSION_ATTEMPT_RECONNECT_DEFAULT, SSE_SESSION_PERSISTENT_DEFAULT } from './constants.ts';

import { EventEmitter, type OffCallback } from '../event-emitter.ts';
import { ExponentialBackoff } from '../exponential-backoff/index.ts';
import { normalizeError } from '../misc.ts';

import { SSEEventParser } from './sse-event-parser.ts';
import { AsyncPushIterator } from './async-push-iterator.ts';
import { HTTPError, ResponseBodyNullError, SSESessionUnsubscribePluginNotAFunctionError } from './errors.ts';

/**
 * A fetch-based Server-Sent Events (SSE) client with reconnect and optional
 * browser tab visibility handling.
 *
 * Each session maintains one HTTP streaming connection at a time. Incoming
 * bytes are parsed into {@link SSEvent} objects and delivered through two
 * surfaces:
 *
 * - **Events** — `"connected"`, `"message"`, `"disconnected"`, `"error"`,
 *   and `"closed"` on the session itself (extends {@link EventEmitter}).
 * - **Messages** — {@link messages}, an async iterable for `for await...of`
 *   consumers.
 *
 * Typical usage:
 *
 * ```ts
 * const session = await SSESession.create("/events");
 *
 * session.on("message", (event) => console.log(event.data));
 *
 * for await (const event of session.messages) {
 *   handle(event);
 * }
 * ```
 *
 * ## Lifecycle
 *
 * - {@link connect} opens (or reopens) the transport. It resolves once the
 *   HTTP stream is established; reading continues in the background.
 * - {@link disconnect} stops the in-flight fetch without ending the session.
 *   Used internally for tab visibility. The {@link messages} iterator stays
 *   open so an existing consumer resumes when the tab becomes visible again.
 * - {@link close} aborts the transport, closes {@link messages}, and emits
 *   `"closed"`.
 *
 * Automatic reconnect is controlled by {@link SSESessionOptions.persistent}
 * (server closed the stream) and
 * {@link SSESessionOptions.attemptReconnect} (transport error).
 *
 * ## Connection supersession
 *
 * Each {@link connect} will create a new controller if one does not exist. Otherwise,
 * it will return immediately.
 */
export class SSESession extends EventEmitter<SSESessionEventMap> {
    /**
     * Creates a session and waits until the first connection is established.
     *
     * @param url - The SSE endpoint URL.
     * @param options - Configuration merged with instance defaults.
     * @returns A connected session.
     * @throws When the initial connection cannot be established.
     */
    static async create(url: string, options: Partial<SSESessionOptions & SSESessionEventCallbacks> = {}): Promise<SSESession> {
        const session = new SSESession(url, options);
        await session.connect();

        return session;
    }

    /**
     * Enables SSE resume semantics by sending `Last-Event-ID` on reconnect.
     *
     * Listens for incoming `"message"` events and remembers the most recent
     * {@link SSEvent.id}. On every subsequent connect or reconnect, the session's
     * {@link onRequest} hook is wrapped so that header is attached when an id is
     * known, allowing the server to replay only events the client has not yet
     * received.
     *
     * The existing {@link onRequest} callback is preserved and runs after the
     * header is applied, so auth or other header mutations continue to work.
     *
     * Since this relies on listening to the `"message"` event, this should ideally
     * be called before "{@link connect} is called so no messages (and no ids) are missed.
     *
     * ```ts
     * const session = new SSESession(url);
     * SSESession.addLastEventIdReconnect(session);
     * await session.connect();
     * // Reconnects send Last-Event-ID once an event with an id is received.
     * ```
     *
     * @param session - The session to instrument.
     * @returns The same session, for chaining.
     */
    static addLastEventIdReconnect(session: SSESession): { session: SSESession; removeListener?: OffCallback } {
        // If the plugin is already installed, we return the session and the remove listener function.
        if (session.plugins.has(SSESession.addLastEventIdReconnect)) {
            // Get the remove listener function from the session.
            const removeListener = session.plugins.get(SSESession.addLastEventIdReconnect) as OffCallback;

            // If the plugin was added with a non-function listener, we throw an error.
            if (typeof removeListener !== 'function') {
                throw new SSESessionUnsubscribePluginNotAFunctionError('addLastEventIdReconnect');
            }

            // If the plugin is already installed, we return the session and the remove listener function.
            return { session, removeListener };
        }

        // Store the last event id so we can send it on reconnect.
        let lastEventId: string | undefined;

        // Define the message listener that will store the last event id and begin listening for 'message' events.
        const messageListener = (event: SSEvent): void => {
            // If the event has no id, we should not update the lastEventId and instead keep the old one.
            lastEventId = event.id ?? lastEventId;
        };

        session.on('message', messageListener);

        // Store the original onRequest callback so we can extend its behavior to write the reconnect header.
        const originalOnRequest = session.options.onRequest;

        // Whether we actively send the Last-Event-ID header on reconnect.
        // Because the `onRequest` callback may be extended further by other code, we can't just roll-back to the `originalOnRequest`
        // Instead, we just use this flag to determine if we should write the Last-Event-ID header.
        let enabled = true;

        // Extend the onRequest callback to write the Last-Event-ID header if it is enabled.
        session.options.onRequest = async (request: RequestInit): Promise<RequestInit> => {
            if (lastEventId && enabled) {
                request.headers = { ...request.headers, 'Last-Event-ID': lastEventId };
            }

            return originalOnRequest(request);
        };

        // Define the remove listener function that will disable the Last-Event-ID header on reconnect.
        const removeListener = (): void => {
            enabled = false;
            session.off('message', messageListener);

            // If the remove listener function is the same as the existing remove listener function, we remove the plugin from the session.
            // This protects against the `removeListener` function being called onto after a new instance of this plugin is added
            const existingRemoveListener = session.plugins.get(SSESession.addLastEventIdReconnect) as OffCallback;
            if (existingRemoveListener === removeListener) {
                session.plugins.delete(SSESession.addLastEventIdReconnect);
            }
        };

        // Store the plugin in the session.
        session.plugins.set(SSESession.addLastEventIdReconnect, removeListener);

        return { session, removeListener };
    }

    /**
     * Pauses and resumes a session based on browser tab visibility.
     *
     * Uses the Page Visibility API (`document.visibilitychange`):
     *
     * - **hidden** — {@link disconnect} stops the active fetch. {@link messages}
     *   stays open; `"disconnected"` fires but `"closed"` does not.
     * - **visible** — {@link connect} re-establishes the stream if needed.
     *
     * This controller will not re-connect if the session was disconnected by something else.
     *
     * No-op in non-browser environments where `document` is undefined.
     *
     * @param session - The session to manage.
     */
    static addBrowserVisibilityHandler(session: SSESession): { session: SSESession; removeListener?: OffCallback } {
        if (typeof document === 'undefined') return { session };

        // If the plugin is already installed, we return the session and the remove listener function.
        if (session.plugins.has(SSESession.addBrowserVisibilityHandler)) {
            // Get the remove listener function from the session.
            const removeListener = session.plugins.get(SSESession.addBrowserVisibilityHandler) as OffCallback;

            // If the plugin was added with a non-function listener, we throw an error.
            if (typeof removeListener !== 'function') {
                throw new SSESessionUnsubscribePluginNotAFunctionError('addBrowserVisibilityHandler');
            }

            // If the plugin is already installed, we return the session and the remove listener function.
            return { session, removeListener };
        }

        // Create a listener that will reconnect the session
        // This listener is added when the hidden->disconnect listener disconnects the session, but will not be added
        // if the session was disconnected by something else, like a manual disconnect, server-disconnect, or another plugin.
        const reconnectOnVisible = async (): Promise<void> => {
            if (document.visibilityState === 'visible') {
                // Remove this listener
                document.removeEventListener('visibilitychange', reconnectOnVisible);

                // Reconnect the session
                await session.connect();
            }
        };

        // Create a listener that will disable the reconnect listener if the session is disconnected by something else.
        const disableOnExternalDisconnect = ({ source }: SSESessionOperationContext): void => {
            // If the source is not the browser visibility handler, we remove the listener.
            if (source !== SSESession.addBrowserVisibilityHandler) {
                document.removeEventListener('visibilitychange', reconnectOnVisible);
            }
        };

        // Create a listener that will live as-long as this session is not closed
        // This listener will create a new 'on visible' listener that will reconnect the session when it disconnects the session.
        // This prevents it from trying to reconnect the session if the session was disconnected by something else
        const handleVisibilityChange = async (): Promise<void> => {
            // If the session is not connected, we do nothing.
            if (!session.active) {
                return;
            }

            // If the session is connected and the visibility is hidden, we disconnect it.
            if (document.visibilityState === 'hidden') {
                // Add the listener for connected event to reconnect the session.
                document.addEventListener('visibilitychange', reconnectOnVisible);

                // Disconnect the session.
                session.disconnect({ source: SSESession.addBrowserVisibilityHandler });
            }
        };

        // Add the listener for visibility change.
        document.addEventListener('visibilitychange', handleVisibilityChange);

        // Add a listener for `beforeDisconnect` event to disable the reconnect listener if the session is disconnected by something else.
        session.on('beforeDisconnect', disableOnExternalDisconnect);

        // Remove the listener for visibility change.
        const removeListener = (): void => {
            document.removeEventListener('visibilitychange', handleVisibilityChange);
            document.removeEventListener('visibilitychange', reconnectOnVisible);

            session.off('beforeDisconnect', disableOnExternalDisconnect);

            // If the remove listener function is the same as the existing remove listener function, we remove the plugin from the session.
            // This protects against the `removeListener` function being called onto after a new instance of this plugin is added
            const existingRemoveListener = session.plugins.get(SSESession.addBrowserVisibilityHandler) as OffCallback;
            if (existingRemoveListener === removeListener) {
                session.plugins.delete(SSESession.addBrowserVisibilityHandler);
            }
        };

        // Store the plugin in the session.
        session.plugins.set(SSESession.addBrowserVisibilityHandler, removeListener);

        // Return the session and the remove listener function.
        return { session, removeListener };
    }

    /** SSE endpoint URL for this session. */
    readonly #url: string;

    /**
     * Per-instance configuration.
     *
     * Defaults live on the instance field (not a shared static) so each session
     * gets its own {@link SSEEventParser} and {@link ExponentialBackoff}.
     */
    public options: SSESessionOptions = {
        // Fetch options
        fetch: (...args) => fetch(...args),
        method: 'GET',
        headers: {
            Accept: 'text/event-stream',
            'Cache-Control': 'no-cache',
        },
        body: new FormData(),

        // callback to mutate the fetch options before the fetch is made.
        onRequest: (request) => Promise.resolve(request),

        // Retry the initial fetch.
        // Use the spread operator so it's impossible for us to override the default options. (default is exponential backoff)
        retry: new ExponentialBackoff({
            ...SSE_SESSION_EXPONENTIAL_BACKOFF_DEFAULTS,
        }),

        // Reconnection options
        attemptReconnect: SSE_SESSION_ATTEMPT_RECONNECT_DEFAULT,
        persistent: SSE_SESSION_PERSISTENT_DEFAULT,

        // Event parser options
        eventParser: new SSEEventParser(),
    };

    /**
     * Registered plugins for the session.
     * This can be used to prevent duplicate plugins from being added to the session.
     * The value for a plugin is arbitrary, for example a function that removes the plugin.
     */
    public readonly plugins: Map<unknown, unknown> = new Map();

    /** AbortController for the currently active fetch, if any. */
    #connectionController: AbortController | null = null;

    /** The server's requested retry interval in milliseconds (per SSE spec) if any */
    #retryInterval: number | null = null;

    /**
     * Asynchronous stream of parsed SSE events for the active connection.
     *
     * Stays open across {@link disconnect} and automatic reconnects so an existing
     * `for await` consumer keeps receiving events after visibility resumes.
     *
     * Closes when:
     * - the server ends the stream and {@link SSESessionOptions.persistent}
     *   is false,
     * - {@link close} is called, or
     * - a transport error occurs with
     *   {@link SSESessionOptions.attemptReconnect} disabled.
     *
     * A later {@link connect} replaces this with a new iterator when the
     * previous one was closed. Consumers should read from `session.messages`
     * rather than caching a reference across terminal disconnects.
     */
    public messages: AsyncPushIterator<SSEvent> = new AsyncPushIterator<SSEvent>();

    public constructor(url: string, options: Partial<SSESessionOptions & SSESessionEventCallbacks> = {}) {
        super();

        const { onConnected, onDisconnected, onError, ...restOptions } = options;

        this.#url = url;
        this.options = {
            ...this.options,
            ...restOptions,
            // Shallow merge would drop default headers when options.headers is set.
            headers: { ...this.options.headers, ...options.headers },
        };

        // Add the onConnected, onDisconnected, and onError as event listeners to the session.
        // We accept these as options to the constructor so that we can both instantiate the session and add the event listeners at the same time,
        // avoiding any potential race-conditions or async/await issues.
        if (onConnected) {
            this.on('connected', onConnected);
        }

        if (onDisconnected) {
            this.on('disconnected', onDisconnected);
        }

        if (onError) {
            this.on('error', onError);
        }
    }

    /**
     * Returns true if the session is active by checking if the controller is not null.
     */
    public get active(): boolean {
        return this.#connectionController !== null;
    }

    /**
     * Connects or reconnects to the SSE endpoint.
     *
     * Resolves once the HTTP stream is established and `"connected"` has been
     * emitted. Body reading continues asynchronously in the background via the
     * internal `#readStream` method.
     *
     * @throws When the fetch retry policy exhausts attempts or the connection
     *   is superseded before the reader is handed off (in the latter case the
     *   promise resolves without throwing).
     */
    public async connect(): Promise<void> {
        // If there is already a controller present, we are already connected.
        if (this.#connectionController) return;

        // Prepare for a fresh transport. Parser state from an abandoned connection
        // must not bleed into the next one; reopen messages if a prior terminal
        // close ended the consumer's iteration loop.
        this.#resetEventParser();
        this.#ensureMessageStreamOpen();

        const connectionController = new AbortController();
        this.#connectionController = connectionController;

        const { method, headers, body } = this.options;

        // Create the fetch options.
        const fetchOptions: RequestInit = {
            method,
            headers: headers || {},
            signal: connectionController.signal,
            cache: 'no-store',
        };

        // If the method is POST, we will set the body of the fetch options.
        // NOTE: RequestInit doesn't allow body to be `undefined`, so if its false-y we will set it to null.
        if (method === 'POST') {
            fetchOptions.body = body || null;
        }

        // Create a reader for the response body.
        let reader: ReadableStreamDefaultReader<Uint8Array>;

        try {
            // Retry the fetch using the provided retry policy. (default is exponential backoff)
            reader = await this.options.retry.run(() => this.#createReader(fetchOptions), {
                signal: connectionController.signal,
            });
        } catch (error) {
            // A newer `connect()` superseded this attempt — leave state to the winner.
            if (this.#connectionController !== connectionController) return;

            // Reset the controller to null to allow for reconnection if needed.
            this.#connectionController = null;

            // Normalize the error so it can be passed to the event listeners.
            const normalizedError = normalizeError(error);

            // Emit the disconnected and error events.
            this.emit('disconnected', { reason: 'error', error: normalizedError });
            this.emit('error', normalizedError);

            // Close the message stream so `for await...of` consumers resolve.
            this.#closeMessageStream();

            // Throw the error so the caller can handle it.
            throw error;
        }

        // Connection succeeded but was already replaced (for example disconnect during fetch).
        if (this.#connectionController !== connectionController) {
            await reader.cancel();

            return;
        }

        this.emit('connected', undefined);

        // Fire-and-forget: connect() resolves while the stream is consumed.
        // Data is emitted as a `message` event and from the `messages` iterator.
        this.#readStream(reader, connectionController).catch((error) => {
            this.emit('error', error);
        });
    }

    /**
     * Disconnects only the currently active transport.
     *
     * `beforeDisconnect` is emitted on every invocation, even when no
     * transport is currently active. This allows observers to react to
     * an explicit disconnect operation.
     *
     * @param context - Context describing the operation initiator.
     * @emits `beforeDisconnect` with the supplied operation context.
     * @emits `disconnected` with reason `"disconnect"` when an active
     * transport is actually disconnected.
     */
    public disconnect(context: SSESessionOperationContext = {}): void {
        this.emit('beforeDisconnect', context);

        if (this.#connectionController) {
            // Grab the current controller to ensure we are aborting the correct one.
            const connectionController = this.#connectionController;
            this.#connectionController = null;

            // Invalidate any in-flight read loop and fetch for this transport.
            connectionController.abort();

            // Emit the disconnected event.
            this.emit('disconnected', {
                reason: 'disconnect',
                source: context.source,
            });
        }

        // Reset the event parser buffer so data is not carried through resulting in malformed events.
        this.#resetEventParser();
    }

    /**
     * Terminates the session and disables attached visibility handling until
     * the same instance is manually {@link connect connected} again.
     *
     * Closes {@link messages} and emits `"closed"`.
     */
    public close(): void {
        // Disconnect the session.
        this.disconnect();

        // Close the message stream so `for await...of` consumers resolve.
        this.#closeMessageStream();

        // Emit the closed event.
        this.emit('closed', undefined);
    }

    /**
     * Performs the HTTP request and returns a reader for the response body.
     *
     * {@link SSESessionOptions.onRequest} may mutate headers (for example auth
     * tokens or `Last-Event-ID`) before the fetch runs.
     */
    async #createReader(fetchOptions: RequestInit): Promise<ReadableStreamDefaultReader<Uint8Array>> {
        const requestOptions = await this.options.onRequest(fetchOptions);
        const response = await this.options.fetch(this.#url, requestOptions);

        // Handle a bad response from the server
        if (!response.ok) {
            // Get the response code and text.
            const responseCode = response.status;
            const responseText = await response.text();

            // Create a new error and emit it.
            const error = new HTTPError(responseCode, responseText);
            this.emit('error', error);

            // Throw the error so the caller can handle it.
            throw error;
        }

        // Handle a response with no body (SSE still uses the response body)
        if (!response.body) {
            const error = new ResponseBodyNullError();
            this.emit('error', error);
            throw error;
        }

        // Return the reader for the response body.
        return response.body.getReader();
    }

    /**
     * Reads bytes from an established stream until it ends, errors, or is
     * superseded by a newer connection.
     */
    async #readStream(reader: ReadableStreamDefaultReader<Uint8Array>, connectionController: AbortController): Promise<void> {
        try {
            while (this.#connectionController === connectionController) {
                const { done, value } = await reader.read();

                // `disconnect()` or a newer connect() may have landed while we were awaiting.
                if (this.#connectionController !== connectionController) return;

                // If the stream is done, we've been disconnected.
                // We need to clean up and either complete or reconnect the session.
                if (done) {
                    // Reset the controller to null to allow for reconnection if needed.
                    this.#connectionController = null;

                    // Emit the disconnected event.
                    this.emit('disconnected', { reason: 'remote' });

                    // If the session is persistent, we will attempt to reconnect.
                    // NOTE: in very rare cases, a malfunctioning server can make this code turn into an infinite loop.
                    if (this.options.persistent) {
                        // Server closed gracefully — reopen unless the consumer opted out.
                        await this.connect();
                    } else {
                        // Close the message stream so `for await...of` consumers resolve.
                        this.#closeMessageStream();
                    }

                    return;
                }

                // Some environments yield `{ done: false, value: undefined }`.
                if (!value) continue;

                // Emit the parsed events to the event listeners and the messages iterator.
                for (const event of this.options.eventParser.parseEvents(value)) {
                    // If the event has a retry interval, we will store it.
                    if (event.retry) {
                        this.#retryInterval = event.retry;
                    }

                    this.emit('message', event);
                    this.messages.push(event);
                }
            }
        } catch (error) {
            // If the controller is different, we already started a new connection and it would be confusing to handle this error.
            // The controller can also be null here if the session was disconnected.
            if (connectionController !== this.#connectionController) return;

            // Invalidate the current controller to allow for reconnection if needed
            this.#connectionController = null;

            const normalizedError = normalizeError(error);
            this.emit('disconnected', { reason: 'error', error: normalizedError });

            // Expected path for `disconnect()` — do not treat as an error or reconnect.
            if (connectionController.signal.aborted) return;

            this.emit('error', normalizedError);

            // If the session is configured to attempt reconnect, we will attempt to reconnect.
            if (this.options.attemptReconnect) {
                // If the server has requested a retry interval, we will wait for it before attempting to reconnect.
                if (this.#retryInterval) {
                    await new Promise((resolve) => setTimeout(resolve, this.#retryInterval!));
                }

                // Attempt to reconnect.
                await this.connect();
            } else {
                // Close the message stream so `for await...of` consumers resolve.
                this.#closeMessageStream();
            }
        }
    }

    /** Clears partial SSE frames left over from an abandoned transport. */
    #resetEventParser(): void {
        this.options.eventParser.reset();
    }

    /**
     * Creates a new {@link messages} iterator when the previous one was closed
     * by a terminal disconnect or server stream end.
     */
    #ensureMessageStreamOpen(): void {
        if (!this.messages.closed) return;

        this.messages = new AsyncPushIterator<SSEvent>();
    }

    /** Ends the message iteration loop for the current connection span. */
    #closeMessageStream(): void {
        if (this.messages.closed) return;

        this.messages.close();
    }
}