'use strict';

const crypto = require('node:crypto');
const { redactSecrets } = require('./redact');
const {
  buildNodeResource, buildNodeExitResource, nodeEmail,
  newNodeClientCredentials, validateEntryProtocol, validateProtocolPair, isRealityProtocol, resolveBindAddress,
  keyMaterialForNode, dropRealityKey,
  SHADOWSOCKS_METHODS,
} = require('./protocols');

const nowIso = () => new Date().toISOString();
const id = (prefix) => `${prefix}_${crypto.randomUUID()}`;

// Minimum gap between two heartbeat-triggered reconciliations of one server.
const HEARTBEAT_RECONCILE_INTERVAL_MS = 120000;

function assertFound(value, label) {
  if (!value) {
    const error = new Error(`${label}不存在`);
    error.statusCode = 404;
    throw error;
  }
  return value;
}

function forwardNode(node) { return (node.topology || 'forward') === 'forward'; }

function tagForNode(node) {
  return `ng-${node.id.replace(/[^a-zA-Z0-9]/g, '').slice(-18)}`;
}

function nodePortInUse(data, serverId, port, excludeNodeId = null) {
  return data.nodes.some((n) => n.id !== excludeNodeId &&
    ((n.serverId === serverId && Number(n.port) === Number(port)) ||
     (n.exitServerId === serverId && Number(n.exitPort) === Number(port))));
}

function allocatePort(data, server) {
  const start = Number(server.portRangeStart || 20000);
  const end = Number(server.portRangeEnd || 50000);
  if (!Number.isInteger(start) || !Number.isInteger(end) || start < 1024 || end > 65535 || start > end) throw new Error(`服务器 ${server.name} 的端口范围无效`);
  const span = end - start + 1;
  const offset = crypto.randomInt(span);
  for (let i = 0; i < span; i += 1) {
    const candidate = start + ((offset + i) % span);
    if (!nodePortInUse(data, server.id, candidate)) return candidate;
  }
  throw new Error(`服务器 ${server.name} 没有可用端口`);
}

function selectPort(data, server, mode, requested, excludeNodeId = null) {
  if (mode !== 'fixed' || !requested) return allocatePort(data, server);
  const port = Number(requested);
  if (!Number.isInteger(port) || port < 1024 || port > 65535) { const error = new Error(`服务器 ${server.name} 的固定端口无效`); error.statusCode = 400; throw error; }
  if (nodePortInUse(data, server.id, port, excludeNodeId)) { const error = new Error(`服务器 ${server.name} 的端口 ${port} 已被占用`); error.statusCode = 409; throw error; }
  return port;
}

function queueJob(data, serverId, nodeId, action, payload) {
  const job = {
    id: id('job'), serverId, nodeId, action, payload, status: 'queued', attempts: 0,
    createdAt: nowIso(), updatedAt: nowIso(), error: null, leaseUntil: null
  };
  data.jobs.push(job);
  return job;
}

function audit(data, actor, action, target, detail = {}) {
  data.activity.unshift({ id: id('evt'), at: nowIso(), actor, action, target, detail });
  data.activity = data.activity.slice(0, 2000);
}

function activeNodeClients(data, node) {
  return data.nodeClients.filter((nc) => nc.nodeId === node.id && nc.status === 'active');
}

async function resourceForNode(data, node, clients) {
  const server = assertFound(data.servers.find((s) => s.id === node.serverId), '服务器');
  const networkMode = node.networkMode || 'ipv4';
  const bindAddress = await resolveBindAddress(networkMode, server);
  const exitServer = forwardNode(node)
    ? assertFound(data.servers.find((s) => s.id === node.exitServerId), '出口服务器')
    : null;
  return buildNodeResource({
    resourceId: node.resourceId, tagPrefix: tagForNode(node), node,
    nodeClients: clients || activeNodeClients(data, node), server, bindAddress, exitServer,
  });
}

async function resourceForNodeExit(data, node) {
  const server = assertFound(data.servers.find((s) => s.id === node.exitServerId), '出口服务器');
  const networkMode = node.networkMode || 'ipv4';
  const bindAddress = await resolveBindAddress(networkMode, server);
  return buildNodeExitResource({
    resourceId: node.exitResourceId, tagPrefix: tagForNode(node), node, server, bindAddress,
  });
}

