/**
 * Browser client for the realtime gateway — **KenEvent over Socket.IO**.
 *
 * 历史：这一层以前实现的是 `GET /ws/v1`（`gitdataai/app/im`，裸 WebSocket +
 * `gitdata.jwt.*` 子协议 + subscribe/history/event 帧）。im 网关已从 KenApi 删除，
 * 实时通道整体换成 KenEvent（RocketMQ → Socket.IO），本文件是迁移后的唯一入口。
 *
 * 保留的对外契约（`hub.ts` / `react.ts` / 页面都用它，迁移时零改动）：
 * - `onStatus`：`WsStatus` 生命周期；
 * - `onFrame`：`ServerFrame`（`event` / `user_event` / `resync_required`）；
 * - `onIssue`：协议层无法自愈的情况（订阅被拒、鉴权失败、连接失败）；
 * - `subscribe` / `unsubscribe`：房间意图，refcount 由 `ChatRealtimeHub` 维护。
 *
 * 与旧实现的三处语义差别，都来自 KenEvent 的契约：
 * 1. **没有 history 快照**：订阅不会回放消息，页面一律用 REST 回拉（
 *    `use-room-stream` 本来就有 HTTP re-seed 路径）。
 * 2. **没有 typing 广播**：`docs/events.json` 里 chat 没有 typing tag，
 *    `sendTyping` 保留为本地 no-op，接口不变以免页面改动。
 * 3. **重连由 Socket.IO 自带**：指数退避 + 连接恢复都在库内，本层只在
 *    `connect` 时重放订阅（服务端订阅是 per-socket 的）。
 */

import { unwrapApi } from '@app/lib'
import type { MessageDto } from '@app/services/api'
import Apis from '@app/services/api'
import { type KenEventEnvelope, KenEventSocket, prop } from './kenEvent'
import { frameSeq, toServerFrame } from './kenEventFrames'
import {
  classifyWsError,
  type RoomEvent,
  type ServerFrame,
  type UserEvent,
  WS_SERVER_DEFAULTS,
  type WsServerErrorKind,
  type WsStatus,
  type WsTypingIndicator,
} from './protocol'
import { TypingRegistry } from './typing'

export interface ChatWsClientOptions {
  /** 覆盖 realtime 入口地址；默认同源（vite dev 与网关都反代 `/socket.io`）。 */
  url?: string
  /** 保留字段：KenEvent 的鉴权走会话 cookie，不再需要单独换 JWT。 */
  fetchJwt?: () => Promise<unknown>
  /** 保留字段：心跳由 Socket.IO 自己完成（30s ping/pong）。 */
  pingIntervalMs?: number
  minReconnectDelayMs?: number
  maxReconnectDelayMs?: number
  maxConnectFailures?: number
  pongTimeoutMs?: number
  /** 保留字段：KenEvent 的订阅上限由服务端强制。 */
  maxSubscriptions?: number
  historyLimitHint?: number
  maxResyncPages?: number
  typingThrottleMs?: number
  correlateSubscribeErrors?: boolean
}

/**
 * 协议层无法自愈的情况。页面只对其中少数几种做反应（forbidden → 收回权限，
 * resync / cursor_lost → HTTP 重新播种），其余用于观测。
 */
export type WsClientIssue =
  | { kind: 'server_error'; error: WsServerErrorKind; message: string; room_id?: string }
  | { kind: 'subscription_limit'; room_id: string; limit: number }
  | { kind: 'gap'; room_id: string; from: number; to: number }
  | { kind: 'resync_required'; room_id: string; reason: string; dropped: boolean }
  | { kind: 'cursor_lost'; room_id: string; reason: string }
  | { kind: 'history_truncated'; room_id: string; pages: number; page_size: number }
  | { kind: 'server_bye'; message: string }
  | { kind: 'socket_closed'; code: number; ready: boolean }
  | { kind: 'pong_timeout'; outstanding_pings: number }
  | { kind: 'unrecognized_frame'; snippet: string }
  | { kind: 'binary_frame' }
  | { kind: 'unparseable_payload'; channel: 'room' | 'user'; room_id?: string }
  | { kind: 'connect_failed'; phase: 'jwt' | 'handshake'; message: string; failures: number }
  | { kind: 'auth_required'; message: string }
  | { kind: 'gave_up'; failures: number }

