NexusGate/agent/agent.js

'use strict';

const fs = require('node:fs');
const path = require('node:path');
const os = require('node:os');
const net = require('node:net');
const crypto = require('node:crypto');
const { spawnSync } = require('node:child_process');

const VERSION = '0.6.8';
const CONTROLLER = String(process.env.NG_CONTROLLER || '').replace(/\/+$/, '');
const AGENT_KEY = process.env.NG_AGENT_KEY || '';
const SINGBOX_BIN = process.env.NG_SINGBOX_BIN || '/usr/local/bin/nexusgate-sing-box';
const SINGBOX_STATS_BIN = process.env.NG_SINGBOX_STATS_BIN || '/usr/local/bin/nexusgate-sing-box-stats';
const ROOT = process.env.NG_CONFIG_DIR || '/etc/nexusgate/agent';
const RESOURCE_DIR = path.join(ROOT, 'resources');
const SINGBOX_CONFIG = process.env.NG_SINGBOX_CONFIG || '/etc/nexusgate/sing-box/config.json';
const SINGBOX_LOG = process.env.NG_SINGBOX_LOG || '/var/log/nexusgate/sing-box-access.log';
const SINGBOX_CURSOR = process.env.NG_SINGBOX_CURSOR || '/etc/nexusgate/sing-box-cursor.json';
const KEY_FILE = process.env.NG_KEY_FILE || '/etc/nexusgate/keys.json';
const POLL_MS = Math.max(3, Number(process.env.NG_POLL_SECONDS || 8)) * 1000;
let lastEngineError = '';
let certificateCache = { fingerprint:'', checkedAt:0, domains:[] };

function sleep(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); }
function log(...args) { console.log(new Date().toISOString(), ...args); }
function safeError(value) {
  return String(value || '').replace(/\bPrivate\s*Key\s*:\s*[^\s,;]+/gi, 'PrivateKey: [REDACTED]')
    .replace(/\bBearer\s+[^\s,;]+/gi, 'Bearer [REDACTED]');
}

async function request(endpoint, options = {}) {
  const controller = new AbortController();
  const timeout = setTimeout(() => controller.abort(), 20000);
  try {
    const response = await fetch(`${CONTROLLER}${endpoint}`, {
      ...options,
      headers: { authorization: `Bearer ${AGENT_KEY}`, 'content-type': 'application/json', ...(options.headers || {}) },
      signal: controller.signal
    });
    const body = await response.json().catch(() => ({}));
    if (!response.ok) throw new Error(body.message || `Controller returned ${response.status}`);
    return body;
  } finally { clearTimeout(timeout); }
}

function ensureDirectories() {
  fs.mkdirSync(RESOURCE_DIR, { recursive: true, mode: 0o700 });
  fs.mkdirSync('/var/log/nexusgate', { recursive: true, mode: 0o750 });
}

function loadKeys() {
  try { return JSON.parse(fs.readFileSync(KEY_FILE, 'utf8')); } catch (error) { if (error.code === 'ENOENT') return {}; throw error; }
}

function saveKeys(keys) {
  const temp = `${KEY_FILE}.tmp`;
  fs.writeFileSync(temp, `${JSON.stringify(keys, null, 2)}\n`, { mode: 0o600 });
  fs.renameSync(temp, KEY_FILE); fs.chmodSync(KEY_FILE, 0o600);
}

function realityKey(keyId) {
  const keys = loadKeys();
  if (keys[keyId]) return keys[keyId];
  const kp = crypto.generateKeyPairSync('x25519');
  const priv = kp.privateKey.export({ format: 'jwk' });
  const pub = kp.publicKey.export({ format: 'jwk' });
  if (!priv.d || !pub.x) throw new Error('X25519 密钥生成失败');
  // JWK 的 base64url 无填充格式与 sing-box 的 reality-keypair 输出一致，直接用
  keys[keyId] = { privateKey: priv.d, publicKey: pub.x, createdAt: new Date().toISOString() };
  saveKeys(keys);
  return keys[keyId];
}

