import {
  classifyWsError,
  isSubscribeError,
  parseServerFrame,
  roomEventSeq,
  WS_SERVER_DEFAULTS,
  asSeq,
  type ClientFrame,
  type RoomEvent,
  type ServerFrame,
  type WsServerErrorKind,
  type WsStatus,
  type WsTypingIndicator,
} from './protocol'
import { requestWsJwt, isWsJwtError, type WsJwt } from './jwt'
import { TypingRegistry } from './typing'

/**
 * Browser client for the chat WebSocket (`GET /ws/v1`, served by the im gateway).
 *
 * Protocol facts it encodes, all traceable to gitdataai/app/im/src:
 * - Auth is a single-use JWT offered as the `gitdata.jwt.<token>` subprotocol
 *   (minted by `GET /auth/jwt-request?appid=chat`, consumed once by the im
 *   gateway via redis `SETNX jti`), so every connect mints a fresh one.
 * - The server sends `ready` first, then `history` / `event` / `user_event`,
 *   `gap` / `resync_required` when stream order breaks, `pong` for app-level
 *   `ping`, and `bye` right before a server-initiated close (:279-336, :550-552,
 *   :577-587).
 * - Idle sockets are reaped after `heartbeat_secs * 3` (:220-224) and the server
 *   never pings on its own, so proving liveness is the client's job.
 * - Subscriptions are per socket (:390-504), so room intents are replayed after a
 *   reconnect and missed messages are pulled with `resync`.
 * - `resync` answers with at most `history_limit` messages ordered **ascending**
 *   from the cursor (:505-531 + services/messages.rs:677-687) and `history`
 *   carries no `has_more`, so a big backlog is walked page by page here.
 * - `error` frames carry no `room_id`; a rejected subscribe is attributed through
 *   the server's in-order frame handling (`handle_client_frame(..).await` inside
 *   the read loop, :242-255) using one ping barrier per subscribe.
 */

const DEFAULT_PATH = '/ws/v1'
const JWT_PROTOCOL_PREFIX = 'gitdata.jwt.'

const DEFAULT_PING_INTERVAL_MS = 25_000 // heartbeat is 30s, the server says bye at 3x
const DEFAULT_MIN_RECONNECT_DELAY_MS = 500
const DEFAULT_MAX_RECONNECT_DELAY_MS = 30_000
const DEFAULT_MAX_CONNECT_FAILURES = 12
/** Snippet size reported for payloads this build does not recognise. */
const UNRECOGNIZED_SNIPPET_CHARS = 256

export interface ChatWsClientOptions {
  /** Full ws:// or wss:// endpoint. Defaults to same-origin `/ws/v1`. */
  url?: string
  /** Provides the single-use handshake JWT. Defaults to `GET /auth/jwt-request?appid=chat`. */
  fetchJwt?: () => Promise<WsJwt>
  /** Interval between app-level `{op:'ping'}` frames; clamped to >= 5s (server floor). */
  pingIntervalMs?: number
  minReconnectDelayMs?: number
  maxReconnectDelayMs?: number
  /**
   * Consecutive failed attempts (jwt, handshake, or a close that never
   * produced `ready`) tolerated before going terminal `failed`. 0 disables.
   */
  maxConnectFailures?: number
  /** How long an unanswered ping may stay outstanding before the socket counts as half-open. */
  pongTimeoutMs?: number
  /** Mirrors `[chat.ws] max_subscriptions`; the client refuses to exceed it locally. */
  maxSubscriptions?: number
  /** Mirrors `[chat.ws] history_limit`; used to detect truncated `history` pages. */
  historyLimitHint?: number
  /** Safety bound for the forward resync walk (pages per gap). */
  maxResyncPages?: number
  /** Client-side typing throttle; the server drops anything inside its own 3s window. */
  typingThrottleMs?: number
  /**
   * Follow each `subscribe` with a `{op:'ping'}` barrier so `error` frames can be
   * attributed to the room that caused them (default true). Turn it off to send
   * the minimum frame set; attribution then degrades to "no room_id".
   */
  correlateSubscribeErrors?: boolean
}

/**
 * Conditions the protocol cannot resolve on its own. Subscribe to these instead
 * of parsing `error` frames: each entry already carries the room when the backend
 * gives enough information to identify it.
 */
