TankIC/3DTank/tools/pair_flow.js

// pair_flow.js - 把抓包里的 c2s / s2c 按时间戳配对成「请求 → 响应」流
// 用法: node pair_flow.js [--root ../packets/plain] [--window 1500] [--out ../names/flow.json]
// 原理:
//   dump_plain.js 已产出 timeline_c2s.txt / timeline_s2c.txt（每帧相对首条消息的毫秒）。
//   对每条 s2c 帧，找「时间上最近的、且相距 <= Window 的」c2s 帧作为触发者；
//   找不到触发者的就是服务端主动推送（ping / 周期 inform）。
// 产物:
//   <space>/flow.txt   人类可读的配对表
//   <out>              机器可读：每空间的 {触发 cmdId -> 回复 cmdId 列表} + 全局统计
// 目的: 让 fake 服务器从「定时吐帧」变成「收到请求才应答」（第 1 层）
const fs = require('fs');
const path = require('path');
const zlib = require('zlib');

const Argv = process.argv.slice(2);
const Arg = (Name, Def) => {
  const I = Argv.indexOf('--' + Name);
  return I >= 0 && Argv[I + 1] ? Argv[I + 1] : Def;
};

const PROJECT = path.resolve(__dirname, '..');
const Root = path.resolve(Arg('root', path.join(PROJECT, 'packets', 'plain')));
const Window = Number(Arg('window', '1500')) || 1500;
const OutPath = path.resolve(Arg('out', path.join(PROJECT, 'names', 'flow.json')));

// 上行空间帧：帧体 = [nullmap][objectId int64 8B][methodId int64 8B][负载]（同 wss_server.js 的 DescribeUplink）
// 判据 = 首字节是 objectId 最高位（0x00 小正数 / 0xff 负数）+ At+8 处命中 names/uplink_table.json 的 methodId。
// 用 methodId 当应答签名：它是客户端源码里的常量，跨会话稳定，于是「某条上行命令」能精确命中它的应答组。
const UplinkByName = new Map();
const UplinkByHex = new Map();
try {
  const Up = JSON.parse(fs.readFileSync(path.join(PROJECT, 'names', 'uplink_table.json'), 'utf8'));
  for (const C of Up.Commands || []) {
    const B = Buffer.alloc(8);
    B.writeBigInt64BE(BigInt(C.MethodId));
    const Hex = B.toString('hex');
    UplinkByHex.set(Hex, C);
    UplinkByName.set(C.MethodId, (C.OwnerName || C.OwnerClass || '?') + '.' + (C.SendFn || '?'));
  }
} catch (Err) {
  console.log('（未加载 names/uplink_table.json，上行空间帧只能按命令 ID 配对）');
}

function SpaceFrame(Body, At) {
  const B0 = Body[At];
  if (B0 !== 0x00 && B0 !== 0xff) return null;
  if (At + 16 > Body.length) return null;
  const Cmd = UplinkByHex.get(Body.slice(At + 8, At + 16).toString('hex'));
  if (!Cmd) return null;
  return { ObjectId: Body.readBigInt64BE(At).toString(), MethodId: Body.readBigInt64BE(At + 8).toString() };
}