function materialize(value, artifacts) {
  if (Array.isArray(value)) return value.map((item) => materialize(item, artifacts));
  if (value && typeof value === 'object') return Object.fromEntries(Object.entries(value).map(([key, item]) => [key, materialize(item, artifacts)]));
  if (typeof value !== 'string') return value;
  const match = value.match(/^\$\{REALITY_PRIVATE:(.+)}$/);
  if (!match) return value;
  const pair = realityKey(match[1]);
  artifacts.realityPublicKey = pair.publicKey;
  return pair.privateKey;
}

function readResources() {
  return fs.readdirSync(RESOURCE_DIR).filter((name) => name.endsWith('.json')).sort().map((name) => JSON.parse(fs.readFileSync(path.join(RESOURCE_DIR, name), 'utf8')));
}

// 云机器的公网 IP 通常是 NAT 地址，不在本地网卡上；直接 bind 会
// "cannot assign requested address" 导致引擎起不来。非本地地址回退到对应栈的通配符。
function normalizeInboundListen(inbound) {
  if (!inbound || typeof inbound !== 'object') return inbound;
  const text = typeof inbound.listen === 'string' ? inbound.listen.trim() : '';
  if (!text || !net.isIP(text) || text === '0.0.0.0' || text === '::') return inbound;
  let local = false;
  for (const list of Object.values(os.networkInterfaces())) {
    for (const info of list || []) if (info && info.address === text) { local = true; break; }
    if (local) break;
  }
  if (local) return inbound;
  const fallback = net.isIPv6(text) ? '::' : '0.0.0.0';
  log(`Listen ${text} 不是本机地址，回退为 ${fallback}`);
  return { ...inbound, listen: fallback };
}

function combinedSingBoxConfig(resources) {
  return {
    log: { level: 'info', output: SINGBOX_LOG, timestamp: true },
    inbounds: resources.flatMap((item) => item.inbounds || []).map(normalizeInboundListen),
    outbounds: [{ type: 'direct', tag: 'ng-fallback-direct' }, ...resources.flatMap((item) => item.outbounds || [])],
    route: { rules: resources.flatMap((item) => item.routingRules || []), final: 'ng-fallback-direct' },
    experimental: { v2ray_api: { listen: `127.0.0.1:${process.env.NG_SINGBOX_API_PORT || 10086}`,
      stats: { enabled: true, inbounds: resources.flatMap((item) => item.inbounds || []).map((item) => item.tag),
        // sing-box 只对显式列出的用户维护 user>>> 计数器；AnyTLS/SS 用 name，VLESS 用 name，SOCKS5 用 username。
        users: resources.flatMap((item) => item.inbounds || []).flatMap((item) => item.users || []).map((item) => item.name || item.username).filter(Boolean) } } }
  };
}

function run(command, args, timeout = 30000) {
  const result = spawnSync(command, args, { encoding: 'utf8', timeout });
  if (result.status !== 0) throw new Error(`${command} ${args.join(' ')} failed: ${safeError(result.stderr || result.stdout || '').slice(0, 1200)}`);
  return result.stdout;
}

function serviceCommand(action, service) {
  if (fs.existsSync('/run/systemd/system')) return run('systemctl', [action, `${service}.service`]);
  return run('rc-service', [service, action]);
}