function cancelQueuedDeletes(data, nodeId, reason) {
  let cancelled = false;
  for (const job of data.jobs.filter((item) => item.nodeId === nodeId && item.action === 'delete_resource' && item.status === 'queued')) {
    job.status = 'failed'; job.error = reason; job.updatedAt = nowIso(); cancelled = true;
  }
  return cancelled;
}

// Rebuild a node's entry resource from its current client list and queue the
// apply job. With no active clients the resource is removed and the node
// returns to draft instead of shipping an empty (and for some protocols
// invalid) inbound. autoRemoving distinguishes this automatic teardown from an
// explicit user undeploy so resume/assign can revive it without a full cycle.
async function rebuildNode(data, node, { includeExit = false } = {}) {
  const clients = activeNodeClients(data, node);
  if (!clients.length) {
    if (node.status !== 'draft' && node.status !== 'removing') {
      queueJob(data, node.serverId, node.id, 'delete_resource', { resourceId: node.resourceId, kind: 'node' });
      if (forwardNode(node) && node.exitResourceId) {
        queueJob(data, node.exitServerId, node.id, 'delete_resource', { resourceId: node.exitResourceId, kind: 'node-exit' });
      }
      node.status = 'removing'; node.autoRemoving = true; node.pendingDelete = false; node.error = null;
    }
    node.updatedAt = nowIso();
    return;
  }
  node.autoRemoving = false;
  const payload = { resource: await resourceForNode(data, node, clients), kind: 'node' };
  // Reality keys are panel-hosted: ship the stored keypair with the deploy so
  // the agent materializes the same private key after any reinstall.
  const keyMaterial = keyMaterialForNode(data, node);
  if (keyMaterial) payload.keyMaterial = keyMaterial;
  queueJob(data, node.serverId, node.id, 'apply_resource', payload);
  if (forwardNode(node) && includeExit) {
    queueJob(data, node.exitServerId, node.id, 'apply_resource', { resource: await resourceForNodeExit(data, node), kind: 'node-exit' });
  }
  node.status = 'deploying'; node.error = null; node.updatedAt = nowIso();
}

function deriveNodeStatus(node) {
  if (!node.entryReady) { if (node.status !== 'failed') node.status = 'deploying'; return; }
  if (forwardNode(node) && !node.exitReady) { if (node.status !== 'degraded') node.status = 'deploying'; return; }
  node.status = 'active'; node.error = null;
}

async function suspendCustomerResources(data, customer, actor = 'system') {
  const affected = new Set();
  for (const nc of data.nodeClients.filter((nc) => nc.customerId === customer.id && nc.status === 'active')) {
    nc.status = 'suspended'; nc.updatedAt = nowIso(); affected.add(nc.nodeId);
  }
  for (const nodeId of affected) {
    const node = data.nodes.find((n) => n.id === nodeId);
    if (node && node.status !== 'draft' && node.status !== 'removing') await rebuildNode(data, node);
  }
  audit(data, actor, 'suspend_customer_resources', customer.id, { nodes: affected.size });
  return affected.size;
}

async function resumeCustomerResources(data, customer, actor = 'admin') {
  const affected = new Set();
  for (const nc of data.nodeClients.filter((nc) => nc.customerId === customer.id && nc.status === 'suspended')) {
    nc.status = 'active'; nc.updatedAt = nowIso(); affected.add(nc.nodeId);
  }
  for (const nodeId of affected) {
    const node = data.nodes.find((n) => n.id === nodeId);
    if (!node) continue;
    // A manual undeploy is respected; only automatic teardowns are revived.
    if (node.status === 'removing' && !node.autoRemoving) continue;
    if (node.status === 'draft' && !node.autoRemoving) continue;
    if (node.status === 'removing' && node.autoRemoving) {
      const deleteRunning = data.jobs.some((job) => job.nodeId === nodeId && job.action === 'delete_resource' && job.status === 'running');
      cancelQueuedDeletes(data, nodeId, '客户恢复，清理任务已取消');
      node.autoRemoving = false;
      if (!deleteRunning) {
        // The resource never left the agent; restore service without re-applying.
        deriveNodeStatus(node);
        node.updatedAt = nowIso();
        continue;
      }
      // A delete is already running; queue the apply behind it.
    }
    await rebuildNode(data, node, { includeExit: forwardNode(node) && !node.exitReady });
  }
  audit(data, actor, 'resume_customer_resources', customer.id, { nodes: affected.size });
  return affected.size;
}

