opendance-windows/build/integration/ws-client.mjs

#!/usr/bin/env node
/**
 * WebSocket 契约一致性客户端（T-WIN-020）
 *
 * 独立的 Node 实现，按 contracts/websocket-keypoints.md **v0.2.0** 逐条跑场景，
 * 对发布包里的 Windows `PoseServer` 做契约一致性验证。
 *
 * 为什么不用 Godot 自己的客户端：同引擎内的自测会共享同样的假设
 * （超时判定、分包、文本/二进制边界、关闭语义、子协议协商）。
 * 真实对端是 Android（Kotlin/OkHttp），所以必须用**另一套实现**来对照。
 *
 * 用法：
 *   node build/integration/ws-client.mjs --url ws://127.0.0.1:18765/pose [--skip-slow]
 *
 * 场景（对应契约章节）：
 *   §1/§3.2 子协议协商 + hello → welcome（含 silentTimeoutMs / heartbeatIntervalMs）
 *   §1.1    heartbeatIntervalMs 必须 < silentTimeoutMs / 2
 *   §2      239 字节二进制帧（magic/version/flags/frame_id/33×7B），含 flags 位
 *   §3.4    stats 文本帧
 *   §3.6    应用层 ping → pong（sentAtMs 原样回传）
 *   §5      同 deviceId 新连接替换旧连接：旧连接收到 close(1000, "replaced…")
 *   §1.1    只发 ping 时连接**不会**被静默超时踢掉（19 秒对照，--skip-slow 可跳过）
 *   §1.1/§7 完全不发数据 → 服务端 close(1001, "timeout")
 *   §3.5    bye → 服务端 close(1000)
 *   §3.1/§6 协议版本不匹配 → bye(reason=protocol_mismatch) + close(1002)
 *   §2      非法二进制帧 → 服务端不崩且计入 malformed（由探针报告核对）
 *
 * 退出码：0 全通过 / 1 有失败 / 2 用法或连接问题。
 */

const args = process.argv.slice(2);
function argValue(name, fallback) {
  const i = args.indexOf(name);
  return i >= 0 && i + 1 < args.length ? args[i + 1] : fallback;
}
const URL_ = argValue('--url', 'ws://127.0.0.1:18765/pose');
const SUBPROTOCOL = 'opendance.pose.v1';
const SKIP_SLOW = args.includes('--skip-slow');

const MAGIC = 0x4f44;
const FRAME_BYTES = 239;

let passed = 0;
let failed = 0;
const failures = [];

function check(name, ok, detail = '') {
  if (ok) {
    passed += 1;
    console.log(`  ✓ ${name}`);
  } else {
    failed += 1;
    failures.push(`${name}${detail ? ` — ${detail}` : ''}`);
    console.log(`  ✗ ${name}${detail ? ` — ${detail}` : ''}`);
  }
}

const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

/** 构造契约 §2 的 239 字节关键点帧 */
function buildFrame({ frameId = 0, flags = 0, mirrored = false, lowConf = false, upperBodyOnly = false } = {}) {
  const buf = new ArrayBuffer(FRAME_BYTES);
  const view = new DataView(buf);
  view.setUint16(0, MAGIC, true);            // magic（小端）
  view.setUint8(2, 0x01);                    // version
  let f = flags;
  if (mirrored) f |= 0x01;
  if (lowConf) f |= 0x02;
  if (upperBodyOnly) f |= 0x04;
  view.setUint8(3, f);
  view.setUint32(4, frameId >>> 0, true);
  for (let i = 0; i < 33; i += 1) {
    const o = 8 + i * 7;
    // 造一组确定性的归一化坐标：x 从 -0.9 到 0.9，y 随索引摆动，z 小量
    const x = Math.round((-0.9 + (1.8 * i) / 32) * 32767);
    const y = Math.round((0.5 * Math.sin(i) + 0.1) * 32767);
    const z = Math.round((0.02 * i - 0.3) * 32767);
    view.setInt16(o, Math.max(-32768, Math.min(32767, x)), true);
    view.setInt16(o + 2, Math.max(-32768, Math.min(32767, y)), true);
    view.setInt16(o + 4, Math.max(-32768, Math.min(32767, z)), true);
    view.setUint8(o + 6, i === 5 ? 100 : 240); // 故意放一个低置信度点（conf<128）
  }
  return new Uint8Array(buf);
}