function activateSingBox() {
  const resources = readResources();
  if (!resources.length) {
    // 无资源时必须确保引擎停下（不能只在配置文件存在时停，否则全新安装会陷入重启循环）
    try { serviceCommand('stop', 'nexusgate-sing-box'); } catch { /* 已经停了 */ }
    if (fs.existsSync(SINGBOX_CONFIG)) fs.rmSync(SINGBOX_CONFIG, { force: true });
    return;
  }
  if (!fs.existsSync(SINGBOX_BIN) || !fs.existsSync(SINGBOX_STATS_BIN)) throw new Error('sing-box 及统计组件未安装；请在入口机器运行 ng-agent engine install');
  const candidate = `${SINGBOX_CONFIG}.candidate.json`;
  const previous = `${SINGBOX_CONFIG}.previous`;
  const next = `${JSON.stringify(combinedSingBoxConfig(resources), null, 2)}\n`;
  if (fs.existsSync(SINGBOX_CONFIG) && fs.readFileSync(SINGBOX_CONFIG, 'utf8') === next && serviceHealth('nexusgate-sing-box').status === 'ready') return;
  fs.mkdirSync(path.dirname(SINGBOX_CONFIG), { recursive: true, mode: 0o700 });
  fs.writeFileSync(candidate, next, { mode: 0o600 });
  try { run(SINGBOX_BIN, ['check', '-c', candidate]); }
  catch (error) { fs.rmSync(candidate, { force: true }); throw error; }
  if (fs.existsSync(SINGBOX_CONFIG)) fs.copyFileSync(SINGBOX_CONFIG, previous);
  fs.renameSync(candidate, SINGBOX_CONFIG);
  try { serviceCommand('restart', 'nexusgate-sing-box'); }
  catch (error) {
    if (fs.existsSync(previous)) fs.copyFileSync(previous, SINGBOX_CONFIG);
    else fs.rmSync(SINGBOX_CONFIG, { force: true });
    if (fs.existsSync(SINGBOX_CONFIG)) { try { serviceCommand('restart', 'nexusgate-sing-box'); } catch { /* preserve original error */ } }
    throw error;
  }
}

function safeResourcePath(resourceId) {
  if (!/^[a-zA-Z0-9_-]{8,100}$/.test(String(resourceId))) throw new Error('Invalid resource id');
  return path.join(RESOURCE_DIR, `${resourceId}.json`);
}

async function flushUsageBeforeChange() {
  const live = readResources().some((resource) =>
    resource.meta?.metricsTag && resource.meta.kind === 'node');
  if (!live || serviceHealth('nexusgate-sing-box').status !== 'ready') return;
  const report = queryUsage();
  if (report.engineErrors.includes('sing-box')) throw new Error('sing-box 重载前无法读取入口计数；请先运行 ng-agent doctor');
  // Await the controller's durable write before a restart destroys the
  // in-memory counters. The periodic loop can report concurrently; the
  // controller ignores repeated and out-of-order samples for one process.
  await request('/api/agent/usage', { method:'POST', body:JSON.stringify(report) });
}

async function flushAllUsageBeforeRestart() {
  await flushUsageBeforeChange();
}

async function applyResource(payload) {
  if (!payload || !payload.resource || !payload.resource.id) throw new Error('Missing resource payload');
  const artifacts = {};
  const resource = materialize(payload.resource, artifacts);
  if (resource.engine && resource.engine !== 'sing-box')
    throw new Error(`Unsupported engine: ${resource.engine}（Xray 已弃用，只支持 sing-box）`);
  for (const inbound of resource.inbounds || []) {
    // Hysteria 2 无认证账户时拒绝启动未受保护的入口。
    if (inbound.type === 'hysteria2') {
      const users = inbound.users || [];
      if (!users.length || users.some((user) => !user.password))
        throw new Error('Hysteria 2 入口没有认证账户；拒绝启动未受保护的入口');
    }
    const tls = inbound.tls;
    if (!tls) continue;
    if (tls.reality && tls.reality.enabled) continue; // Reality 无证书，不走证书检查
    const domain = tls.server_name;
    if (!/^(?:[a-z0-9-]+\.)+[a-z]{2,63}$/.test(domain || '')) throw new Error('TLS 域名无效');
    const expected = `/etc/nexusgate/tls/${domain}/`;
    const cert = { certificateFile: tls.certificate_path, keyFile: tls.key_path };
    if (!cert.certificateFile || !cert.keyFile)
      throw new Error(`TLS certificate for ${domain} is missing; run ng-agent cert issue ${domain} EMAIL or ng-agent cert import`);
    if (!cert.certificateFile.startsWith(expected) || !cert.keyFile.startsWith(expected) ||
        !fs.existsSync(cert.certificateFile) || !fs.existsSync(cert.keyFile)) {
      throw new Error(`TLS certificate for ${domain} is missing; run ng-agent cert issue ${domain} EMAIL or ng-agent cert import`);
    }
    run('openssl', ['x509', '-in', cert.certificateFile, '-noout', '-checkend', '86400']);
    const nameCheck = run('openssl', ['x509', '-in', cert.certificateFile, '-noout', '-checkhost', domain]);
    if (!nameCheck.includes('does match')) throw new Error(`TLS certificate does not match ${domain}`);
  }
  const target = safeResourcePath(resource.id);
  const backup = fs.existsSync(target) ? fs.readFileSync(target) : null;
  await flushUsageBeforeChange();
  fs.writeFileSync(target, `${JSON.stringify(resource, null, 2)}\n`, { mode: 0o600 });
  try {
    activateSingBox();
    lastEngineError = '';
  }
  catch (error) {
    if (backup) fs.writeFileSync(target, backup, { mode: 0o600 }); else fs.rmSync(target, { force: true });
    try { activateSingBox(); } catch { /* preserve original error */ }
    throw error;
  }
  return { artifacts };
}

