/**
 * KenEvent transport: the Socket.IO wire contract of the realtime gateway
 * (`KenEvent/src/socket`, Engine.IO v4, path `/socket.io`).
 *
 * 与旧 im 网关（`GET /ws/v1`，裸 WebSocket + `gitdata.jwt.*` 子协议）的区别：
 * - **连接**：Socket.IO，默认路径 `/socket.io`；浏览器握手带不了自定义头，
 *   所以 token 由会话 cookie（`gitdata_session`）或 CONNECT 包的 `auth.token`
 *   提供，KenEvent 侧见 `socket/response.rs` 与 `socket/session/state.rs`。
 * - **订阅**：客户端发 `subscribe {topic, tag, filter, id}`，服务端按
 *   RocketMQ 的 topic/tag 精确匹配（`events/subscriptions.rs:validate`），
 *   没有通配，所以一个房间要按关心的 tag 各订阅一条。
 * - **推送**：服务端只发两种事件名——`event` 与 `resync`，载荷是
 *   `{topic, tag, user_id, recipient_user_id, properties, body}` 信封
 *   （`events/action/common.rs:payload`）。**详情一律走 REST 回拉**。
 * - **个人事件**：`gitdata-notify`/`notification-create` 与 chat 的
 *   mention/read_state/presence 等由服务端强制绑定 `filter.user_id` 自己。
 */
import { io, type Socket } from 'socket.io-client'

export const KEN_EVENT_PATH = '/socket.io'
export const CHAT_TOPIC = 'gitdata-chat'
export const NOTIFY_TOPIC = 'gitdata-notify'

/** 房间级事件：一条订阅一个 tag，`filter` 固定为 `{room_id}`。 */
export const ROOM_TAGS = [
  'message-create',
  'message-update',
  'message-delete',
  'messages-delete',
  'pin-create',
  'pin-delete',
  'reaction-update',
  'room-update',
  'member-create',
  'member-delete',
  'thread-update',
  'voice-presence',
  'voice-hand',
  'voice-layout',
  'voice-moderate',
] as const

/** 用户级事件：不带 room 过滤，服务端会补上 `filter.user_id = 自己`。 */
export const PERSONAL_CHAT_TAGS = [
  'mention-create',
  'reminder-create',
  'membership-create',
  'read_state-update',
  'presence-update',
] as const

export const PERSONAL_NOTIFY_TAGS = ['notification-create'] as const

/** 服务端推送的信封（`events/action/common.rs:80`）。 */
export interface KenEventEnvelope {
  topic: string
  tag: string
  user_id?: number | null
  recipient_user_id?: number
  properties?: Record<string, unknown> | null
  body?: unknown
  /** 断流补齐时 KenEvent 会把事件名换成 `resync` 并置这个标记。 */
  resync_required?: boolean
}

export interface SubscriptionAck {
  ok?: boolean
  data?: unknown
  error?: string
}

export interface KenEventSocketOptions {
  /** 默认同源（`window.location.origin`）；vite dev 与网关都由 `/socket.io` 反代。 */
  url?: string
  onOpen: (info: { sid: string }) => void
  onClose: (reason: string, wasOpen: boolean) => void
  onAuthRequired: (message: string) => void
  onConnectError: (message: string) => void
  onEnvelope: (envelope: KenEventEnvelope) => void
  onResync: (envelope: KenEventEnvelope) => void
  onServerError: (message: string) => void
  onSubscribeError: (subscription: string, message: string) => void
}

const DEVICE_KEY = 'gitdata.ws.device_id'

/** 设备标识：KenEvent 用它做多端会话隔离（`session/state.rs`）。 */
export function deviceId(): string {
  try {
    const stored = window.localStorage.getItem(DEVICE_KEY)
    if (stored) return stored
    const created = crypto.randomUUID()
    window.localStorage.setItem(DEVICE_KEY, created)
    return created
  } catch {
    // 隐私模式 / SSR：退化成会话级 id，重连后仍能复用（同一模块生命周期内）。
    return crypto.randomUUID()
  }
}

function asRecord(value: unknown): Record<string, unknown> | null {
  return typeof value === 'object' && value !== null ? (value as Record<string, unknown>) : null
}

