/**
 * React binding for the agent gateway's session stream. Browser-direct by
 * design: the backend only mints the session JWT (`GET …/tasks/{id}/token`),
 * and this hook opens `GET /api/v1/sessions/{session}/events` itself.
 *
 * Three things the raw `AgentClient` does not do:
 *
 *  - **Token lifetime.** The JWT lives 600s, which is shorter than a long run.
 *    The hook silently re-mints shortly before `expiresAt` and reopens the
 *    stream at the same watermark, so a subscriber never sees the gap.
 *  - **Resume across reconnects.** `Last-Event-ID` is derived from the frames
 *    the caller has already been told about, and it survives `fetchToken`
 *    rotation.
 *  - **State for React.** Messages are deduplicated by `seq` and kept ordered,
 *    because a resumed stream legitimately replays frames.
 */
import { useCallback, useEffect, useRef, useState } from 'react'
import { AgentClient } from './client'
import { AgentSdkError } from './errors'
import type { MessageDto, SseEvent } from './types'

/** What the token endpoint hands the browser. */
export interface AgentStreamToken {
  sessionId: string
  token: string
  baseUrl: string
  /** ISO-8601; the hook refreshes 60s before this. */
  expiresAt: string
}

export type AgentStreamStatus =
  | 'idle'
  | 'connecting'
  | 'live'
  | 'reconnecting'
  | 'expired'
  | 'error'
  | 'stopped'

export interface UseAgentStreamOptions {
  /** Gateway session behind the task; `null` keeps the hook idle. */
  sessionId?: string | null
  /** Callers can hold the stream closed (e.g. the panel is collapsed). */
  enabled?: boolean
  fetchToken: () => Promise<AgentStreamToken>
  onMessage?: (message: MessageDto) => void
  onEvent?: (event: SseEvent) => void
  onRunState?: (state: string) => void
}

export interface AgentStreamState {
  status: AgentStreamStatus
  /** Live frames only, `seq`-ascending. Merge with the archive for display. */
  messages: MessageDto[]
  runState?: string
  error?: string
  /** Highest `message_added` seq seen — the resume cursor. */
  watermark: number
}

export interface UseAgentStreamResult extends AgentStreamState {
  /** POST the text straight to the gateway (the API proxy has no send route). */
  send: (text: string) => Promise<void>
  /** Drop the token and reopen the stream at the current watermark. */
  reconnect: () => void
}

const REFRESH_LEAD_MS = 60_000

/**
 * Keep one session stream open for as long as the caller is mounted. The
 * returned `messages` are the frames this client actually saw; the caller owns
 * the historical ones.
 */