export class ChatWsClient {
  private transport: KenEventSocket | null = null
  private status: WsStatus = 'idle'
  private selfIdValue: string | undefined
  private readonly rooms = new Set<string>()
  private readonly sequences = new Map<string, number>()
  private readonly typing = new TypingRegistry()
  private readonly statusListeners = new Set<(status: WsStatus) => void>()
  private readonly frameListeners = new Set<(frame: ServerFrame) => void>()
  private readonly issueListeners = new Set<(issue: WsClientIssue) => void>()
  private readonly typingListeners = new Set<(roomId: string) => void>()
  private readonly seenNotifications = new Set<string>()
  private failures = 0
  private typingNoticeShown = false

  private readonly options: ChatWsClientOptions

  constructor(options: ChatWsClientOptions = {}) {
    this.options = options
    this.typing.onTyping((roomId) => {
      for (const listener of [...this.typingListeners]) listener(roomId)
    })
  }

  /** `ready.user_id`：KenEvent 不在帧里回传，保留接口供页面消费（Undefined 即未知）。 */
  get userId(): string | undefined {
    return this.selfIdValue
  }

  get connectionStatus(): WsStatus {
    return this.status
  }

  onStatus(listener: (status: WsStatus) => void): () => void {
    this.statusListeners.add(listener)
    return () => this.statusListeners.delete(listener)
  }

  onFrame(listener: (frame: ServerFrame) => void): () => void {
    this.frameListeners.add(listener)
    return () => this.frameListeners.delete(listener)
  }

  onIssue(listener: (issue: WsClientIssue) => void): () => void {
    this.issueListeners.add(listener)
    return () => this.issueListeners.delete(listener)
  }

  onTypingChange(listener: (roomId: string) => void): () => void {
    this.typingListeners.add(listener)
    return () => this.typingListeners.delete(listener)
  }

  /** 幂等：已连接或连接中时直接返回。 */
  start(): void {
    if (this.transport && this.status !== 'closed' && this.status !== 'failed') return
    this.failures = 0
    this.setStatus('connecting')

    const transport = new KenEventSocket({
      url: this.options.url,
      onOpen: ({ sid }) => {
        this.failures = 0
        void sid
        this.setStatus('open')
        // 个人通道（通知/提及/已读）每个连接订阅一次。
        transport.subscribePersonal()
        for (const roomId of this.rooms) transport.subscribeRoom(roomId)
      },
      onClose: (_reason, wasOpen) => {
        if (this.status === 'auth_required' || this.status === 'failed') return
        // Socket.IO 会自己重连；这里只反映状态，不主动重建。
        this.setStatus(wasOpen ? 'reconnecting' : 'closed')
        this.emitIssue({ kind: 'socket_closed', code: 1006, ready: wasOpen })
      },
      onAuthRequired: (message) => {
        // 终态：等页面在登录后显式 `start()`（hub 的 watchRoom 会这么做）。
        this.setStatus('auth_required')
        this.emitIssue({ kind: 'auth_required', message })
        transport.stop()
      },
      onConnectError: (message) => {
        this.failures += 1
        this.emitIssue({ kind: 'connect_failed', phase: 'handshake', message, failures: this.failures })
        this.setStatus('reconnecting')
      },
      onEnvelope: (envelope) => this.handleEnvelope(envelope),
      onResync: (envelope) => {
        const roomId = prop(envelope, 'room_id')
        if (!roomId) return
        this.emitFrame({ type: 'resync_required', room_id: roomId, reason: 'lagged' })
        this.emitIssue({ kind: 'resync_required', room_id: roomId, reason: 'lagged', dropped: true })
      },
      onServerError: (message) => {
        const error = classifyServerError(message)
        this.emitIssue({ kind: 'server_error', error, message })
      },
      onSubscribeError: (subscription, message) => {
        const roomId = subscription.split(':')[0]
        const error = classifyServerError(message)
        this.emitIssue({ kind: 'server_error', error, message, room_id: roomId })
      },
    })

    this.transport = transport
    transport.start()
  }

