feat(订单实时): 前端 SSE 单例工厂（退避重连 + 看门狗 + 重连补拉）

frontend/scripts/verify-order-stream.mjs

@@ -0,0 +1,289 @@
+#!/usr/bin/env node
+/**
+ * 订单实时流（src/utils/orderStream.ts）的执行级验证。
+ * 用法：cd frontend && node scripts/verify-order-stream.mjs
+ *
+ * 断言失败打印 FAIL 并以非 0 退出；每条 PASS 都打名字（读数要看得见）。
+ */
+import { build } from "esbuild";
+import assert from "node:assert/strict";
+import path from "node:path";
+import { fileURLToPath, pathToFileURL } from "node:url";
+import fs from "node:fs/promises";
+import os from "node:os";
+
+const here = path.dirname(fileURLToPath(import.meta.url));
+const root = path.resolve(here, "..");
+const tmp = await fs.mkdtemp(path.join(os.tmpdir(), "verify-order-stream-"));
+const bundle = path.join(tmp, "orderStream.mjs");
+
+await build({
+  entryPoints: [path.join(root, "src/utils/orderStream.ts")],
+  outfile: bundle,
+  bundle: true,
+  format: "esm",
+  platform: "neutral",
+  logLevel: "error"
+});
+
+const { createOrderStream } = await import(pathToFileURL(bundle).href);
+
+// ─────────────────────────── 假件 ───────────────────────────
+
+class FakeEventSource {
+  constructor(url, at) {
+    this.url = url;
+    this.at = at;              // 建连时刻（假时钟读数）
+    this.closed = false;
+    this.listeners = new Map();
+  }
+  addEventListener(type, fn) {
+    if (!this.listeners.has(type)) this.listeners.set(type, []);
+    this.listeners.get(type).push(fn);
+  }
+  close() {
+    this.closed = true;
+  }
+  fire(type, payload = {}) {
+    for (const fn of this.listeners.get(type) || []) fn(payload);
+  }
+}
+
+function fakeClock() {
+  let t = 0;
+  const jobs = [];
+  let nextId = 1;
+  return {
+    now: () => t,
+    setTimer: (fn, ms) => {
+      const id = nextId++;
+      jobs.push({ id, at: t + ms, fn });
+      return id;
+    },
+    clearTimer: id => {
+      const i = jobs.findIndex(j => j.id === id);
+      if (i >= 0) jobs.splice(i, 1);
+    },
+    advance(ms) {
+      const end = t + ms;
+      for (;;) {
+        const due = jobs.filter(j => j.at <= end).sort((a, b) => a.at - b.at)[0];
+        if (!due) break;
+        jobs.splice(jobs.indexOf(due), 1);
+        t = due.at;
+        due.fn();
+      }
+      t = end;
+    }
+  };
+}
+
+function harness() {
+  const clock = fakeClock();
+  const built = [];
+  const state = { urlCalls: 0 };
+  const stream = createOrderStream({
+    url: () => {
+      state.urlCalls += 1;
+      return `http://x/api/v1/payment/stream/orders?token=T${state.urlCalls}`;
+    },
+    createEventSource: url => {
+      const es = new FakeEventSource(url, clock.now());
+      built.push(es);
+      return es;
+    },
+    now: clock.now,
+    setTimer: clock.setTimer,
+    clearTimer: clock.clearTimer,
+    backoffBaseMs: 1000,
+    backoffMaxMs: 30000,
+    watchdogMs: 50000
+  });
+  return { clock, built, state, stream };
+}
+
+let failed = 0;
+function check(name, fn) {
+  try {
+    fn();
+    console.log("  PASS  " + name);
+  } catch (e) {
+    failed += 1;
+    console.error("  FAIL  " + name + "\n        " + (e && e.message));
+  }
+}
+
+// ─────────────────────────── 断言 ───────────────────────────
+
+check("subscribe 建一条连接，URL 来自 url()", () => {
+  const { built, stream } = harness();
+  const off = stream.subscribe("lsy_order", () => {});
+  assert.equal(built.length, 1);
+  assert.equal(built[0].url, "http://x/api/v1/payment/stream/orders?token=T1");
+  off();
+});
+
+check("changed 帧只触发该通道的订阅者", () => {
+  const { built, stream } = harness();
+  let order = 0;
+  let refund = 0;
+  const off1 = stream.subscribe("lsy_order", () => { order += 1; });
+  const off2 = stream.subscribe("lsy_refund", () => { refund += 1; });
+
+  built[0].fire("changed", { data: JSON.stringify({ channel: "lsy_order" }) });
+  assert.equal(order, 1);
+  assert.equal(refund, 0, "别的通道的帧不许触发本通道");
+
+  built[0].fire("changed", { data: JSON.stringify({ channel: "har_order" }) });
+  assert.equal(order, 1);
+  assert.equal(refund, 0, "没人订阅的通道：静默丢弃");
+
+  off1();
+  off2();
+});
+
+check("坏帧不抛异常、不影响后续帧", () => {
+  const { built, stream } = harness();
+  let n = 0;
+  const off = stream.subscribe("lsy_order", () => { n += 1; });
+
+  built[0].fire("changed", { data: "{oops" });
+  built[0].fire("changed", {});
+  built[0].fire("changed", { data: JSON.stringify({}) });
+  built[0].fire("changed", { data: JSON.stringify({ channel: "lsy_order" }) });
+
+  assert.equal(n, 1, "只有最后那条合法帧该生效");
+  off();
+});
+
+check("首次 open 不补拉；重连后的 open 补拉一次", () => {
+  const { clock, built, stream } = harness();
+  let resync = 0;
+  const offR = stream.onResync(() => { resync += 1; });
+  const off = stream.subscribe("lsy_order", () => {});
+
+  built[0].fire("open", {});
+  assert.equal(resync, 0, "首次连接时页面自己正在加载，不该再多拉一次");
+
+  built[0].fire("error", {});
+  clock.advance(1000);
+  assert.equal(built.length, 2);
+  built[1].fire("open", {});
+  assert.equal(resync, 1, "断线期间漏掉的帧必须靠重连补拉兜回来");
+  assert.equal(built[0].closed, true, "旧连接要显式 close（浏览器自带重连与我们的退避不能两条腿跑）");
+
+  off();
+  offR();
+});
+
+check("退避 1s→2s→4s→…→30s 封顶", () => {
+  const { clock, built, stream } = harness();
+  const off = stream.subscribe("lsy_order", () => {});
+
+  let expected = 1;
+  for (const delay of [1000, 2000, 4000, 8000, 16000, 30000, 30000]) {
+    built[built.length - 1].fire("error", {});
+    clock.advance(delay - 1);
+    assert.equal(built.length, expected, `${delay}ms 还没到就重连了`);
+    clock.advance(1);
+    assert.equal(built.length, expected + 1, `${delay}ms 到了还没重连`);
+    expected += 1;
+  }
+  off();
+});
+
+check("成功连接后退避重置回 1s（连续断线不把惩罚累积上去）", () => {
+  const { clock, built, stream } = harness();
+  const off = stream.subscribe("lsy_order", () => {});
+
+  built[0].fire("error", {});                 // 第一次断线：1s
+  clock.advance(1000);
+  assert.equal(built.length, 2);
+
+  built[1].fire("error", {});                 // 还没连上过：2s
+  clock.advance(1999);
+  assert.equal(built.length, 2, "还没到 2000ms 就重连了");
+  clock.advance(1);
+  assert.equal(built.length, 3);
+
+  built[2].fire("open", {});                  // 连上了 —— 下一次断线必须回到 1s
+  built[2].fire("error", {});
+  clock.advance(999);
+  assert.equal(built.length, 3, "999ms 就重连了（退避被重置过头？）");
+  clock.advance(1);
+  assert.equal(built.length, 4,
+    "成功连接后退避没重置：仍在按上一轮的 4s 计，页面会在毫无必要的情况下越等越久");
+
+  off();
+});
+
+check("每次重连都重新调 url()（令牌换新）", () => {
+  const { clock, built, state, stream } = harness();
+  const off = stream.subscribe("lsy_order", () => {});
+  built[0].fire("open", {});
+  assert.equal(state.urlCalls, 1);
+
+  built[0].fire("error", {});
+  clock.advance(1000);
+  assert.equal(state.urlCalls, 2, "重连必须重新拼 URL —— 旧令牌可能已经失效");
+  assert.notEqual(built[0].url, built[1].url);
+  off();
+});
+
+check("50s 没有任何帧（ping 也算）→ 弃掉重建", () => {
+  const { clock, built, stream } = harness();
+  const off = stream.subscribe("lsy_order", () => {});
+  built[0].fire("open", {});
+
+  clock.advance(20000);
+  built[0].fire("ping", {});          // 服务端 20s 一个 ping：看门狗被喂饱
+
+  clock.advance(49999);
+  assert.equal(built[0].closed, false, "ping 之后不到 50s 不该重建");
+  clock.advance(1);
+  assert.equal(built[0].closed, true, "50s 没有帧 = 半死连接（TCP 还在、对端不推了），必须重建");
+  clock.advance(1000);
+  assert.equal(built.length, 2);
+  off();
+});
+
+check("stop() 关连接、清定时器，此后不再重连", () => {
+  const { clock, built, stream } = harness();
+  const off = stream.subscribe("lsy_order", () => {});
+  built[0].fire("open", {});
+
+  stream.stop();
+
+  assert.equal(built[0].closed, true);
+  clock.advance(600000);
+  assert.equal(built.length, 1, "stop 之后不许再有任何新连接（登出后 token 已清，重连只会被 403 挡回）");
+  off();
+});
+
+check("最后一个订阅者退订 → 连接关闭", () => {
+  const { built, stream } = harness();
+  const off1 = stream.subscribe("lsy_order", () => {});
+  const off2 = stream.subscribe("lsy_refund", () => {});
+
+  off1();
+  assert.equal(built[0].closed, false, "还有订阅者时不许断");
+  off2();
+  assert.equal(built[0].closed, true);
+});
+
+check("三个通道名的线上契约（与后端 OrderEventPublisher 逐字一致）", () => {
+  const CHANNELS = ["lsy_order", "lsy_refund", "har_order"];
+  const { built, stream } = harness();
+  const hits = { lsy_order: 0, lsy_refund: 0, har_order: 0 };
+  const offs = CHANNELS.map(c => stream.subscribe(c, () => { hits[c] += 1; }));
+
+  for (const c of CHANNELS) {
+    built[0].fire("changed", { data: JSON.stringify({ channel: c }) });
+  }
+
+  assert.deepEqual(hits, { lsy_order: 1, lsy_refund: 1, har_order: 1 });
+  offs.forEach(off => off());
+});
+
+console.log(failed === 0 ? "\n全部通过" : `\n${failed} 条失败`);
+if (failed > 0) process.exit(1);

