inkwatch/shared/protocol.js

/**
 * 墨观 协议规范 v1 —— 客户端与服务端的共同契约
 *
 * 设计约束（见 V4.0 方案第 5 章）：
 *  1. 所有消息携带版本号 v，服务端兼容 N-1 版本
 *  2. payload 字段白名单制，禁止透传未声明字段
 *  3. 上行 ts 由服务端接收时改写（服务器时间权威）
 *  4. 高频指标帧默认 JSON，可通过 hello 协商 MessagePack
 *  5. 客户端对未知字段一律忽略（向后兼容）
 */

'use strict';

const {
  METRIC_WHITELIST,
  METRIC_KEYS,
  METRICS_SCHEMA,
  bytesPerSecToMbps,
} = require('./metrics-schema');

/** 协议当前版本 */
const PROTOCOL_VERSION = 1;

/** 服务端支持兼容的最低版本（N-1 策略） */
const MIN_SUPPORTED_VERSION = 1;

/**
 * 指标白名单不再在本文件硬编码 —— 改由 `shared/metrics-schema.js` 派生。
 *
 * 历史教训（已修复）：本文件曾自持一份手写白名单，与 rust-core 采集字段、
 * 时序表列名三处命名互不一致，交集仅 3 个字段，导致真实采集帧被
 * `sanitizePayload()` 静默过滤掉 CPU/内存/磁盘/网络全部数值。
 * 一致性现由 `tests/schema-consistency-test.js` 强制锁定。
 */

/** 上行事件名 */
const UPSTREAM_EVENTS = new Set([
  'agent:hello',
  'agent:metrics',
  'agent:heartbeat',
  'viewer:subscribe',
  // 退订：局域网直连时查看端改走内网，必须能**主动退掉云端订阅**，
  // 否则"省流量"只是句空话 —— 云端仍会把每一帧推过来。
  'viewer:unsubscribe',
  // 设备间传文件的"发布"：告诉对端"我这儿有个文件，凭这些信息来取"。
  // 采集端发的广播给该设备的订阅者；查看端发的定向给目标设备的采集端。
  // ⚠️ 载荷里**分开发** `lan_endpoints`（数组，服务端照旧校验只收私有网段）
  // 与 `token`；**服务端刻意不解析那个"领取码"的格式** ——
  // 码是客户端之间的约定，服务端多懂一层就多一处要跟着改的地方。
  'agent:offer',
  'viewer:offer',
]);

/** 下行事件名 */
const DOWNSTREAM_EVENTS = new Set([
  'server:config',
  'server:snapshot',
  'server:metrics',
  'server:device_status',
  // 心跳应答。服务端必须回一声：采集端连接的常态是"只上行、无下行"，
  // 若心跳只去不回，客户端的半开检测会把**完全健康**的连接判成"疑似离线"
  // （因为能写不能证明对端活着 —— TCP 会把数据静默缓冲进死连接）。
  'server:pong',
  // "有人要给你一个文件，凭这些信息去取"。与 server:metrics 同构：
  // 都是"某个设备的订阅者会收到的东西"，复用同一套订阅与转发。
  'server:offer',
  // 退订回执 —— 查看端据此确认"云端确实不再推了"，才敢切到内网。
  'server:unsubscribed',
  'error',
]);

/** 统一错误码 */
const ERROR_CODES = {
  BAD_ENVELOPE: 'E_BAD_ENVELOPE',
  BAD_VERSION: 'E_BAD_VERSION',
  BAD_TYPE: 'E_BAD_TYPE',
  UNAUTHORIZED: 'E_UNAUTHORIZED',
  NOT_PAIRED: 'E_NOT_PAIRED',
  RATE_LIMITED: 'E_RATE_LIMITED',
  RATE_LIMIT_PAIRING: 'E_PAIRING_LOCKED',
  INTERNAL: 'E_INTERNAL',
  /**
   * MessagePack 量化路径指纹未知（服务端重启导致指纹表清空）。
   * 客户端应带完整路径清单重发一帧，或本轮退回 JSON —— 不要静默丢弃整帧。
   * 仅下行（服务端 → 客户端）。
   */
  UNKNOWN_PATH_FP: 'E_UNKNOWN_PATH_FP',
};