// Shared delta/epoch accounting used for per-customer and per-node counters.
// holder: { usageCounters, retiredUsageEpochs }. Returns null when the sample
// must be skipped (retired epoch or out-of-order replay).
function applyDelta(holder, uplink, downlink, sampleEpoch) {
  const previous = holder.usageCounters || { uplink: 0, downlink: 0 };
  const retiredEpochs = Array.isArray(holder.retiredUsageEpochs) ? holder.retiredUsageEpochs : [];
  if (sampleEpoch && retiredEpochs.includes(sampleEpoch)) return null;
  const restarted = Boolean(sampleEpoch && previous.epoch && sampleEpoch !== previous.epoch);
  if (!restarted && sampleEpoch && previous.epoch && !sampleEpoch.endsWith(':unknown') &&
      (uplink < previous.uplink || downlink < previous.downlink)) return null;
  // Counters reset when the engine restarts. Replayed reports have zero delta.
  const upDelta = !restarted && uplink >= previous.uplink ? uplink - previous.uplink : uplink;
  const downDelta = !restarted && downlink >= previous.downlink ? downlink - previous.downlink : downlink;
  if (restarted && previous.epoch) {
    holder.retiredUsageEpochs = [...retiredEpochs, previous.epoch].slice(-8);
  }
  holder.usageCounters = { uplink, downlink, epoch: sampleEpoch || null };
  return { upDelta, downDelta };
}

// Merge agent-reported keypairs (artifacts.keyMaterial) into the panel key
// store. The panel store always wins: an agent report only fills a gap (e.g.
// a node created before panel hosting, whose key only exists on the node).
// A conflicting report is audited, never applied — the next deploy re-ships
// the panel key and the agent overwrites its local copy with it.
function mergeKeyMaterialBackfill(data, node, artifacts) {
  if (!artifacts || typeof artifacts !== 'object') return;
  const reported = artifacts.keyMaterial;
  // keyMaterial is transport data for the store, not a node artifact.
  delete artifacts.keyMaterial;
  if (reported && typeof reported === 'object') {
    data.keyStore ||= {};
    for (const [keyId, pair] of Object.entries(reported)) {
      if (!pair || typeof pair.privateKey !== 'string' || typeof pair.publicKey !== 'string') continue;
      const existing = data.keyStore[keyId];
      if (!existing) {
        data.keyStore[keyId] = { privateKey: pair.privateKey, publicKey: pair.publicKey, updatedAt: nowIso() };
      } else if (existing.privateKey !== pair.privateKey || existing.publicKey !== pair.publicKey) {
        audit(data, 'agent', 'reality_key_conflict', node.id, { keyId });
      }
    }
  }
  // The subscription-facing publicKey always comes from the panel store, so
  // it is identical before and after any agent report.
  const keyId = node.reality && node.reality.keyId;
  if (keyId && data.keyStore && data.keyStore[keyId]) {
    artifacts.realityPublicKey = data.keyStore[keyId].publicKey;
  }
}

function validateNodeDeploy(data, node) {
  const direct = !forwardNode(node);
  if (direct) validateEntryProtocol(node.protocol);
  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), '出口服务器');
  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; }
    if (!server.engine?.certificates?.includes(server.tlsDomain)) {
      const error = new Error('入口证书未由 Agent 确认。在入口机运行 ng-agent update && ng-agent cert，证书申请成功后刷新设备状态');
      error.statusCode = 409; throw error;
    }
  }
  if (node.protocol === 'anytls') {
    const agent = data.agents.find((item) => item.serverId === server.id && item.status === 'active');
    const parts = String(agent?.version || '').split('.').map(Number);
    if (!agent || !(parts[0] > 0 || (parts[0] === 0 && (parts[1] > 6 || (parts[1] === 6 && parts[2] >= 1))))) {
      const error = new Error(`AnyTLS 入口需要新版 Agent：请在 ${server.name} 运行 ng-agent update`); error.statusCode = 409; throw error;
    }
    if (!server.engine?.singBoxInstalled) {
      const error = new Error(`AnyTLS 入口缺少 sing-box 及统计组件：请在 ${server.name} 运行 ng-agent engine install，完成后等待 Agent 心跳并刷新页面`);
      error.statusCode = 409; throw error;
    }
  }
  const servers = direct ? [server] : [server, exitServer];
  const unavailable = servers.filter((s) => s.status !== 'online' || Date.now() - Date.parse(s.lastSeenAt || 0) > 120000);
  if (unavailable.length) {
    const error = new Error(`目标设备尚未就绪：${unavailable.map((s) => `${s.name}（Agent 离线；请在该机器运行 ng-agent doctor）`).join('、')}`);
    error.statusCode = 409; throw error;
  }
  if ((node.networkMode || 'ipv4') === 'ipv6') {
    const missing = servers.filter((s) => !s.publicAddressV6);
    if (missing.length) { const error = new Error(`IPv6 模式需要先为这些服务器填写 IPv6 地址：${missing.map((s) => s.name).join('、')}`); error.statusCode = 400; throw error; }
  }
  if (nodePortInUse(data, node.serverId, node.port, node.id)) { const error = new Error(`端口 ${node.port} 已被该服务器上的其他节点占用`); error.statusCode = 409; throw error; }
  if (!direct && nodePortInUse(data, node.exitServerId, node.exitPort, node.id)) { const error = new Error(`出口端口 ${node.exitPort} 已被该服务器上的其他节点占用`); error.statusCode = 409; throw error; }
}