export type WsClientIssue =
  /** The server sent `{type:'error'}`; `room_id` is attributed best-effort. */
  | { 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 }
  /** No local cursor to resume from: re-seed the transcript over HTTP. */
  | { kind: 'cursor_lost'; room_id: string; reason: string }
  /** The forward resync walk hit `maxResyncPages`; older messages are still missing. */
  | { kind: 'history_truncated'; room_id: string; pages: number; page_size: number }
  | { kind: 'server_bye'; message: string }
  /** The socket dropped; `ready: false` means the upgrade never completed. */
  | { kind: 'socket_closed'; code: number; ready: boolean }
  | { kind: 'pong_timeout'; outstanding_pings: number }
  /** Inbound payload that cannot be mapped onto `ServerFrame` (protocol drift). */
  | { kind: 'unrecognized_frame'; snippet: string }
  /** The backend only writes text frames, so binary data means a broken hop. */
  | { kind: 'binary_frame' }
  /** A broadcast whose payload even the server could not parse (frames.rs:36-41). */
  | { 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 }

interface SubscribeBarrier {
  sent_at: number
  /** room_id -> subscribe epoch collected since the previous barrier. */
  rooms: Map<string, number>
}

export class ChatWsClient {
  private readonly url: string
  private readonly fetchJwt: () => Promise<WsJwt>
  private readonly pingIntervalMs: number
  private readonly minReconnectDelayMs: number
  private readonly maxReconnectDelayMs: number
  private readonly maxConnectFailures: number
  private readonly pongTimeoutMs: number
  private readonly maxSubscriptions: number
  private readonly historyLimitHint: number
  private readonly maxResyncPages: number
  private readonly typingThrottleMs: number
  private readonly correlateSubscribeErrors: boolean

  private ws: WebSocket | null = null
  private connectionStatus: WsStatus = 'idle'
  private started = false
  private generation = 0
  private reconnectAttempts = 0
  private connectFailures = 0
  private attemptReachedReady = false
  private reconnectTimer: ReturnType<typeof setTimeout> | undefined
  private pingTimer: ReturnType<typeof setInterval> | undefined
  private lastTypingSentAt = 0

  /** Room subscription intents; replayed on every `ready`. */
  private readonly rooms = new Set<string>()
  /** Resync requests that must survive a reconnect. */
  private readonly forcedResync = new Set<string>()
  /** Rooms with an outstanding forward resync walk. */
  private readonly awaitingResync = new Set<string>()
  private readonly lastSeq = new Map<string, number>()
  /** Pages consumed by the current resync walk, per room. */
  private readonly resyncPages = new Map<string, number>()

  private barriers: SubscribeBarrier[] = []
  private barrierRooms = new Map<string, number>()
  private readonly subscribeEpoch = new Map<string, number>()

  private currentUserId: string | undefined
  private currentSessionId: string | undefined
  private currentJwtTtlSecs: number | undefined

  private readonly typingRegistry = new TypingRegistry()
  private readonly frameHandlers = new Set<(frame: ServerFrame) => void>()
  private readonly statusHandlers = new Set<(status: WsStatus) => void>()
  private readonly issueHandlers = new Set<(issue: WsClientIssue) => void>()

  constructor(options: ChatWsClientOptions = {}) {
    this.url = options.url ?? sameOriginWsUrl()
    this.fetchJwt = options.fetchJwt ?? requestWsJwt
    this.pingIntervalMs = Math.max(options.pingIntervalMs ?? DEFAULT_PING_INTERVAL_MS, 5_000)
    this.minReconnectDelayMs = options.minReconnectDelayMs ?? DEFAULT_MIN_RECONNECT_DELAY_MS
    this.maxReconnectDelayMs = options.maxReconnectDelayMs ?? DEFAULT_MAX_RECONNECT_DELAY_MS
    this.maxConnectFailures = options.maxConnectFailures ?? DEFAULT_MAX_CONNECT_FAILURES
    this.pongTimeoutMs = options.pongTimeoutMs ?? this.pingIntervalMs * 2 + 5_000
    this.maxSubscriptions = options.maxSubscriptions ?? WS_SERVER_DEFAULTS.maxSubscriptions
    this.historyLimitHint = options.historyLimitHint ?? WS_SERVER_DEFAULTS.historyLimit
    this.maxResyncPages = options.maxResyncPages ?? 10
    this.typingThrottleMs = options.typingThrottleMs ?? WS_SERVER_DEFAULTS.typingThrottleSecs * 1000
    this.correlateSubscribeErrors = options.correlateSubscribeErrors ?? true
  }