async function deleteResource(payload) {
  const target = safeResourcePath(payload.resourceId);
  if (!fs.existsSync(target)) return { alreadyAbsent: true };
  const backup = fs.readFileSync(target);
  await flushUsageBeforeChange();
  fs.rmSync(target);
  try { activateSingBox(); }
  catch (error) { fs.writeFileSync(target, backup, { mode: 0o600 }); throw error; }
  return { removed: true };
}

async function execute(job) {
  log('Executing', job.id, job.action);
  try {
    let result;
    if (job.action === 'apply_resource') result = await applyResource(job.payload);
    else if (job.action === 'delete_resource') result = await deleteResource(job.payload);
    else throw new Error(`Unsupported job action: ${job.action}`);
    await request(`/api/agent/jobs/${encodeURIComponent(job.id)}/complete`, { method: 'POST', body: JSON.stringify({ success: true, result }) });
    log('Completed', job.id);
  } catch (error) {
    log('Failed', job.id, safeError(error.message));
    try { await request(`/api/agent/jobs/${encodeURIComponent(job.id)}/complete`, { method: 'POST', body: JSON.stringify({ success: false, error: safeError(error.message) }) }); }
    catch (reportError) { log('Unable to report failure', reportError.message); }
  }
}

function systemInfo() {
  const cpus = os.cpus();
  let memoryAvailable = os.freemem();
  try {
    const match = fs.readFileSync('/proc/meminfo', 'utf8').match(/^MemAvailable:\s+(\d+) kB/m);
    if (match) memoryAvailable = Number(match[1]) * 1024;
  } catch { /* Other supported systems still report free memory. */ }
  let diskTotal = null, diskAvailable = null;
  try {
    const disk = fs.statfsSync('/');
    diskTotal = disk.blocks * disk.bsize;
    diskAvailable = disk.bavail * disk.bsize;
  } catch { /* Older Node builds cannot report filesystem capacity. */ }
  return {
    hostname: os.hostname(), platform: os.platform(), release: os.release(), arch: os.arch(),
    uptimeSeconds: Math.round(os.uptime()), load1: Number(os.loadavg()[0].toFixed(2)),
    cpuCount: cpus.length, memoryTotal: os.totalmem(), memoryFree: os.freemem(),
    memoryAvailable, diskTotal, diskAvailable
  };
}

