payment-platform/frontend/scripts/verify-order-stream.mjs

#!/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);