  /* ------------------------------------------------------------------ state */

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

  /** `ready.user_id`; undefined until the first successful handshake. */
  get userId(): string | undefined {
    return this.currentUserId
  }

  /** `ready.session_id` (ws/mod.rs:135), a fresh v7 uuid per socket: correlate with server logs. */
  get sessionId(): string | undefined {
    return this.currentSessionId
  }

  /** `expires_in_secs` of the JWT used for the current/last handshake (services/jwt.rs). */
  get jwtTtlSecs(): number | undefined {
    return this.currentJwtTtlSecs
  }

  /** Highest `seq` observed for a room (from `history` / `event`), if any. */
  lastKnownSeq(roomId: string): number | undefined {
    return this.lastSeq.get(roomId)
  }

  isSubscribed(roomId: string): boolean {
    return this.rooms.has(roomId)
  }

  subscribedRooms(): string[] {
    return [...this.rooms]
  }

  /** Live typing indicators for a room, with this client's own echo filtered out. */
  typingUsers(roomId: string, options: { threadId?: string | null } = {}): WsTypingIndicator[] {
    return this.typingRegistry.active(roomId, { ...options, excludeUserId: this.currentUserId })
  }

  /* ---------------------------------------------------------- subscriptions */

  /** Observe every server frame. Handlers run after the client reconciled its own state. */
  onFrame(handler: (frame: ServerFrame) => void): () => void {
    this.frameHandlers.add(handler)
    return () => this.frameHandlers.delete(handler)
  }

  /** Observe connection lifecycle. Returns an unsubscribe function. */
  onStatus(handler: (status: WsStatus) => void): () => void {
    this.statusHandlers.add(handler)
    handler(this.connectionStatus)
    return () => this.statusHandlers.delete(handler)
  }

  /** Observe conditions the protocol cannot resolve on its own. */
  onIssue(handler: (issue: WsClientIssue) => void): () => void {
    this.issueHandlers.add(handler)
    return () => this.issueHandlers.delete(handler)
  }

  /** Fires with the room id whenever its live typing set changes. */
  onTypingChange(handler: (roomId: string) => void): () => void {
    return this.typingRegistry.onTyping(handler)
  }

  /**
   * Joins a room stream. Unless `resync: false`, the server answers with a
   * `history` frame holding the newest `history_limit` messages (ws/mod.rs:470-495).
   * The intent survives reconnects; history can repeat messages already seen, so
   * dedupe by `id`.
   */
  subscribe(roomId: string, options: { resync?: boolean } = {}): void {
    if (this.rooms.has(roomId)) return
    if (this.rooms.size >= this.maxSubscriptions) {
      // The server's limit error carries no room id, so enforce the cap while the
      // offending room is still known.
      this.emitIssue({ kind: 'subscription_limit', room_id: roomId, limit: this.maxSubscriptions })
      return
    }
    this.rooms.add(roomId)
    this.sendSubscribe(roomId, options.resync ?? true)
  }

  unsubscribe(roomId: string): void {
    if (!this.rooms.delete(roomId)) return
    this.forgetRoom(roomId)
    this.write({ op: 'unsubscribe', room_id: roomId })
  }

  /**
   * Asks for messages with `seq > afterSeq`; the reply is a `history` frame.
   * Forward-only and ascending, so pass the cursor you already hold (or 0 to walk
   * the room from its first message). Full pages are followed automatically up to
   * `maxResyncPages`. If the socket is down the pull is replayed after `ready`,
   * but only for rooms that still have a subscription intent.
   */
  resync(roomId: string, afterSeq: number): void {
    this.resyncPages.delete(roomId)
    this.awaitingResync.add(roomId)
    this.startResync(roomId, afterSeq)
  }