export function useAgentStream(options: UseAgentStreamOptions): UseAgentStreamResult {
  const { sessionId, enabled = true, fetchToken } = options

  const [state, setState] = useState<AgentStreamState & { sessionId: string | null }>({
    status: 'idle',
    messages: [],
    watermark: 0,
    sessionId: null,
  })
  const [generation, setGeneration] = useState(0)

  const handlers = useRef(options)
  const fetchTokenRef = useRef(fetchToken)
  const tokenRef = useRef<AgentStreamToken | null>(null)
  const clientRef = useRef<AgentClient | null>(null)
  const watermarkRef = useRef(0)
  const watermarkSessionRef = useRef<string | null>(null)
  const sessionRef = useRef<string | null>(sessionId ?? null)

  // The latest callbacks and inputs live in refs so the stream is opened once per
  // generation: a re-render (a keystroke in the composer) must not reconnect.
  useEffect(() => {
    handlers.current = options
    fetchTokenRef.current = fetchToken
    sessionRef.current = sessionId ?? null
  })

  const ensureClient = useCallback(async (): Promise<AgentClient> => {
    const session = sessionRef.current
    if (!session) throw new AgentSdkError('会话尚未创建，暂时无法连接智能体网关')
    const current = clientRef.current
    if (current !== null && current.sessionId === session) return current
    // Minting a JWT is a round trip to the API host, so the panel says
    // "connecting" from here rather than from the effect that opened the stream.
    setState(prev => ({ ...prev, status: 'connecting', error: undefined }))
    const minted = tokenRef.current ?? (await fetchTokenRef.current())
    tokenRef.current = minted
    const client = new AgentClient({
      baseUrl: minted.baseUrl,
      token: minted.token,
      sessionId: minted.sessionId || session,
    })
    clientRef.current = client
    return client
  }, [])

  const reconnect = useCallback(() => {
    tokenRef.current = null
    clientRef.current = null
    setGeneration(n => n + 1)
  }, [])

  useEffect(() => {
    if (!enabled || !sessionId) return
    const controller = new AbortController()
    let timer: ReturnType<typeof setTimeout> | undefined

    void (async () => {
      try {
        const client = await ensureClient()
        if (controller.signal.aborted) return
        // The resume cursor belongs to one gateway session: reusing it across
        // tasks would ask the new session for frames it never had.
        if (watermarkSessionRef.current !== sessionId) {
          watermarkRef.current = 0
          watermarkSessionRef.current = sessionId
        }
        setState(prev => ({
          ...prev,
          status: 'live',
          error: undefined,
          sessionId,
          watermark: watermarkRef.current,
          messages: prev.sessionId === sessionId ? prev.messages : [],
        }))

        const minted = tokenRef.current
        if (minted !== null) {
          // Silent rotation: the SDK's own retry loop would replay a dead JWT.
          const until = Date.parse(minted.expiresAt) - Date.now() - REFRESH_LEAD_MS
          timer = setTimeout(
            () => {
              tokenRef.current = null
              setGeneration(n => n + 1)
            },
            Math.max(1_000, Number.isFinite(until) ? until : 0),
          )
        }

        await client.streamEvents({
          signal: controller.signal,
          lastEventId: watermarkRef.current,
          onMessage: message => {
            watermarkRef.current = Math.max(watermarkRef.current, message.seq)
            setState(prev => ({
              ...prev,
              watermark: watermarkRef.current,
              messages: prev.messages.some(existing => existing.seq === message.seq)
                ? prev.messages
                : [...prev.messages, message].sort((a, b) => a.seq - b.seq),
            }))
            handlers.current.onMessage?.(message)
          },
          onEvent: event => handlers.current.onEvent?.(event),
          onError: error => {
            if (error.isUnauthorized || error.isForbidden) {
              tokenRef.current = null
              clientRef.current = null
              setState(prev => ({ ...prev, status: 'expired', error: error.message }))
              setGeneration(n => n + 1)
              return false
            }
            setState(prev => ({ ...prev, status: 'reconnecting', error: error.message }))
            return undefined
          },
        })
        if (!controller.signal.aborted) setState(prev => ({ ...prev, status: 'stopped' }))
      } catch (cause) {
        if (controller.signal.aborted) return
        setState(prev => ({
          ...prev,
          status: 'error',
          error: cause instanceof Error ? cause.message : String(cause),
        }))
      }
    })()

    return () => {
      controller.abort()
      if (timer !== undefined) clearTimeout(timer)
      clientRef.current = null
    }
  }, [enabled, sessionId, generation, ensureClient])

  const send = useCallback(
    async (text: string) => {
      const client = await ensureClient()
      await client.sendMessage(text)
    },
    [ensureClient],
  )

  // A stream that is not being held open for this session reports `idle` and no
  // frames, whatever the previous session left behind.
  const attached = enabled && Boolean(sessionId) && state.sessionId === sessionId
  return {
    ...state,
    status: attached ? state.status : 'idle',
    messages: attached ? state.messages : [],
    watermark: attached ? state.watermark : 0,
    send,
    reconnect,
  }
}

/**
 * Merge the archive (rows the backend's subscriber persisted) with the live
 * frames, newest per `seq`, oldest first.
 */
export function mergeAgentMessages<T extends { seq: number }>(
  archive: readonly T[],
  live: readonly T[],
): T[] {
  if (live.length === 0) return [...archive]
  const bySeq = new Map<number, T>()
  for (const message of archive) bySeq.set(message.seq, message)
  for (const message of live) bySeq.set(message.seq, message)
  return [...bySeq.values()].sort((a, b) => a.seq - b.seq)
}