// Re-queue apply/delete jobs so the agent's local resources converge to the
// panel's desired state. onlyNodeIds restricts the pass to those nodes (used
// by heartbeat reconciliation to heal just the nodes with missing resources);
// failInFlight marks queued/running jobs as superseded (enrollment semantics:
// a re-registering agent abandons its old leases). Heartbeat reconciliation
// 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), '服务器');
  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();
    }
  }
  let count = 0;
  for (const node of data.nodes.filter((n) => n.serverId === serverId || n.exitServerId === serverId)) {
    if (onlyNodeIds && !onlyNodeIds.has(node.id)) continue;
    const isEntry = node.serverId === serverId;
    const isExit = node.exitServerId === serverId && forwardNode(node);
    if (!isEntry && !isExit) continue;
    if (node.status === 'draft') continue;
    if (node.status === 'removing') {
      if (isEntry) queueJob(data, serverId, node.id, 'delete_resource', { resourceId: node.resourceId, kind: 'node' });
      if (isExit && node.exitResourceId) queueJob(data, serverId, node.id, 'delete_resource', { resourceId: node.exitResourceId, kind: 'node-exit' });
    } else {
      const clients = activeNodeClients(data, node);
      if (!clients.length) continue;
      if (isEntry) {
        const payload = { resource: await resourceForNode(data, node, clients), kind: 'node' };
        const keyMaterial = keyMaterialForNode(data, node);
        if (keyMaterial) payload.keyMaterial = keyMaterial;
        queueJob(data, serverId, node.id, 'apply_resource', payload);
      }
      if (isExit) queueJob(data, serverId, node.id, 'apply_resource', { resource: await resourceForNodeExit(data, node), kind: 'node-exit' });
      node.status = 'deploying';
    }
    node.updatedAt = nowIso(); count += 1;
  }
  audit(data, actor, 'reconcile_server', server.id, { nodes: count });
  return count;
}

class Orchestrator {
  constructor(store) {
    this.store = store;
    // serverId -> epoch ms of the last heartbeat-triggered reconciliation.
    this.heartbeatReconcileAt = new Map();
  }

  async deployNode(nodeId, actor = 'admin') {
    return this.store.transaction(async (data) => {
      const node = assertFound(data.nodes.find((n) => n.id === nodeId), '节点');
      validateNodeDeploy(data, node);
      await rebuildNode(data, node, { includeExit: true });
      audit(data, actor, 'deploy_node', node.id, { name: node.name });
      return structuredClone(node);
    });
  }