function engineHealth() {
  const resources = readResources();
  const installed = fs.existsSync(SINGBOX_BIN) && fs.existsSync(SINGBOX_STATS_BIN);
  const service = serviceHealth('nexusgate-sing-box');
  const result = {
    status: installed ? service.status : 'error',
    detail: installed ? service.detail : 'sing-box 及统计组件未安装；请在入口机器运行 ng-agent engine install',
    singBoxInstalled: installed,
    certificates: installedCertificates(),
    singBoxStatus: service.status,
  };
  if (result.status === 'ready' && resources.length) {
    try {
      const missing = missingHysteriaListeners(resources);
      if (missing.length) {
        result.status = 'error';
        result.detail = `Hysteria 2 UDP 未监听：${missing.join(', ')}`;
      }
    } catch (error) {
      result.status = 'error';
      result.detail = safeError(error.message);
    }
  }
  return result;
}

function installedCertificates() {
  const root = process.env.NG_TLS_DIR || '/etc/nexusgate/tls';
  let domains;
  try { domains = fs.readdirSync(root).slice(0, 50); } catch { return []; }
  const fingerprint = JSON.stringify(domains.map((domain) => {
    try {
      const cert = fs.statSync(path.join(root, domain, 'fullchain.pem'));
      const key = fs.statSync(path.join(root, domain, 'privkey.pem'));
      return [domain, cert.mtimeMs, cert.size, key.mtimeMs, key.size];
    } catch { return [domain]; }
  }));
  if (certificateCache.fingerprint === fingerprint && Date.now() - certificateCache.checkedAt < 300000)
    return [...certificateCache.domains];
  const valid = domains.filter((domain) => {
    if (!/^(?:[a-z0-9-]+\.)+[a-z]{2,63}$/.test(domain)) return false;
    const cert = path.join(root, domain, 'fullchain.pem');
    const key = path.join(root, domain, 'privkey.pem');
    if (!fs.existsSync(cert) || !fs.existsSync(key)) return false;
    const expiry = spawnSync('openssl', ['x509', '-in', cert, '-noout', '-checkend', '86400'],
      { encoding:'utf8', timeout:3000 });
    const host = spawnSync('openssl', ['x509', '-in', cert, '-noout', '-checkhost', domain],
      { encoding:'utf8', timeout:3000 });
    if (expiry.status !== 0 || host.status !== 0 || !/does match/.test(host.stdout)) return false;
    const verify = spawnSync('openssl', ['verify', '-verify_hostname', domain, '-untrusted', cert, cert],
      { encoding:'utf8', timeout:3000 });
    if (verify.status !== 0) return false;
    const certKey = spawnSync('openssl', ['x509', '-in', cert, '-pubkey', '-noout'], { encoding:'utf8', timeout:3000 });
    const privateKey = spawnSync('openssl', ['pkey', '-in', key, '-pubout'], { encoding:'utf8', timeout:3000 });
    return certKey.status === 0 && privateKey.status === 0 && certKey.stdout.trim() === privateKey.stdout.trim();
  });
  certificateCache = { fingerprint, checkedAt:Date.now(), domains:valid };
  return [...valid];
}

function serviceHealth(service) {
  const result = fs.existsSync('/run/systemd/system')
    ? spawnSync('systemctl', ['is-active', `${service}.service`], { encoding:'utf8', timeout:3000 })
    : spawnSync('rc-service', [service, 'status'], { encoding:'utf8', timeout:3000 });
  return { status: result.status === 0 ? 'ready' : 'error', detail: (result.status === 0 ? result.stdout : lastEngineError || result.stderr || result.stdout || '').trim().slice(0, 400) };
}

async function heartbeat() {
  const engine = engineHealth();
  // Report the resources actually held locally; the panel re-dispatches any
  // expected resource missing from this list (e.g. after a state wipe).
  const resources = readResources().map((resource) => resource && resource.id).filter((value) => typeof value === 'string' && value);
  await request('/api/agent/heartbeat', { method: 'POST', body: JSON.stringify({ version: VERSION, system: systemInfo(), engine, resources }) });
  try { fs.writeFileSync('/etc/nexusgate/last-heartbeat.json', `${JSON.stringify({ at:new Date().toISOString() })}\n`, { mode:0o600 }); }
  catch (error) { log('Heartbeat reached controller but readiness file failed', error.message); }
}