  /**
   * Broadcasts a typing signal. The throttle is per socket — mirroring the
   * server's own 3s per-socket window (ws/mod.rs:533-549) — because a frame sent
   * inside that window is dropped regardless of which room it names.
   */
  sendTyping(roomId: string, threadId?: string | null): void {
    const now = Date.now()
    if (now - this.lastTypingSentAt < this.typingThrottleMs) return
    this.lastTypingSentAt = now
    this.write({ op: 'typing', room_id: roomId, thread_id: threadId ?? null })
  }

  /* -------------------------------------------------------------- lifecycle */

  /** Opens the socket (and keeps it open until {@link stop}). Idempotent. */
  start(): void {
    if (this.started) return
    this.started = true
    this.reconnectAttempts = 0
    this.connectFailures = 0
    void this.connect()
  }

  /**
   * Closes the socket and stops reconnecting. Room intents are kept, so a later
   * {@link start} rejoins them. Terminal states (`auth_required`, `failed`) also
   * need an explicit `start()` to resume.
   */
  stop(): void {
    if (!this.started) {
      if (this.connectionStatus === 'auth_required' || this.connectionStatus === 'failed') {
        this.teardownSocket()
        this.setStatus('closed')
      }
      return
    }
    this.started = false
    this.teardownSocket()
    this.setStatus('closed')
  }

  /** Release timers and handlers. Only for teardown paths and tests. */
  dispose(): void {
    this.stop()
    this.typingRegistry.dispose()
    this.frameHandlers.clear()
    this.statusHandlers.clear()
    this.issueHandlers.clear()
  }

  private teardownSocket(): void {
    this.generation += 1
    this.clearReconnectTimer()
    this.clearPingTimer()
    const ws = this.ws
    this.ws = null
    this.resetBarrierState()
    this.typingRegistry.clear()
    try {
      ws?.close(1000)
    } catch {
      // already closing/closed
    }
  }

  /* ---------------------------------------------------------------- connect */

  private async connect(): Promise<void> {
    if (!this.started) return
    this.clearReconnectTimer()
    this.setStatus(this.reconnectAttempts === 0 ? 'connecting' : 'reconnecting')
    this.attemptReachedReady = false
    const generation = ++this.generation

    let jwt: WsJwt
    try {
      jwt = await this.fetchJwt()
    } catch (error) {
      if (generation !== this.generation || !this.started) return
      if (isWsJwtError(error) && error.kind === 'auth') {
        // 401/403 means the session is gone; the upgrade would be rejected too, so
        // a backoff loop only burns the jwt endpoint.
        this.started = false
        this.connectFailures += 1
        this.emitIssue({ kind: 'auth_required', message: error.message })
        this.setStatus('auth_required')
        return
      }
      this.registerFailure('jwt', error)
      this.scheduleReconnect()
      return
    }
    if (generation !== this.generation || !this.started) return
    this.currentJwtTtlSecs = jwt.expiresInSecs

    let ws: WebSocket
    try {
      // The upgrade response re-offers the same protocol name (ws/mod.rs) and
      // JWTs only contain base64url / '.' characters, so they are valid
      // subprotocol tokens.
      ws = new WebSocket(this.url, [`${JWT_PROTOCOL_PREFIX}${jwt.jwt}`])
    } catch (error) {
      if (generation !== this.generation) return
      this.registerFailure('handshake', error)
      this.scheduleReconnect()
      return
    }
    this.ws = ws
    this.lastTypingSentAt = 0 // the server throttle is per connection
    this.resetBarrierState()

    ws.onopen = () => {
      if (generation !== this.generation || this.ws !== ws) return
      this.startPingLoop()
    }
    ws.onmessage = event => {
      if (generation !== this.generation || this.ws !== ws) return
      this.handleMessage(event.data)
    }
    ws.onclose = event => {
      if (generation !== this.generation || this.ws !== ws) return
      this.ws = null
      this.clearPingTimer()
      this.resetBarrierState()
      this.typingRegistry.clear() // peers may have stopped typing while we were away
      if (!this.started) return
      this.emitIssue({ kind: 'socket_closed', code: event.code, ready: this.attemptReachedReady })
      // A socket that never produced `ready` is a failed attempt: every rejection
      // in ws/mod.rs:99-132 is a plain HTTP 400, which browsers hide from JS.
      if (!this.attemptReachedReady) {
        this.registerFailure('handshake', { message: `handshake closed (${event.code})` })
      }
      this.scheduleReconnect()
    }
  }

