observer/web/src/api/socket.ts

/**
 * The live connection.
 *
 * One WebSocket for the whole tab. It reconnects on its own, resubscribes to
 * whatever the page was watching, and hands React a snapshot object that only
 * changes identity when something in it actually changed — so `useSyncExternalStore`
 * re-renders exactly the components whose data moved.
 *
 * Why not observables: this needs one stream, one reducer and one subscriber
 * list. RxJS would add a dependency and an idiom to every component to express
 * what `useSyncExternalStore` already does natively, and React's own store
 * contract is the thing that gets concurrent rendering right.
 */

import type { ChainInfo } from './generated/ChainInfo'
import type { ChainSummary } from './generated/ChainSummary'
import type { ClientMessage } from './generated/ClientMessage'
import type { LeaderboardRow } from './generated/LeaderboardRow'
import type { RecentBlock } from './generated/RecentBlock'
import type { ServerMessage } from './generated/ServerMessage'
import type { Window as WindowName } from './generated/Window'

/** How the socket is doing, for the status light in the header. */
export type Connection = 'connecting' | 'live' | 'reconnecting' | 'offline'

/** Everything the UI renders, in one immutable object. */
export interface ObserverState {
  connection: Connection
  /** Every configured chain, in the order the backend lists them. */
  chains: ChainInfo[]
  /** Which chain the page is watching. */
  chain: string | null
  /** Which leaderboard window. */
  window: WindowName
  summary: ChainSummary | null
  leaderboard: LeaderboardRow[]
  /**
   * What the window actually spanned, by the chain's own clock.
   *
   * `null` until the first snapshot, or where too few blocks in the window
   * carry a timestamp to span anything. Never inferred from the block count
   * and the current interval — that is the arithmetic that told readers 3,600
   * Planck blocks was six hours when it was twenty-nine.
   */
  windowSeconds: number | null
  /** Newest first — the order the ticker renders in. */
  blocks: RecentBlock[]
  /** True once the first snapshot for the current subscription has landed. */
  ready: boolean
}

const INITIAL: ObserverState = {
  connection: 'connecting',
  chains: [],
  chain: null,
  window: '3600',
  summary: null,
  leaderboard: [],
  windowSeconds: null,
  blocks: [],
  ready: false,
}

/** Blocks kept in the ticker. Matches the backend's replay length. */
const TICKER_LIMIT = 40

/**
 * Reconnect backoff. Capped low: this is a live scoreboard, and a miner
 * watching their standing would rather the page retry briskly than back off to
 * a minute after a brief network blip.
 */
const RECONNECT_MIN_MS = 1_000
const RECONNECT_MAX_MS = 15_000

/**
 * Application-level keepalive.
 *
 * Browsers cannot send WebSocket ping frames from JavaScript, so a socket that
 * has been silently dropped by an intermediary looks identical to a quiet chain
 * — and on a chain averaging a block every few seconds, "quiet" is a real
 * state. This ping is the only way the tab can tell the two apart.
 */
const PING_INTERVAL_MS = 25_000

function socketUrl(): string {
  const configured = import.meta.env.VITE_WS_URL
  if (configured) return configured
  // Same origin by default: in production nginx serves the bundle and proxies
  // /v1 to the API, and in development Vite proxies it. Deriving the URL means
  // no build-time value to get wrong per environment.
  const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'
  return `${protocol}//${window.location.host}/v1/ws`
}

/**
 * The store. One instance per tab, created in `main.tsx`.
 */
export class Observer {
  private state: ObserverState = INITIAL
  private listeners = new Set<() => void>()
  private socket: WebSocket | null = null
  private reconnectDelay = RECONNECT_MIN_MS
  private reconnectTimer: number | null = null
  private pingTimer: number | null = null
  private closed = false

  /** `useSyncExternalStore` subscribe. */
  subscribe = (listener: () => void): (() => void) => {
    this.listeners.add(listener)
    return () => {
      this.listeners.delete(listener)
    }
  }

  /**
   * `useSyncExternalStore` snapshot.
   *
   * Must return a stable reference between changes — returning a fresh object
   * here would re-render every subscriber on every tick and, in React 19, throw
   * for an unstable snapshot.
   */
  getSnapshot = (): ObserverState => this.state