  // Undeploy (keep the node record) or delete it entirely once agents confirm cleanup.
  async removeNode(nodeId, actor = 'admin', deleteRecord = false) {
    return this.store.transaction((data) => {
      const node = assertFound(data.nodes.find((n) => n.id === nodeId), '节点');
      node.pendingDelete = deleteRecord;
      node.autoRemoving = false;
      if (node.status !== 'draft') {
        queueJob(data, node.serverId, node.id, 'delete_resource', { resourceId: node.resourceId, kind: 'node' });
        if (forwardNode(node) && node.exitResourceId) {
          queueJob(data, node.exitServerId, node.id, 'delete_resource', { resourceId: node.exitResourceId, kind: 'node-exit' });
        }
        node.status = 'removing'; node.error = null;
      } else if (deleteRecord) {
        dropRealityKey(data, node);
        data.nodes = data.nodes.filter((n) => n.id !== node.id);
        data.nodeClients = data.nodeClients.filter((nc) => nc.nodeId !== node.id);
      }
      node.updatedAt = nowIso();
      audit(data, actor, deleteRecord ? 'delete_node' : 'remove_node', node.id, { name: node.name });
      return true;
    });
  }

  async repairNode(nodeId, actor = 'admin') {
    return this.store.transaction(async (data) => {
      const node = assertFound(data.nodes.find((n) => n.id === nodeId), '节点');
      if (!['failed', 'degraded'].includes(node.status)) { const error = new Error('节点状态正常，无需修复'); error.statusCode = 409; throw error; }
      await rebuildNode(data, node, { includeExit: true });
      audit(data, actor, 'repair_node', node.id, { name: node.name });
      return true;
    });
  }

  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);
    });
  }

  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;
    });
  }

  async reconcileServer(serverId, actor = 'system', options = {}) {
    return this.store.transaction(async (data) => reconcileServerData(data, serverId, actor, options));
  }

  // Heartbeat-driven reconciliation. Newer agents report the resource IDs
  // they actually hold; when a resource the panel expects is missing locally
  // (e.g. the node's /etc/nexusgate state was wiped and only the agent was
  // restarted, so no re-enrollment ever happens), re-queue applies for just
  // the affected nodes. Old agents never send `resources`, and callers only
  // invoke this for array reports, so their behavior is unchanged. Resources
  // the agent holds but the panel no longer expects are NOT cleaned up here.
  // Gates: never run while the server has queued/running jobs (an in-flight
  // deploy explains the gap), and at most one trigger per server per
  // HEARTBEAT_RECONCILE_INTERVAL_MS so a node that cannot heal does not spin.
  async reconcileFromHeartbeat(serverId, reportedResourceIds) {
    if (!Array.isArray(reportedResourceIds)) return 0;
    const last = this.heartbeatReconcileAt.get(serverId) || 0;
    if (Date.now() - last < HEARTBEAT_RECONCILE_INTERVAL_MS) return 0;
    const reported = new Set(reportedResourceIds.slice(0, 1000).filter((value) => typeof value === 'string' && value));
    let triggered = false;
    const count = await this.store.transaction(async (data) => {
      const server = data.servers.find((item) => item.id === serverId);
      if (!server) return 0;
      const missingNodeIds = new Set();
      for (const node of data.nodes) {
        if (node.status === 'draft' || node.status === 'removing') continue;
        if (!activeNodeClients(data, node).length) continue;
        if (node.serverId === serverId && node.resourceId && !reported.has(node.resourceId)) missingNodeIds.add(node.id);
        if (node.exitServerId === serverId && forwardNode(node) && node.exitResourceId && !reported.has(node.exitResourceId)) missingNodeIds.add(node.id);
      }
      if (!missingNodeIds.size) return 0;
      const inFlight = data.jobs.some((job) => job.serverId === serverId && ['queued', 'running'].includes(job.status));
      if (inFlight) return 0;
      triggered = true;
      return reconcileServerData(data, serverId, 'system:heartbeat', { onlyNodeIds: missingNodeIds, failInFlight: false });
    });
    if (triggered) this.heartbeatReconcileAt.set(serverId, Date.now());
    return count;
  }

  async completeJob(jobId, success, result = {}, errorMessage = null) {
    return this.store.transaction((data) => {
      const job = assertFound(data.jobs.find((item) => item.id === jobId), '任务');
      if (job.status !== 'running') {
        if (job.status === 'completed' && success) return structuredClone(job);
        const stale = new Error('任务已被重试或取消，请等待当前任务完成'); stale.statusCode = 409; throw stale;
      }
      job.status = success ? 'completed' : 'failed'; job.updatedAt = nowIso(); job.completedAt = nowIso(); job.leaseUntil = null;
      job.error = success ? null : redactSecrets(errorMessage || 'Agent operation failed').slice(0, 1800);
      const node = data.nodes.find((n) => n.id === job.nodeId);
      if (!node) return structuredClone(job);
      const kind = job.payload && (job.payload.kind || (job.payload.resource && job.payload.resource.meta && job.payload.resource.meta.kind));
      if (success && job.action === 'apply_resource') {
        if (kind === 'node-exit') node.exitReady = true;
        else {
          node.entryReady = true; node.artifacts = result.artifacts || {};
          mergeKeyMaterialBackfill(data, node, node.artifacts);
        }
        deriveNodeStatus(node);
      } else if (success && job.action === 'delete_resource') {
        if (kind === 'node-exit') node.exitReady = false; else node.entryReady = false;
        const pending = data.jobs.some((item) => item.nodeId === node.id && item.action === 'delete_resource' && ['queued', 'running'].includes(item.status));
        if (!pending && node.status === 'removing') {
          if (node.pendingDelete) {
            dropRealityKey(data, node);
            data.nodes = data.nodes.filter((n) => n.id !== node.id);
            data.nodeClients = data.nodeClients.filter((nc) => nc.nodeId !== node.id);
          } else {
            node.status = 'draft'; node.error = null;
          }
        }
      } else if (!success) {
        node.status = kind === 'node-exit' ? 'degraded' : 'failed';
        node.error = job.error;
      }
      node.updatedAt = nowIso();
      return structuredClone(job);
    });
  }

  async recordUsage(serverId, samples, error = null, epoch = null) {
    // Empty reports carry diagnostic status, not billable counters. Keep them
    // visible without rewriting the full database on every idle Agent tick.
    if (!Array.isArray(samples) || samples.length === 0) {
      return this.store.transient((data) => {
        const server = data.servers.find((item) => item.id === serverId);
        if (server) server.usage = { lastReportAt: nowIso(), sampleCount: 0,
          error: error ? String(error).slice(0, 240) : null };
      });
    }
    return this.store.transaction((data) => {
      const reportedAt = nowIso();
      const server = data.servers.find((item) => item.id === serverId);
      if (server) server.usage = { lastReportAt: reportedAt, sampleCount: Array.isArray(samples) ? samples.length : 0,
        error: error ? String(error).slice(0, 240) : null };
      for (const sample of Array.isArray(samples) ? samples.slice(0, 1000) : []) {
        const uplink = Number(sample.uplink);
        const downlink = Number(sample.downlink);
        if (![uplink, downlink].every((value) => Number.isSafeInteger(value) && value >= 0)) continue;
        const sampleEpoch = typeof sample.epoch === 'string' ? sample.epoch.slice(0, 80) : epoch;
        if (sample.kind === 'user') {
          const customer = data.customers.find((item) => item.id === sample.id);
          if (!customer) continue;
          // Counters are per engine process, hence per server. Traffic itself
          // accumulates on the customer across all servers.
          const key = `srv:${serverId}`;
          customer.serverUsage = customer.serverUsage || {};
          const slot = customer.serverUsage[key] || (customer.serverUsage[key] = {});
          const delta = applyDelta(slot, uplink, downlink, sampleEpoch);
          if (!delta) continue;
          customer.usedUplinkBytes = (customer.usedUplinkBytes || 0) + delta.upDelta;
          customer.usedDownlinkBytes = (customer.usedDownlinkBytes || 0) + delta.downDelta;
          customer.usedBytes = (customer.usedBytes || 0) + delta.upDelta + delta.downDelta;
          customer.lastUsageAt = reportedAt;
          customer.updatedAt = reportedAt;
        } else if (sample.kind === 'inbound') {
          // Node-level traffic comes from entry inbounds only; the exit leg is
          // 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 delta = applyDelta(node, uplink, downlink, sampleEpoch);
          if (!delta) continue;
          const metered = node.meteredTraffic || { uplink: 0, downlink: 0 };
          node.meteredTraffic = { uplink: metered.uplink + delta.upDelta, downlink: metered.downlink + delta.downDelta };
          node.lastUsageAt = reportedAt;
        }
      }
    });
  }
}

module.exports = { Orchestrator, allocatePort, selectPort, nodePortInUse, queueJob, audit, id, nowIso, cancelQueuedDeletes, resourceForNode,
  resourceForNodeExit, rebuildNode, applyDelta, forwardNode, activeNodeClients,
  suspendCustomerResources, resumeCustomerResources, HEARTBEAT_RECONCILE_INTERVAL_MS };