/**
 * 校验并规范化上行消息信封
 * @returns {{ok:true, msg:object} | {ok:false, code:string, reason:string}}
 */
function validateEnvelope(raw, now) {
  if (raw === null || typeof raw !== 'object' || Array.isArray(raw)) {
    return { ok: false, code: ERROR_CODES.BAD_ENVELOPE, reason: '消息必须为 JSON 对象' };
  }

  const v = raw.v;
  if (typeof v !== 'number' || !Number.isInteger(v) || v < MIN_SUPPORTED_VERSION) {
    return { ok: false, code: ERROR_CODES.BAD_VERSION, reason: `不支持的协议版本: ${v}` };
  }
  if (v > PROTOCOL_VERSION) {
    return { ok: false, code: ERROR_CODES.BAD_VERSION, reason: `客户端版本过新: ${v}` };
  }

  const type = raw.type;
  if (typeof type !== 'string' || !UPSTREAM_EVENTS.has(type)) {
    return { ok: false, code: ERROR_CODES.BAD_TYPE, reason: `未知上行事件: ${type}` };
  }

  // seq 仅对指标/心跳类要求；hello 阶段设备尚无 seq 概念时允许缺省
  const seq = Number.isInteger(raw.seq) ? raw.seq : 0;
  if (seq < 0) {
    return { ok: false, code: ERROR_CODES.BAD_ENVELOPE, reason: 'seq 必须为非负整数' };
  }

  const deviceId = typeof raw.device_id === 'string' ? raw.device_id : '';

  // 服务器时间权威：上行 ts 一律由服务端改写
  return {
    ok: true,
    msg: {
      v: Math.min(v, PROTOCOL_VERSION),
      type,
      ts: now,
      seq,
      device_id: deviceId,
      client_ts: Number.isInteger(raw.ts) ? raw.ts : null, // 仅用于诊断
      payload: sanitizePayload(raw.payload, type),
    },
  };
}

/** 单层键数上限（数组元素同样受此限） */
const MAX_PAYLOAD_KEYS = 64;
/** 允许的嵌套深度：1 = 顶层键的值可以是对象，但其子值不能再是对象 */
const MAX_NEST_DEPTH = 1;
/** 字符串长度上限 */
const MAX_STRING_LEN = 512;

/**
 * 递归净化单个值 —— 数据最小化与"控制字段要能通过"之间的折中。
 *
 * ⚠️ 历史缺陷（已修）：早期实现对非指标事件只放行 string/number/boolean/数组，
 * **对象一律丢弃**。后果是 `agent:hello` 里的 `capabilities` / `keepalive` /
 * `reconnect_policy` 三个嵌套对象全部被剥掉 —— 客户端老老实实上报了能力集，
 * 服务端 `msg.payload.capabilities` 恒为 `{}`，device 表里 capabilities 是空对象，
 * 查看端因此无法按能力隐藏卡片（方案 7.1 优雅降级直接失效），且**不报任何错**。
 * 与指标字段命名分裂、msgpack 未解包同属一类：白名单过严导致静默丢数据。
 *
 * 现在允许**一层**嵌套对象（够表达能力集/保活状态这类结构化控制字段），
 * 但仍有界：深度 1、单层 ≤64 键、字符串 ≤512、仅标量或标量数组作叶子。
 * 更深的结构、函数、Symbol 等一律丢弃。
 */
