blake-john/agent-config/websocket-engineer

import { createServer } from "http";
import { Server } from "socket.io";
import { createAdapter } from "@socket.io/redis-adapter";
import { createClient } from "redis";
import jwt from "jsonwebtoken";
const httpServer = createServer();
const io = new Server(httpServer, {
  cors: { origin: process.env.ALLOWED_ORIGIN, credentials: true },
  pingTimeout: 20000,
  pingInterval: 25000,
});
// 认证中间件 — 在建立连接前执行
io.use((socket, next) => {
  const token = socket.handshake.auth.token;
  if (!token) return next(new Error("Authentication required"));
  try {
    socket.data.user = jwt.verify(token, process.env.JWT_SECRET);
    next();
  } catch {
    next(new Error("Invalid token"));
  }
});
// 用于水平扩展的 Redis adapter
const pubClient = createClient({ url: process.env.REDIS_URL });
const subClient = pubClient.duplicate();
await Promise.all([pubClient.connect(), subClient.connect()]);
io.adapter(createAdapter(pubClient, subClient));
io.on("connection", (socket) => {
  const { userId } = socket.data.user;
  console.log(`connected: ${userId} (${socket.id})`);
  // 在线状态：标记用户在线
  pubClient.hSet("presence", userId, socket.id);
  socket.on("join-room", (roomId) => {
    socket.join(roomId);
    socket.to(roomId).emit("user-joined", { userId });
  });
  socket.on("message", ({ roomId, text }) => {
    io.to(roomId).emit("message", { userId, text, ts: Date.now() });
  });
  socket.on("disconnect", () => {
    pubClient.hDel("presence", userId);
    console.log(`disconnected: ${userId}`);
  });
});
httpServer.listen(3000);

import { io } from "socket.io-client";
const socket = io("wss://api.example.com", {
  auth: { token: getAuthToken() },
  reconnection: true,
  reconnectionAttempts: 10,
  reconnectionDelay: 1000,       // 初始延迟（毫秒）
  reconnectionDelayMax: 30000,   // 上限 30 秒
  randomizationFactor: 0.5,      // 抖动避免惊群效应
});
// 断线时排队消息
let messageQueue = [];
socket.on("connect", () => {
  console.log("connected:", socket.id);
  // 发送队列中的消息
  messageQueue.forEach((msg) => socket.emit("message", msg));
  messageQueue = [];
});
socket.on("disconnect", (reason) => {
  console.warn("disconnected:", reason);
  if (reason === "io server disconnect") socket.connect(); // 手动重连
});
socket.on("connect_error", (err) => {
  console.error("connection error:", err.message);
});
function sendMessage(roomId, text) {
  const msg = { roomId, text };
  if (socket.connected) {
    socket.emit("message", msg);
  } else {
    messageQueue.push(msg); // 缓存直到重连
  }
}