/**
 * High-level transport factory for pydantic-AI–style backends.
 *
 * Composes:
 *   - `parseSSE` for spec-compliant SSE framing (multi-line `data:`, comments, idle timeout).
 *   - `createPydanticAISSEMap` for normalizing pydantic-AI events into `ChatStreamEvent`.
 *
 * Use when your backend speaks the canonical pydantic-AI stream shape
 * (`text_delta` / `tool_call` / `tool_result` / `done` / `error`).
 * URL building, auth headers, history loading, and optional session
 * bootstrapping are caller responsibilities — pass them in.
 */

import type {
  ChatMessage,
  ChatStreamEvent,
  ChatTransport,
  CreateSessionOptions,
  HistoryPage,
  SendOptions,
  SessionInfo,
  StreamOptions,
} from '../../types';
import { TransportError } from './types';
import { parseSSE } from './sse';
import {
  createPydanticAISSEMap,
  mapPydanticAIEvent,
  createToolIdQueue,
  type PydanticAIEvent,
} from './mappers';

export interface PydanticAIChatTransportOpts {
  /**
   * Build the SSE stream URL for a user message turn.
   * @example (sessionId, message) => `${base}/stream?session_id=${sessionId}&message=${encodeURIComponent(message)}`
   */
  buildStreamUrl: (sessionId: string, message: string) => string | URL;

  /** Optional history loader. If omitted, `loadHistory` returns an empty page. */
  loadHistory?: (sessionId: string, cursor?: string | null) => Promise<HistoryPage>;

  /**
   * Optional session bootstrap. Called from `createSession`. Useful for
   * backends that need a `POST /sessions` round-trip or want to pre-seed
   * history.
   */
  bootstrapSession?: (opts?: CreateSessionOptions) => Promise<SessionInfo>;

  /** Optional session teardown. */
  closeSession?: (sessionId: string) => Promise<void>;

  /**
   * Optional non-streaming send (for hosts that need a buffered fallback,
   * e.g. when the user disables streaming). Defaults to throwing — most
   * hosts only use streaming.
   */
  send?: (
    sessionId: string,
    content: string,
    options?: SendOptions,
  ) => Promise<ChatMessage>;

  /** Request headers (Authorization, content-type, etc.). */
  buildHeaders?: () => HeadersInit | Promise<HeadersInit>;

  /** Override fetch (tests, retry layers). */
  fetchImpl?: typeof fetch;

  /**
   * HTTP method for the stream request. Defaults to `'POST'` with
   * `{ content, attachments, metadata }` JSON body. Set to `'GET'` if
   * your backend embeds the message in the URL via `buildStreamUrl`.
   */
  streamMethod?: 'GET' | 'POST';

  /** Idle timeout for the SSE connection, in ms. Forwarded to `parseSSE`. */
  idleTimeoutMs?: number;

  /**
   * Side-channel for events that don't translate to `ChatStreamEvent`
   * (e.g. `approval_required` — surfaces interactive prompts outside the
   * normal message stream). Called synchronously while parsing the SSE
   * frame; mutate caller-owned state, don't `await` long work here.
   */
  onPydanticEvent?: (event: PydanticAIEvent) => void;
}

const DEFAULT_SESSION_ID = 'default';

function mapStatusToCode(status: number): string {
  if (status === 401 || status === 403) return 'unauthorized';
  if (status === 404) return 'not_found';
  if (status === 408) return 'timeout';
  if (status === 429) return 'rate_limited';
  if (status >= 500) return 'server_error';
  return 'error';
}

export function createPydanticAIChatTransport(
  opts: PydanticAIChatTransportOpts,
): ChatTransport {
  const fetchImpl = opts.fetchImpl ?? fetch.bind(globalThis);
  const streamMethod = opts.streamMethod ?? 'POST';

  async function resolvedHeaders(extra?: Record<string, string>): Promise<Headers> {
    const base = opts.buildHeaders ? await opts.buildHeaders() : {};
    const headers = new Headers(base as HeadersInit);
    if (extra) {
      for (const [k, v] of Object.entries(extra)) headers.set(k, v);
    }
    return headers;
  }

  return {
    async createSession(createOpts) {
      if (opts.bootstrapSession) return opts.bootstrapSession(createOpts);
      return { sessionId: DEFAULT_SESSION_ID };
    },

    async loadHistory(sessionId, cursor) {
      if (opts.loadHistory) return opts.loadHistory(sessionId, cursor);
      return { messages: [], hasMore: false, nextCursor: null };
    },

    async *stream(
      sessionId: string,
      content: string,
      options: StreamOptions,
    ): AsyncGenerator<ChatStreamEvent, void, void> {
      const url = opts.buildStreamUrl(sessionId, content);

      const headers = await resolvedHeaders({ Accept: 'text/event-stream' });
      const init: RequestInit = {
        method: streamMethod,
        headers,
        signal: options.signal,
      };
      if (streamMethod === 'POST') {
        headers.set('Content-Type', 'application/json');
        init.body = JSON.stringify({
          content,
          attachments: options.attachments ?? [],
          metadata: options.metadata ?? {},
        });
      }

      const res = await fetchImpl(typeof url === 'string' ? url : url.toString(), init);

      if (!res.ok) {
        const text = await res.text().catch(() => '');
        throw new TransportError(
          `stream failed (${res.status}): ${text || res.statusText}`,
          mapStatusToCode(res.status),
        );
      }

      const sideChannel = opts.onPydanticEvent;
      if (!sideChannel) {
        yield* parseSSE(res, {
          signal: options.signal,
          idleTimeoutMs: opts.idleTimeoutMs,
          map: createPydanticAISSEMap(),
        });
        return;
      }

      // Side-channel mode: parse the raw pydantic event, fire callback,
      // then forward through the canonical mapper.
      const toolIds = createToolIdQueue();
      yield* parseSSE(res, {
        signal: options.signal,
        idleTimeoutMs: opts.idleTimeoutMs,
        map: (raw) => {
          if (!raw.data) return null;
          let parsed: PydanticAIEvent;
          try {
            parsed = JSON.parse(raw.data) as PydanticAIEvent;
          } catch {
            return null;
          }
          try {
            sideChannel(parsed);
          } catch {
            // Side-channel handler must not break the stream.
          }
          const out: ChatStreamEvent[] = [];
          for (const evt of mapPydanticAIEvent(parsed, toolIds)) out.push(evt);
          if (out.length === 0) return null;
          if (out.length === 1) return out[0]!;
          return out;
        },
      });
    },

    async send(sessionId, content, sendOpts) {
      if (opts.send) return opts.send(sessionId, content, sendOpts);
      throw new TransportError(
        'Buffered send is not supported by this transport',
        'unsupported',
      );
    },

    async closeSession(sessionId) {
      if (opts.closeSession) await opts.closeSession(sessionId);
    },
  };
}