export function asEnvelope(value: unknown): KenEventEnvelope | null {
  const record = asRecord(value)
  if (!record) return null
  const topic = record.topic
  const tag = record.tag
  if (typeof topic !== 'string' || typeof tag !== 'string') return null
  return record as unknown as KenEventEnvelope
}

/** `properties` 里的字符串字段（RocketMQ 用户属性全是字符串）。 */
export function prop(envelope: KenEventEnvelope, key: string): string | null {
  const value = envelope.properties?.[key]
  if (typeof value === 'string' && value.length > 0) return value
  const fromBody = asRecord(envelope.body)?.[key]
  return typeof fromBody === 'string' && fromBody.length > 0 ? fromBody : null
}

/**
 * 订阅管理器：把「打开了哪些房间」这份意图翻译成 KenEvent 的逐 tag 订阅。
 *
 * 重连后服务端的订阅是空的（per-socket 状态），所以每次 `connect` 都要重放；
 * 房间的 refcount 由上层 hub 维护，这里只负责 idempotent 增删。
 */
export class KenEventSocket {
  private socket: Socket | null = null
  /** roomId -> tag -> 订阅 id（KenEvent 要求 id 是 UUID，由客户端生成）。 */
  private readonly rooms = new Map<string, Map<string, string>>()
  /** `${topic}|${tag}` -> 个人订阅 id。 */
  private readonly personalIds = new Map<string, string>()
  /** 通用资源订阅：key -> 订阅意图（issue / pull / org / kenshell 等）。 */
  private readonly topics = new Map<
    string,
    { topic: string; tag: string; filter: Record<string, string>; refs: number; id: string }
  >()
  private personal = false
  private disposed = false

  private readonly options: KenEventSocketOptions

  constructor(options: KenEventSocketOptions) {
    this.options = options
  }

  get connected(): boolean {
    return this.socket?.connected ?? false
  }

  /** 幂等：连接中/已连接时什么也不做。 */
  start(): void {
    if (this.disposed || this.socket) return
    const socket = io(this.options.url, {
      path: KEN_EVENT_PATH,
      // 先 polling 再升级 websocket：网关（KenGateWay）不代理 Upgrade，会回 501，
      // 从 websocket 起步会直接连不上；polling 起步在两种部署下都能连上，
      // 拿到连接后 socket.io 会自己尝试升级到 WebSocket（nginx 环境即可升级成功）。
      transports: ['polling', 'websocket'],
      // 兜底：某个传输被拒时继续尝试下一个，而不是直接失败。
      tryAllTransports: true,
      withCredentials: true,
      auth: (cb) => cb({ device_id: deviceId() }),
      reconnection: true,
      reconnectionDelay: 500,
      reconnectionDelayMax: 30_000,
      timeout: 20_000,
    })
    this.socket = socket

    socket.on('connect', () => {
      this.options.onOpen({ sid: socket.id ?? '' })
      this.replay()
    })
    socket.on('disconnect', (reason) => {
      this.options.onClose(reason, true)
    })
    socket.on('connect_error', (error) => {
      const message = error instanceof Error ? error.message : String(error)
      // KenEvent 在鉴权失败时发 `connect_error('authentication failed')`
      // （`socket/session/state.rs:connect`）。
      if (/auth|unauthor|missing access token/i.test(message)) {
        this.options.onAuthRequired(message)
        return
      }
      this.options.onConnectError(message)
    })
    socket.on('event', (payload: unknown) => {
      const envelope = asEnvelope(payload)
      if (!envelope) {
        this.options.onServerError('unrecognized event envelope')
        return
      }
      if (envelope.resync_required) {
        this.options.onResync(envelope)
        return
      }
      this.options.onEnvelope(envelope)
    })
    socket.on('resync', (payload: unknown) => {
      const envelope = asEnvelope(payload)
      if (envelope) this.options.onResync(envelope)
    })
    socket.on('error', (payload: unknown) => {
      const message = (asRecord(payload)?.message as string | undefined) ?? 'realtime gateway error'
      this.options.onServerError(message)
    })
  }

  stop(): void {
    const socket = this.socket
    this.socket = null
    if (!socket) return
    socket.removeAllListeners()
    socket.disconnect()
  }

  dispose(): void {
    this.disposed = true
    this.rooms.clear()
    this.personal = false
    this.stop()
  }

