feat(panel): reskin backend — node sortIndex, presence/speed, /api/system, credentials, subscription aggregation (Node+Go parity)

lib/extsub.js

@@ -0,0 +1,284 @@
+'use strict';
+
+// External subscription aggregation (panel-go/extsub.go mirrors this
+// file; design: SUB-AGGREGATION-DESIGN.md). The panel fetches a
+// customer's external subscription sources, parses link lists, caches
+// the result in data.subSourceCache and merges manual + external items
+// into the subscription output behind the customer's own nodes.
+//
+// Safety: http/https only, SSRF guard rejects non-global IPs unless
+// settings.allowPrivateSubSources is on, every redirect hop is
+// re-validated, the connection is pinned to the validated IP (no DNS
+// rebinding window), 10s timeout, 1MiB body cap, 200 items per source.
+
+const http = require('node:http');
+const https = require('node:https');
+const dns = require('node:dns/promises');
+const net = require('node:net');
+const crypto = require('node:crypto');
+const { isGlobalIPv4, isGlobalIPv6 } = require('./sysstat');
+
+const LINK_SCHEMES = ['vless:', 'ss:', 'hysteria2:', 'anytls:', 'socks5:'];
+const MAX_BODY_BYTES = 1024 * 1024;
+const MAX_ITEMS = 200;
+const FETCH_TIMEOUT_MS = 10000;
+const MAX_REDIRECTS = 3;
+const DEFAULT_UA = 'v2rayN/6.55';
+
+function isGlobalIP(ip) {
+  const family = net.isIP(ip);
+  if (family === 4) return isGlobalIPv4(ip);
+  if (family === 6) return isGlobalIPv6(ip.replace(/^\[|\]$/g, ''));
+  return false;
+}
+
+function validateLinkUri(uri) {
+  if (typeof uri !== 'string' || !uri || uri.length > 2048) return false;
+  let url;
+  try { url = new URL(uri); } catch { return false; }
+  return LINK_SCHEMES.includes(url.protocol) && Boolean(url.hostname);
+}
+
+// Parse a fetched body into a normalized link list. Accepts a base64
+// blob or plaintext lines; unsupported schemes and unparseable lines are
+// dropped and counted, never fatal.
+function parseLinkList(body) {
+  let text = String(body == null ? '' : body);
+  const compact = text.replace(/\s+/g, '');
+  if (compact.length >= 16 && /^[A-Za-z0-9+/=_-]+$/.test(compact)) {
+    try {
+      const normalized = compact.replace(/-/g, '+').replace(/_/g, '/');
+      const decoded = Buffer.from(normalized, 'base64').toString('utf8');
+      if (decoded.includes('://')) text = decoded;
+    } catch { /* keep plaintext */ }
+  }
+  const items = [];
+  let dropped = 0;
+  for (const rawLine of text.split(/\r?\n/)) {
+    const line = rawLine.trim();
+    if (!line) continue;
+    if (items.length >= MAX_ITEMS) { dropped += 1; continue; }
+    if (!validateLinkUri(line)) { dropped += 1; continue; }
+    let name = '';
+    const hash = line.lastIndexOf('#');
+    if (hash >= 0) {
+      try { name = decodeURIComponent(line.slice(hash + 1)); } catch { name = line.slice(hash + 1); }
+    }
+    items.push({ uri: line, name: name.slice(0, 120) });
+  }
+  return { items, dropped };
+}
+
+async function resolveChecked(hostname, allowPrivate) {
+  let addresses;
+  if (net.isIP(hostname)) {
+    addresses = [hostname];
+  } else {
+    const records = await dns.lookup(hostname, { all: true });
+    addresses = records.map((record) => record.address);
+  }
+  if (!addresses.length) throw new Error('域名解析无结果');
+  if (!allowPrivate && addresses.some((ip) => !isGlobalIP(ip))) {
+    throw new Error('订阅源指向非公网地址，已拒绝');
+  }
+  return addresses[0];
+}
+
+function fetchOnce(urlString, { userAgent, allowPrivate, deadline }) {
+  return new Promise((resolve, reject) => {
+    (async () => {
+      const url = new URL(urlString);
+      if (!['http:', 'https:'].includes(url.protocol)) throw new Error('仅支持 http/https 订阅源');
+      const hostname = url.hostname.replace(/^\[|\]$/g, '');
+      const ip = await resolveChecked(hostname, allowPrivate);
+      const isTls = url.protocol === 'https:';
+      const port = Number(url.port || (isTls ? 443 : 80));
+      const hostHeader = url.port ? `${hostname}:${url.port}` : hostname;
+      const options = {
+        hostname: ip, port, path: `${url.pathname}${url.search}`, method: 'GET',
+        headers: { host: hostHeader, 'user-agent': userAgent || DEFAULT_UA, accept: '*/*' },
+        timeout: Math.max(1000, deadline - Date.now()),
+      };
+      if (isTls) options.servername = hostname;
+      const transport = isTls ? https : http;
+      const req = transport.request(options, (res) => {
+        if (res.statusCode >= 300 && res.statusCode < 400 && res.headers.location) {
+          res.resume();
+          resolve({ redirect: new URL(res.headers.location, urlString).toString() });
+          return;
+        }
+        if (res.statusCode !== 200) {
+          res.resume();
+          reject(new Error(`订阅源返回 HTTP ${res.statusCode}`));
+          return;
+        }
+        const chunks = [];
+        let size = 0;
+        res.on('data', (chunk) => {
+          size += chunk.length;
+          if (size > MAX_BODY_BYTES) {
+            req.destroy(new Error('订阅源响应超过 1MiB'));
+            return;
+          }
+          chunks.push(chunk);
+        });
+        res.on('end', () => resolve({ body: Buffer.concat(chunks).toString('utf8') }));
+        res.on('error', reject);
+      });
+      req.on('timeout', () => req.destroy(new Error('订阅源请求超时')));
+      req.on('error', reject);
+      req.end();
+    })().catch(reject);
+  });
+}
+
+async function fetchSourceBody(source, settings) {
+  const allowPrivate = Boolean(settings && settings.allowPrivateSubSources);
+  const deadline = Date.now() + FETCH_TIMEOUT_MS;
+  let current = source.url;
+  for (let hop = 0; hop <= MAX_REDIRECTS; hop += 1) {
+    if (Date.now() > deadline) throw new Error('订阅源请求超时');
+    const result = await fetchOnce(current, { userAgent: source.userAgent, allowPrivate, deadline });
+    if (!result.redirect) return result.body;
+    if (hop === MAX_REDIRECTS) throw new Error('订阅源重定向次数过多');
+    current = result.redirect;
+  }
+  throw new Error('订阅源请求失败');
+}
+
+// Fetch + parse one source. Never throws: returns { ok, items, dropped, error }.
+async function fetchExternalSource(source, settings) {
+  try {
+    const body = await fetchSourceBody(source, settings || {});
+    const { items, dropped } = parseLinkList(body);
+    return { ok: true, items, dropped, error: null };
+  } catch (error) {
+    return { ok: false, items: [], dropped: 0, error: String(error && error.message || error).slice(0, 240) };
+  }
+}
+
+function refreshMinutes(settings) {
+  const value = Number(settings && settings.externalSubRefreshMinutes);
+  if (!Number.isFinite(value) || value <= 0) return 60;
+  return Math.min(720, Math.max(15, value));
+}
+
+// Deterministic ±10% jitter from the source id so both implementations
+// and restarts agree, and sources do not all fire in the same pass.
+function jitterFactor(sourceId) {
+  const hash = crypto.createHash('sha256').update(String(sourceId)).digest();
+  return 1 + ((hash[0] % 21) - 10) / 100;
+}
+
+// Collect sources due for a refresh pass: [{ customerId, source }].
+function dueSources(data, nowMs, limit = 3) {
+  const out = [];
+  const cache = data.subSourceCache || {};
+  for (const customer of data.customers) {
+    for (const source of customer.externalSources || []) {
+      if (!source.enabled) continue;
+      const entry = cache[source.id];
+      if (!entry || !entry.fetchedAt) { out.push({ customerId: customer.id, source }); continue; }
+      const base = refreshMinutes(data.settings) * 60000;
+      const backoff = Math.min(Math.pow(2, Math.min(entry.failCount || 0, 10)) * base, 24 * 3600 * 1000);
+      const due = Date.parse(entry.fetchedAt) + backoff * jitterFactor(source.id);
+      if (nowMs >= due) out.push({ customerId: customer.id, source });
+      if (out.length >= limit) return out;
+    }
+    if (out.length >= limit) break;
+  }
+  return out.slice(0, limit);
+}
+
+// Apply one fetch result to the cache inside a store transaction draft.
+function applyFetchResult(data, sourceId, result, nowIso) {
+  data.subSourceCache ||= {};
+  const previous = data.subSourceCache[sourceId] || { fetchedAt: null, okAt: null, failCount: 0, lastError: null, droppedCount: 0, items: [] };
+  const entry = { ...previous, fetchedAt: nowIso };
+  if (result.ok) {
+    entry.okAt = nowIso;
+    entry.failCount = 0;
+    entry.lastError = null;
+    entry.droppedCount = result.dropped;
+    entry.items = result.items;
+  } else {
+    entry.failCount = (previous.failCount || 0) + 1;
+    entry.lastError = result.error;
+  }
+  data.subSourceCache[sourceId] = entry;
+  return entry;
+}
+
+// ---------- save-time normalization (shared by create/PATCH) ----------
+
+function cleanTextLocal(value, max) {
+  return String(value == null ? '' : value).trim().slice(0, max);
+}
+
+function normalizeExternalSources(value, idFn) {
+  if (value == null) return undefined;
+  if (!Array.isArray(value)) { const e = new Error('外部订阅格式无效'); e.statusCode = 400; throw e; }
+  if (value.length > 5) { const e = new Error('每个客户最多 5 个外部订阅'); e.statusCode = 400; throw e; }
+  return value.map((raw) => {
+    const item = raw && typeof raw === 'object' ? raw : {};
+    const url = cleanTextLocal(item.url, 2048);
+    let parsed;
+    try { parsed = new URL(url); } catch { parsed = null; }
+    if (!parsed || !['http:', 'https:'].includes(parsed.protocol) || !parsed.hostname) {
+      const e = new Error('外部订阅地址无效'); e.statusCode = 400; throw e;
+    }
+    const next = {
+      id: typeof item.id === 'string' && item.id ? item.id.slice(0, 80) : idFn('src'),
+      url, remark: cleanTextLocal(item.remark, 80), enabled: item.enabled !== false,
+    };
+    const ua = cleanTextLocal(item.userAgent, 120);
+    if (ua) next.userAgent = ua;
+    return next;
+  });
+}
+
+function normalizeManualLinks(value, idFn) {
+  if (value == null) return undefined;
+  if (!Array.isArray(value)) { const e = new Error('外部链接格式无效'); e.statusCode = 400; throw e; }
+  if (value.length > 50) { const e = new Error('每个客户最多 50 条外部链接'); e.statusCode = 400; throw e; }
+  return value.map((raw) => {
+    const item = raw && typeof raw === 'object' ? raw : {};
+    const uri = cleanTextLocal(item.uri, 2048);
+    if (!validateLinkUri(uri)) { const e = new Error('外部链接无效或协议不受支持'); e.statusCode = 400; throw e; }
+    return {
+      id: typeof item.id === 'string' && item.id ? item.id.slice(0, 80) : idFn('lnk'),
+      uri, remark: cleanTextLocal(item.remark, 80), enabled: item.enabled !== false,
+    };
+  });
+}
+
+// ---------- merge helpers (used by lib/subscriptions.js) ----------
+
+function dedupKey(uri) {
+  try {
+    const url = new URL(uri);
+    let userinfo = '';
+    try { userinfo = `${decodeURIComponent(url.username)}:${decodeURIComponent(url.password)}`; }
+    catch { userinfo = `${url.username}:${url.password}`; }
+    const hash = crypto.createHash('sha256').update(userinfo).digest('hex').slice(0, 16);
+    return `${url.protocol}//${url.hostname.toLowerCase()}:${url.port}#${hash}`;
+  } catch { return null; }
+}
+
+function withName(uri, name) {
+  const separator = uri.lastIndexOf('#');
+  const base = separator >= 0 ? uri.slice(0, separator) : uri;
+  return `${base}#${encodeURIComponent(name)}`;
+}
+
+function nameFromUri(uri) {
+  const hash = uri.lastIndexOf('#');
+  if (hash < 0) return '';
+  try { return decodeURIComponent(uri.slice(hash + 1)); } catch { return uri.slice(hash + 1); }
+}
+
+module.exports = {
+  LINK_SCHEMES, DEFAULT_UA, validateLinkUri, parseLinkList, fetchExternalSource,
+  refreshMinutes, dueSources, applyFetchResult, normalizeExternalSources, normalizeManualLinks,
+  dedupKey, withName, nameFromUri,
+};