function sanitizeValue(val, depth) {
  if (val === null) return null;
  switch (typeof val) {
    case 'string':
      return val.length > MAX_STRING_LEN ? val.slice(0, MAX_STRING_LEN) : val;
    case 'number':
      return Number.isFinite(val) ? val : null; // NaN/Infinity 归 null
    case 'boolean':
      return val;
    default:
      break;
  }
  if (Array.isArray(val)) {
    return val
      .filter((x) => ['string', 'number', 'boolean'].includes(typeof x))
      .slice(0, MAX_PAYLOAD_KEYS)
      .map((x) => (typeof x === 'string' && x.length > MAX_STRING_LEN ? x.slice(0, MAX_STRING_LEN) : x));
  }
  if (typeof val === 'object' && depth < MAX_NEST_DEPTH) {
    const out = {};
    let n = 0;
    for (const [k, v] of Object.entries(val)) {
      if (n >= MAX_PAYLOAD_KEYS) break;
      if (typeof k !== 'string' || k.length > 64) continue;
      const cleaned = sanitizeValue(v, depth + 1);
      if (cleaned === undefined) continue;
      out[k] = cleaned;
      n++;
    }
    return out;
  }
  return undefined; // 深度超限 / 不可序列化类型 → 丢弃
}

/**
 * payload 白名单过滤 —— 数据最小化原则（见 V4.0 方案 9.6）
 */
function sanitizePayload(payload, type) {
  if (payload === null || typeof payload !== 'object' || Array.isArray(payload)) {
    return {};
  }
  const out = {};

  if (type === 'agent:metrics') {
    for (const [k, val] of Object.entries(payload)) {
      if (!METRIC_WHITELIST.has(k)) continue; // 未声明字段一律丢弃
      out[k] = normalizeMetricValue(k, val);
    }
    return out;
  }

  // 其他事件：限制键数与嵌套深度，防夹带；但保留一层对象以承载结构化控制字段
  let n = 0;
  for (const [k, val] of Object.entries(payload)) {
    if (n >= MAX_PAYLOAD_KEYS) break;
    if (typeof k !== 'string' || k.length > 64) continue;
    const cleaned = sanitizeValue(val, 0);
    if (cleaned === undefined) continue;
    out[k] = cleaned;
    n++;
  }
  return out;
}

/** 指标值规整：数值或 null，NaN/Infinity 一律归 null */
function normalizeMetricValue(key, val) {
  if (val === null || val === undefined) return null;
  if (Array.isArray(val)) {
    return val
      .filter((x) => typeof x === 'number' && Number.isFinite(x))
      .slice(0, 64)
      .map(round2);
  }
  if (typeof val === 'number' && Number.isFinite(val)) return round2(val);
  return null;
}

function round2(n) {
  return Math.round(n * 100) / 100;
}

/**
 * 构造下行消息信封
 */
function makeEnvelope(type, payload, extra = {}) {
  return {
    v: PROTOCOL_VERSION,
    type,
    ts: Date.now(),
    ...extra,
    payload: payload || {},
  };
}

/** 构造错误消息 */
function makeError(code, reason) {
  return { v: PROTOCOL_VERSION, type: 'error', ts: Date.now(), payload: { code, reason } };
}

/** 协商编码：客户端声明列表与服务端支持列表取交集，缺省回退 json */
function negotiateEncoding(clientEncodings) {
  const supported = ['json', 'msgpack'];
  if (!Array.isArray(clientEncodings)) return 'json';
  for (const enc of clientEncodings) {
    if (supported.includes(enc)) return enc;
  }
  return 'json';
}

module.exports = {
  PROTOCOL_VERSION,
  MIN_SUPPORTED_VERSION,
  // 指标白名单与清单：由 metrics-schema 单一真相源派生后原样透出，
  // 调用方既可只取白名单，也可拿到完整 schema 做展示/校验。
  METRIC_WHITELIST,
  METRIC_KEYS,
  METRICS_SCHEMA,
  bytesPerSecToMbps,
  UPSTREAM_EVENTS,
  DOWNSTREAM_EVENTS,
  ERROR_CODES,
  validateEnvelope,
  sanitizePayload,
  makeEnvelope,
  makeError,
  negotiateEncoding,
};