  /** 加入一个房间：按 `ROOM_TAGS` 逐条订阅。 */
  subscribeRoom(roomId: string): void {
    if (this.rooms.has(roomId)) return
    this.rooms.set(roomId, new Map())
    this.start()
    if (this.connected) this.emitRoom(roomId)
  }

  unsubscribeRoom(roomId: string): void {
    const ids = this.rooms.get(roomId)
    if (!ids) return
    this.rooms.delete(roomId)
    if (!this.connected) return
    for (const [tag, id] of ids) this.emitUnsubscribe(CHAT_TOPIC, tag, id)
  }

  /**
   * 订阅任意 topic/tag（带资源过滤）。同一 `key` 只发一次订阅，引用计数归零时退订；
   * 返回的函数可安全重复调用。
   */
  subscribeTopic(spec: { key: string; topic: string; tag: string; filter?: Record<string, string> }): () => void {
    const existing = this.topics.get(spec.key)
    if (existing) {
      existing.refs += 1
      return () => this.releaseTopic(spec.key)
    }
    const entry = {
      topic: spec.topic,
      tag: spec.tag,
      filter: spec.filter ?? {},
      refs: 1,
      id: crypto.randomUUID(),
    }
    this.topics.set(spec.key, entry)
    this.start()
    if (this.connected) this.emit(entry.topic, entry.tag, { id: entry.id, filter: entry.filter })
    return () => this.releaseTopic(spec.key)
  }

  private releaseTopic(key: string): void {
    const entry = this.topics.get(key)
    if (!entry) return
    entry.refs -= 1
    if (entry.refs > 0) return
    this.topics.delete(key)
    if (this.connected) this.emitUnsubscribe(entry.topic, entry.tag, entry.id)
  }

  /** 个人通道（通知、提及、已读游标）：一次连接订阅一次。 */
  subscribePersonal(): void {
    if (this.personal) return
    this.personal = true
    this.start()
    if (this.connected) this.emitPersonal()
  }

  /** 重放全部订阅意图；每次 `connect` 都会调用。 */
  private replay(): void {
    this.personalIds.clear()
    for (const ids of this.rooms.values()) ids.clear()
    if (this.personal) this.emitPersonal()
    for (const roomId of this.rooms.keys()) this.emitRoom(roomId)
    for (const entry of this.topics.values()) {
      entry.id = crypto.randomUUID()
      this.emit(entry.topic, entry.tag, { id: entry.id, filter: entry.filter })
    }
  }

  private emitRoom(roomId: string): void {
    const ids = this.rooms.get(roomId)
    if (!ids) return
    for (const tag of ROOM_TAGS) {
      if (ids.has(tag)) continue
      const id = crypto.randomUUID()
      ids.set(tag, id)
      this.emit(CHAT_TOPIC, tag, { id, filter: { room_id: roomId } })
    }
  }

  private emitPersonal(): void {
    for (const tag of PERSONAL_CHAT_TAGS) this.emitPersonalOne(CHAT_TOPIC, tag)
    for (const tag of PERSONAL_NOTIFY_TAGS) this.emitPersonalOne(NOTIFY_TOPIC, tag)
  }

  private emitPersonalOne(topic: string, tag: string): void {
    const key = `${topic}|${tag}`
    if (this.personalIds.has(key)) return
    const id = crypto.randomUUID()
    this.personalIds.set(key, id)
    this.emit(topic, tag, { id })
  }

  private emit(topic: string, tag: string, extra: { id: string; filter?: unknown }): void {
    const socket = this.socket
    if (!socket) return
    const payload = { topic, tag, ...extra }
    // ack 里的 `{ok:false,error}` 是唯一能归因到具体订阅的失败信号。
    socket.timeout(10_000).emit('subscribe', payload, (error: Error | null, ack?: SubscriptionAck) => {
      if (error) {
        this.options.onSubscribeError(`${topic}/${tag}`, 'subscribe ack timeout')
        return
      }
      if (ack && ack.ok === false) {
        this.options.onSubscribeError(`${topic}/${tag}`, ack.error ?? 'subscribe rejected')
      }
    })
  }

  private emitUnsubscribe(topic: string, tag: string, id: string): void {
    this.socket?.emit('unsubscribe', { topic, tag, id })
  }
}