lib/orchestrator.js

@@ -240,6 +240,9 @@ function validateNodeDeploy(data, node) {
   else validateProtocolPair(node.protocol, node.exitProtocol);
   const server = assertFound(data.servers.find((s) => s.id === node.serverId), '服务器');
   const exitServer = direct ? null : assertFound(data.servers.find((s) => s.id === node.exitServerId), '出口服务器');
+  for (const target of direct ? [server] : [server, exitServer]) {
+    if (target.enabled === false) { const error = new Error(`设备 ${target.name} 已停用`); error.statusCode = 409; throw error; }
+  }
   if (!activeNodeClients(data, node).length) { const error = new Error('请先给节点分配至少一个客户再部署'); error.statusCode = 400; throw error; }
   if (['vless-ws-tls', 'hysteria2', 'anytls'].includes(node.protocol)) {
     if (!server.tlsDomain) { const error = new Error(`${node.protocol} 需要先填写设备 TLS 域名并在节点上安装证书`); error.statusCode = 400; throw error; }
@@ -281,6 +284,8 @@ function validateNodeDeploy(data, node) {
 // gates on "no in-flight jobs" instead and passes failInFlight: false.
 async function reconcileServerData(data, serverId, actor, { onlyNodeIds = null, failInFlight = true } = {}) {
   const server = assertFound(data.servers.find((item) => item.id === serverId), '服务器');
+  // A disabled device receives no new deployments, including reconciles.
+  if (server.enabled === false) return 0;
   if (failInFlight) {
     for (const job of data.jobs.filter((item) => item.serverId === serverId && ['queued', 'running'].includes(item.status))) {
       job.status = 'failed'; job.error = 'Agent 重新注册，旧任务已由对账任务替代'; job.updatedAt = nowIso();
@@ -314,6 +319,90 @@ async function reconcileServerData(data, serverId, actor, { onlyNodeIds = null,
   return count;
 }
 
+// Data-level assign/unassign (the transaction-free cores of the
+// Orchestrator methods) so the customer form's node alignment can run
+// inside the customer create/PATCH transaction.
+async function assignCustomerData(data, nodeId, customerId, actor = 'admin') {
+  const node = assertFound(data.nodes.find((n) => n.id === nodeId), '节点');
+  const customer = assertFound(data.customers.find((c) => c.id === customerId), '客户');
+  if (customer.status !== 'active') { const error = new Error('只能给状态正常的客户分配节点'); error.statusCode = 409; throw error; }
+  let nc = data.nodeClients.find((entry) => entry.nodeId === nodeId && entry.customerId === customerId);
+  if (nc && nc.status === 'active') { const error = new Error('该客户已分配到此节点'); error.statusCode = 409; throw error; }
+  if (nc) {
+    nc.status = 'active'; nc.updatedAt = nowIso();
+  } else {
+    nc = {
+      id: id('ncl'), nodeId, customerId, credentials: newNodeClientCredentials(node.protocol),
+      status: 'active', createdAt: nowIso(), updatedAt: nowIso(),
+    };
+    data.nodeClients.push(nc);
+  }
+  // A manually undeployed node stays draft until explicitly deployed.
+  // Automatic teardowns (suspend) are revived when clients return.
+  if ((node.status === 'removing' || node.status === 'draft') && !node.autoRemoving) {
+    // record the assignment only
+  } else {
+    if (node.status === 'removing' && node.autoRemoving) {
+      cancelQueuedDeletes(data, node.id, '已取消：节点恢复服务');
+      node.autoRemoving = false;
+    }
+    await rebuildNode(data, node);
+  }
+  audit(data, actor, 'assign_customer', node.id, { customerId, customer: customer.name });
+  return structuredClone(nc);
+}
+
+async function unassignCustomerData(data, nodeId, customerId, actor = 'admin') {
+  const node = assertFound(data.nodes.find((n) => n.id === nodeId), '节点');
+  const index = data.nodeClients.findIndex((entry) => entry.nodeId === nodeId && entry.customerId === customerId);
+  if (index < 0) { const error = new Error('该客户未分配到此节点'); error.statusCode = 404; throw error; }
+  data.nodeClients.splice(index, 1);
+  if (node.status !== 'draft' && node.status !== 'removing') await rebuildNode(data, node);
+  audit(data, actor, 'unassign_customer', node.id, { customerId });
+  return true;
+}
+
+// Full-set alignment used by the customer form's 关联入站 multi-select:
+// bindings not in desiredIds are removed, missing ones are created.
+// Suspended customers get suspended bindings (mirroring their resource
+// state) without touching node resources.
+async function alignCustomerNodes(data, customer, desiredIds, actor = 'admin') {
+  const desired = [...new Set((desiredIds || []).map(String).filter(Boolean))];
+  for (const nodeId of desired) {
+    const node = data.nodes.find((n) => n.id === nodeId && !n.pendingDelete);
+    if (!node) { const error = new Error('节点不存在'); error.statusCode = 400; throw error; }
+  }
+  const current = data.nodeClients.filter((nc) => nc.customerId === customer.id);
+  const rebuild = new Set();
+  for (const nc of [...current]) {
+    if (desired.includes(nc.nodeId)) continue;
+    data.nodeClients.splice(data.nodeClients.indexOf(nc), 1);
+    audit(data, actor, 'unassign_customer', nc.nodeId, { customerId: customer.id });
+    rebuild.add(nc.nodeId);
+  }
+  for (const nodeId of desired) {
+    if (current.some((nc) => nc.nodeId === nodeId)) continue;
+    const node = data.nodes.find((n) => n.id === nodeId);
+    const nc = {
+      id: id('ncl'), nodeId, customerId: customer.id, credentials: newNodeClientCredentials(node.protocol),
+      status: customer.status === 'active' ? 'active' : 'suspended', createdAt: nowIso(), updatedAt: nowIso(),
+    };
+    data.nodeClients.push(nc);
+    audit(data, actor, 'assign_customer', node.id, { customerId: customer.id, customer: customer.name });
+    if (nc.status === 'active') rebuild.add(nodeId);
+  }
+  for (const nodeId of rebuild) {
+    const node = data.nodes.find((n) => n.id === nodeId);
+    if (!node) continue;
+    if ((node.status === 'removing' || node.status === 'draft') && !node.autoRemoving) continue;
+    if (node.status === 'removing' && node.autoRemoving) {
+      cancelQueuedDeletes(data, node.id, '已取消：节点恢复服务');
+      node.autoRemoving = false;
+    }
+    await rebuildNode(data, node);
+  }
+}
+
 class Orchestrator {
   constructor(store) {
     this.store = store;
@@ -365,47 +454,11 @@ class Orchestrator {
   }
 
   async assignCustomer(nodeId, customerId, actor = 'admin') {
-    return this.store.transaction(async (data) => {
-      const node = assertFound(data.nodes.find((n) => n.id === nodeId), '节点');
-      const customer = assertFound(data.customers.find((c) => c.id === customerId), '客户');
-      if (customer.status !== 'active') { const error = new Error('只能给状态正常的客户分配节点'); error.statusCode = 409; throw error; }
-      let nc = data.nodeClients.find((entry) => entry.nodeId === nodeId && entry.customerId === customerId);
-      if (nc && nc.status === 'active') { const error = new Error('该客户已分配到此节点'); error.statusCode = 409; throw error; }
-      if (nc) {
-        nc.status = 'active'; nc.updatedAt = nowIso();
-      } else {
-        nc = {
-          id: id('ncl'), nodeId, customerId, credentials: newNodeClientCredentials(node.protocol),
-          status: 'active', createdAt: nowIso(), updatedAt: nowIso(),
-        };
-        data.nodeClients.push(nc);
-      }
-      // A manually undeployed node stays draft until explicitly deployed.
-      // Automatic teardowns (suspend) are revived when clients return.
-      if ((node.status === 'removing' || node.status === 'draft') && !node.autoRemoving) {
-        // record the assignment only
-      } else {
-        if (node.status === 'removing' && node.autoRemoving) {
-          cancelQueuedDeletes(data, node.id, '已取消：节点恢复服务');
-          node.autoRemoving = false;
-        }
-        await rebuildNode(data, node);
-      }
-      audit(data, actor, 'assign_customer', node.id, { customerId, customer: customer.name });
-      return structuredClone(nc);
-    });
+    return this.store.transaction(async (data) => assignCustomerData(data, nodeId, customerId, actor));
   }
 
   async unassignCustomer(nodeId, customerId, actor = 'admin') {
-    return this.store.transaction(async (data) => {
-      const node = assertFound(data.nodes.find((n) => n.id === nodeId), '节点');
-      const index = data.nodeClients.findIndex((entry) => entry.nodeId === nodeId && entry.customerId === customerId);
-      if (index < 0) { const error = new Error('该客户未分配到此节点'); error.statusCode = 404; throw error; }
-      data.nodeClients.splice(index, 1);
-      if (node.status !== 'draft' && node.status !== 'removing') await rebuildNode(data, node);
-      audit(data, actor, 'unassign_customer', node.id, { customerId });
-      return true;
-    });
+    return this.store.transaction(async (data) => unassignCustomerData(data, nodeId, customerId, actor));
   }
 
   async reconcileServer(serverId, actor = 'system', options = {}) {
@@ -516,8 +569,27 @@ class Orchestrator {
           const key = `srv:${serverId}`;
           customer.serverUsage = customer.serverUsage || {};
           const slot = customer.serverUsage[key] || (customer.serverUsage[key] = {});
+          const prevCounters = slot.usageCounters;
+          const prevSampleAt = slot.lastSampleAt;
           const delta = applyDelta(slot, uplink, downlink, sampleEpoch);
           if (!delta) continue;
+          slot.lastSampleAt = reportedAt;
+          // Presence + approximate speed from the same delta (no extra
+          // sampling channel): a positive delta is activity; speed is the
+          // delta over the gap since the previous processed sample.
+          const restarted = Boolean(sampleEpoch && prevCounters && prevCounters.epoch && sampleEpoch !== prevCounters.epoch);
+          if (delta.upDelta + delta.downDelta > 0) {
+            customer.lastOnlineAt = reportedAt;
+            slot.lastActiveAt = reportedAt;
+          }
+          if (prevSampleAt && !restarted) {
+            const dt = (Date.parse(reportedAt) - Date.parse(prevSampleAt)) / 1000;
+            if (dt >= 5) {
+              slot.speedUpBps = Math.round(delta.upDelta / dt);
+              slot.speedDownBps = Math.round(delta.downDelta / dt);
+              slot.speedAt = reportedAt;
+            }
+          }
           customer.usedUplinkBytes = (customer.usedUplinkBytes || 0) + delta.upDelta;
           customer.usedDownlinkBytes = (customer.usedDownlinkBytes || 0) + delta.downDelta;
           customer.usedBytes = (customer.usedBytes || 0) + delta.upDelta + delta.downDelta;
@@ -528,8 +600,21 @@ class Orchestrator {
           // transport and is never double counted (same rule as before).
           const node = data.nodes.find((item) => item.id === sample.id && item.serverId === serverId);
           if (!node) continue;
+          const prevCounters = node.usageCounters;
+          const prevSampleAt = node.lastSampleAt;
           const delta = applyDelta(node, uplink, downlink, sampleEpoch);
           if (!delta) continue;
+          node.lastSampleAt = reportedAt;
+          const restarted = Boolean(sampleEpoch && prevCounters && prevCounters.epoch && sampleEpoch !== prevCounters.epoch);
+          if (delta.upDelta + delta.downDelta > 0) node.lastActiveAt = reportedAt;
+          if (prevSampleAt && !restarted) {
+            const dt = (Date.parse(reportedAt) - Date.parse(prevSampleAt)) / 1000;
+            if (dt >= 5) {
+              node.speedUpBps = Math.round(delta.upDelta / dt);
+              node.speedDownBps = Math.round(delta.downDelta / dt);
+              node.speedAt = reportedAt;
+            }
+          }
           const metered = node.meteredTraffic || { uplink: 0, downlink: 0 };
           node.meteredTraffic = { uplink: metered.uplink + delta.upDelta, downlink: metered.downlink + delta.downDelta };
           node.lastUsageAt = reportedAt;
@@ -541,4 +626,5 @@ class Orchestrator {
 
 module.exports = { Orchestrator, allocatePort, selectPort, nodePortInUse, queueJob, audit, id, nowIso, cancelQueuedDeletes, resourceForNode,
   resourceForNodeExit, rebuildNode, applyDelta, forwardNode, activeNodeClients,
-  suspendCustomerResources, resumeCustomerResources, HEARTBEAT_RECONCILE_INTERVAL_MS };
+  suspendCustomerResources, resumeCustomerResources, HEARTBEAT_RECONCILE_INTERVAL_MS,
+  assignCustomerData, unassignCustomerData, alignCustomerNodes };

lib/presence.js

@@ -0,0 +1,81 @@
+'use strict';
+
+// Presence & speed derivation (panel-go/presence.go mirrors this file).
+//
+// Signals (agent protocol unchanged):
+// - Agent usage reports carry cumulative per-user counters aggregated per
+//   server (NOT per node) and cumulative per-inbound (per node) counters.
+//   A processed sample whose delta is > 0 marks activity: per-server for
+//   the customer (serverUsage slot lastActiveAt + customer.lastOnlineAt),
+//   per-node for the inbound (node.lastActiveAt).
+// - AnyTLS observations confirm customer activity and also stamp
+//   customer.lastOnlineAt at ingest time.
+// A customer counts as "online on node N" when they hold an active
+// binding on N and their last activity on N's server is inside the
+// window. With several nodes on one server this attributes the activity
+// to each of them — the reports cannot separate nodes on a server.
+
+const PRESENCE_WINDOW_MS = 180000; // 3 minutes
+const SPEED_STALE_MS = 150000;
+
+function parseMs(value) {
+  if (!value) return 0;
+  const ms = Date.parse(value);
+  return Number.isFinite(ms) ? ms : 0;
+}
+
+function customerLastActiveAt(customer) {
+  return customer.lastOnlineAt || null;
+}
+
+function customerOnline(customer, nowMs) {
+  const ms = parseMs(customer.lastOnlineAt);
+  return ms > 0 && nowMs - ms <= PRESENCE_WINDOW_MS;
+}
+
+// Sum of the customer's per-server speeds that are still fresh.
+function customerSpeedBps(customer, nowMs) {
+  let up = 0;
+  let down = 0;
+  for (const slot of Object.values(customer.serverUsage || {})) {
+    if (!slot || !slot.speedAt) continue;
+    if (nowMs - parseMs(slot.speedAt) > SPEED_STALE_MS) continue;
+    up += Number(slot.speedUpBps || 0);
+    down += Number(slot.speedDownBps || 0);
+  }
+  return { up, down };
+}
+
+function nodeSpeedBps(node, nowMs) {
+  if (!node.speedAt || nowMs - parseMs(node.speedAt) > SPEED_STALE_MS) return { up: 0, down: 0 };
+  return { up: Number(node.speedUpBps || 0), down: Number(node.speedDownBps || 0) };
+}
+
+// Customers (ids) with an active binding on this node whose last
+// activity on the node's server is inside the presence window. The
+// per-server usage slot is authoritative; when the customer has never
+// reported usage from this server, the customer-level lastOnlineAt is
+// used as a fallback (AnyTLS observations stamp it without creating a
+// usage slot, and they are validated against this server at ingest).
+function nodeOnlineCustomerIds(data, node, nowMs) {
+  const out = [];
+  for (const nc of data.nodeClients) {
+    if (nc.nodeId !== node.id || nc.status !== 'active') continue;
+    const customer = data.customers.find((item) => item.id === nc.customerId);
+    if (!customer) continue;
+    const slot = (customer.serverUsage || {})[`srv:${node.serverId}`];
+    const ms = slot && slot.lastActiveAt ? parseMs(slot.lastActiveAt) : parseMs(customer.lastOnlineAt);
+    if (ms > 0 && nowMs - ms <= PRESENCE_WINDOW_MS) out.push(customer.id);
+  }
+  return out;
+}
+
+function countOnlineCustomers(data, nowMs) {
+  return data.customers.filter((customer) => customerOnline(customer, nowMs)).length;
+}
+
+module.exports = {
+  PRESENCE_WINDOW_MS, SPEED_STALE_MS, parseMs,
+  customerLastActiveAt, customerOnline, customerSpeedBps, nodeSpeedBps,
+  nodeOnlineCustomerIds, countOnlineCustomers,
+};

lib/reset.js

@@ -0,0 +1,96 @@
+'use strict';
+
+// Monthly traffic reset — Node port of panel-go/reset.go (the two panel
+// implementations must stay behavior-identical). A customer with
+// trafficResetDay (1-31) gets the cycle counters zeroed once per monthly
+// period. Due-ness derives only from the persisted trafficResetPeriod
+// marker, so a reset missed while the panel was down fires at the first
+// housekeeping pass after it is back. All dates are panel-local.
+
+function daysInMonth(year, monthIndex) {
+  return new Date(Date.UTC(year, monthIndex + 1, 0)).getUTCDate();
+}
+
+function resetInstant(year, monthIndex, day) {
+  const dim = daysInMonth(year, monthIndex);
+  return new Date(year, monthIndex, Math.min(day, dim), 0, 0, 0, 0);
+}
+
+function periodKeyOf(date) {
+  const pad = (n) => String(n).padStart(2, '0');
+  return `${date.getFullYear()}-${pad(date.getMonth() + 1)}`;
+}
+
+// The "YYYY-MM" of the most recent scheduled reset instant at or before now.
+function trafficResetPeriodKey(now, day) {
+  let year = now.getFullYear();
+  let month = now.getMonth();
+  let cand = resetInstant(year, month, day);
+  if (now.getTime() < cand.getTime()) {
+    month -= 1;
+    if (month < 0) { month = 11; year -= 1; }
+    cand = resetInstant(year, month, day);
+  }
+  return periodKeyOf(cand);
+}
+
+// (Re)arm a customer's period marker for its current reset day WITHOUT
+// zeroing usage: the schedule takes effect from the next reset instant.
+function armTrafficReset(customer, now = new Date()) {
+  const day = Number(customer.trafficResetDay || 0);
+  if (!Number.isInteger(day) || day < 1 || day > 31) {
+    delete customer.trafficResetPeriod;
+    return;
+  }
+  customer.trafficResetPeriod = trafficResetPeriodKey(now, day);
+}
+
+// Validate a trafficResetDay request value: 0 disables, 1-31 is the day of
+// month. null/absent means 0. Mirrors panel-go ParseResetDay.
+function parseResetDay(value) {
+  if (value == null) return 0;
+  const number = Number(value);
+  if (!Number.isFinite(number) || number < 0 || !Number.isInteger(number)) {
+    const error = new Error('每月重置日必须是非负整数');
+    error.statusCode = 400;
+    throw error;
+  }
+  if (number > 31) {
+    const error = new Error('每月重置日须在 0–31 之间（0 表示不重置）');
+    error.statusCode = 400;
+    throw error;
+  }
+  return number;
+}
+
+// Perform due monthly resets for all customers in data as of now. Returns
+// the list of customers whose counters were zeroed (the caller mirrors the
+// manual reset-usage semantics: a traffic_limit suspension may lift).
+// auditFn(data, customer, previousUsed, day, period) records the event.
+function applyMonthlyTrafficResets(data, now, auditFn) {
+  const reset = [];
+  for (const customer of data.customers) {
+    const day = Number(customer.trafficResetDay || 0);
+    if (!Number.isInteger(day) || day < 1 || day > 31) continue;
+    const key = trafficResetPeriodKey(now, day);
+    if (!customer.trafficResetPeriod) {
+      // First sight: arm only — usage already accrued in the current
+      // period must survive until the next reset day.
+      customer.trafficResetPeriod = key;
+      continue;
+    }
+    if (customer.trafficResetPeriod === key) continue;
+    const previousUsed = customer.usedBytes || 0;
+    customer.usedBytes = 0;
+    customer.usedUplinkBytes = 0;
+    customer.usedDownlinkBytes = 0;
+    customer.trafficResetPeriod = key;
+    customer.trafficResetAt = now.toISOString();
+    customer.updatedAt = now.toISOString();
+    if (auditFn) auditFn(data, customer, previousUsed, day, key);
+    reset.push(customer);
+  }
+  return reset;
+}
+
+module.exports = { trafficResetPeriodKey, armTrafficReset, parseResetDay, applyMonthlyTrafficResets, daysInMonth, resetInstant };

lib/store.js

@@ -13,7 +13,7 @@ function emptyData() {
     schemaVersion: 1,
     createdAt: new Date().toISOString(),
     settings: {
-      brandName: 'NexusGate',
+      brandName: 'amo',
       sessionHours: 12,
       observationTtlMinutes: 10,
       activityRetentionDays: 30,

lib/subscriptions.js

@@ -1,6 +1,12 @@
 'use strict';
 
 const { buildClientUri, nodeEmail } = require('./protocols');
+const { dedupKey, withName, nameFromUri } = require('./extsub');
+
+function nodeSortIndex(data, nodeId) {
+  const node = data.nodes.find((item) => item.id === nodeId);
+  return node && Number.isFinite(node.sortIndex) ? node.sortIndex : Number.MAX_SAFE_INTEGER;
+}
 
 // Client URLs are only issued for confirmed, running nodes. Each link is built
 // on the fly from the node's current config and the customer's own credentials
@@ -22,7 +28,11 @@ function nodeClientUri(data, nodeClient) {
 function entries(data, customerId) {
   const seen = new Map();
   const items = [];
-  for (const nodeClient of data.nodeClients.filter((nc) => nc.customerId === customerId && nc.status === 'active')) {
+  // Own nodes follow the panel-wide custom order (sortIndex); the sort is
+  // stable, so equal/unset indexes keep the binding creation order.
+  const bindings = data.nodeClients.filter((nc) => nc.customerId === customerId && nc.status === 'active');
+  bindings.sort((a, b) => nodeSortIndex(data, a.nodeId) - nodeSortIndex(data, b.nodeId));
+  for (const nodeClient of bindings) {
     const uri = nodeClientUri(data, nodeClient);
     if (!uri) continue;
     const node = data.nodes.find((item) => item.id === nodeClient.nodeId);
@@ -38,6 +48,47 @@ function entries(data, customerId) {
   return items;
 }
 
+// Full subscription item list: own nodes (sortIndex order), then manual
+// links (list order), then external sources (source order, cached order).
+// Global dedup by scheme+host+port+credential fingerprint, own first.
+// Name collisions across segments get the " (n)" suffix treatment.
+function aggregateEntries(data, customer) {
+  const items = entries(data, customer.id);
+  const usedKeys = new Set();
+  const usedNames = new Set();
+  for (const item of items) {
+    const key = dedupKey(item.clientUri);
+    if (key) usedKeys.add(key);
+    usedNames.add(nameFromUri(item.clientUri));
+  }
+  const append = (uri, rawName) => {
+    const key = dedupKey(uri);
+    if (key && usedKeys.has(key)) return;
+    let name = rawName || '未命名节点';
+    if (usedNames.has(name)) {
+      let index = 2;
+      while (usedNames.has(`${name} (${index})`)) index += 1;
+      name = `${name} (${index})`;
+    }
+    if (key) usedKeys.add(key);
+    usedNames.add(name);
+    items.push({ nodeId: '', clientUri: withName(uri, name) });
+  };
+  (customer.manualLinks || []).forEach((link, index) => {
+    if (!link.enabled) return;
+    append(link.uri, link.remark || nameFromUri(link.uri) || `手动节点 ${index + 1}`);
+  });
+  for (const source of customer.externalSources || []) {
+    if (!source.enabled) continue;
+    const cache = (data.subSourceCache || {})[source.id];
+    if (!cache || !Array.isArray(cache.items)) continue;
+    for (const cached of cache.items) {
+      append(cached.uri, `${source.remark ? `[${source.remark}] ` : ''}${cached.name || '未命名节点'}`);
+    }
+  }
+  return items;
+}
+
 function q(value) { return JSON.stringify(String(value)); }
 
 function clashProxy(item) {
@@ -149,7 +200,7 @@ function makeSingBox(items) {
 }
 
 function formatSubscription(data, customer, format, userAgent = '') {
-  const items = entries(data, customer.id);
+  const items = aggregateEntries(data, customer);
   if (!items.length) return null;
   const selected = format === 'auto' ? (/clash|mihomo/i.test(userAgent) ? 'clash' : 'base64') : format;
   if (selected === 'raw') return { body:items.map((item) => item.clientUri).join('\n') + '\n', contentType:'text/plain; charset=utf-8' };
@@ -160,4 +211,4 @@ function formatSubscription(data, customer, format, userAgent = '') {
   return undefined;
 }
 
-module.exports = { entries, nodeClientUri, clashProxy, makeClash, makeSingBox, makeSurge, formatSubscription };
+module.exports = { entries, aggregateEntries, nodeClientUri, clashProxy, makeClash, makeSingBox, makeSurge, formatSubscription };

lib/sysstat.js

@@ -0,0 +1,189 @@
+'use strict';
+
+// Host/system sampling for the system-status page (panel-go/sysstat.go
+// mirrors this file). Samples are taken at most every ~30s (housekeeping
+// cadence and on-demand from /api/system when stale) and kept in a
+// memory-only ring buffer (240 entries ≈ 4h at 60s cadence, tens of KB).
+// Nothing here writes to the data file: the "today" traffic baseline is
+// also memory-only, so a panel restart restarts the day counter from
+// the traffic seen since boot (named trade-off, see UI-RESKIN-REPORT).
+
+const fs = require('node:fs');
+const os = require('node:os');
+const { customerOnline } = require('./presence');
+
+const HISTORY_CAP = 240;
+const SAMPLE_MIN_INTERVAL_MS = 25000;
+
+function readCpuTimes() {
+  try {
+    const line = fs.readFileSync('/proc/stat', 'utf8').split('\n')[0];
+    const parts = line.trim().split(/\s+/).slice(1).map(Number);
+    if (parts.length < 5 || parts.some((n) => !Number.isFinite(n))) return null;
+    const idle = parts[3] + (parts[4] || 0);
+    const total = parts.reduce((a, b) => a + b, 0);
+    return { idle, total };
+  } catch { return null; }
+}
+
+function readMem() {
+  try {
+    const text = fs.readFileSync('/proc/meminfo', 'utf8');
+    const grab = (key) => {
+      const match = new RegExp(`^${key}:\\s+(\\d+)`, 'm').exec(text);
+      return match ? Number(match[1]) * 1024 : 0;
+    };
+    const total = grab('MemTotal');
+    const available = grab('MemAvailable');
+    return { total, used: total > 0 ? total - available : 0 };
+  } catch { return { total: 0, used: 0 }; }
+}
+
+function isGlobalIPv4(ip) {
+  const parts = String(ip).split('.').map(Number);
+  if (parts.length !== 4 || parts.some((n) => !Number.isInteger(n) || n < 0 || n > 255)) return false;
+  const [a, b] = parts;
+  if (a === 0 || a === 10 || a === 127) return false;
+  if (a === 172 && b >= 16 && b <= 31) return false;
+  if (a === 192 && b === 168) return false;
+  if (a === 100 && b >= 64 && b <= 127) return false; // CGNAT
+  if (a === 169 && b === 254) return false; // link-local
+  if (a >= 224) return false; // multicast / reserved
+  return true;
+}
+
+function isGlobalIPv6(ip) {
+  const lower = String(ip).toLowerCase();
+  if (lower === '::1' || lower === '::') return false;
+  if (/^f[cd]/.test(lower)) return false; // ULA fc00::/7
+  if (/^fe[89ab]/.test(lower)) return false; // link-local fe80::/10
+  if (lower.startsWith('ff')) return false; // multicast
+  return lower.includes(':');
+}
+
+// Pick the host's global addresses from the interface list (3x-ui
+// 口径: local interface addresses, no external lookup of any kind).
+function hostIPs() {
+  let ipv4 = null;
+  let ipv6 = null;
+  const interfaces = os.networkInterfaces();
+  for (const list of Object.values(interfaces)) {
+    for (const info of list || []) {
+      if (info.internal) continue;
+      if (info.family === 'IPv4' && !ipv4 && isGlobalIPv4(info.address)) ipv4 = info.address;
+      if (info.family === 'IPv6' && !ipv6 && isGlobalIPv6(info.address)) ipv6 = info.address;
+    }
+  }
+  return { ipv4, ipv6 };
+}
+
+class SystemMonitor {
+  constructor(dataDir) {
+    this.dataDir = dataDir;
+    this.startedAt = Date.now();
+    this.prevCpu = null;
+    this.cpuPercent = null;
+    this.history = [];
+    this.lastSampleAt = 0;
+    this.dayKey = null;
+    this.dayBaseUp = 0;
+    this.dayBaseDown = 0;
+    this.todayUp = 0;
+    this.todayDown = 0;
+    this.current = null;
+  }
+
+  sample(data, force = false) {
+    const now = Date.now();
+    if (!force && now - this.lastSampleAt < SAMPLE_MIN_INTERVAL_MS) return this.current;
+    this.lastSampleAt = now;
+    const cpu = readCpuTimes();
+    if (cpu && this.prevCpu) {
+      const totalDelta = cpu.total - this.prevCpu.total;
+      const idleDelta = cpu.idle - this.prevCpu.idle;
+      if (totalDelta > 0) this.cpuPercent = Math.max(0, Math.min(100, (1 - idleDelta / totalDelta) * 100));
+    }
+    if (cpu) this.prevCpu = cpu;
+    const mem = readMem();
+    let disk = { total: 0, used: 0 };
+    try {
+      const stat = fs.statfsSync(this.dataDir);
+      disk = { total: stat.blocks * stat.bsize, used: (stat.blocks - stat.bfree) * stat.bsize };
+    } catch { /* keep zeros */ }
+    let totalUp = 0;
+    let totalDown = 0;
+    let upBps = 0;
+    let downBps = 0;
+    for (const node of data.nodes || []) {
+      if (node.meteredTraffic) {
+        totalUp += Number(node.meteredTraffic.uplink || 0);
+        totalDown += Number(node.meteredTraffic.downlink || 0);
+      }
+      if (node.speedAt && now - Date.parse(node.speedAt) <= 150000) {
+        upBps += Number(node.speedUpBps || 0);
+        downBps += Number(node.speedDownBps || 0);
+      }
+    }
+    const dayKey = new Date(now).toLocaleDateString('en-CA');
+    if (this.dayKey !== dayKey) {
+      this.dayKey = dayKey;
+      this.dayBaseUp = totalUp;
+      this.dayBaseDown = totalDown;
+    }
+    this.todayUp = Math.max(0, totalUp - this.dayBaseUp);
+    this.todayDown = Math.max(0, totalDown - this.dayBaseDown);
+    const connections = (data.customers || []).filter((c) => customerOnline(c, now)).length;
+    const entry = {
+      at: new Date(now).toISOString(),
+      cpu: this.cpuPercent == null ? null : Math.round(this.cpuPercent * 10) / 10,
+      mem: mem.total > 0 ? Math.round((mem.used / mem.total) * 1000) / 10 : null,
+      upBps, downBps, connections,
+    };
+    this.history.push(entry);
+    if (this.history.length > HISTORY_CAP) this.history.splice(0, this.history.length - HISTORY_CAP);
+    this.current = {
+      at: entry.at, cpuPercent: entry.cpu, mem, disk,
+      upBps, downBps, connections, totalUp, totalDown,
+    };
+    return this.current;
+  }
+
+  status(data, version) {
+    const current = this.sample(data) || {
+      cpuPercent: null, mem: readMem(), disk: { total: 0, used: 0 },
+      upBps: 0, downBps: 0, connections: 0, totalUp: 0, totalDown: 0,
+    };
+    const mem = current.mem;
+    const disk = current.disk;
+    const onlineNodes = (data.nodes || []).filter((n) => n.status === 'active').length;
+    return {
+      version,
+      startedAt: new Date(this.startedAt).toISOString(),
+      panelUptimeSec: Math.floor((Date.now() - this.startedAt) / 1000),
+      hostUptimeSec: Math.floor(os.uptime()),
+      panelMemBytes: process.memoryUsage().rss,
+      host: {
+        cpuPercent: current.cpuPercent,
+        cpuCount: os.cpus().length,
+        load1: os.loadavg()[0],
+        memTotal: mem.total, memUsed: mem.used,
+        memPercent: mem.total > 0 ? Math.round((mem.used / mem.total) * 1000) / 10 : null,
+        diskTotal: disk.total, diskUsed: disk.used,
+        diskPercent: disk.total > 0 ? Math.round((disk.used / disk.total) * 1000) / 10 : null,
+      },
+      network: {
+        upBps: current.upBps, downBps: current.downBps,
+        connections: current.connections,
+        onlineCustomers: current.connections,
+        onlineNodes, nodesTotal: (data.nodes || []).length,
+        customersTotal: (data.customers || []).length,
+      },
+      today: { upBytes: this.todayUp, downBytes: this.todayDown },
+      totals: { upBytes: current.totalUp, downBytes: current.totalDown },
+      ips: hostIPs(),
+      history: this.history,
+    };
+  }
+}
+
+module.exports = { SystemMonitor, hostIPs, isGlobalIPv4, isGlobalIPv6, readCpuTimes, readMem };