frontend/src/utils/orderStream.ts

@@ -0,0 +1,198 @@
+/**
+ * 订单实时信号（SSE）—— 与后端 `/api/v1/payment/stream/orders` 的单条长连接。
+ *
+ * 三条纪律（每条都有对应断言，见 scripts/verify-order-stream.mjs）：
+ * 1. **零用户可见状态**：不显示连接状态、不弹提示、断线也不跳登录页。
+ * 2. **url() 每次（重）连现取**：令牌会换（滑动过期、重新登录），只在建连时算一次
+ *    等于拿旧令牌反复撞墙。
+ * 3. **重连成功后补拉一次**（resync）：断线期间漏掉的帧靠它兜；首次 open 不补拉
+ *    ——那会儿页面自己正在加载。
+ *
+ * 本模块是**纯的**（零 import、依赖全注入），所以 node 脚本能原样打包它来跑；
+ * 应用级单例在 `@/utils/orderStreamInstance`。
+ */
+
+export interface EventSourceLike {
+  addEventListener(type: string, fn: (event: any) => void): void;
+  close(): void;
+}
+
+export interface OrderStreamDeps {
+  /** 完整 URL（`?token=` 由调用方拼好）。**每次（重）连都会重新调**。 */
+  url: () => string;
+  createEventSource: (url: string) => EventSourceLike;
+  now?: () => number;
+  setTimer?: (fn: () => void, ms: number) => number;
+  clearTimer?: (id: number) => void;
+  /** 退避起点，默认 1000ms。 */
+  backoffBaseMs?: number;
+  /** 退避上限，默认 30000ms。 */
+  backoffMaxMs?: number;
+  /** 看门狗阈值，默认 50000ms（服务端 20s 一个 ping：两个没来就重建）。 */
+  watchdogMs?: number;
+}
+
+export interface OrderStream {
+  /** 订阅一个通道；返回退订函数。第一个订阅者触发建连，最后一个退订触发断连。 */
+  subscribe(channel: string, handler: () => void): () => void;
+  /** 重连成功（非首次）时回调一次，用来补拉。 */
+  onResync(handler: () => void): () => void;
+  /** 断连且不再重连（登出时调用）。 */
+  stop(): void;
+  /** 测试用：当前有没有连接。**页面上不许显示**。 */
+  isRunning(): boolean;
+}
+
+const DEFAULT_BACKOFF_BASE_MS = 1000;
+const DEFAULT_BACKOFF_MAX_MS = 30000;
+const DEFAULT_WATCHDOG_MS = 50000;
+
+export function createOrderStream(deps: OrderStreamDeps): OrderStream {
+  const backoffBase = deps.backoffBaseMs ?? DEFAULT_BACKOFF_BASE_MS;
+  const backoffMax = deps.backoffMaxMs ?? DEFAULT_BACKOFF_MAX_MS;
+  const watchdogMs = deps.watchdogMs ?? DEFAULT_WATCHDOG_MS;
+  const setTimer = deps.setTimer ?? ((fn, ms) => setTimeout(fn, ms) as unknown as number);
+  const clearTimer = deps.clearTimer ?? (id => clearTimeout(id as unknown as ReturnType<typeof setTimeout>));
+
+  const handlers = new Map<string, Set<() => void>>();
+  const resyncHandlers = new Set<() => void>();
+
+  let source: EventSourceLike | null = null;
+  let reconnectTimer: number | null = null;
+  let watchdogTimer: number | null = null;
+  let backoff = backoffBase;
+  let everConnected = false;
+
+  function clearTimers(): void {
+    if (reconnectTimer !== null) {
+      clearTimer(reconnectTimer);
+      reconnectTimer = null;
+    }
+    if (watchdogTimer !== null) {
+      clearTimer(watchdogTimer);
+      watchdogTimer = null;
+    }
+  }
+
+  function closeSource(): void {
+    if (source) {
+      try {
+        source.close();
+      } catch {
+        /* 关不上就算了：连接已经死了 */
+      }
+      source = null;
+    }
+  }
+
+  function armWatchdog(): void {
+    if (watchdogTimer !== null) {
+      clearTimer(watchdogTimer);
+    }
+    // 半死连接（TCP 还在、对端不推了）不会自己报 error，只能靠「多久没听到声音」判
+    watchdogTimer = setTimer(dropAndReconnect, watchdogMs);
+  }
+
+  function dropAndReconnect(): void {
+    clearTimers();
+    closeSource();
+    scheduleReconnect();
+  }
+
+  function scheduleReconnect(): void {
+    if (handlers.size === 0) {
+      return; // 没人订阅了就别连（登出后就是这个状态）
+    }
+    const delay = backoff;
+    backoff = Math.min(backoff * 2, backoffMax);
+    reconnectTimer = setTimer(connect, delay);
+  }
+
+  function connect(): void {
+    reconnectTimer = null;
+    const es = deps.createEventSource(deps.url()); // ← 每次（重）连都现取 URL
+    source = es;
+
+    es.addEventListener("open", () => {
+      backoff = backoffBase; // 连上了就重置退避，别把上次的惩罚带进下一次断线
+      armWatchdog();
+      if (everConnected) {
+        for (const fn of resyncHandlers) {
+          fn();
+        }
+      }
+      everConnected = true;
+    });
+
+    es.addEventListener("changed", (event: any) => {
+      armWatchdog();
+      let channel: unknown;
+      try {
+        channel = JSON.parse(event?.data ?? "{}")?.channel;
+      } catch {
+        return; // 坏帧静默丢弃
+      }
+      if (typeof channel !== "string") {
+        return;
+      }
+      const set = handlers.get(channel);
+      if (set) {
+        for (const fn of set) {
+          fn();
+        }
+      }
+    });
+
+    es.addEventListener("ping", () => {
+      armWatchdog();
+    });
+
+    es.addEventListener("error", () => {
+      // 不区分「网络断」与「被 401 挡回」：都是退避重连，永远不弹窗、不跳登录页
+      if (source === es) {
+        dropAndReconnect();
+      }
+    });
+  }
+
+  const stream: OrderStream = {
+    subscribe(channel, handler) {
+      let set = handlers.get(channel);
+      if (!set) {
+        set = new Set();
+        handlers.set(channel, set);
+      }
+      set.add(handler);
+      if (source === null && reconnectTimer === null) {
+        connect();
+      }
+      return () => {
+        const current = handlers.get(channel);
+        current?.delete(handler);
+        if (current && current.size === 0) {
+          handlers.delete(channel);
+        }
+        if (handlers.size === 0) {
+          stream.stop();
+        }
+      };
+    },
+    onResync(handler) {
+      resyncHandlers.add(handler);
+      return () => {
+        resyncHandlers.delete(handler);
+      };
+    },
+    stop() {
+      clearTimers();
+      closeSource();
+      backoff = backoffBase;
+      everConnected = false;
+    },
+    isRunning() {
+      return source !== null;
+    }
+  };
+
+  return stream;
+}