/**
 * 打开一条连接，返回 { ws, state, next(), closed(), counts }
 *
 * @param omitDeviceId 为 true 时 `hello` **不带 deviceId 字段**（契约 §3.1 标注为可选，
 *   §5 要求此时接收端"不做去重，按连接计数"）
 * @param hello 为 false 时连 `hello` 都不发（测接收端对"只发二进制帧"的对端是否宽容）
 */
function connect({
  deviceId = '',
  device = 'Node 契约客户端',
  protocol = SUBPROTOCOL,
  hello = true,
  omitDeviceId = false,
} = {}) {
  return new Promise((resolve, reject) => {
    const ws = new WebSocket(URL_, [SUBPROTOCOL]);
    ws.binaryType = 'arraybuffer';
    const inbox = [];
    const waiters = [];
    const state = { closeInfo: null, error: null };
    /** 收到的消息类型计数（用于断言"config 不是周期性刷屏"这类事实） */
    const counts = {};

    const push = (msg) => {
      const t = msg?.type ?? '(未知)';
      counts[t] = (counts[t] ?? 0) + 1;
      const w = waiters.shift();
      if (w) w(msg);
      else inbox.push(msg);
    };

    ws.addEventListener('message', (ev) => {
      if (typeof ev.data === 'string') {
        try {
          push(JSON.parse(ev.data));
        } catch {
          push({ type: '(非 JSON 文本)', raw: ev.data });
        }
      } else {
        push({ type: '(二进制)', byteLength: ev.data.byteLength });
      }
    });
    ws.addEventListener('close', (ev) => {
      state.closeInfo = { code: ev.code, reason: ev.reason };
      const w = waiters.shift();
      if (w) w({ type: '(closed)', ...state.closeInfo });
      else inbox.push({ type: '(closed)', ...state.closeInfo });
    });
    ws.addEventListener('error', (ev) => {
      state.error = ev?.message ?? 'error';
    });

    const opened = new Promise((res, rej) => {
      ws.addEventListener('open', res);
      const t = setTimeout(() => rej(new Error('连接超时（3s）')), 3000);
      t.unref?.();
    });

    opened
      .then(() => {
        if (hello) {
          const payload = {
            type: 'hello', protocol, device,
            appVersion: '0.2.0', cameraFacing: 'front', targetFps: 30, resolution: '640x480',
          };
          // 契约 §3.1：deviceId 可选 —— 这里刻意支持"整个字段都不出现"的合规对端
          if (!omitDeviceId) payload.deviceId = deviceId;
          ws.send(JSON.stringify(payload));
        }
        resolve({
          ws,
          state,
          counts,
          next: (timeoutMs = 3000) => new Promise((res) => {
            if (inbox.length) return res(inbox.shift());
            const t = setTimeout(() => res({ type: '(超时)', timeoutMs }), timeoutMs);
            t.unref?.();
            waiters.push((m) => {
              clearTimeout(t);
              res(m);
            });
          }),
          /**
           * 按**类型**取消息，跳过其它类型（并把它们计入 counts）。
           * 契约没有规定消息顺序，客户端必须顺序无关 —— 尤其是接收端现在会在
           * `welcome` 之后立刻补一条 `config`，"第一条消息是 welcome" 不再是保证。
           */
          async nextOfType(type, timeoutMs = 3000) {
            const skipped = [];
            const deadline = Date.now() + timeoutMs;
            while (Date.now() < deadline) {
              const msg = await this.next(Math.max(50, deadline - Date.now()));
              if (msg?.type === type) return msg;
              skipped.push(msg?.type);
              if (msg?.type === '(closed)') return msg;
            }
            return { type: '(超时)', expected: type, skipped };
          },
          async closed(timeoutMs = 3000) {
            if (state.closeInfo) return state.closeInfo;
            const t0 = Date.now();
            while (Date.now() - t0 < timeoutMs) {
              if (state.closeInfo) return state.closeInfo;
              await sleep(50);
            }
            return null;
          },
        });
      })
      .catch(reject);
  });
}