  private registerFailure(phase: 'jwt' | 'handshake', error: unknown): void {
    this.connectFailures += 1
    this.emitIssue({
      kind: 'connect_failed',
      phase,
      message: error instanceof Error ? error.message : String(error),
      failures: this.connectFailures,
    })
    if (this.maxConnectFailures > 0 && this.connectFailures >= this.maxConnectFailures) {
      this.started = false
      this.clearReconnectTimer()
      this.emitIssue({ kind: 'gave_up', failures: this.connectFailures })
      this.setStatus('failed')
    }
  }

  /* --------------------------------------------------------------- incoming */

  private handleMessage(data: unknown): void {
    if (typeof data !== 'string') {
      this.emitIssue({ kind: 'binary_frame' })
      return
    }
    const frame = parseServerFrame(data)
    if (!frame) {
      this.emitIssue({
        kind: 'unrecognized_frame',
        snippet: data.slice(0, UNRECOGNIZED_SNIPPET_CHARS),
      })
      return
    }
    this.applyFrameEffects(frame)
    for (const handler of [...this.frameHandlers]) handler(frame)
  }

  /** Keeps seq cursors, typing state and room intents consistent with the server. */
  private applyFrameEffects(frame: ServerFrame): void {
    switch (frame.type) {
      case 'ready':
        // Subscriptions are per socket, so re-offer every room intent.
        this.attemptReachedReady = true
        this.connectFailures = 0
        this.reconnectAttempts = 0
        this.currentUserId = frame.user_id
        this.currentSessionId = frame.session_id
        this.resetBarrierState()
        this.setStatus('open')
        this.resubscribeRooms()
        break
      case 'history':
        this.applyHistory(frame.room_id, frame.messages)
        break
      case 'event':
        this.applyRoomEvent(frame.room_id, frame.event)
        break
      case 'user_event':
        if (frame.event === null) this.emitIssue({ kind: 'unparseable_payload', channel: 'user' })
        break
      case 'gap':
        // The server noticed dropped events: backfill from our own cursor and let
        // the app dedupe by id/seq.
        this.emitIssue({ kind: 'gap', room_id: frame.room_id, from: frame.from, to: frame.to })
        if (this.rooms.has(frame.room_id)) {
          this.resync(
            frame.room_id,
            Number.isFinite(frame.from) ? frame.from - 1 : 0,
          )
        }
        break
      case 'resync_required': {
        // `access revoked` means the server already dropped the subscription
        // (ws/mod.rs:318-341); anything else (currently `lagged`, :289-298) means
        // it also forgot its cursor for this room, so the client must re-pull.
        const dropped = frame.reason === 'access revoked'
        if (dropped) {
          this.dropRoomIntent(frame.room_id)
        } else if (!this.lastSeq.has(frame.room_id)) {
          this.emitIssue({ kind: 'cursor_lost', room_id: frame.room_id, reason: frame.reason })
        }
        this.emitIssue({
          kind: 'resync_required',
          room_id: frame.room_id,
          reason: frame.reason,
          dropped,
        })
        if (!dropped && this.rooms.has(frame.room_id)) {
          this.resync(frame.room_id, this.lastSeq.get(frame.room_id) ?? 0)
        }
        break
      }
      case 'error':
        this.applyServerError(frame.message)
        break
      case 'pong':
        // Pongs arrive in ping order, so the head barrier is fully settled: every
        // response (including errors) for its frames has already been delivered.
        this.barriers.shift()
        break
      case 'bye':
        // Only `idle timeout` / `internal error` are produced, both retriable, so
        // `onclose` keeps driving the reconnect while `started`.
        this.emitIssue({ kind: 'server_bye', message: frame.message })
        break
    }
  }

