import { useEffect, useRef, useState } from 'react'

export type SseConnectionStatus = 'idle' | 'connecting' | 'open' | 'closed'

export interface SseEvent<T> {
  id?: string
  event?: string
  data: T
}

export interface UseSseStreamOptions<T> {
  /** Stream URL, or `null` to stay idle. */
  url: string | null
  /** Forwarded on every (re)connect attempt; read from a ref, so an inline object is safe. */
  headers?: Record<string, string>
  /** Invoked for each parsed `data:` frame. Ref-stable — changing its identity does not reconnect. */
  onMessage: (data: T) => void
  /** Invoked with parsed SSE metadata after `onMessage` successfully receives its payload. */
  onEvent?: (event: SseEvent<T>) => void
  /** Invoked on every status transition. Ref-stable like `onMessage`. */
  onStatus?: (status: SseConnectionStatus) => void
  /** Injectable for tests; defaults to the global `fetch`. */
  fetchImpl?: typeof fetch
  /** Delay before retrying after a stream ends or errors; defaults to 3000ms. */
  retryDelayMs?: number
}

const DEFAULT_RETRY_DELAY_MS = 3000
const MAX_RECENT_EVENT_IDS = 1024
const MAX_FRAME_BYTES = 64 * 1024
const FRAME_BOUNDARY = /\r\n\r\n|\r\n\n|\r\r\n|\r\r|\n\r\n|\n\r|\n\n/g

type ParsedFrame<T> =
  | { kind: 'ignore' }
  | { kind: 'malformed' }
  | { kind: 'event'; id?: string; event?: string; payload: T }

function parseFrame<T>(frame: string): ParsedFrame<T> {
  let id: string | undefined
  let event: string | undefined
  const data: string[] = []

  for (const line of frame.split(/\r\n|[\r\n]/)) {
    if (line === '' || line.startsWith(':')) continue
    const separator = line.indexOf(':')
    const field = separator === -1 ? line : line.slice(0, separator)
    const value = separator === -1 ? '' : line.slice(separator + 1).replace(/^ /, '')
    if (field === 'data') data.push(value)
    if (field === 'event') event = value
    if (field === 'id' && !value.includes('\0')) id = value
  }

  if (data.length === 0) return { kind: 'ignore' }
  try {
    return { kind: 'event', id, event, payload: JSON.parse(data.join('\n')) as T }
  } catch {
    return { kind: 'malformed' }
  }
}

/**
 * Authenticated SSE client via `fetch()` + a manually-decoded byte stream — native
 * `EventSource` cannot set an `Authorization` header, which the collector requires on
 * every request including `/events`. Parses `data: <json>\n\n` frames and
 * auto-reconnects after the stream ends or errors, until unmounted or `url` becomes null.
 */
export function useSseStream<T>(options: UseSseStreamOptions<T>): SseConnectionStatus {
  const { url, retryDelayMs = DEFAULT_RETRY_DELAY_MS, fetchImpl = fetch } = options
  const [status, setStatus] = useState<SseConnectionStatus>(url ? 'connecting' : 'idle')

  const onMessageRef = useRef(options.onMessage)
  const onEventRef = useRef(options.onEvent)
  const onStatusRef = useRef(options.onStatus)
  const headersRef = useRef(options.headers)
  useEffect(() => {
    onMessageRef.current = options.onMessage
    onEventRef.current = options.onEvent
    onStatusRef.current = options.onStatus
    headersRef.current = options.headers
  })

  useEffect(() => {
    if (url === null) {
      setStatus('idle')
      return
    }
    // Rebind: TS doesn't narrow a closed-over param through hoisted function declarations below.
    const streamUrl: string = url

    let disposed = false
    const controller = new AbortController()
    const deliveredIds = new Set<string>()
    const deliveredIdOrder: string[] = []
    let lastDeliveredId: string | undefined

    const transition = (next: SseConnectionStatus) => {
      if (disposed) return
      setStatus(next)
      onStatusRef.current?.(next)
    }

    async function readStream(body: ReadableStream<Uint8Array>): Promise<boolean> {
      const reader = body.getReader()
      const decoder = new TextDecoder()
      let buffer = ''
      while (true) {
        const { done, value } = await reader.read()
        if (done) return false
        buffer += decoder.decode(value, { stream: true })
        let boundary: RegExpExecArray | null
        const frameBoundary = new RegExp(FRAME_BOUNDARY)
        let start = 0
        while ((boundary = frameBoundary.exec(buffer)) !== null) {
          const frame = buffer.slice(start, boundary.index)
          start = boundary.index + boundary[0].length
          if (new TextEncoder().encode(frame).byteLength > MAX_FRAME_BYTES) {
            await reader.cancel('SSE frame exceeds maximum size')
            return true
          }
          const parsed = parseFrame<T>(frame)
          if (parsed.kind === 'ignore' || (parsed.kind === 'event' && parsed.id !== undefined && deliveredIds.has(parsed.id))) continue
          if (parsed.kind === 'malformed') {
            await reader.cancel('Malformed SSE data frame')
            return true
          }
          try {
            onMessageRef.current(parsed.payload)
            if (parsed.id !== undefined) {
              deliveredIds.add(parsed.id)
              deliveredIdOrder.push(parsed.id)
              if (deliveredIdOrder.length > MAX_RECENT_EVENT_IDS) {
                const expiredId = deliveredIdOrder.shift()
                if (expiredId !== undefined) deliveredIds.delete(expiredId)
              }
              lastDeliveredId = parsed.id
            }
            onEventRef.current?.({ id: parsed.id, event: parsed.event, data: parsed.payload })
          } catch {
            // Callback failures do not interrupt the stream.
          }
        }
        buffer = buffer.slice(start)
        if (new TextEncoder().encode(buffer).byteLength > MAX_FRAME_BYTES) {
          await reader.cancel('SSE frame exceeds maximum size')
          return true
        }
      }
    }

    async function connectOnce(): Promise<boolean> {
      transition('connecting')
      let res: Response
      try {
        res = await fetchImpl(streamUrl, {
          headers: lastDeliveredId === undefined
            ? headersRef.current
            : { ...headersRef.current, 'Last-Event-ID': lastDeliveredId },
          signal: controller.signal,
          cache: 'no-store',
        })
      } catch {
        return false
      }
      if (disposed || !res.ok || !res.body) return false
      transition('open')
      try {
        return await readStream(res.body)
      } catch {
        // aborted on unmount, or the connection dropped — fall through to reconnect
      }
      return false
    }

    async function loop(): Promise<void> {
      while (!disposed) {
        const terminal = await connectOnce()
        if (disposed) break
        transition('closed')
        if (terminal) break
        await new Promise((resolve) => setTimeout(resolve, retryDelayMs))
      }
    }

    void loop()

    return () => {
      disposed = true
      controller.abort()
    }
  }, [url, retryDelayMs, fetchImpl])

  return status
}