function badFrame() {
  // 契约要求恰好 239 字节；这里故意发 100 字节
  return new Uint8Array(100).fill(0x7f);
}

async function main() {
  console.log(`\n== WebSocket 契约一致性（websocket-keypoints v0.2.0）@ ${URL_} ==\n`);

  // ---------- §1/§3.2 握手 ----------
  console.log('[1] 子协议协商 + hello → welcome（§1 / §3.1 / §3.2）');
  const a = await connect({ deviceId: 'aaaa000000000001', device: 'Pixel 8' });
  const welcome = await a.nextOfType('welcome');
  check('收到 welcome', welcome?.type === 'welcome', JSON.stringify(welcome));
  check('welcome.protocol 与契约一致', welcome?.protocol === SUBPROTOCOL, String(welcome?.protocol));
  check('welcome.sessionId 非空', typeof welcome?.sessionId === 'string' && welcome.sessionId.length > 0);
  check('welcome.expectedFps = 30', welcome?.expectedFps === 30, String(welcome?.expectedFps));
  const silent = Number(welcome?.silentTimeoutMs);
  const heartbeat = Number(welcome?.heartbeatIntervalMs);
  check('welcome 声明 silentTimeoutMs', Number.isFinite(silent) && silent > 0, String(welcome?.silentTimeoutMs));
  check('welcome 声明 heartbeatIntervalMs', Number.isFinite(heartbeat) && heartbeat > 0, String(welcome?.heartbeatIntervalMs));
  check('heartbeatIntervalMs < silentTimeoutMs / 2（§1.1 硬性）',
    heartbeat < silent / 2, `心跳 ${heartbeat} ms，超时 ${silent} ms`);
  check('子协议已协商成功（未回落到空串）', a.ws.protocol === SUBPROTOCOL, `实际 '${a.ws.protocol}'`);

  // ---------- §2 二进制帧 ----------
  console.log('\n[2] 239 字节二进制帧（§2）');
  for (let i = 0; i < 60; i += 1) {
    a.ws.send(buildFrame({ frameId: i, mirrored: i % 2 === 0, lowConf: i % 5 === 0, upperBodyOnly: false }));
    await sleep(8);
  }
  // 故意跳号（丢包检测：§8 应进入 degraded 并可自动恢复）
  for (let i = 63; i < 75; i += 1) {
    a.ws.send(buildFrame({ frameId: i }));
    await sleep(8);
  }
  check('已发送 72 帧（含跳号 60→63）', true);
  a.ws.send(badFrame());
  check('已发送 1 个非法长度帧（100 字节，服务端应计入 malformed 且不崩）', true);
  await sleep(200);

  // ---------- §3.4 / §3.6 ----------
  console.log('\n[3] stats 与 ping/pong（§3.4 / §3.6）');
  a.ws.send(JSON.stringify({ type: 'stats', fps: 29.6, dropped: 3, latencyMs: 42, confidenceAvg: 0.93 }));
  const sentAt = 1730000000000;
  a.ws.send(JSON.stringify({ type: 'ping', sentAtMs: sentAt }));
  const pong = await a.nextOfType('pong');
  check('收到 pong', pong?.type === 'pong', JSON.stringify(pong));
  check('pong.sentAtMs 原样回传', pong?.sentAtMs === sentAt, String(pong?.sentAtMs));
  check('pong.serverAtMs 已填写', Number.isFinite(Number(pong?.serverAtMs)) && Number(pong?.serverAtMs) > 0,
    String(pong?.serverAtMs));

  // ---------- §5 设备去重 ----------
  console.log('\n[4] 同 deviceId 替换旧连接（§5）');
  const a2 = await connect({ deviceId: 'aaaa000000000001', device: 'Pixel 8' });
  const a2welcome = await a2.nextOfType('welcome');
  check('新连接收到 welcome', a2welcome?.type === 'welcome', JSON.stringify(a2welcome));
  const oldClose = await a.closed(4000);
  check('旧连接被服务端关闭', oldClose !== null, '旧连接仍在打开');
  check('旧连接关闭码为 1000（契约 §5）', oldClose?.code === 1000, String(oldClose?.code));
  check('旧连接关闭原因为 replaced by new connection',
    String(oldClose?.reason ?? '').includes('replaced'), String(oldClose?.reason));
  check('新连接保持打开', a2.ws.readyState === WebSocket.OPEN, String(a2.ws.readyState));

  // ---------- §1.1 保活：只发 ping 不会被踢 ----------
  console.log('\n[5] 保活：只发应用层 ping，连接不应被静默超时踢掉（§1.1）');
  if (SKIP_SLOW) {
    console.log('  · --skip-slow：跳过 19 秒对照');
  } else {
    const k = await connect({ deviceId: 'bbbb000000000002', device: 'Keepalive' });
    const kw = await k.nextOfType('welcome');
    check('保活连接已握手', kw?.type === 'welcome', JSON.stringify(kw));
    const deadline = Date.now() + (silent + 4000); // 超过静默超时 4 秒
    let pings = 0;
    while (Date.now() < deadline) {
      if (k.ws.readyState !== WebSocket.OPEN) break;
      k.ws.send(JSON.stringify({ type: 'ping', sentAtMs: Date.now() }));
      pings += 1;
      await sleep(Math.max(500, Math.floor(heartbeat / 2)));
    }
    check(`持续 ${Math.round((silent + 4000) / 100) / 10}s 发 ping（${pings} 次）后连接仍打开`,
      k.ws.readyState === WebSocket.OPEN,
      `readyState=${k.ws.readyState}（静默超时 ${silent} ms，期望心跳 ${heartbeat} ms）`);
    k.ws.close(1000, 'done');
    await k.closed(2000);
  }

  // ---------- §1.1 / §7 静默超时 ----------
  console.log('\n[6] 完全不发数据 → close(1001, "timeout")（§1.1 / §7）');
  if (SKIP_SLOW) {
    console.log('  · --skip-slow：跳过静默超时对照');
  } else {
    const s = await connect({ deviceId: 'cccc000000000003', device: 'Silent' });
    const sw = await s.nextOfType('welcome');
    check('静默连接已握手', sw?.type === 'welcome', JSON.stringify(sw));
    const t0 = Date.now();
    const closeInfo = await s.closed(silent + 6000);
    const elapsed = Date.now() - t0;
    check('服务端关闭了静默连接', closeInfo !== null, `${elapsed} ms 内未关闭`);
    check('关闭码为 1001（going away）', closeInfo?.code === 1001, String(closeInfo?.code));
    check('关闭原因为 timeout', String(closeInfo?.reason ?? '').includes('timeout'), String(closeInfo?.reason));
    check(`关闭时机接近 silentTimeoutMs（实测 ${elapsed} ms，阈值 ${silent} ms）`,
      closeInfo !== null && elapsed >= silent * 0.8 && elapsed <= silent + 4000, `${elapsed} ms`);
  }

  // ---------- §3.5 bye ----------
  console.log('\n[7] bye → 服务端 close(1000)（§3.5）');
  const b = await connect({ deviceId: 'dddd000000000004', device: 'Bye' });
  await b.nextOfType('welcome');
  b.ws.send(JSON.stringify({ type: 'bye', reason: 'user_stop' }));
  const byeClose = await b.closed(4000);
  check('服务端关闭了连接', byeClose !== null);
  check('关闭码为 1000', byeClose?.code === 1000, String(byeClose?.code));

  // ---------- §3.1 / §6 协议不匹配 ----------
  console.log('\n[8] 协议版本不匹配 → bye + close(1002)（§3.1 / §6）');
  const c = await connect({ deviceId: 'eeee000000000005', device: 'Old', protocol: 'opendance.pose.v0' });
  const byeMsg = await c.nextOfType('bye');
  check('收到 bye', byeMsg?.type === 'bye', JSON.stringify(byeMsg));
  check('bye.reason 说明协议不匹配', String(byeMsg?.reason ?? '').includes('protocol_mismatch'), String(byeMsg?.reason));
  const mismatchClose = await c.closed(4000);
  check('关闭码为 1002（协议错误）', mismatchClose?.code === 1002, String(mismatchClose?.code));

  // ---------- §1.1 向后兼容：v0.1.0 风格对端（只发帧，从不 ping）----------
  console.log('\n[9] v0.1.0 风格对端：只发二进制帧、不发 ping / stats、hello 不带 deviceId（§1.1 / §3.1）');
  if (SKIP_SLOW) {
    console.log('  · --skip-slow：跳过 17 秒帧保活对照');
  } else {
    const legacy = await connect({ omitDeviceId: true, device: 'Legacy v0.1.0' });
    const lw = await legacy.nextOfType('welcome');
    check('v0.1.0 风格对端也收到 welcome', lw?.type === 'welcome', JSON.stringify(lw));
    const legacyFrames = Math.ceil((silent + 2000) / 500);   // 每 500ms 一帧
    for (let i = 0; i < legacyFrames; i += 1) {
      if (legacy.ws.readyState !== WebSocket.OPEN) break;
      legacy.ws.send(buildFrame({ frameId: i }));
      await sleep(500);
    }
    check(`只发帧（每 500ms 一帧、共 ${legacyFrames} 帧）持续 ${((silent + 2000) / 1000).toFixed(1)}s 后连接仍打开`,
      legacy.ws.readyState === WebSocket.OPEN,
      `readyState=${legacy.ws.readyState}（契约 §1.1：每帧都算活动，ping 只是兜底）`);
    check('没有收到过 pong 也能正常推流（本场景从不发 ping）',
      (legacy.counts.pong ?? 0) === 0, JSON.stringify(legacy.counts));
    // §3.3 的语义是"运行中调整时下发"，不是心跳：17 秒内不应出现周期性刷屏
    const cfgCount = legacy.counts.config ?? 0;
    check(`17 秒内 config 条数应在合理范围（收到 ${cfgCount} 条；周期广播会是 ~8 条）`,
      cfgCount <= 3, `实际 ${cfgCount} 条 —— config 应为变更驱动，而非心跳`);
    legacy.ws.close(1000, 'done');
    await legacy.closed(2000);
  }

  // ---------- §5 缺 deviceId → 不做去重 ----------
  console.log('\n[10] 两个都不带 deviceId 的连接不应互相顶掉（§5）');
  const n1 = await connect({ omitDeviceId: true, device: 'No-Id A' });
  const n1w = await n1.nextOfType('welcome');
  check('第一条无 deviceId 连接握手成功', n1w?.type === 'welcome', JSON.stringify(n1w));
  const n2 = await connect({ omitDeviceId: true, device: 'No-Id B' });
  const n2w = await n2.nextOfType('welcome');
  check('第二条无 deviceId 连接握手成功', n2w?.type === 'welcome', JSON.stringify(n2w));
  await sleep(500);
  check('第一条仍在打开（未被去重顶掉）', n1.ws.readyState === WebSocket.OPEN, String(n1.ws.readyState));
  check('第二条仍在打开', n2.ws.readyState === WebSocket.OPEN, String(n2.ws.readyState));
  n1.ws.close(1000, 'done');
  n2.ws.close(1000, 'done');
  await n1.closed(2000);
  await n2.closed(2000);

  // ---------- 不打招呼的对端（连 hello 都不发）----------
  console.log('\n[11] 完全不发 hello、只发二进制帧的对端应被宽容对待');
  const raw = await connect({ hello: false });
  const rawFrames = 5;
  for (let i = 0; i < rawFrames; i += 1) {
    raw.ws.send(buildFrame({ frameId: 1000 + i }));
    await sleep(30);
  }
  await sleep(300);
  check('接收端没有因此断开连接', raw.ws.readyState === WebSocket.OPEN, String(raw.ws.readyState));
  check('一个控制帧都没发过', Object.keys(raw.counts).length === 0, JSON.stringify(raw.counts));
  raw.ws.close(1000, 'done');
  await raw.closed(2000);

  // ---------- 收尾 ----------
  console.log('\n[12] 收尾');
  a2.ws.close(1000, 'done');
  await a2.closed(2000);
  await sleep(300);

  console.log(`\n== 结果：${passed} 通过 / ${failed} 失败 ==`);
  if (failed > 0) {
    console.log('失败项：');
    for (const f of failures) console.log(`  - ${f}`);
    process.exit(1);
  }
  console.log('全部通过 ✅');
  process.exit(0);
}

main().catch((err) => {
  console.error('✗ 客户端异常：', err?.message ?? err);
  process.exit(2);
});