  stop(): void {
    this.transport?.stop()
    this.transport = null
    this.setStatus('closed')
  }

  dispose(): void {
    this.transport?.dispose()
    this.transport = null
    this.rooms.clear()
    this.sequences.clear()
    this.typing.clear()
    this.statusListeners.clear()
    this.frameListeners.clear()
    this.issueListeners.clear()
    this.typingListeners.clear()
    this.setStatus('idle')
  }

  subscribe(roomId: string): void {
    if (this.rooms.has(roomId)) return
    this.rooms.add(roomId)
    this.start()
    this.transport?.subscribeRoom(roomId)
  }

  unsubscribe(roomId: string): void {
    this.rooms.delete(roomId)
    this.transport?.unsubscribeRoom(roomId)
    this.typing.clear(roomId)
  }

  /**
   * typing 广播在 KenEvent 里没有对应 tag（`docs/events.json` 的 `gitdata-chat`
   * 不含 typing），所以这里保持 no-op：接口不变，页面调用无害，等服务端补上
   * 事件后只需在这里 `emit('typing', …)`。
   */
  sendTyping(roomId: string, threadId?: string | null): void {
    void roomId
    void threadId
    if (!this.typingNoticeShown && import.meta.env.DEV) {
      this.typingNoticeShown = true
      console.info('[ws] typing 事件在 KenEvent 侧尚未定义，已忽略 sendTyping()')
    }
  }

  typingUsers(roomId: string, options: { threadId?: string | null } = {}): WsTypingIndicator[] {
    return this.typing.active(roomId, {
      excludeUserId: this.selfIdValue,
      threadId: options.threadId ?? null,
    })
  }

  lastKnownSeq(roomId: string): number | undefined {
    return this.sequences.get(roomId)
  }

  private handleEnvelope(envelope: KenEventEnvelope): void {
    void this.pump(envelope)
  }

  /**
   * 信封 → 帧。KenApi 的事件体只是摘要（`summary_body` = id/kind/occurred_at/
   * summary），消息正文必须回拉 `GET /chat/rooms/{room}/messages/{id}` 才能
   * 交给 transcript reducer，这也是 KenEvent 文档里"详情走 REST"的含义。
   */
  private async pump(envelope: KenEventEnvelope): Promise<void> {
    const hydrated = await this.hydrate(envelope)
    let frame = toServerFrame(hydrated)

    if (!frame) {
      // 已知的 chat 事件映射不出来（比如 reaction-update 只带摘要）时，
      // 退化成一次房间级软刷新，页面自己回拉最新状态。
      const roomId = prop(envelope, 'room_id')
      if (roomId && envelope.topic === 'gitdata-chat') {
        frame = { type: 'event', room_id: roomId, event: { kind: 'room_updated', room_id: roomId } }
      } else {
        // 未消费的事件（issue/pull/kenshell/oauth…）静默丢弃：它们有各自的页面。
        return
      }
    }

    const seq = frameSeq(frame)
    if (seq) {
      const current = this.sequences.get(seq.roomId) ?? 0
      if (seq.seq > current) this.sequences.set(seq.roomId, seq.seq)
    }
    this.emitFrame(frame)
  }