function parseUsageStats(output) {
  const parsed = JSON.parse(output);
  // 零流量时 Go 端输出 {"stat":null}，视为无数据而非格式错误。
  const statList = parsed && typeof parsed === 'object' ? (parsed.stat == null ? [] : parsed.stat) : null;
  if (!Array.isArray(statList))
    throw new Error('统计接口返回的 JSON stat 格式无效');
  const inbounds = new Map(), users = new Map();
  for (const stat of statList) {
    const inboundMatch = /^inbound>>>(.+)>>>traffic>>>(uplink|downlink)$/.exec(stat.name || '');
    const userMatch = inboundMatch ? null : /^user>>>(.+)>>>traffic>>>(uplink|downlink)$/.exec(stat.name || '');
    if (!inboundMatch && !userMatch) continue;
    // Protobuf JSON omits a scalar with its default value (zero). The engine can
    // therefore return a named counter without a `value` field until traffic.
    const value = Object.hasOwn(stat, 'value') ? Number(stat.value) : 0;
    if (!Number.isSafeInteger(value) || value < 0 || stat.value === null || stat.value === '') {
      const reason = typeof stat.value === 'string' && /^-?\d+$/.test(stat.value)
        ? (stat.value.startsWith('-') ? '负数' : '超出安全整数范围') : `格式 ${typeof stat.value}`;
      throw new Error(`入口计数异常（${reason}）：${stat.name}；请运行 ng-agent doctor 查看原始统计`);
    }
    const bucket = inboundMatch ? inbounds : users;
    const key = (inboundMatch || userMatch)[1];
    const direction = (inboundMatch || userMatch)[2];
    let entry = bucket.get(key);
    if (!entry) { entry = { uplink: 0, downlink: 0 }; bucket.set(key, entry); }
    entry[direction] = value;
  }
  return {
    inbounds: [...inbounds].map(([tag, counters]) => ({ tag, ...counters })),
    users: [...users].map(([name, counters]) => ({ name, ...counters })),
  };
}

function serviceEpoch(service = 'nexusgate-sing-box') {
  let pid;
  if (fs.existsSync('/run/systemd/system')) {
    const result = spawnSync('systemctl', ['show', '-p', 'MainPID', '--value', `${service}.service`], { encoding:'utf8', timeout:3000 });
    if (result.status === 0) pid = Number(result.stdout.trim());
  } else {
    try { pid = Number(fs.readFileSync(`/run/${service}.pid`, 'utf8').trim()); } catch { /* service is not running */ }
  }
  if (!Number.isSafeInteger(pid) || pid <= 0) return null;
  try {
    // PID alone can be reused; the Linux process start tick distinguishes a new sing-box instance.
    const stat = fs.readFileSync(`/proc/${pid}/stat`, 'utf8');
    const startTick = stat.slice(stat.lastIndexOf(') ') + 2).trim().split(/\s+/)[19];
    return startTick ? `${pid}:${startTick}` : null;
  } catch { return null; }
}

function udpPortsForPid(pid) {
  const sockets = new Set();
  for (const fd of fs.readdirSync(`/proc/${pid}/fd`)) {
    try {
      const match = fs.readlinkSync(`/proc/${pid}/fd/${fd}`).match(/^socket:\[(\d+)\]$/);
      if (match) sockets.add(match[1]);
    } catch { /* a descriptor closed during inspection */ }
  }
  const ports = new Set();
  for (const table of ['/proc/net/udp', '/proc/net/udp6']) {
    const rows = fs.readFileSync(table, 'utf8').trim().split('\n').slice(1);
    for (const row of rows) {
      const fields = row.trim().split(/\s+/);
      if (fields[3] === '07' && sockets.has(fields[9]))
        ports.add(parseInt(fields[1].split(':').at(-1), 16));
    }
  }
  return ports;
}

