AgentServer/packages/client/src/http.ts

// 请求与 SSE 的公共部分：AgentClient（/v1）与 BotClient（/bot/v1）共用
// SSE 用 fetch + ReadableStream 实现：可携带 Authorization 头，断线后按 Last-Event-ID 续传
import type { AgentEvent, ApiErrorBody, BrowserTokenDto } from '@agent-server/shared'

export interface AgentClientOptions {
  /** 服务地址，例如 https://agent.example.com；同源时可为空字符串 */
  baseUrl: string
  /** Agent Key：只在业务平台的后端使用，不要放进浏览器代码 */
  apiKey?: string
  /**
   * 浏览器中使用：返回浏览器令牌（由平台后端用 Agent Key 换取，见 browserTokenProvider）。
   * 令牌失效（401）时会以 { refresh: true } 再调用一次并重试
   */
  getToken?: (opts: { refresh: boolean }) => string | Promise<string>
  fetch?: typeof fetch
}

export class AgentApiError extends Error {
  constructor(readonly status: number, readonly code: string, message: string, readonly details?: unknown) {
    super(message)
  }
}

export class HttpClient {
  private readonly fetchImpl: typeof fetch

  constructor(protected readonly opts: AgentClientOptions) {
    this.fetchImpl = opts.fetch ?? globalThis.fetch.bind(globalThis)
  }

  private async headers(extra?: Record<string, string>, refresh = false): Promise<Record<string, string>> {
    const token = this.opts.getToken ? await this.opts.getToken({ refresh }) : this.opts.apiKey
    return { ...(token ? { authorization: `Bearer ${token}` } : {}), ...extra }
  }

  /** 发请求；使用浏览器令牌时，令牌失效则刷新后重试一次。Cookie 只在同源（后台）时携带 */
  protected async send(path: string, init: RequestInit & { headers?: Record<string, string> }): Promise<Response> {
    const go = async (refresh: boolean) => this.fetchImpl(`${this.opts.baseUrl}${path}`, {
      ...init, credentials: 'same-origin', headers: await this.headers(init.headers, refresh),
    })
    const res = await go(false)
    if (res.status === 401 && this.opts.getToken) {
      await res.body?.cancel()
      return go(true)
    }
    return res
  }

  async request<T>(method: string, path: string, body?: unknown): Promise<T> {
    const isForm = typeof FormData !== 'undefined' && body instanceof FormData
    const res = await this.send(path, {
      method,
      headers: body !== undefined && !isForm ? { 'content-type': 'application/json' } : undefined,
      body: body === undefined ? undefined : isForm ? body : JSON.stringify(body),
    })
    if (!res.ok) {
      const err = (await res.json().catch(() => null)) as ApiErrorBody | null
      throw new AgentApiError(res.status, err?.error.code ?? 'http_error', err?.error.message ?? `HTTP ${res.status}`, err?.error.details)
    }
    if (res.status === 204) return undefined as T
    return (await res.json()) as T
  }

  /** 订阅 SSE 事件流；断线自动重连并从最后收到的 seq 续传。返回取消函数 */
  protected stream<E extends { seq: number }>(path: string, handlers: StreamHandlers<E>, after: number): () => void {
    let lastSeq = after
    let stopped = false
    let controller: AbortController | null = null
    let retry = 0

    const connect = async () => {
      while (!stopped) {
        controller = new AbortController()
        try {
          const res = await this.send(path, {
            headers: { accept: 'text/event-stream', 'last-event-id': String(lastSeq) },
            signal: controller.signal,
          })
          if (!res.ok || !res.body) {
            const err = (await res.json().catch(() => null)) as ApiErrorBody | null
            throw new AgentApiError(res.status, err?.error.code ?? 'stream_error', err?.error.message ?? `HTTP ${res.status}`)
          }
          retry = 0
          handlers.onStatus?.('open')
          for await (const event of parseSse<E>(res.body)) {
            lastSeq = Math.max(lastSeq, event.seq)
            handlers.onEvent(event)
          }
        } catch (err) {
          if (stopped) return
          if (err instanceof AgentApiError && (err.status === 401 || err.status === 404)) {
            handlers.onStatus?.('closed')
            handlers.onError?.(err)
            return
          }
          handlers.onError?.(err)
        }
        if (stopped) return
        handlers.onStatus?.('reconnecting')
        await new Promise(r => setTimeout(r, Math.min(1000 * 2 ** retry++, 15000)))
      }
    }
    void connect()
    return () => {
      stopped = true
      controller?.abort()
      handlers.onStatus?.('closed')
    }
  }
}

/**
 * 浏览器令牌缓存：fetchToken 调用平台自己的后端（后端再用 Agent Key 调 issueBrowserToken）。
 * 在过期前 60 秒自动换新，并发请求共用同一次获取
 */
export function browserTokenProvider(fetchToken: () => Promise<BrowserTokenDto>): NonNullable<AgentClientOptions['getToken']> {
  let current: BrowserTokenDto | null = null
  let pending: Promise<BrowserTokenDto> | null = null
  return async ({ refresh }) => {
    const fresh = current && Date.parse(current.expiresAt) - Date.now() > 60_000
    if (fresh && !refresh) return current!.token
    pending ??= fetchToken().finally(() => { pending = null })
    current = await pending
    return current.token
  }
}

export interface StreamHandlers<E = AgentEvent> {
  onEvent: (e: E) => void
  onStatus?: (s: 'open' | 'reconnecting' | 'closed') => void
  onError?: (err: unknown) => void
}

async function* parseSse<E>(body: ReadableStream<Uint8Array>): AsyncGenerator<E> {
  const reader = body.getReader()
  const decoder = new TextDecoder()
  let buf = ''
  while (true) {
    const { value, done } = await reader.read()
    if (done) return
    buf += decoder.decode(value, { stream: true })
    let idx: number
    while ((idx = buf.indexOf('\n\n')) >= 0) {
      const chunk = buf.slice(0, idx)
      buf = buf.slice(idx + 2)
      const data = chunk.split('\n').filter(l => l.startsWith('data: ')).map(l => l.slice(6)).join('\n')
      if (data) yield JSON.parse(data) as E
    }
  }
}

export function query(q: Record<string, string | number | boolean | undefined>) {
  const s = Object.entries(q).filter(([, v]) => v !== undefined).map(([k, v]) => `${k}=${encodeURIComponent(String(v))}`).join('&')
  return s ? `?${s}` : ''
}