  private applyHistory(roomId: string, messages: readonly { seq: number }[]): void {
    if (messages.length > 0) {
      let max = this.lastSeq.get(roomId) ?? 0
      for (const message of messages) {
        const seq = asSeq(message.seq)
        if (seq !== undefined && seq > max) max = seq
      }
      this.lastSeq.set(roomId, max)
    }
    // `history` only ever follows an authorised read, so any subscribe still
    // pending for this room cannot have failed on permissions.
    this.settleRoom(roomId)

    if (!this.awaitingResync.has(roomId)) return
    const pages = (this.resyncPages.get(roomId) ?? 0) + 1
    if (messages.length < this.historyLimitHint) {
      this.endResync(roomId)
      return
    }
    if (pages > this.maxResyncPages) {
      this.endResync(roomId)
      this.emitIssue({
        kind: 'history_truncated',
        room_id: roomId,
        pages: pages - 1,
        page_size: this.historyLimitHint,
      })
      return
    }
    // A full page means more may be ahead: continue the forward walk.
    this.resyncPages.set(roomId, pages)
    this.write({ op: 'resync', room_id: roomId, after_seq: this.lastSeq.get(roomId) ?? 0 })
  }

  private applyRoomEvent(roomId: string, event: RoomEvent | null): void {
    if (event === null) {
      this.emitIssue({ kind: 'unparseable_payload', channel: 'room', room_id: roomId })
      return
    }
    const seq = roomEventSeq(event)
    if (seq !== undefined && seq > (this.lastSeq.get(roomId) ?? 0)) {
      this.lastSeq.set(roomId, seq)
    }
    if (event.kind === 'typing') this.typingRegistry.record(event)
    this.settleRoom(roomId)
  }

  private applyServerError(message: string): void {
    const error = classifyWsError(message)
    const roomId = isSubscribeError(error) ? this.attributeSubscribeError() : undefined
    if (roomId !== undefined) this.dropRoomIntent(roomId)
    this.emitIssue({ kind: 'server_error', error, message, room_id: roomId })
  }

  /* ------------------------------------------------------ intent bookkeeping */

  private dropRoomIntent(roomId: string): void {
    this.rooms.delete(roomId)
    this.forgetRoom(roomId)
  }

  private forgetRoom(roomId: string): void {
    this.forcedResync.delete(roomId)
    this.awaitingResync.delete(roomId)
    this.resyncPages.delete(roomId)
    this.lastSeq.delete(roomId)
    this.typingRegistry.clear(roomId)
    // Keep the newest epoch: it may still sit in an un-acked barrier, and the
    // server's `error` reply carries no room_id, so forgetting it would make
    // attribution impossible.
    const keep = this.subscribeEpoch.get(roomId)
    this.subscribeEpoch.clear()
    if (keep !== undefined) this.subscribeEpoch.set(roomId, keep)
    for (const barrier of this.barriers) {
      if (keep !== undefined && barrier.rooms.get(roomId) === keep) continue
      barrier.rooms.delete(roomId)
    }
    if (keep !== undefined && this.barrierRooms.get(roomId) === keep) {
      this.barrierRooms.delete(roomId)
    }
  }

  private endResync(roomId: string): void {
    this.awaitingResync.delete(roomId)
    this.resyncPages.delete(roomId)
    this.forcedResync.delete(roomId)
  }

  /** Remember a resync that could not be written, so `ready` can replay it. */
  private startResync(roomId: string, afterSeq: number): void {
    if (this.write({ op: 'resync', room_id: roomId, after_seq: afterSeq })) return
    this.forcedResync.add(roomId)
  }

  /* ------------------------------------------------------- subscribe barriers */

  private resubscribeRooms(): void {
    for (const roomId of this.rooms) {
      const afterSeq = this.lastSeq.get(roomId)
      const forced = this.forcedResync.has(roomId)
      // Skip the snapshot when we already know the cursor; the explicit resync
      // below fetches exactly the missed range.
      this.sendSubscribe(roomId, afterSeq === undefined)
      if (afterSeq !== undefined || forced) {
        this.resyncPages.delete(roomId)
        this.awaitingResync.add(roomId)
        this.write({ op: 'resync', room_id: roomId, after_seq: afterSeq ?? 0 })
      }
    }
  }

  private sendSubscribe(roomId: string, resync: boolean): void {
    if (!this.rooms.has(roomId)) return
    const epoch = (this.subscribeEpoch.get(roomId) ?? 0) + 1
    this.subscribeEpoch.set(roomId, epoch)
    if (!this.write({ op: 'subscribe', room_id: roomId, resync })) return
    if (!this.correlateSubscribeErrors) return
    // One barrier per subscribe keeps error attribution unambiguous.
    this.barrierRooms.set(roomId, epoch)
    this.armPingBarrier()
  }