// 帧 → { Cmd, Size }；Compressed 帧先解压再取 nullmap/命令 ID
function FrameInfo(Buf, Direction) {
  const B0 = Buf[0];
  const HeaderSize = B0 & 0x80 ? 4 : 2;
  // 4 字节头：bit6 是压缩位，长度取低 6 位 + 后 3 字节（注意不是 & 0x7f）
  const Total =
    HeaderSize === 4
      ? 4 + ((((B0 & 0x3f) << 24) | (Buf[1] << 16) | (Buf[2] << 8) | Buf[3]) >>> 0)
      : 2 + ((((B0 & 0x3f) << 8) | Buf[1]) >>> 0);
  let Body = Buf.slice(HeaderSize);
  if (B0 & 0x40) {
    try {
      Body = zlib.inflateSync(Body);
    } catch (Err) {
      return { Cmd: -2, Size: Total, Note: 'zlib 失败' };
    }
  }
  const Nm = Body[0];
  let At;
  if (Nm & 0x80) At = Nm & 0x40 ? 4 + (((Nm & 0x3f) << 16) | (Body[1] << 8) | Body[2]) : 1 + (Nm & 0x3f);
  else At = 1 + ((Nm & 0x60) >>> 5);
  if (At >= Body.length) return { Cmd: -1, Size: Total, Note: '保活 ping' };
  if (Direction === 'c2s') {
    const Sp = SpaceFrame(Body, At);
    if (Sp) return { Cmd: 0, Size: Total, Space: true, ...Sp, Payload: Body.slice(At + 16) };
  }
  return { Cmd: Body[At], Size: Total, Payload: Body.slice(At + 1) };
}

function ReadTimeline(Dir, Direction) {
  const P = path.join(Dir, 'timeline_' + Direction + '.txt');
  if (!fs.existsSync(P)) return [];
  const Out = [];
  for (const Line of fs.readFileSync(P, 'utf8').split('\n')) {
    const M = /^(\S+)\t(\d+)$/.exec(Line.trim());
    if (M) Out.push({ Name: M[1], Ms: Number(M[2]) });
  }
  return Out;
}

// 帧签名：空间帧用 methodId（S:<methodId>），其余用「命令 ID + 负载前 4 字节」
// 必须与 wss_server.js 的 FrameSignature 完全一致
function Signature(Info) {
  if (Info.Space) return 'S:' + Info.MethodId;
  const Head = Info.Payload ? Info.Payload.slice(0, 4).toString('hex') : '';
  return Info.Cmd + (Head ? ':' + Head : '');
}

// 签名 → 人类可读标签（flow.txt / 日志用）
function SigLabel(Sig) {
  const M = /^S:(\d+)$/.exec(Sig);
  if (!M) return 'cmd=' + Sig.split(':')[0];
  const Name = UplinkByName.get(M[1]);
  return '上行=' + (Name || '?') + '(' + M[1] + ')';
}

const Spaces = [];
const Global = { AsTrigger: {}, AsReply: {}, Push: {}, Spaces: 0, Pairs: 0, S2c: 0 };