function missingHysteriaListeners(resources) {
  const expected = [...new Set(resources.flatMap((item) => item.inbounds || [])
    .filter((item) => item.type === 'hysteria2').map((item) => item.listen_port))];
  if (!expected.length) return [];
  const epoch = serviceEpoch('nexusgate-sing-box');
  if (!epoch) throw new Error('无法确认 sing-box 进程，无法验证 Hysteria 2 UDP 监听');
  const actual = udpPortsForPid(Number(epoch.split(':')[0]));
  return expected.filter((port) => !actual.has(port));
}

function assertHysteriaListeners(resources) {
  if (!resources.some((item) => (item.inbounds || []).some((inbound) => inbound.protocol === 'hysteria'))) return;
  for (let attempt = 0; attempt < 10; attempt++) {
    try {
      const missing = missingHysteriaListeners(resources);
      if (!missing.length) return;
    } catch { /* systemd may not have published its MainPID yet */ }
    if (attempt < 9) spawnSync('sleep', ['0.2']);
  }
  const missing = missingHysteriaListeners(resources);
  throw new Error(`Hysteria 2 未监听 UDP ${missing.join(', ')}；检查 sing-box 配置与本机端口冲突`);
}

function queryUsage() {
  // Count client ingress once; counting the exit as well doubles forwarded traffic.
  const resources = readResources().filter((resource) => resource.meta && resource.meta.metricsTag &&
    resource.meta.kind === 'node');
  const samples = [], errors = [], engineErrors = [];
  if (!resources.length) return { samples, error: null, engineErrors };
  const service = 'nexusgate-sing-box';
  const before = serviceEpoch(service);
  const port = process.env.NG_SINGBOX_API_PORT || 10086;
  const result = spawnSync(SINGBOX_STATS_BIN, [`--server=127.0.0.1:${port}`], { encoding: 'utf8', timeout: 15000 });
  const after = serviceEpoch(service);
  let failed = false;
  if (before && after && before !== after) { errors.push('sing-box restarted during statistics query'); engineErrors.push('sing-box'); failed = true; }
  else if (result.status !== 0) { errors.push(`sing-box statsquery failed: ${safeError(result.stderr || result.error?.message || result.stdout).slice(0, 160)}`); engineErrors.push('sing-box'); failed = true; }
  const parsed = { inbounds: [], users: [] };
  if (!failed) {
    try {
      const part = parseUsageStats(result.stdout);
      parsed.inbounds.push(...part.inbounds);
      parsed.users.push(...part.users);
    } catch (error) { errors.push(`sing-box: ${safeError(error.message)}`); engineErrors.push('sing-box'); failed = true; }
  }
  if (!failed) {
    const epoch = `sing-box:${before || serviceEpoch(service) || 'unknown'}`;
    const tagNodes = new Map(resources.map((item) => [item.meta.metricsTag, item.meta.nodeId]));
    for (const inbound of parsed.inbounds) {
      const nodeId = tagNodes.get(inbound.tag);
      if (!nodeId) continue;
      samples.push({ kind: 'inbound', id: nodeId, uplink: inbound.uplink, downlink: inbound.downlink, epoch });
    }
    for (const user of parsed.users) {
      const match = /^ng:(.+)$/.exec(user.name);
      if (!match) continue;
      samples.push({ kind: 'user', id: match[1], uplink: user.uplink, downlink: user.downlink, epoch });
    }
    if (!parsed.inbounds.length && !parsed.users.length) errors.push('sing-box 入口尚无统计记录');
  }
  return { samples, error: errors.join('; ') || null, engineErrors };
}

