TS 实时通信：WebSocket 与 SSE 的类型安全设计与生产实践

type ClientEvent = "chat.send" | "room.join" | "room.leave" | "typing"
type ServerEvent = "chat.message" | "room.joined" | "user.typing" | "error"

interface ServerPayloads {
  "chat.message": { id: string; roomId: string; text: string; at: number }
  "room.joined": { roomId: string; members: string[] }
  "user.typing": { roomId: string; userId: string }
  error: { code: string; message: string }
}
type Handler<K extends keyof ServerPayloads> = (payload: ServerPayloads[K]) => void

class TypedSocket {
  constructor(private ws: WebSocket) {}
  on<K extends keyof ServerPayloads>(ev: K, fn: Handler<K>) {
    this.ws.addEventListener("message", (e) => {
      const msg = JSON.parse(e.data) as { type: K; data: ServerPayloads[K] }
      if (msg.type === ev) fn(msg.data)
    })
  }
}

app.get("/events", (req, res) => {
  res.writeHead(200, {
    "Content-Type": "text/event-stream",
    "Cache-Control": "no-cache, no-transform",
    Connection: "keep-alive",
  })
  const send = (event: string, data: unknown, id?: string) => {
    if (id) res.write(`id: ${id}\n`)
    res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`)
  }
})

const es = new EventSource("/events")
es.addEventListener("chat.message", (e) => {
  const payload = JSON.parse((e as MessageEvent).data)
  render(payload)
})
es.onerror = () => { /* 浏览器会自动按 Retry-After 重连 */ }

type ClientMessage =
  | { type: "chat.send"; roomId: string; text: string }
  | { type: "room.join"; roomId: string }
  | { type: "room.leave"; roomId: string }
  | { type: "typing"; roomId: string }

function handle(msg: ClientMessage) {
  switch (msg.type) {
    case "chat.send": return sendChat(msg)
    case "room.join": return joinRoom(msg)
    case "room.leave": return leaveRoom(msg)
    case "typing": return typing(msg)
    default: return assertNever(msg)   // 新增分支时此处编译报错
  }
}
const assertNever = (x: never): never => { throw new Error(String(x)) }

function startHeartbeat(ws: WebSocket, interval = 30_000) {
  let alive = true
  ws.on("pong", () => { alive = true })
  const timer = setInterval(() => {
    if (!alive) { ws.terminate(); return }
    alive = false
    ws.ping()
  }, interval)
  ws.on("close", () => clearInterval(timer))
}

function backoff(attempt: number, base = 500, cap = 30_000) {
  const exp = Math.min(cap, base * 2 ** attempt)
  const jitter = exp * (0.5 + Math.random() * 0.5)  // 抖动避免惊群
  return jitter
}

interface Envelope<T> { seq: number; type: string; data: T }

// 客户端重连
socket.emit("resume", { roomId, lastSeq })
// 服务端
const missed = history.filter((m) => m.seq > lastSeq)
missed.forEach((m) => send(m))

class Sender {
  private queue: string[] = []
  private flushing = false
  constructor(private ws: WebSocket, private high = 1000) {}
  push(msg: string) {
    if (this.queue.length >= this.high) {
      this.dropOldest()   // 或断开连接，按业务选择
    }
    this.queue.push(msg)
    void this.flush()
  }
  private async flush() {
    if (this.flushing) return
    this.flushing = true
    while (this.queue.length) {
      const chunk = this.queue.shift()!
      if (!this.ws.write(chunk)) {
        await once(this.ws, "drain")   // 等待底层缓冲排空
      }
    }
    this.flushing = false
  }
}

type ClientMessage =
  | { type: "sub"; channel: string; params?: Record<string, unknown> }
  | { type: "unsub"; channel: string }
type ServerMessage =
  | { type: "data"; channel: string; data: unknown }
  | { type: "ack"; channel: string; subId: string }

const subscriptions = new Map<string, Map<string, Set<string>>>()
//            channel -> (subId -> clientIds)
function subscribe(channel: string, subId: string, clientId: string) { /* ... */ }
function unsubscribeAll(clientId: string) { /* 断线时清理，防泄漏 */ }

class Room {
  private members = new Set<Client>()
  join(c: Client) { this.members.add(c) }
  leave(c: Client) { this.members.delete(c) }
  broadcast(msg: ServerMessage, except?: Client) {
    const raw = JSON.stringify(msg)
    for (const m of this.members) if (m !== except) m.send(raw)
  }
}