  /** Close the current barrier group with a ping; its `pong` settles the group. */
  private armPingBarrier(): void {
    const rooms = this.barrierRooms
    this.barrierRooms = new Map()
    if (this.write({ op: 'ping' })) {
      this.barriers.push({ sent_at: Date.now(), rooms })
      return
    }
    // Nothing reached the server, so keep the intents for the next round.
    for (const entry of rooms) this.barrierRooms.set(entry[0], entry[1])
  }

  private settleRoom(roomId: string): void {
    const epoch = this.subscribeEpoch.get(roomId)
    if (epoch === undefined) return
    this.subscribeEpoch.delete(roomId)
    for (const barrier of this.barriers) {
      if (barrier.rooms.get(roomId) === epoch) barrier.rooms.delete(roomId)
    }
    if (this.barrierRooms.get(roomId) === epoch) this.barrierRooms.delete(roomId)
  }

  /**
   * The server answers client frames strictly in order, so an `error` belongs to
   * the oldest un-acked barrier. Attribute only when that barrier holds exactly
   * one subscribe; otherwise nothing is guessed.
   */
  private attributeSubscribeError(): string | undefined {
    const head = this.barriers[0]
    if (!head || head.rooms.size !== 1) return undefined
    const entry = [...head.rooms.entries()][0]
    if (!entry) return undefined
    const roomId = entry[0]
    const epoch = entry[1]
    head.rooms.delete(roomId)
    if (this.subscribeEpoch.get(roomId) !== epoch) return undefined
    return roomId
  }

  private resetBarrierState(): void {
    this.barriers = []
    this.barrierRooms = new Map()
  }

  /* ------------------------------------------------------------------ timers */

  private startPingLoop(): void {
    this.clearPingTimer()
    this.pingTimer = setInterval(() => {
      const ws = this.ws
      if (!ws || ws.readyState !== WebSocket.OPEN) return
      const head = this.barriers[0]
      if (head && Date.now() - head.sent_at > this.pongTimeoutMs) {
        // A ping never came back, so the socket is half-open. Room traffic is no
        // proof of liveness: only an outstanding barrier can trip this.
        const outstanding = this.barriers.length
        this.barriers = []
        this.setStatus('reconnecting')
        ws.close(1000)
        this.emitIssue({ kind: 'pong_timeout', outstanding_pings: outstanding })
        return
      }
      this.armPingBarrier()
    }, this.pingIntervalMs)
  }

  private scheduleReconnect(): void {
    if (!this.started || this.reconnectTimer !== undefined) return
    this.setStatus('reconnecting')
    const exponential = this.minReconnectDelayMs * 2 ** Math.min(this.reconnectAttempts, 16)
    this.reconnectAttempts += 1
    const delay = Math.min(exponential, this.maxReconnectDelayMs) * (0.8 + Math.random() * 0.4)
    this.reconnectTimer = setTimeout(() => {
      this.reconnectTimer = undefined
      void this.connect()
    }, delay)
  }

  /* ------------------------------------------------------------------ write */

  private canWrite(): boolean {
    const ws = this.ws
    return ws !== null && ws.readyState === WebSocket.OPEN
  }

  /**
   * Write one frame. There is deliberately no send queue: `subscribe` intents are
   * replayed from `ready`, `resync` is replayed through `forcedResync`, and
   * `typing` / `ping` are ephemeral. Queuing them used to mean both duplicate
   * history bursts and silent eviction of a queued subscribe.
   */
  private write(frame: ClientFrame): boolean {
    if (!this.canWrite()) return false
    const ws = this.ws
    if (!ws) return false
    try {
      ws.send(JSON.stringify(frame))
      return true
    } catch {
      return false
    }
  }

  /* ----------------------------------------------------------------- status */

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

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

  private clearReconnectTimer(): void {
    if (this.reconnectTimer !== undefined) {
      clearTimeout(this.reconnectTimer)
      this.reconnectTimer = undefined
    }
  }

  private clearPingTimer(): void {
    if (this.pingTimer !== undefined) {
      clearInterval(this.pingTimer)
      this.pingTimer = undefined
    }
  }
}

function sameOriginWsUrl(): string {
  const proto = window.location.protocol === 'https:' ? 'wss' : 'ws'
  return `${proto}://${window.location.host}${DEFAULT_PATH}`
}