function readSingBoxObservations() {
  if (!fs.existsSync(SINGBOX_LOG)) return { observations: [], offset: null };
  let cursor = { offset: 0 };
  try { cursor = JSON.parse(fs.readFileSync(SINGBOX_CURSOR, 'utf8')); } catch { /* first read */ }
  const size = fs.statSync(SINGBOX_LOG).size;
  if (size < cursor.offset) cursor.offset = 0;
  const start = Math.max(cursor.offset, size - 2 * 1024 * 1024);
  if (start >= size) return { observations: [], offset: null };
  const buffer = Buffer.alloc(size - start);
  const fd = fs.openSync(SINGBOX_LOG, 'r');
  try { fs.readSync(fd, buffer, 0, buffer.length, start); } finally { fs.closeSync(fd); }
  const end = buffer.lastIndexOf(10);
  if (end < 0) return { observations: [], offset: null };
  // AnyTLS 用户名本身就是 ng:<customerId>，身份直接写在日志行里，无需 tag 映射。
  const entryTags = new Set(readResources().filter((item) =>
    item.meta?.kind === 'node').map((item) => item.meta.metricsTag));
  const sourceById = new Map(), unique = new Map();
  for (const line of buffer.subarray(0, end).toString('utf8').split('\n')) {
    const context = line.match(/\[(\d+)\s+\d+ms\]\s+inbound\/anytls\[([^\]]+)\]:\s+(.*)$/);
    if (!context || !entryTags.has(context[2])) continue;
    const key = `${context[2]}:${context[1]}`;
    const source = context[3].match(/inbound connection from\s+(\[[^\]]+\]|[\d.]+):\d+/);
    if (source) sourceById.set(key, source[1].replace(/^\[|\]$/g, ''));
    // Only count authenticated AnyTLS sessions, never unauthenticated port probes.
    const authed = context[3].match(/\[ng:([^\]]+)\] inbound connection to /);
    if (authed && sourceById.has(key)) {
      const customerId = authed[1];
      const ip = sourceById.get(key);
      unique.set(`${customerId}|${ip}`, { customerId, ip });
    }
  }
  return { observations: [...unique.values()], offset: start + end + 1 };
}

async function usageLoop() {
  while (true) {
    await sleep(60000);
    try {
      let report;
      try { report = queryUsage(); }
      catch (error) { report = { samples: [], error: safeError(error.message) }; }
      await request('/api/agent/usage', { method: 'POST', body: JSON.stringify(report) });
    } catch (error) { log('Usage report failed', safeError(error.message)); }
    try {
      const sing = readSingBoxObservations();
      if (sing.observations.length) await request('/api/agent/observations', { method: 'POST', body: JSON.stringify({ observations: sing.observations }) });
      // A failed upload must not advance the cursor; retry the same log lines next cycle.
      if (sing.offset !== null) fs.writeFileSync(SINGBOX_CURSOR, `${JSON.stringify({ offset: sing.offset })}\n`, { mode: 0o600 });
    } catch (error) { log('IP observation report failed', safeError(error.message)); }
  }
}

async function pollLoop() {
  while (true) {
    try {
      const result = await request('/api/agent/poll', { method: 'POST', body: '{}' });
      if (result.job) await execute(result.job);
      else await sleep(POLL_MS);
    } catch (error) { log('Poll failed', error.message); await sleep(Math.max(POLL_MS, 10000)); }
  }
}

async function main() {
  if (!CONTROLLER || !AGENT_KEY) throw new Error('NG_CONTROLLER and NG_AGENT_KEY are required');
  ensureDirectories();
  log(`NexusGate Agent v${VERSION} starting`);
  // Keep the control channel alive even when a stale node configuration cannot start.
  // The administrator can then see the engine error and a repair job can be claimed.
  try { activateSingBox(); } catch (error) { lastEngineError = safeError(error.message); log('Initial sing-box activation failed', lastEngineError); }
  try { await heartbeat(); } catch (error) { log('Initial heartbeat failed', error.message); }
  setInterval(() => heartbeat().catch((error) => log('Heartbeat failed', error.message)), 30000).unref();
  usageLoop();
  await pollLoop();
}

if (require.main === module) {
  const action = process.argv[2] === 'flush-usage' ? flushAllUsageBeforeRestart() : main();
  action.catch((error) => { console.error(safeError(error.message)); process.exitCode = 1; });
}

module.exports = { activateSingBox, combinedSingBoxConfig, applyResource, realityKey,
  readSingBoxObservations, parseUsageStats, queryUsage, installedCertificates,
  udpPortsForPid, missingHysteriaListeners, systemInfo };