  /** 消息类事件回拉完整 DTO；失败时原样返回信封，由上层降级处理。 */
  private async hydrate(envelope: KenEventEnvelope): Promise<KenEventEnvelope> {
    const hydrated = await this.hydrateNotification(envelope)
    if (hydrated) return hydrated
    return this.hydrateMessage(envelope)
  }

  /**
   * 通知事件同样只带摘要，而 `WsNotification` 要渲染标题/正文，所以回拉最新一条
   * （`GET /me/notifications?limit=1`）。同一条只处理一次，避免重连补发时重复弹窗。
   */
  private async hydrateNotification(envelope: KenEventEnvelope): Promise<KenEventEnvelope | null> {
    if (envelope.topic !== 'gitdata-notify' || envelope.tag !== 'notification-create') return null
    if (isNotificationDto(envelope.body)) return envelope
    try {
      const page = await unwrapApi<{ items?: unknown[] }>(
        Apis.notifications.list_notifications({ params: { limit: 1, offset: 0 } }),
      )
      const latest = page?.items?.[0]
      if (!isNotificationDto(latest)) return null
      if (this.seenNotifications.has(latest.id)) return null
      this.seenNotifications.add(latest.id)
      return { ...envelope, body: latest }
    } catch {
      return null
    }
  }

  private async hydrateMessage(envelope: KenEventEnvelope): Promise<KenEventEnvelope> {
    const needsMessage = envelope.tag === 'message-create' || envelope.tag === 'message-update'
    if (!needsMessage) return envelope
    if (isMessageDto(envelope.body)) return envelope

    const roomId = prop(envelope, 'room_id')
    const messageId = prop(envelope, 'message_id')
    if (!roomId || !messageId) return envelope
    try {
      const message = await unwrapApi<MessageDto>(
        Apis.chat.get_message({ pathParams: { room_id: roomId, message_id: messageId } }),
      )
      return { ...envelope, body: message }
    } catch {
      return envelope
    }
  }

  private emitFrame(frame: ServerFrame): void {
    for (const listener of [...this.frameListeners]) listener(frame)
  }

  private emitIssue(issue: WsClientIssue): void {
    for (const listener of [...this.issueListeners]) listener(issue)
  }

  private setStatus(status: WsStatus): void {
    if (this.status === status) return
    this.status = status
    for (const listener of [...this.statusListeners]) listener(status)
  }
}

/** 可渲染的通知 DTO（摘要体只有 `{id, kind, occurred_at, summary}`）。 */
function isNotificationDto(value: unknown): value is { id: string; notification_type: string } {
  if (typeof value !== 'object' || value === null) return false
  const body = value as Record<string, unknown>
  return typeof body.id === 'string' && typeof body.notification_type === 'string'
}

/** 真正可渲染的消息 DTO（摘要体只有 `{id, kind, occurred_at, summary}`）。 */
function isMessageDto(value: unknown): boolean {
  if (typeof value !== 'object' || value === null) return false
  const body = value as Record<string, unknown>
  return typeof body.id === 'string' && typeof body.room_id === 'string' && typeof body.author_id === 'number'
}

function classifyServerError(message: string): WsServerErrorKind {
  // 先按旧协议的四条固定文案匹配，再按 KenEvent 的措辞兜底
  // （`events/subscriptions.rs:validate` 会回 "unsupported subscription topic or tag"）。
  const known = classifyWsError(message)
  if (known !== 'unknown') return known
  const text = message.toLowerCase()
  if (text.includes('limit')) return 'subscription_limit'
  if (text.includes('forbidden') || text.includes('permission') || text.includes('not allowed')) {
    return 'forbidden'
  }
  if (text.includes('subscri')) return 'subscribe_failed'
  if (text.includes('invalid') || text.includes('unsupported') || text.includes('unrecognized')) {
    return 'invalid_frame'
  }
  return 'unknown'
}

/** Re-exported for consumers that still type against `RoomEvent`/`UserEvent`. */
export type { RoomEvent, ServerFrame, UserEvent }
export { WS_SERVER_DEFAULTS }