for (const Name of fs.readdirSync(Root).sort()) {
  const Dir = path.join(Root, Name);
  if (!fs.statSync(Dir).isDirectory()) continue;
  const C2s = ReadTimeline(Dir, 'c2s');
  const S2c = ReadTimeline(Dir, 's2c');
  if (!C2s.length || !S2c.length) continue;
  Global.Spaces++;

  const C2sFrames = C2s.map((X) => ({ ...X, Info: FrameInfo(fs.readFileSync(path.join(Dir, X.Name)), 'c2s') }));
  const S2cFrames = S2c.map((X) => ({ ...X, Info: FrameInfo(fs.readFileSync(path.join(Dir, X.Name)), 's2c') }));

  const Lines = ['空间 ' + Name + '   请求 ' + C2s.length + ' 条 / 响应 ' + S2c.length + ' 条（窗口 ' + Window + 'ms）', ''];
  const Triggers = [];
  let Pushed = 0;
  let Pairs = 0;

  for (const S of S2cFrames) {
    if (S.Info.Cmd === -1) continue; // ping 单独统计
    Global.S2c++;
    // 找最近的、时间不超过窗口的 c2s
    let Best = null;
    for (const C of C2sFrames) {
      if (C.Ms <= S.Ms && S.Ms - C.Ms <= Window) Best = C; // 列表按时间升序，取最后一个满足的
    }
    if (Best) {
      const Key = Signature(Best.Info);
      let Group = Triggers.find((G) => G.Trigger === Key);
      if (!Group) {
        Group = { Trigger: Key, TriggerName: Best.Name, TriggerMs: Best.Ms, Replies: [] };
        Triggers.push(Group);
      }
      Group.Replies.push({ Name: S.Name, Cmd: S.Info.Cmd, Ms: S.Ms, Delay: S.Ms - Best.Ms, Size: S.Info.Size });
      Global.AsTrigger[Key] = (Global.AsTrigger[Key] || 0) + 1;
      Global.AsReply[S.Info.Cmd] = (Global.AsReply[S.Info.Cmd] || 0) + 1;
      Pairs++;
      Global.Pairs++;
    } else {
      Pushed++;
      Global.Push[S.Info.Cmd] = (Global.Push[S.Info.Cmd] || 0) + 1;
    }
  }

  for (const G of Triggers) {
    Lines.push(
      '触发 c2s ' + G.TriggerName + '  ' + SigLabel(G.Trigger) + '  @+' + G.TriggerMs + 'ms  →  ' + G.Replies.length + ' 条回复',
    );
    for (const R of G.Replies) {
      Lines.push('    +' + String(R.Delay).padStart(6) + 'ms  ' + R.Name + '  cmd=' + R.Cmd + '  ' + R.Size + 'B');
    }
  }
  Lines.push('', '无触发的主动推送: ' + Pushed + ' 条');
  const PushCount = {};
  for (const S of S2cFrames) if (S.Info.Cmd !== -1) PushCount[S.Info.Cmd] = (PushCount[S.Info.Cmd] || 0) + 1;
  Lines.push('  各命令总条数: ' + JSON.stringify(PushCount));
  fs.writeFileSync(path.join(Dir, 'flow.txt'), Lines.join('\n') + '\n', 'utf8');

  // 给 fake 服务器用的极简配对表：每行「s2c 文件 \t 触发签名|- \t 相对触发的延迟ms \t 原时间线ms」
  const Tsv = [];
  for (const S of S2cFrames) {
    if (S.Info.Cmd === -1) continue; // 保活 ping 交给服务器自己的定时器
    let Trigger = '-';
    let Delay = 0;
    for (const C of C2sFrames) {
      if (C.Ms <= S.Ms && S.Ms - C.Ms <= Window) {
        Trigger = Signature(C.Info);
        Delay = S.Ms - C.Ms;
      }
    }
    Tsv.push(S.Name + '\t' + Trigger + '\t' + Delay + '\t' + S.Ms);
  }
  fs.writeFileSync(path.join(Dir, 'flow_s2c.tsv'), Tsv.join('\n') + '\n', 'utf8');

  Spaces.push({
    Space: Name,
    Triggers: Triggers.map((G) => ({
      Trigger: G.Trigger,
      TriggerLabel: SigLabel(G.Trigger),
      TriggerName: G.TriggerName,
      Replies: G.Replies.map((R) => ({ Cmd: R.Cmd, Delay: R.Delay, Size: R.Size })),
    })),
    Pushed,
  });
  console.log(
    Name.padEnd(20) + ' c2s ' + String(C2s.length).padStart(3) + '  s2c ' + String(S2c.length).padStart(3) +
      '  配对 ' + String(Pairs).padStart(4) + '  推送 ' + Pushed,
  );
}

fs.writeFileSync(
  OutPath,
  JSON.stringify(
    {
      Note: '由 pair_flow.js 生成：s2c 帧按时间就近归到触发的 c2s 帧；无触发的记为主动推送',
      Window,
      Stats: Global,
      Spaces,
    },
    null,
    2,
  ),
  'utf8',
);
console.log('\n全局统计:');
console.log('  作为触发的 c2s 命令: ' + JSON.stringify(Global.AsTrigger));
console.log('  作为回复的 s2c 命令: ' + JSON.stringify(Global.AsReply));
console.log('  主动推送的 s2c 命令: ' + JSON.stringify(Global.Push));
console.log('  已写出: ' + OutPath);