  /** Open the socket. Idempotent. */
  connect(): void {
    if (this.socket || this.closed) return
    const socket = new WebSocket(socketUrl())
    this.socket = socket

    socket.onopen = () => {
      this.reconnectDelay = RECONNECT_MIN_MS
      this.patch({ connection: 'live' })
      // The server does not remember subscriptions across sockets, so the
      // reconnect has to restate what this tab is watching. Without it a
      // recovered connection would deliver the chain list and then nothing.
      if (this.state.chain) {
        this.send({ type: 'subscribe', chain: this.state.chain, window: this.state.window })
      }
      this.pingTimer = window.setInterval(() => this.send({ type: 'ping' }), PING_INTERVAL_MS)
    }

    socket.onmessage = (event: MessageEvent<string>) => {
      let message: ServerMessage
      try {
        message = JSON.parse(event.data) as ServerMessage
      } catch {
        return
      }
      this.apply(message)
    }

    socket.onclose = () => {
      this.socket = null
      if (this.pingTimer !== null) {
        window.clearInterval(this.pingTimer)
        this.pingTimer = null
      }
      if (this.closed) return
      this.patch({ connection: 'reconnecting', ready: false })
      this.reconnectTimer = window.setTimeout(() => this.connect(), this.reconnectDelay)
      this.reconnectDelay = Math.min(this.reconnectDelay * 2, RECONNECT_MAX_MS)
    }

    // `onerror` is always followed by `onclose`, so reconnection is handled
    // there and this only exists to stop the browser logging an unhandled one.
    socket.onerror = () => {}
  }

  /** Close for good. Used when the app unmounts. */
  disconnect(): void {
    this.closed = true
    if (this.reconnectTimer !== null) window.clearTimeout(this.reconnectTimer)
    if (this.pingTimer !== null) window.clearInterval(this.pingTimer)
    this.socket?.close()
    this.socket = null
  }

  /** Watch a chain at a window. Safe to call before the socket is open. */
  watch(chain: string, windowName: WindowName): void {
    if (this.state.chain === chain && this.state.window === windowName) return
    // Clear the previous chain's data rather than letting it linger under the
    // new chain's name: a stale leaderboard under a different heading is worse
    // than an empty one, because it looks correct.
    this.patch({
      chain,
      window: windowName,
      summary: null,
      leaderboard: [],
      windowSeconds: null,
      blocks: [],
      ready: false,
    })
    this.send({ type: 'subscribe', chain, window: windowName })
  }

  private send(message: ClientMessage): void {
    if (this.socket?.readyState === WebSocket.OPEN) {
      this.socket.send(JSON.stringify(message))
    }
  }

  private apply(message: ServerMessage): void {
    switch (message.type) {
      case 'chains':
        this.patch({ chains: message.chains })
        break

      case 'snapshot':
        // A snapshot for a chain the page has since navigated away from would
        // overwrite the current one; ignore it rather than flicker.
        if (message.chain !== this.state.chain) return
        this.patch({
          summary: message.summary,
          leaderboard: message.leaderboard,
          windowSeconds: message.window_seconds,
          blocks: [...message.recent_blocks].reverse(),
          ready: true,
        })
        break

      case 'summary':
        if (message.chain !== this.state.chain) return
        this.patch({ summary: message.summary })
        break

      case 'leaderboard':
        if (message.chain !== this.state.chain || message.window !== this.state.window) return
        this.patch({ leaderboard: message.rows, windowSeconds: message.window_seconds })
        break

      case 'block': {
        if (message.chain !== this.state.chain) return
        // A reorg replaces the block at a height rather than appending a second
        // one, so the ticker matches the chain rather than accumulating both
        // sides of every fork.
        const rest = this.state.blocks.filter((b) => b.height !== message.block.height)
        this.patch({ blocks: [message.block, ...rest].slice(0, TICKER_LIMIT) })
        break
      }

      case 'chain_status':
        this.patch({
          chains: this.state.chains.map((c) => (c.id === message.chain ? message.info : c)),
        })
        break

      case 'error':
        console.warn('[observer]', message.message)
        break

      case 'pong':
        break
    }
  }

  private patch(next: Partial<ObserverState>): void {
    this.state = { ...this.state, ...next }
    for (const listener of this.listeners) listener()
  }
}