/**
 * Sarmad Protocol VPS coordinator
 * --------------------------------
 * Role: heartbeat liveness cursor, frozen cycle walk, bootstrap peers, public status,
 *       WebSocket block/tx relay for browser nodes, block body store + sync APIs.
 * Queue ORDER comes from the chain (JOIN set or bonded set + tip-hash seed).
 * The VPS only skips addresses that stopped heartbeating so a dead phone does not stall 33s.
 * Phones/browsers derive the same shuffle from seed + the frozen on-chain address list.
 *
 * Endpoints:
 *   POST /v1/heartbeat
 *   GET  /v1/cycle
 *   GET  /v1/status
 *   GET  /v1/cluster          → multi-VPS scaffold fields (clusterEnabled=false with 1 VPS)
 *   GET  /v1/bootstrap
 *   POST /v1/block-report   (optional body.block with full block; followers forward to leader)
 *   POST /v1/tx             (followers forward to leader)
 *   GET  /v1/blocks?from=&limit=
 *   GET  /v1/mempool
 *   WS   /v1/ws
 *   POST /v1/genesis        (DEV: publish network genesis timestamp; requires reset token)
 *   POST /v1/reset
 *   POST /v1/hub/block|tx|heartbeat → equal-hub mesh gossip (no phone rate limits)
 *   GET  /v1/hub/peers             → learned hub baseUrls
 *   POST /v1/coordinator-announce  (ignored in single-VPS mode; kept for old clients)
 *   GET  /v1/coordinators          (always empty — phones use this directory only)
 *   GET  /v1/directory             → signed VPS URL list (clients cache for failover)
 *   GET  /js/* /css/*           → engine assets (bundled for srmd.exe UI)
 *   GET  /srmd.html               → desktop producing-node UI (served to EXE package)
 *   GET  /coordinator.html        → redirects to /srmd.html
 *   GET  /                     → status page
 */

const http = require("http");
const fs = require("fs");
const path = require("path");
const crypto = require("crypto");
const { URL } = require("url");
const { WebSocketServer } = require("ws");
const relay = require("./relay_store");
const coordRewards = require("./coord_rewards");
const hubGossip = require("./hub_gossip");
const { createCluster } = require("./vps_cluster");
const snapshotStore = require("./snapshot_store");
const broadcastStore = require("./broadcast_store");
const releaseStore = require("../coordinator_exe/release_store");
let directoryStore;
try {
  directoryStore = require("../coordinator_exe/directory_store");
} catch (_) {
  directoryStore = require("./directory_store");
}
const { DATA_DIR, dataPath } = require("./data_dir");
const blockValidator = require("./block_validator");
const sarfeesBridge = require("./sarfees_bridge");
let exeWallet;
try {
  exeWallet = require("../coordinator_exe/exe_wallet");
} catch (_) {
  try {
    exeWallet = require("./exe_wallet");
  } catch {
    exeWallet = null;
  }
}

const PORT = Number(process.env.SARMAD_COORD_PORT || 9334);
const HOST = process.env.SARMAD_COORD_HOST || "0.0.0.0";
const HEARTBEAT_TTL_MS = Number(process.env.SARMAD_HEARTBEAT_TTL_MS || 90_000);
const COORDINATOR_TTL_MS = Number(process.env.SARMAD_COORDINATOR_TTL_MS || HEARTBEAT_TTL_MS);
/** When set, this process is an EXE/edge coordinator and announces itself to the directory. */
let DIRECTORY_URL = String(process.env.SARMAD_DIRECTORY_URL || "").replace(/\/$/, "");
let DIRECTORY_LABEL = String(process.env.SARMAD_DIRECTORY_LABEL || (DIRECTORY_URL ? "primary" : "self"));
/** Backup directory list for announce failover (EXE only). */
let DIRECTORY_URLS = [];
try {
  DIRECTORY_URLS = require("./directory_urls").directoryUrls();
} catch (_) {
  DIRECTORY_URLS = DIRECTORY_URL ? [DIRECTORY_URL] : [];
}
const PUBLIC_BASE_URL = String(process.env.SARMAD_PUBLIC_BASE_URL || "").replace(/\/$/, "");
const COORD_VERSION = String(process.env.SARMAD_COORD_VERSION || "1.84");
const RELEASE_BASE = String(process.env.SARMAD_RELEASE_BASE || "https://sarmad.pro/downloads").replace(/\/$/, "");
const BLOCK_TIME_SECONDS = 33;
/** Keep in sync with ProtocolConstants.kt / protocol.js */
const PRODUCER_ADMISSION_HEIGHT = 3_333_333;
const MAX_PRODUCERS_PER_IP = 5;
/**
 * Readiness lookahead. The queue is deterministic, so the producer after the current
 * one is known a full slot ahead and is published as `nextProducer` so it can sync.
 * Every producer gets a full block-time slot (no early 8s burn).
 */
const ON_DECK_MIN_WARN_MS = Number(process.env.SARMAD_ON_DECK_MIN_WARN_MS || 15_000);
/** Transport limits. Block reports carry up to 200 blocks, so keep headroom. */
const MAX_BODY_BYTES = 8 * 1024 * 1024;
const MAX_BODY_WAIT_MS = 30_000;
const MAX_RESPONSE_BYTES = 64 * 1024 * 1024;
const RATE_LIMIT_SWEEP_MS = 60_000;
const RATE_LIMIT_BUCKET_TTL_MS = 10 * 60_000;
const HUB_RATE_LIMIT_PER_MIN = 1200;
/** Anti-fork: stagger VPS↔VPS / VPS↔client sync between 3–33s. */
const ANTI_FORK_DELAY_MIN_MS = Math.max(
  3_000,
  Number(process.env.SARMAD_ANTI_FORK_DELAY_MIN_MS || 3_000) || 3_000
);
const ANTI_FORK_DELAY_MAX_MS = Math.max(
  ANTI_FORK_DELAY_MIN_MS,
  Number(process.env.SARMAD_ANTI_FORK_DELAY_MAX_MS || 33_000) || 33_000
);

function antiForkDelayMs() {
  const span = ANTI_FORK_DELAY_MAX_MS - ANTI_FORK_DELAY_MIN_MS;
  return ANTI_FORK_DELAY_MIN_MS + Math.floor(Math.random() * (span + 1));
}

function sleep(ms) {
  return new Promise((resolve) => setTimeout(resolve, Math.max(0, Number(ms) || 0)));
}

/** Multi-VPS scaffold: empty / single peer → always leader (today's behaviour). */
const vpsCluster = createCluster({
  selfUrl: PUBLIC_BASE_URL || process.env.SARMAD_VPS_SELF_URL || "",
  clusterCsv: process.env.SARMAD_VPS_CLUSTER || "",
});

vpsCluster.setHooks({
  log: console.log,
  getLocalTipHash: () => {
    try {
      const t = relay.getTip && relay.getTip();
      return t && t.hash ? String(t.hash) : "";
    } catch (_) {
      return "";
    }
  },
  getLocalHeight: () => {
    try {
      const t = relay.getTip && relay.getTip();
      return Math.max(Number(cycle && cycle.chainHeight) || 0, Number(t && t.height) || 0);
    } catch (_) {
      return 0;
    }
  },
  getLocalProducerCount: () => {
    try {
      const t = relay.getTip && relay.getTip();
      const h = Math.max(Number(cycle && cycle.chainHeight) || 0, Number(t && t.height) || 0);
      return (snapshotStore.chainProducerSet(h) || []).length;
    } catch (_) {
      return null;
    }
  },
  /** Failover choice B: new leader opens a fresh shuffle of whoever is live now. */
  onBecameLeader: (epoch) => {
    reopenCycleOnBecameLeader(epoch);
  },
  onBecameFollower: (leaderUrl, epoch) => {
    console.log(
      `[vps-cluster] now follower of ${leaderUrl} epoch=${epoch} — watch-only (no skip clock)`
    );
  },
  syncFromLeader: async (leaderUrl) => {
    const st = await httpGetJson(leaderUrl, "/v1/status");
    if (!st || st.status < 200 || st.status >= 300) {
      throw new Error(`leader status HTTP ${st && st.status}`);
    }
    const j = (st && st.json) || {};
    await syncChainFromHub(leaderUrl);
    // If we are ahead of the leader (restart race), push our tip forward so the cluster
    // does not permanently fork at a lower height.
    const localTip = relay.getTip();
    const leaderHeight = Math.max(
      Number(j.chainHeight) || 0,
      Number(j.tip && j.tip.height) || 0
    );
    const leaderGenesis =
      j.genesisTimestamp == null || j.genesisTimestamp === ""
        ? null
        : Number(j.genesisTimestamp);
    // After Reset Network the leader is empty with no genesis. Pushing our leftover
    // blocks would resurrect the wiped chain (phones then "resume" old height).
    if (localTip.height > leaderHeight && leaderHeight === 0 && leaderGenesis == null) {
      console.log(
        "[vps-cluster] leader tip empty after reset — wiping local chain (not pushing old blocks)"
      );
      resetCoordinatorState();
      return;
    }
    if (localTip.height > leaderHeight) {
      let from = leaderHeight + 1;
      let pushed = 0;
      while (from <= localTip.height && pushed < 200) {
        const block = relay.getBlocksFrom(from, 1)[0];
        if (!block) break;
        const body = {
          producerAddress: "",
          blockHash: block.hash,
          height: block.height,
          tipTimestamp: block.header && block.header.timestamp,
          block,
        };
        const res = await httpPostJson(leaderUrl, "/v1/block-report", body, {
          "X-Sarmad-Cluster-Forward": "1",
        });
        if (!res || res.status < 200 || res.status >= 300) break;
        const stored = res.json && res.json.blockStored;
        if (stored === false) break;
        pushed += 1;
        from += 1;
      }
      if (pushed > 0) {
        console.log(
          `[vps-cluster] pushed ${pushed} local block(s) to leader ${leaderUrl} (localTip=${localTip.height})`
        );
      }
    }

    // Anti-fork ACK barrier: if the leader has a pending handoff and we are already at
    // the leader's recorded tip hash, ACK it so the leader can switch safely.
    try {
      const pendingId = String(j.pendingHandoffId || "");
      const pendingEpoch = Number(j.pendingHandoffEpoch || 0);
      const pendingTipHash = String(j.pendingHandoffTipHash || "");
      const localTipHash = relay.getTip && relay.getTip().hash ? String(relay.getTip().hash) : "";
      if (pendingId && pendingTipHash && localTipHash && localTipHash === pendingTipHash) {
        const self = vpsCluster.selfUrl && vpsCluster.selfUrl();
        await httpPostJson(leaderUrl, "/v1/cluster/handoff-ack", {
          pendingHandoffId: pendingId,
          pendingHandoffEpoch: pendingEpoch,
          tipHash: localTipHash,
          fromUrl: self || PUBLIC_BASE_URL || "",
        });
      }
    } catch (_) {}

    if (j.leaderUrl && j.leaderEpoch) {
      vpsCluster.adoptRemoteLeader(j.leaderUrl, j.leaderEpoch);
    } else if (j.isLeader && (j.selfUrl || leaderUrl) && j.leaderEpoch) {
      vpsCluster.adoptRemoteLeader(j.selfUrl || leaderUrl, j.leaderEpoch);
    }
    // Adopt genesis from leader so followers share the same network identity.
    if (j.genesisTimestamp != null && Number.isFinite(Number(j.genesisTimestamp))) {
      setGenesisTimestamp(Number(j.genesisTimestamp));
    }
    if (j.genesisEpochSeconds != null && Number.isFinite(Number(j.genesisEpochSeconds))) {
      genesisEpochSeconds = Number(j.genesisEpochSeconds);
    }
  },
  forwardToLeader: async (leader, pathname, body, headers = {}) => {
    return httpPostJson(leader, pathname, body, {
      "X-Sarmad-Cluster-Forward": "1",
      ...headers,
    });
  },
});

function isClusterForward(req) {
  return String(req.headers["x-sarmad-cluster-forward"] || "") === "1";
}

async function authoritativeCyclePayload() {
  if (vpsCluster.enabled() && !vpsCluster.isLeader()) {
    try {
      const leader = String((vpsCluster.statusFields(0) || {}).leaderUrl || "").replace(/\/$/, "");
      const self = String((vpsCluster.statusFields(0) || {}).selfUrl || "").replace(/\/$/, "");
      if (leader && leader !== self) {
        const r = await httpGetJson(leader, "/v1/cycle?local=1");
        if (r && r.status >= 200 && r.status < 300 && r.json && r.json.cycleId != null) {
          return r.json;
        }
      }
    } catch (_) {}
  }
  return cyclePayload();
}

async function maybeForwardFollowerWrite(req, res, pathname, body) {
  if (isClusterForward(req)) return false;
  const fwd = await vpsCluster.forwardIfFollower(pathname, body);
  if (!fwd.forwarded) return false;
  const r = fwd.result || {};
  sendJson(res, r.status && r.status >= 100 ? r.status : 502, r.json || { error: "cluster forward failed" });
  return true;
}

function requestRemoteIp(req) {
  const raw = String((req.socket && req.socket.remoteAddress) || "");
  return raw.replace(/^::ffff:/, "");
}

/** Allow /v1/cluster/merge-nodes only from this host or configured cluster peers. */
function isClusterPeerRequest(req) {
  const ip = requestRemoteIp(req);
  if (!ip) return false;
  if (ip === "127.0.0.1" || ip === "::1" || ip === "localhost") return true;
  let peers = [];
  try {
    peers = vpsCluster.peers() || [];
  } catch (_) {
    peers = [];
  }
  for (const p of peers) {
    try {
      const host = new URL(p).hostname;
      if (host && (host === ip || host === "127.0.0.1" || host === "localhost")) return true;
    } catch (_) {}
  }
  return false;
}

function selfHubBaseUrl() {
  return String(exeConfig.publicBaseUrl || PUBLIC_BASE_URL || "")
    .trim()
    .replace(/\/$/, "");
}

function selfHubAddress() {
  return String(exeConfig.address || "");
}

function fanoutBlock(report) {
  // Deferred gossip: let the local tip settle before peers ingest (anti-fork).
  const wait = antiForkDelayMs();
  setTimeout(() => {
    hubGossip.gossipBlock(selfHubBaseUrl(), report, console.warn, selfHubAddress());
  }, wait);
}

function fanoutTx(tx) {
  hubGossip.gossipTx(selfHubBaseUrl(), tx, console.warn, selfHubAddress());
}

const lastHbFanoutMs = new Map();

function upsertNodeFromHeartbeat(body, extra = {}) {
  const address = String(body.address || "");
  if (!address) return false;
  const prev = nodes.get(address);
  const incomingSeen = Number(body.lastSeenMs || 0);
  const lastSeenMs =
    incomingSeen > 0
      ? Math.max(incomingSeen, prev && prev.lastSeenMs ? prev.lastSeenMs : 0)
      : Date.now();
  const sourceIp = extra.sourceIp || (prev && prev.sourceIp) || null;
  const wantProduce =
    extra.producingEnabled !== undefined
      ? extra.producingEnabled !== false
      : body.producingEnabled !== undefined
        ? body.producingEnabled !== false
        : prev
          ? prev.producingEnabled !== false
          : true;
  nodes.set(address, {
    address,
    publicKey: String(body.publicKey || (prev && prev.publicKey) || ""),
    peerHost: body.peerHost || (prev && prev.peerHost) || null,
    peerPort: body.peerPort
      ? Number(body.peerPort)
      : prev && prev.peerPort
        ? Number(prev.peerPort)
        : null,
    chainHeight: Number(body.chainHeight || (prev && prev.chainHeight) || 0),
    tipHash: String(body.tipHash || (prev && prev.tipHash) || ""),
    tipTimestamp: Number(body.tipTimestamp || (prev && prev.tipTimestamp) || 0),
    producingEnabled: wantProduce,
    sourceIp,
    transport: String(body.transport || (prev && prev.transport) || "tcp"),
    clientVersion: String(body.clientVersion || (prev && prev.clientVersion) || ""),
    platform: String(body.platform || (prev && prev.platform) || ""),
    lastSeenMs,
  });
  if (wantProduce && snapshotStore.unparkProducer(address)) {
    pendingNewcomerAdmit = true;
    console.log(
      `[sarmad-coordinator] unparked ${address} — rejoins queue as a newcomer after this slot`
    );
  }
  // ensureCycle() seeds the producer shuffle from cycle.tipHash, so the cycle tip has
  // to come from our own relay chain — a heartbeat body is client-supplied and must
  // never steer the seed. The node record above still keeps the claimed tip.
  const relayTip = relay.getTip();
  if (relayTip && relayTip.hash) cycle.tipHash = String(relayTip.hash);
  if (relayTip && relayTip.timestamp) cycle.tipTimestamp = Number(relayTip.timestamp);
  if (body.chainHeight != null) {
    cycle.chainHeight = Math.min(
      Math.max(cycle.chainHeight, Number(body.chainHeight) || 0),
      (relayTip && relayTip.height) || 0
    );
  }
  cycle.blockTimeSeconds = effectiveBlockTimeSeconds(cycle.tipTimestamp);
  ensureCycle();
  return true;
}

function fanoutHeartbeatToCluster(body) {
  if (!vpsCluster.enabled()) return;
  const self = normalizeClusterSelf();
  for (const peer of vpsCluster.peers()) {
    if (!peer || peer === self) continue;
    httpPostJson(peer, "/v1/cluster/merge-nodes", {
      nodes: [
        {
          address: body.address,
          publicKey: body.publicKey,
          peerHost: body.peerHost,
          peerPort: body.peerPort,
          chainHeight: body.chainHeight,
          tipHash: body.tipHash,
          tipTimestamp: body.tipTimestamp,
          producingEnabled: body.producingEnabled !== false,
          transport: body.transport,
          clientVersion: body.clientVersion,
          platform: body.platform,
          lastSeenMs: Date.now(),
        },
      ],
    }).catch(() => {});
  }
}

function normalizeClusterSelf() {
  try {
    return String(vpsCluster.selfUrl() || "").replace(/\/$/, "");
  } catch (_) {
    return "";
  }
}

function fanoutHeartbeat(body) {
  const address = String(body && body.address || "");
  if (!address) return;
  const now = Date.now();
  const last = lastHbFanoutMs.get(address) || 0;
  if (now - last < 12_000) return;
  lastHbFanoutMs.set(address, now);
  // Addresses come and go over months; drop entries for nodes we stopped hearing from.
  if (lastHbFanoutMs.size > 1000) {
    for (const [addr, seen] of lastHbFanoutMs) {
      if (now - seen > HEARTBEAT_TTL_MS) lastHbFanoutMs.delete(addr);
    }
  }
  hubGossip.gossipHeartbeat(selfHubBaseUrl(), body, console.warn, selfHubAddress());
  fanoutHeartbeatToCluster(body);
}

/** Push full live registry to cluster peers so leader/follower active-sets stay aligned. */
async function syncActiveSetWithCluster() {
  if (!vpsCluster.enabled()) return;
  const self = normalizeClusterSelf();
  const snapshot = attachedNodes().map((n) => ({
    address: n.address,
    publicKey: n.publicKey || "",
    peerHost: n.peerHost || null,
    peerPort: n.peerPort || null,
    chainHeight: n.chainHeight || 0,
    tipHash: n.tipHash || "",
    tipTimestamp: n.tipTimestamp || 0,
    producingEnabled: n.producingEnabled !== false,
    transport: n.transport || "tcp",
    clientVersion: n.clientVersion || "",
    platform: n.platform || "",
    lastSeenMs: n.lastSeenMs || Date.now(),
  }));
  if (!snapshot.length) return;
  for (const peer of vpsCluster.peers()) {
    if (!peer || peer === self) continue;
    try {
      await httpPostJson(peer, "/v1/cluster/merge-nodes", {
        nodes: snapshot,
        parkedProducers: snapshotStore.parkedProducers(),
      });
    } catch (_) {}
  }
}

function applyPhoneHeartbeat(body, req) {
  return releaseStore.verifyClientHeartbeat(body).then((auth) => {
    if (!auth.ok) return auth;
    const sourceIp = admissionIp(req ? clientIp(req) : "");
    const producing = gatedProducingEnabled(body, sourceIp);
    upsertNodeFromHeartbeat(body, { sourceIp, producingEnabled: producing });
    if (body.producingEnabled !== false && !producing) {
      return { ok: true, producingEnabled: false, admission: admissionHint(body.address) };
    }
    return { ok: true, producingEnabled: producing };
  });
}

function admissionIp(raw) {
  let ip = String(raw || "unknown").trim();
  if (ip.startsWith("::ffff:")) ip = ip.slice(7);
  if (ip.includes(":")) {
    const compact = ip.split("%")[0];
    const parts = compact.split(":").filter((p) => p.length);
    return parts.slice(0, 4).join(":") + "::/64";
  }
  return ip;
}

function liveProducerCountForIp(ip, exceptAddr) {
  pruneDead();
  return [...nodes.values()].filter(
    (n) =>
      n.producingEnabled !== false &&
      n.sourceIp === ip &&
      n.address !== exceptAddr
  ).length;
}

function gatedProducingEnabled(body, sourceIp) {
  const addr = String(body.address || "");
  if (!addr) return false;
  if (body.producingEnabled === false) return false;
  const prev = nodes.get(addr);
  const stillLive =
    prev &&
    prev.producingEnabled !== false &&
    Date.now() - (prev.lastSeenMs || 0) <= HEARTBEAT_TTL_MS;
  if (sourceIp && sourceIp !== "unknown" && !stillLive) {
    if (liveProducerCountForIp(sourceIp, addr) >= MAX_PRODUCERS_PER_IP) return false;
  }
  const height = relay.getTip().height;
  if (height >= PRODUCER_ADMISSION_HEIGHT && !snapshotStore.isProducerBonded(addr)) {
    return false;
  }
  return true;
}

function admissionHint(address) {
  const height = relay.getTip().height;
  const needBond = height >= PRODUCER_ADMISSION_HEIGHT && !snapshotStore.isProducerBonded(address);
  return {
    producerAdmissionHeight: PRODUCER_ADMISSION_HEIGHT,
    maxProducersPerIp: MAX_PRODUCERS_PER_IP,
    needBond,
  };
}

/** Cluster-trusted merge (no client signature) — only for /v1/cluster/merge-nodes. */
function mergeClusterNodes(list, parked) {
  if (Array.isArray(parked)) {
    snapshotStore.replaceParkedProducers(parked);
  }
  if (!Array.isArray(list)) return 0;
  let n = 0;
  for (const row of list) {
    if (!row || !row.address) continue;
    if (upsertNodeFromHeartbeat(row)) n += 1;
  }
  return n;
}

/**
 * Live block-report / hub block ingest.
 * Validates structure + tx signatures + scheduled producer before storing.
 * Marks producer completed only after a successful store (fixes burned-slot-on-reject).
 * @param {object} body
 * @param {{ skipProducerTurn?: boolean }} [opts]  cluster catch-up / hub backfill
 */
function ingestReportedBlock(body, opts = {}) {
  const producer = String(body.producerAddress || "");
  const height = Number(body.height || 0);
  const tip = relay.getTip();
  let stored = null;
  const skipProducerTurn = Boolean(opts.skipProducerTurn);

  if (body.block) {
    // Coordinator owns wall pacing: refuse produce before this turn's scheduled open.
    if (!skipProducerTurn && Date.now() < Number(cycle.slotOpenedAtMs || 0)) {
      return {
        ok: false,
        error: "slot not open yet",
        tip,
        secondsToNextSlot: secondsToNextSlot(),
      };
    }
    const remaining = remainingOrder();
    const activeAddresses = cycle.activeNodes.map((n) => n.address);
    const check = blockValidator.validateForRelayIngest(
      body.block,
      tip,
      height || tip.height + 1,
      {
        activeAddresses,
        completedAddresses: cycle.completed,
        remaining,
        claimedProducerAddress: producer,
      },
      {
        // VPS moves skipped producers into completed; phones use allowedSkipDepth=0.
        allowedSkipDepth: 0,
        strictProducer:
          !skipProducerTurn && process.env.SARMAD_RELAY_SKIP_PRODUCER !== "1",
      }
    );
    if (!check.valid) {
      console.warn(
        `[sarmad-coordinator] block-report rejected h=${height}: ${check.message}`
      );
      return { ok: false, error: check.message || "invalid block", tip };
    }
    const canonicalHash = check.computedHash || body.blockHash;
    stored = relay.storeBlock(body.block, {
      height,
      blockHash: canonicalHash,
      tipTimestamp: body.tipTimestamp,
    });
    if (stored && stored.ok) {
      const resolvedProducer = check.producerAddress || producer;
      if (!stored.duplicate) {
        const short =
          resolvedProducer && resolvedProducer.length > 20
            ? resolvedProducer.slice(0, 12) + "…" + resolvedProducer.slice(-6)
            : resolvedProducer || "?";
        console.log(
          `[sarmad-coordinator] stored block h=${height}` +
            ` producer=${short}` +
            (skipProducerTurn ? " (hub/catch-up)" : "")
        );
      }
      if (
        !skipProducerTurn &&
        resolvedProducer &&
        !cycle.completed.includes(resolvedProducer)
      ) {
        cycle.completed.push(resolvedProducer);
      }
      if (!skipProducerTurn) {
        openNextSlot();
      }
      if (!stored.duplicate) {
        try {
          coordRewards.observeBlock(body.block);
        } catch (_) {}
        try {
          snapshotStore.onBlockStored(relay, {
            height,
            hash: canonicalHash || (body.block && body.block.hash),
            header: body.block.header,
            transactions: body.block.transactions || [],
          });
        } catch (_) {}
      }
      const newTip = relay.getTip();
      cycle.chainHeight = Math.max(cycle.chainHeight, newTip.height);
      cycle.tipHash = newTip.hash;
      cycle.tipTimestamp = newTip.timestamp;
    }
  } else {
    // Metadata-only report (legacy): do not advance completed without a stored body.
    cycle.chainHeight = Math.max(cycle.chainHeight, Math.min(height, tip.height));
    if (body.blockHash && tip.height >= height) cycle.tipHash = String(body.blockHash);
    if (body.tipTimestamp != null && tip.height >= height) {
      cycle.tipTimestamp = Number(body.tipTimestamp);
    }
  }
  cycle.blockTimeSeconds = effectiveBlockTimeSeconds(cycle.tipTimestamp);
  // Slot finished — admit waiting newcomers into a fresh shuffle (don't wait for full drain).
  if (stored && stored.ok && !skipProducerTurn) {
    if (pendingNewcomerAdmit) {
      pendingNewcomerAdmit = false;
      if (!reopenCycleForNewcomers()) ensureCycle();
    } else if (!reopenCycleForNewcomers()) {
      ensureCycle();
    }
  }
  return stored;
}

/** @type {Map<string, any>} */
const nodes = new Map(); // address -> node record
/** JOIN'd wallets that missed one offline slot live in AccountState.producerParked. */
/** Unparked miners wait for the next slot boundary so they rejoin as newcomers. */
let pendingNewcomerAdmit = false;

/** @type {Map<string, any>} */
const coordinators = new Map(); // address -> EXE coordinator record

/** Local EXE operator identity (set via /v1/exe/config when running as edge coordinator). */
let exeConfig = {
  address: "",
  publicKey: "",
  publicBaseUrl: PUBLIC_BASE_URL,
  signature: "",
  inviterAddress: "",
};

function getResetToken() {
  return String(process.env.SARMAD_RESET_TOKEN || "");
}

/**
 * Real-world moment genesis represents (Unix seconds), captured once at launch and persisted
 * so restarts keep the same anchor. Display only: phones convert network seconds to a date with it.
 * A dev reset captures a new anchor because reset means a new genesis.
 */
const genesisEpochFile = dataPath("genesis_epoch.json");
let genesisEpochSeconds = 0;

/**
 * Authoritative genesis block timestamp (network seconds) for the whole network.
 * Every node must build the same genesis from this value, otherwise chains diverge and
 * a second device would start its own network instead of joining.
 * null means "not published yet"; 0 is a valid timestamp (canonical genesis).
 */
let genesisTimestamp = null;

function loadGenesisEpoch() {
  try {
    const raw = fs.readFileSync(genesisEpochFile, "utf8");
    const parsed = JSON.parse(raw);
    const value = Number(parsed && parsed.genesisEpochSeconds);
    const ts = Number(parsed && parsed.genesisTimestamp);
    return {
      epoch: Number.isFinite(value) && value > 0 ? Math.floor(value) : 0,
      timestamp:
        parsed && parsed.genesisTimestamp != null && Number.isFinite(ts) && ts >= 0
          ? Math.floor(ts)
          : null,
    };
  } catch (_) {
    return { epoch: 0, timestamp: null };
  }
}

function saveGenesisEpoch() {
  try {
    fs.writeFileSync(
      genesisEpochFile,
      JSON.stringify({ genesisEpochSeconds, genesisTimestamp }, null, 2),
      "utf8"
    );
  } catch (e) {
    console.error("[sarmad-coordinator] could not persist genesis epoch:", e && e.message);
  }
}

function captureGenesisEpoch() {
  genesisEpochSeconds = Math.floor(Date.now() / 1000);
  saveGenesisEpoch();
  console.log(
    `[sarmad-coordinator] genesis anchored at ${new Date(genesisEpochSeconds * 1000).toISOString()}`
  );
}

function setGenesisTimestamp(ts) {
  if (ts == null) return false;
  const value = Math.floor(Number(ts));
  if (!Number.isFinite(value) || value < 0) return false;
  genesisTimestamp = value;
  saveGenesisEpoch();
  console.log(`[sarmad-coordinator] network genesis timestamp set to ${genesisTimestamp}`);
  return true;
}

function ensureGenesisEpoch() {
  const stored = loadGenesisEpoch();
  if (stored.timestamp != null) genesisTimestamp = stored.timestamp;
  if (genesisEpochSeconds > 0) {
    // fall through
  } else if (stored.epoch > 0) {
    genesisEpochSeconds = stored.epoch;
  } else {
    captureGenesisEpoch();
  }
  // Published genesis with an empty tip is leftover after a wipe — clear so phones can re-anchor.
  if (relay.getTip().height === 0 && genesisTimestamp != null && !DIRECTORY_URL) {
    console.log(
      "[sarmad-coordinator] directory tip empty — clearing stale genesisTimestamp=" + genesisTimestamp
    );
    genesisTimestamp = null;
    saveGenesisEpoch();
  }
}

function createGenesisCycle() {
  return {
    cycleId: 1,
    seed: "0".repeat(64),
    activeNodes: [], // {address, publicKey}
    completed: [], // addresses produced or skipped this cycle
    chainHeight: 0,
    tipHash: "0".repeat(64),
    tipTimestamp: 0,
    slotOpenedAtMs: Date.now(),
    blockTimeSeconds: BLOCK_TIME_SECONDS,
  };
}

let cycle = createGenesisCycle();

function resetCoordinatorState() {
  nodes.clear();
  coordinators.clear();
  cycle = createGenesisCycle();
  // The resetting device publishes its new genesis right after; until then nodes keep
  // whatever genesis they have instead of adopting a stale one.
  genesisTimestamp = null;
  captureGenesisEpoch();
  relay.resetStore();
  try {
    snapshotStore.resetSnapshots();
  } catch (_) {}
  try {
    broadcastStore.resetBroadcasts();
  } catch (_) {}
  try {
    coordRewards.resetRewards();
  } catch (_) {}
}

/** POST /v1/reset (and genesis) to every other cluster peer. Token is not logged. */
async function fanoutClusterAuthPost(pathname, body, token) {
  const peers = typeof vpsCluster.peers === "function" ? vpsCluster.peers() : [];
  const self = typeof vpsCluster.selfUrl === "function" ? vpsCluster.selfUrl() || "" : "";
  for (const peer of peers) {
    if (!peer || peer === self) continue;
    try {
      const r = await httpPostJson(peer, pathname, body, {
        "X-Sarmad-Cluster-Forward": "1",
        "X-Sarmad-Reset-Token": token,
      });
      console.log(
        `[sarmad-coordinator] cluster fanout ${pathname} → ${peer} HTTP ${r && r.status}`
      );
    } catch (e) {
      console.warn(
        `[sarmad-coordinator] cluster fanout ${pathname} → ${peer} failed: ${
          e && e.message ? e.message : e
        }`
      );
    }
  }
}

function httpGetJson(baseUrl, pathname) {
  return new Promise((resolve, reject) => {
    let url;
    try {
      url = new URL(pathname, baseUrl.endsWith("/") ? baseUrl : baseUrl + "/");
    } catch (e) {
      reject(e);
      return;
    }
    const lib = url.protocol === "https:" ? require("https") : require("http");
    const req = lib.get(
      {
        hostname: url.hostname,
        port: url.port || (url.protocol === "https:" ? 443 : 80),
        path: url.pathname + url.search,
        timeout: 15_000,
      },
      (res) => {
        const chunks = [];
        let size = 0;
        let aborted = false;
        res.on("data", (c) => {
          if (aborted) return;
          size += c.length;
          // A hostile or broken peer answering /v1/blocks must not be able to
          // OOM this coordinator during hub sync.
          if (size > MAX_RESPONSE_BYTES) {
            aborted = true;
            req.destroy();
            reject(new Error("response too large"));
            return;
          }
          chunks.push(c);
        });
        res.on("end", () => {
          if (aborted) return;
          try {
            resolve({
              status: res.statusCode || 0,
              json: JSON.parse(Buffer.concat(chunks).toString("utf8") || "{}"),
            });
          } catch (e) {
            reject(e);
          }
        });
      }
    );
    req.on("error", reject);
    req.on("timeout", () => {
      req.destroy();
      reject(new Error("timeout"));
    });
  });
}

function httpPostJson(baseUrl, pathname, body, extraHeaders = {}) {
  return new Promise((resolve, reject) => {
    let url;
    try {
      url = new URL(pathname, baseUrl.endsWith("/") ? baseUrl : baseUrl + "/");
    } catch (e) {
      reject(e);
      return;
    }
    const payload = Buffer.from(JSON.stringify(body == null ? {} : body), "utf8");
    const lib = url.protocol === "https:" ? require("https") : require("http");
    const req = lib.request(
      {
        hostname: url.hostname,
        port: url.port || (url.protocol === "https:" ? 443 : 80),
        path: url.pathname + url.search,
        method: "POST",
        headers: {
          "Content-Type": "application/json",
          "Content-Length": payload.length,
          ...extraHeaders,
        },
        timeout: 15_000,
      },
      (res) => {
        const chunks = [];
        res.on("data", (c) => chunks.push(c));
        res.on("end", () => {
          let json = {};
          try {
            json = JSON.parse(Buffer.concat(chunks).toString("utf8") || "{}");
          } catch (_) {}
          resolve({ status: res.statusCode || 0, json });
        });
      }
    );
    req.on("error", reject);
    req.on("timeout", () => {
      req.destroy();
      reject(new Error("timeout"));
    });
    req.write(payload);
    req.end();
  });
}

async function syncChainFromHub(hubUrl) {
  const base = String(hubUrl || "").replace(/\/$/, "");
  if (!base) return;
  try {
    const st = await httpGetJson(base, "/v1/status");
    if (!st || st.status < 200 || st.status >= 300) return;
    const remote = st.json || {};
    const remoteHeight = Math.max(
      Number(remote.chainHeight) || 0,
      Number(remote.tip && remote.tip.height) || 0
    );
    let from = relay.getTip().height + 1;
    if (from > remoteHeight) return;
    // Stagger VPS↔VPS pull so both sides do not race the same tip window.
    await sleep(antiForkDelayMs());
    let pulled = 0;
    while (from <= remoteHeight) {
      const batch = await httpGetJson(base, `/v1/blocks?from=${from}&limit=100`);
      if (!batch || batch.status < 200 || batch.status >= 300) break;
      const blocks = (batch.json && batch.json.blocks) || [];
      if (!blocks.length) break;
      let progressed = false;
      for (const b of blocks.slice().sort((a, c) => a.height - c.height)) {
        const localTip = relay.getTip();
        const struct = blockValidator.validateBlockStructure(b, localTip, b.height);
        if (!struct.valid) {
          console.warn(
            `[sarmad-coordinator] hub backfill rejected h=${b.height}: ${struct.message}`
          );
          return;
        }
        const stored = relay.storeBlock(b, {
          height: b.height,
          blockHash: struct.computedHash || b.hash,
          tipTimestamp: b.header && b.header.timestamp,
        });
        if (stored && stored.ok) {
          progressed = true;
          if (!stored.duplicate) {
            pulled += 1;
            try {
              snapshotStore.onBlockStored(relay, {
                height: b.height,
                hash: struct.computedHash || b.hash,
                header: b.header,
                transactions: b.transactions || [],
              });
            } catch (_) {}
          }
        } else {
          return;
        }
      }
      if (!progressed) break;
      from = relay.getTip().height + 1;
    }
    if (pulled > 0) {
      const tip = relay.getTip();
      cycle.chainHeight = Math.max(cycle.chainHeight, tip.height);
      cycle.tipHash = tip.hash;
      cycle.tipTimestamp = tip.timestamp;
      console.log(`[sarmad-coordinator] backfilled ${pulled} block(s) from hub ${base} → tip=${tip.height}`);
    }
  } catch (e) {
    console.warn("[sarmad-coordinator] hub sync failed:", e && e.message ? e.message : e, "@", base);
  }
}

/**
 * EXE edge nodes: adopt directory genesis when local tip is empty, then pull missing blocks.
 * Also used after gossip so hubs that were offline can catch up via pull.
 */
async function syncChainFromDirectory() {
  if (!DIRECTORY_URL) return;
  try {
    const st = await httpGetJson(DIRECTORY_URL, "/v1/status");
    if (!st || st.status < 200 || st.status >= 300) return;
    const remote = st.json || {};
    const remoteGenesis =
      remote.genesisTimestamp != null && Number.isFinite(Number(remote.genesisTimestamp))
        ? Math.floor(Number(remote.genesisTimestamp))
        : null;
    const localTip = relay.getTip();
    const remoteHeight = Math.max(
      Number(remote.chainHeight) || 0,
      Number(remote.tip && remote.tip.height) || 0
    );

    if (remoteHeight <= 0 && localTip.height === 0) {
      if (genesisTimestamp != null) {
        genesisTimestamp = null;
        saveGenesisEpoch();
        console.log(
          "[sarmad-coordinator] cleared stale genesis (directory and local tip empty)"
        );
      }
      return;
    }

    if (remoteGenesis != null && remoteHeight > 0) {
      if (genesisTimestamp == null) {
        setGenesisTimestamp(remoteGenesis);
      } else if (genesisTimestamp !== remoteGenesis && localTip.height === 0) {
        genesisTimestamp = remoteGenesis;
        saveGenesisEpoch();
        console.log(
          `[sarmad-coordinator] adopted directory genesis ${remoteGenesis} (local tip empty)`
        );
      } else if (genesisTimestamp !== remoteGenesis && localTip.height > 0) {
        console.warn(
          `[sarmad-coordinator] genesis mismatch vs directory (local=${genesisTimestamp} remote=${remoteGenesis}) — not wiping local tip=${localTip.height}`
        );
        return;
      }
    }

    if (remote.genesisEpochSeconds && Number(remote.genesisEpochSeconds) > 0 && remoteHeight > 0) {
      const ep = Math.floor(Number(remote.genesisEpochSeconds));
      if (ep !== genesisEpochSeconds && localTip.height === 0) {
        genesisEpochSeconds = ep;
        saveGenesisEpoch();
      }
    }

    await syncChainFromHub(DIRECTORY_URL);
    try {
      coordRewards.sanitizeAgainstTip(relay.getTip().timestamp);
    } catch (_) {}
  } catch (e) {
    console.warn(
      "[sarmad-coordinator] directory sync failed:",
      e && e.message ? e.message : e
    );
  }
}

function pruneDeadCoordinators() {
  const now = Date.now();
  for (const [addr, c] of coordinators.entries()) {
    if (now - c.lastSeenMs > COORDINATOR_TTL_MS) coordinators.delete(addr);
  }
}

function liveCoordinators() {
  pruneDeadCoordinators();
  return [...coordinators.values()].sort((a, b) => {
    if (a.activeCount !== b.activeCount) return a.activeCount - b.activeCount;
    return String(a.address).localeCompare(String(b.address));
  });
}

function isValidCoordBaseUrl(url) {
  try {
    const u = new URL(String(url || ""));
    if (u.protocol !== "http:" && u.protocol !== "https:") return false;
    if (!u.hostname) return false;
    return true;
  } catch (_) {
    return false;
  }
}

function coordinatorPublicPayload(c) {
  return {
    id: c.address,
    address: c.address,
    publicKey: c.publicKey || "",
    baseUrl: c.baseUrl,
    activeCount: Number(c.activeCount || 0),
    relayHeight: Number(c.relayHeight || 0),
    version: c.version || "",
    lastSeenMs: c.lastSeenMs,
  };
}

/**
 * Authorize POST /v1/reset. Never logs the token.
 * @param {import("http").IncomingMessage} req
 * @returns {{ ok: true } | { ok: false, code: number, error: string }}
 */
function authorizeReset(req) {
  const expected = getResetToken();
  if (!expected) {
    return { ok: false, code: 503, error: "reset disabled" };
  }
  const provided = req.headers["x-sarmad-reset-token"];
  if (provided == null || provided === "") {
    return { ok: false, code: 403, error: "forbidden" };
  }
  const a = Buffer.from(String(provided), "utf8");
  const b = Buffer.from(expected, "utf8");
  if (a.length !== b.length || !crypto.timingSafeEqual(a, b)) {
    return { ok: false, code: 403, error: "forbidden" };
  }
  return { ok: true };
}

function effectiveBlockTimeSeconds(_networkTimestampSeconds) {
  return BLOCK_TIME_SECONDS;
}

function skipGraceSeconds(_blockTime) {
  // Exact 33s deadline — no extra grace. Wall-clock pacing is owned here via openNextSlot().
  return 0;
}

/**
 * After a block or a full-slot skip: open the next turn on a 33s wall-clock cadence.
 * Early submits must not collapse the gap (clients may finish at second 0).
 * Late finishes (at/after deadline) start the next turn immediately.
 */
function openNextSlot() {
  const btMs = (cycle.blockTimeSeconds || BLOCK_TIME_SECONDS) * 1000;
  const scheduledEnd = Number(cycle.slotOpenedAtMs || 0) + btMs;
  cycle.slotOpenedAtMs = Math.max(Date.now(), scheduledEnd);
}

function pruneDead() {
  const now = Date.now();
  for (const [addr, n] of nodes.entries()) {
    if (now - n.lastSeenMs > HEARTBEAT_TTL_MS) nodes.delete(addr);
  }
}

function liveNodes() {
  pruneDead();
  return [...nodes.values()].filter((n) => n.producingEnabled !== false);
}

function liveProducerRecords() {
  return liveNodes()
    .filter((n) => !snapshotStore.isParked(n.address))
    .map((n) => ({ address: n.address, publicKey: n.publicKey || "" }))
    .sort((a, b) => a.address.localeCompare(b.address));
}

/** On-chain shuffle set, or null when empty (bootstrap: walk live heartbeats). */
function chainQueueNodes() {
  const tip = relay.getTip();
  const height = Math.max(Number(cycle.chainHeight) || 0, Number(tip && tip.height) || 0);
  const addrs = snapshotStore.chainProducerSet(height) || [];
  if (!addrs.length) return null;
  return addrs
    .map((address) => {
      const n = nodes.get(address);
      return {
        address,
        publicKey: (n && n.publicKey) || snapshotStore.producerPublicKey(address) || "",
      };
    })
    .sort((a, b) => a.address.localeCompare(b.address));
}

/**
 * Prefer on-chain producers that are currently live with producingEnabled.
 * Exact-tip clients send producingEnabled=false while catching up — exclude them
 * from the produce queue. When the chain set is non-empty, never invent miners from
 * bare live heartbeats (clients reject that via vpsCycleSetAllowed). If no JOIN'd
 * wallet is live, use the full chain queue so offline slots can be skipped.
 * Bootstrap only: empty chain set → live heartbeats.
 */
function cycleActiveFromChainOrLive() {
  const chain = chainQueueNodes();
  if (!chain) return liveProducerRecords();
  const live = liveProducerRecords();
  const liveSet = new Set(live.map((n) => n.address));
  const intersected = chain.filter((n) => liveSet.has(n.address));
  return intersected.length ? intersected : chain;
}

function queueSource() {
  const chain = chainQueueNodes();
  if (!chain) return "live-bootstrap";
  const live = liveProducerRecords();
  const liveSet = new Set(live.map((n) => n.address));
  const intersected = chain.filter((n) => liveSet.has(n.address));
  return intersected.length ? "chain" : "chain-offline";
}

/** All heartbeating clients (producers + listeners) — used for sticky/queue load hints. */
function attachedNodes() {
  pruneDead();
  return [...nodes.values()];
}

function shuffleOrder(addresses, seedHex) {
  const sorted = [...new Set(addresses)].sort();
  if (sorted.length <= 1) return sorted;
  const order = [...sorted];
  let state = seedToState(seedHex);
  for (let i = order.length - 1; i >= 1; i--) {
    state = nextState(state);
    const j = Number(state % BigInt(i + 1));
    const tmp = order[i];
    order[i] = order[j];
    order[j] = tmp;
  }
  return order;
}

function seedToState(seedHex) {
  const normalized = String(seedHex || "").toLowerCase().padStart(64, "0").slice(0, 64);
  const hash = crypto.createHash("sha3-256").update(normalized, "utf8").digest();
  let value = 0n;
  for (let i = 0; i < 8; i++) value = (value << 8n) | BigInt(hash[i]);
  return value === 0n ? 1n : value;
}

function nextState(state) {
  let x = state;
  x ^= x >> 12n;
  x ^= x << 25n;
  x ^= x >> 27n;
  return BigInt.asUintN(64, x * 0x2545f4914f6cdd1dn);
}

function remainingOrder() {
  const order = shuffleOrder(
    cycle.activeNodes.map((n) => n.address),
    cycle.seed
  );
  const done = new Set(cycle.completed);
  return order.filter((a) => !done.has(a));
}

function maybeOpenNewCycle(tipHash) {
  if (cycle.activeNodes.length === 0 || remainingOrder().length === 0) {
    const scheduledOpen = Number(cycle.slotOpenedAtMs) || 0;
    const active = cycleActiveFromChainOrLive();
    cycle = {
      ...cycle,
      cycleId: cycle.cycleId + (cycle.activeNodes.length === 0 && cycle.cycleId === 1 ? 0 : 1),
      seed: tipHash || cycle.tipHash || cycle.seed,
      activeNodes: active,
      completed: [],
      // Keep openNextSlot() pacing — never collapse a scheduled future open to now.
      slotOpenedAtMs: Math.max(Date.now(), scheduledOpen),
      blockTimeSeconds: effectiveBlockTimeSeconds(cycle.tipTimestamp),
    };
    if (cycle.cycleId === 1 && cycle.activeNodes.length && cycle.completed.length === 0) {
      // keep cycleId 1 for first real set
    } else if (active.length && remainingOrder().length === active.length && cycle.completed.length === 0) {
      // freshly opened
    }
  }
}

function ensureCycle() {
  pruneDead();
  if (cycle.activeNodes.length === 0) {
    cycle.activeNodes = cycleActiveFromChainOrLive();
    cycle.seed = cycle.tipHash || cycle.seed;
    cycle.completed = [];
    cycle.slotOpenedAtMs = Date.now();
    cycle.blockTimeSeconds = effectiveBlockTimeSeconds(cycle.tipTimestamp);
  } else if (remainingOrder().length === 0) {
    const scheduledOpen = Number(cycle.slotOpenedAtMs) || 0;
    cycle.cycleId += 1;
    cycle.activeNodes = cycleActiveFromChainOrLive();
    cycle.seed = cycle.tipHash || cycle.seed;
    cycle.completed = [];
    // Preserve openNextSlot() wall cadence across cycle rollover.
    cycle.slotOpenedAtMs = Math.max(Date.now(), scheduledOpen);
    cycle.blockTimeSeconds = effectiveBlockTimeSeconds(cycle.tipTimestamp);
  }
}

/**
 * After a slot finishes (block or skip): if any live phone is missing from the frozen
 * set, open a fresh cycle immediately so newcomers get turns without waiting for a
 * full queue drain (can be many minutes when many devices are online).
 * Call only at slot boundaries — never mid-slot from a random heartbeat.
 *
 * @param {{ immediate?: boolean }} opts  immediate=true → next producer may go now
 *   (used after skipping offline nodes so we do not stack another full 33s wait).
 */
function reopenCycleForNewcomers(opts = {}) {
  pruneDead();
  if (cycle.activeNodes.length === 0) return false;
  // Newcomers and returning skipped miners wait until the bag empties (or failover B).
  // Do not reshuffle mid-cycle just because someone joined.
  if (remainingOrder().length > 0) return false;
  const known = new Set(cycle.activeNodes.map((n) => n.address));
  const chain = chainQueueNodes();
  const nextActive = cycleActiveFromChainOrLive();
  const newCount = nextActive.reduce((n, node) => n + (known.has(node.address) ? 0 : 1), 0);
  const extra = chain
    ? cycle.activeNodes.some((n) => !chain.some((c) => c.address === n.address))
    : 0;
  if (newCount === 0 && !extra) return false;
  const scheduledOpen = Number(cycle.slotOpenedAtMs) || 0;
  cycle.cycleId += 1;
  cycle.activeNodes = nextActive;
  cycle.seed = cycle.tipHash || cycle.seed;
  cycle.completed = [];
  cycle.slotOpenedAtMs = opts.immediate
    ? Date.now()
    : Math.max(Date.now(), scheduledOpen);
  cycle.blockTimeSeconds = effectiveBlockTimeSeconds(cycle.tipTimestamp);
  console.log(
    `[sarmad-coordinator] cycle ${cycle.cycleId} reopened ` +
      `(${queueSource()} new=${newCount} extra=${Boolean(extra)}); ` +
      `active=${nextActive.length} immediate=${Boolean(opts.immediate)}`
  );
  return true;
}

/**
 * Hot-standby failover (choice B): discard any stale local cursor and open a new
 * cycle from currently live / on-chain producers. Call only when this node becomes leader.
 */
function reopenCycleOnBecameLeader(epoch) {
  pruneDead();
  const tip = relay.getTip && relay.getTip();
  if (tip && tip.hash) cycle.tipHash = String(tip.hash);
  if (tip && tip.height != null) cycle.chainHeight = Number(tip.height) || cycle.chainHeight;
  if (tip && tip.header && tip.header.timestamp != null) {
    cycle.tipTimestamp = Number(tip.header.timestamp) || cycle.tipTimestamp;
  }
  const active = cycleActiveFromChainOrLive();
  cycle.cycleId += 1;
  cycle.activeNodes = active;
  cycle.seed = cycle.tipHash || cycle.seed;
  cycle.completed = [];
  cycle.slotOpenedAtMs = Date.now();
  cycle.blockTimeSeconds = effectiveBlockTimeSeconds(cycle.tipTimestamp);
  console.log(
    `[vps-cluster] became leader epoch=${epoch} — fresh shuffle cycle=${cycle.cycleId} ` +
      `active=${active.length} seed=${String(cycle.seed || "").slice(0, 12)}…`
  );
}

/**
 * Drop any cycle addresses not in the on-chain JOIN set immediately (do not wait
 * for bag drain / reopenCycleForNewcomers). Keeps clients from stalling on
 * "Waiting for chain producer set" while ghost heartbeats burn 33s slots.
 */
function reconcileCycleWithChain() {
  const chain = chainQueueNodes();
  if (!chain) return;
  const chainSet = new Set(chain.map((n) => n.address));
  const before = cycle.activeNodes.length;
  const filtered = cycle.activeNodes.filter((n) => chainSet.has(n.address));
  if (filtered.length === before) return;
  const removed = before - filtered.length;
  cycle.activeNodes = filtered;
  cycle.completed = cycle.completed.filter((a) => chainSet.has(a));
  console.log(
    `[sarmad-coordinator] stripped ${removed} non-JOIN address(es) from cycle ` +
      `(active=${filtered.length} source=${queueSource()})`
  );
  if (filtered.length === 0 || remainingOrder().length === 0) {
    ensureCycle();
  }
}

/** address -> ms it was first published as the on-deck producer. Reset per cycle. */
const onDeckSinceMs = new Map();

/** The producer after the current one; published so it can sync before its slot opens. */
function onDeckProducer() {
  return remainingOrder()[1] || "";
}

function noteOnDeckProducer() {
  const remaining = remainingOrder();
  // Anything no longer awaiting a turn (produced, skipped, or reshuffled away) is stale.
  if (onDeckSinceMs.size) {
    const pending = new Set(remaining);
    for (const addr of [...onDeckSinceMs.keys()]) {
      if (!pending.has(addr)) onDeckSinceMs.delete(addr);
    }
  }
  const next = remaining[1] || "";
  if (next && !onDeckSinceMs.has(next)) onDeckSinceMs.set(next, Date.now());
}

/** True when `address` has had a full slot's notice that its turn was coming. */
function hadOnDeckWarning(address) {
  const since = onDeckSinceMs.get(address) || 0;
  return since > 0 && Date.now() - since >= ON_DECK_MIN_WARN_MS;
}

function advanceSkipsIfNeeded() {
  // One cycle owner: the follower must not skip on its own (that desynced EXE
  // and opened extra slots while A/B handed off).
  if (vpsCluster.enabled() && !vpsCluster.isLeader()) return;
  ensureCycle();
  reconcileCycleWithChain();
  noteOnDeckProducer();
  const bt = cycle.blockTimeSeconds || BLOCK_TIME_SECONDS;
  const liveSet = new Set(liveNodes().map((n) => n.address));
  const remaining = remainingOrder();
  if (remaining.length === 0) {
    ensureCycle();
    return;
  }
  const current = remaining[0];
  const elapsedMs = Date.now() - cycle.slotOpenedAtMs;
  const slotMs = bt * 1000;
  if (!liveSet.has(current)) {
    // Registered but producingEnabled=false (syncing / listener / tip lag):
    // full block slot to catch up — never early-burn.
    // Still do not park: only truly offline/absent nodes get parked below.
    const registered = nodes.get(current);
    if (registered && registered.producingEnabled === false) {
      if (elapsedMs < slotMs) return;
      console.log(
        `[sarmad-coordinator] burn turn for syncing/listener ${current} ` +
          `after full ${bt}s (producingEnabled=false) — not parking`
      );
      onDeckSinceMs.delete(current);
      cycle.completed.push(current);
      openNextSlot();
      if (pendingNewcomerAdmit) {
        pendingNewcomerAdmit = false;
        if (!reopenCycleForNewcomers()) ensureCycle();
      } else if (remainingOrder().length === 0) {
        ensureCycle();
      }
      return;
    }
    if (elapsedMs < slotMs) return;
    console.log(
      `[sarmad-coordinator] skip offline producer ${current} after full ${bt}s slot — dropped from queue`
    );
    snapshotStore.parkProducer(current);
    cycle.activeNodes = cycle.activeNodes.filter((n) => n.address !== current);
    cycle.completed = cycle.completed.filter((a) => a !== current);
    openNextSlot();
    if (pendingNewcomerAdmit) {
      pendingNewcomerAdmit = false;
      if (!reopenCycleForNewcomers()) ensureCycle();
    } else if (remainingOrder().length === 0) {
      ensureCycle();
    }
    return;
  }
  if (elapsedMs < slotMs) return;
  console.log(`[sarmad-coordinator] skip missed-slot producer ${current}`);
  cycle.completed.push(current);
  openNextSlot();
  if (pendingNewcomerAdmit) {
    pendingNewcomerAdmit = false;
    if (!reopenCycleForNewcomers()) ensureCycle();
  } else if (remainingOrder().length === 0) {
    ensureCycle();
  }
}

function secondsToNextSlot() {
  advanceSkipsIfNeeded();
  const bt = cycle.blockTimeSeconds || BLOCK_TIME_SECONDS;
  // Time left until this turn's deadline (may be > bt if the slot is scheduled in the future).
  const remainingMs = Number(cycle.slotOpenedAtMs || 0) + bt * 1000 - Date.now();
  return Math.max(0, Math.ceil(remainingMs / 1000));
}

function syncCycleTipFromRelay() {
  const tip = relay.getTip();
  if (tip && Number(tip.height) > 0) {
    cycle.chainHeight = Math.max(Number(cycle.chainHeight) || 0, Number(tip.height) || 0);
    if (tip.hash) cycle.tipHash = tip.hash;
    if (tip.timestamp) cycle.tipTimestamp = tip.timestamp;
  }
}

function cyclePayload() {
  const wait = secondsToNextSlot();
  syncCycleTipFromRelay();
  const remaining = remainingOrder();
  const order = shuffleOrder(
    cycle.activeNodes.map((n) => n.address),
    cycle.seed
  );
  const cursor = Math.max(0, order.length - remaining.length);
  const tip = relay.getTip();
  return {
    cycleId: cycle.cycleId,
    seed: cycle.seed,
    activeNodes: cycle.activeNodes,
    completed: cycle.completed,
    cursor,
    currentProducer: remaining[0] || "",
    // Published so the next producer can sync and pre-build before its slot opens.
    nextProducer: remaining[1] || "",
    // Historical field: early-burn window removed — every producer gets a full block slot.
    readyWindowSeconds: 0,
    blockTimeSeconds: cycle.blockTimeSeconds,
    chainHeight: Math.max(Number(cycle.chainHeight) || 0, Number(tip && tip.height) || 0),
    activeCount: liveNodes().length,
    secondsToNextSlot: wait,
    slotOpenedAtMs: cycle.slotOpenedAtMs,
    slotDeadlineEpochMs: cycle.slotOpenedAtMs + cycle.blockTimeSeconds * 1000,
    genesisEpochSeconds,
    genesisTimestamp,
    parkedProducers: snapshotStore.parkedProducers(),
    queueSource: queueSource(),
    chainProducerCount: (snapshotStore.chainProducerSet(
      Math.max(Number(cycle.chainHeight) || 0, Number(tip && tip.height) || 0)
    ) || []).length,
  };
}

function statusPayload() {
  advanceSkipsIfNeeded();
  const tip = relay.getTip();
  const activeCount = liveNodes().length;
  const queueLength = attachedNodes().length;
  const published = releaseStore.peekManifest && releaseStore.peekManifest();
  const publishedVersion =
    published && published.version ? String(published.version).trim() : "";
  // Prefer signed release_manifest so clients never see a stale COORD_VERSION
  // download link ahead of the APK/zip actually published on the site.
  const advertiseVersion = publishedVersion || COORD_VERSION;
  const apkUrl =
    (published &&
      published.artifacts &&
      published.artifacts.android &&
      published.artifacts.android.url) ||
    `${RELEASE_BASE}/SRMD${advertiseVersion}.apk`;
  const exeUrl =
    (published &&
      published.artifacts &&
      published.artifacts.windows &&
      published.artifacts.windows.url) ||
    `${RELEASE_BASE}/SRMD-Coordinator-${advertiseVersion}.zip`;
  return {
    chainHeight: Math.max(cycle.chainHeight, tip.height),
    activeCount,
    secondsToNextSlot: secondsToNextSlot(),
    blockTimeSeconds: cycle.blockTimeSeconds,
    cycleId: cycle.cycleId,
    genesisEpochSeconds,
    genesisTimestamp,
    tipHash: tip.hash,
    tipTimestamp: tip.timestamp,
    wsPath: "/v1/ws",
    mempoolSize: relay.getMempool().length,
    role: DIRECTORY_URL ? "exe" : vpsCluster.isLeader() ? "directory" : "directory-follower",
    coordinatorCount: liveCoordinators().length,
    version: advertiseVersion,
    apkDownloadUrl: apkUrl,
    exeDownloadUrl: exeUrl,
    releasePublicKey: releaseStore.getPublicKeyHex(),
    releaseManifestUrl: `${RELEASE_BASE}/release_manifest.json`,
    directoryManifestUrl: `${RELEASE_BASE}/vps_directory.json`,
    releaseStrict: process.env.SARMAD_RELEASE_STRICT !== "0",
    ...vpsCluster.statusFields(queueLength),
    ...snapshotStore.statusFields(),
    ...broadcastStore.statusFields(),
    producerAdmissionHeight: PRODUCER_ADMISSION_HEIGHT,
    producerAdmissionBondSrmd: 3,
    maxProducersPerIp: MAX_PRODUCERS_PER_IP,
    producerStorageReserveGb: 1,
    queueSource: queueSource(),
    chainProducerCount: (snapshotStore.chainProducerSet(
      Math.max(Number(cycle.chainHeight) || 0, Number(tip && tip.height) || 0)
    ) || []).length,
    producerRamReserveMb: 416,
  };
}

const SRMD_B58 = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz";

function isValidSrmdAddress(address) {
  const a = String(address || "").trim();
  if (!a.startsWith("SRMD")) return false;
  if (a.length < 10 || a.length > 64) return false;
  for (let i = 4; i < a.length; i++) {
    if (!SRMD_B58.includes(a[i])) return false;
  }
  return true;
}

/** Bind is address-only: Telegram account ↔ SRMD wallet. Signature optional/ignored. */
function verifySarfeesBind(body) {
  const telegramId = Number(body && body.telegramId);
  let address = String((body && body.address) || "").trim();
  if (address.includes("|")) address = address.split("|")[0].trim();
  if (!telegramId || !isValidSrmdAddress(address)) {
    return { ok: false, error: "bad address" };
  }
  return { ok: true, telegramId, address };
}

function readBody(req) {
  return new Promise((resolve, reject) => {
    const chunks = [];
    let size = 0;
    let settled = false;
    // Without a cap and a deadline a single POST can hold the coordinator's
    // memory and a socket open indefinitely.
    const timer = setTimeout(() => {
      if (settled) return;
      settled = true;
      req.destroy();
      reject(new Error("body timeout"));
    }, MAX_BODY_WAIT_MS);
    const fail = (err) => {
      if (settled) return;
      settled = true;
      clearTimeout(timer);
      reject(err);
    };
    req.on("data", (c) => {
      if (settled) return;
      size += c.length;
      if (size > MAX_BODY_BYTES) {
        req.destroy();
        fail(new Error("body too large"));
        return;
      }
      chunks.push(c);
    });
    req.on("end", () => {
      if (settled) return;
      settled = true;
      clearTimeout(timer);
      try {
        const raw = Buffer.concat(chunks).toString("utf8") || "{}";
        resolve(JSON.parse(raw));
      } catch (e) {
        reject(e);
      }
    });
    req.on("error", fail);
  });
}

function sendJson(res, code, obj) {
  const body = JSON.stringify(obj);
  res.writeHead(code, {
    "Content-Type": "application/json; charset=utf-8",
    "Access-Control-Allow-Origin": "*",
    "Cache-Control": "no-store",
  });
  res.end(body);
}

function sendText(res, code, text, type) {
  res.writeHead(code, {
    "Content-Type": type,
    "Access-Control-Allow-Origin": "*",
    "Cache-Control": "no-store",
  });
  res.end(text);
}

/** @type {Map<string, number[]>} */
const rateLimitBuckets = new Map();
let lastRateLimitSweepMs = 0;

/** Buckets are keyed by IP x endpoint, so without a sweep the map never shrinks. */
function sweepRateLimitBuckets(now) {
  if (now - lastRateLimitSweepMs < RATE_LIMIT_SWEEP_MS) return;
  lastRateLimitSweepMs = now;
  for (const [key, bucket] of rateLimitBuckets) {
    if (!bucket.length || bucket[bucket.length - 1] <= now - RATE_LIMIT_BUCKET_TTL_MS) {
      rateLimitBuckets.delete(key);
    }
  }
}

function clientIp(req) {
  const xff = req.headers["x-forwarded-for"];
  if (xff) {
    const first = String(xff).split(",")[0].trim();
    if (first) return first;
  }
  return (req.socket && req.socket.remoteAddress) || "unknown";
}

function rateLimit(key, max, windowMs) {
  const now = Date.now();
  sweepRateLimitBuckets(now);
  let bucket = rateLimitBuckets.get(key);
  if (!bucket) {
    bucket = [];
    rateLimitBuckets.set(key, bucket);
  }
  while (bucket.length && bucket[0] <= now - windowMs) {
    bucket.shift();
  }
  if (bucket.length >= max) return false;
  bucket.push(now);
  return true;
}

/**
 * Direct request from this machine, not forwarded. A loopback check alone is not
 * enough: nginx proxies /v1/* from the public internet and its peer address is
 * also 127.0.0.1, so a forwarding header disqualifies the request.
 */
function isDirectLoopbackRequest(req) {
  if (req.headers["x-forwarded-for"] || req.headers["x-real-ip"]) return false;
  let a = String((req.socket && req.socket.remoteAddress) || "")
    .trim()
    .toLowerCase()
    .replace(/^\[/, "")
    .replace(/\]$/, "");
  if (a.startsWith("::ffff:")) a = a.slice(7);
  return a === "::1" || a === "127.0.0.1" || a.startsWith("127.");
}

function isSecureRequest(req) {
  const proto = String(req.headers["x-forwarded-proto"] || "")
    .split(",")[0]
    .trim()
    .toLowerCase();
  if (proto === "https") return true;
  const host = String(req.headers.host || "").toLowerCase();
  if (host.endsWith(":443")) return true;
  return false;
}

function wsUrlHintForRequest(req) {
  return relay.wsUrlHint(req.headers.host, PORT, { secure: isSecureRequest(req) });
}

function enforceRateLimit(req, res, endpoint, max, windowMs = 60_000) {
  const key = `${endpoint}:${clientIp(req)}`;
  if (!rateLimit(key, max, windowMs)) {
    sendJson(res, 429, { error: "rate limited" });
    return false;
  }
  return true;
}

const publicDir = path.join(__dirname, "public");

let statusPageCache = null;
function statusPageHtml() {
  if (statusPageCache != null) return statusPageCache;
  statusPageCache = fs.readFileSync(path.join(publicDir, "index.html"), "utf8");
  return statusPageCache;
}
const webNodeDirCandidates = [
  path.join(__dirname, "web_node"),
  path.join(__dirname, "..", "web_node"),
];
const webNodeDir =
  webNodeDirCandidates.find((d) => fs.existsSync(path.join(d, "node.html"))) ||
  webNodeDirCandidates[0];

const MIME = {
  ".html": "text/html; charset=utf-8",
  ".js": "text/javascript; charset=utf-8",
  ".css": "text/css; charset=utf-8",
  ".json": "application/json; charset=utf-8",
  ".png": "image/png",
  ".svg": "image/svg+xml",
  ".apk": "application/vnd.android.package-archive",
  ".zip": "application/zip",
  ".exe": "application/octet-stream",
};

function safeJoin(root, reqPath) {
  const cleaned = path.normalize(reqPath).replace(/^([/\\])+/, "");
  const full = path.join(root, cleaned);
  if (!full.startsWith(path.resolve(root))) return null;
  return full;
}

function sendFile(res, filePath) {
  if (!filePath || !fs.existsSync(filePath) || fs.statSync(filePath).isDirectory()) {
    return sendJson(res, 404, { error: "not found" });
  }
  const ext = path.extname(filePath).toLowerCase();
  const type = MIME[ext] || "application/octet-stream";
  res.writeHead(200, {
    "Content-Type": type,
    "Access-Control-Allow-Origin": "*",
    "Cache-Control": ext === ".html" ? "no-store" : "public, max-age=60",
  });
  fs.createReadStream(filePath).pipe(res);
}

const server = http.createServer(async (req, res) => {
  try {
    const url = new URL(req.url || "/", `http://${req.headers.host || "localhost"}`);
    const p = url.pathname;

    if (req.method === "OPTIONS") {
      res.writeHead(204, {
        "Access-Control-Allow-Origin": "*",
        "Access-Control-Allow-Methods": "GET,POST,OPTIONS",
        "Access-Control-Allow-Headers": "Content-Type, X-Sarmad-Reset-Token",
      });
      return res.end();
    }

    if (req.method === "GET" && (p === "/" || p === "/index.html")) {
      // Cached: a synchronous read per request blocks the event loop under traffic.
      return sendText(res, 200, statusPageHtml(), "text/html; charset=utf-8");
    }

    // Browser full node (same-origin with API — avoids mixed content when opened via VPS URL)
    if (req.method === "GET" && (p === "/node.html" || p === "/node")) {
      // Browser product retired — redirect to srmd desktop UI when available.
      const srmd = path.join(publicDir, "srmd.html");
      if (fs.existsSync(srmd)) {
        res.writeHead(302, { Location: "/srmd.html" });
        return res.end();
      }
      return sendFile(res, path.join(webNodeDir, "node.html"));
    }
    if (req.method === "GET" && p.startsWith("/web/")) {
      return sendFile(res, safeJoin(webNodeDir, p.slice("/web/".length)));
    }
    if (req.method === "GET" && (p.startsWith("/js/") || p.startsWith("/css/"))) {
      return sendFile(res, safeJoin(webNodeDir, p.slice(1)));
    }
    if (req.method === "GET" && p.startsWith("/downloads/")) {
      return sendFile(res, safeJoin(path.join(publicDir, "downloads"), p.slice("/downloads/".length)));
    }

    if (req.method === "GET" && p === "/v1/status") {
      return sendJson(res, 200, statusPayload());
    }

    if (req.method === "GET" && p === "/v1/cluster") {
      return sendJson(res, 200, vpsCluster.statusFields(liveNodes().length));
    }

    if (req.method === "GET" && p === "/v1/cycle") {
      if (url.searchParams.get("local") === "1") {
        return sendJson(res, 200, cyclePayload());
      }
      return sendJson(res, 200, await authoritativeCyclePayload());
    }

    if (req.method === "GET" && p === "/v1/sarfees/hunt-proof") {
      const address = String(url.searchParams.get("address") || "").trim();
      let hunt = address
        ? await sarfeesBridge.lookupHunting(address)
        : { bound: false, hunting: false };
      if (hunt && hunt.dbMissing && DIRECTORY_URL) {
        try {
          const r = await httpGetJson(
            DIRECTORY_URL,
            `/v1/sarfees/hunt-proof?address=${encodeURIComponent(address)}`
          );
          if (r && r.json) return sendJson(res, 200, r.json);
        } catch (_) {}
      }
      const tip = relay.getTip();
      const window = sarfeesBridge.huntWindowId(Number(tip && tip.timestamp) || 0);
      const hmac = hunt && hunt.hunting ? sarfeesBridge.huntHmacHex(address, window) : "";
      return sendJson(res, 200, {
        bound: !!(hunt && hunt.bound),
        hunting: !!(hunt && hunt.hunting),
        window,
        hmac,
        driverState: (hunt && hunt.driverState) || "",
      });
    }

    if (req.method === "POST" && p === "/v1/sarfees/bind") {
      if (!sarfeesBridge.checkInternalToken(req)) {
        return sendJson(res, 403, { error: "forbidden" });
      }
      let body;
      try {
        body = await readBody(req);
      } catch (_) {
        return sendJson(res, 400, { error: "bad json" });
      }
      const verified = verifySarfeesBind(body);
      if (!verified.ok) return sendJson(res, 400, { error: verified.error || "bad proof" });
      const saved = await sarfeesBridge.bindAddress(verified.telegramId, verified.address);
      return sendJson(res, saved.ok ? 200 : 500, saved);
    }

    if (req.method === "GET" && p === "/v1/bootstrap") {
      pruneDead();
      const peers = [...nodes.values()]
        .filter((n) => n.peerHost && n.peerPort)
        .slice(0, 33)
        .map((n) => ({
          host: n.peerHost,
          port: n.peerPort,
          peerId: n.address,
        }));
      return sendJson(res, 200, {
        peers,
        wsUrl: wsUrlHintForRequest(req),
        tip: relay.getTip(),
      });
    }

    if (req.method === "GET" && p === "/v1/blocks") {
      if (!enforceRateLimit(req, res, "blocks", 60)) return;
      const from = Number(url.searchParams.get("from") || 0);
      const limit = Number(url.searchParams.get("limit") || 100);
      return sendJson(res, 200, {
        tip: relay.getTip(),
        blocks: relay.getBlocksFrom(from, limit),
      });
    }

    if (req.method === "GET" && p === "/v1/sync-plan") {
      if (!enforceRateLimit(req, res, "sync-plan", 30)) return;
      const localHeight = Number(url.searchParams.get("localHeight") || 0);
      const mode = String(url.searchParams.get("mode") || "full");
      return sendJson(res, 200, snapshotStore.buildSyncPlan(localHeight, mode));
    }

    if (req.method === "GET" && p === "/v1/release/manifest") {
      if (!enforceRateLimit(req, res, "release", 30)) return;
      const manifest = await releaseStore.getManifest();
      if (!manifest) {
        return sendJson(res, 404, {
          error: "release manifest not published",
          releasePublicKey: releaseStore.getPublicKeyHex(),
        });
      }
      return sendJson(res, 200, manifest);
    }

    if (req.method === "GET" && p === "/v1/directory") {
      if (!enforceRateLimit(req, res, "directory", 60)) return;
      const doc = await directoryStore.getDirectory();
      if (!doc) {
        return sendJson(res, 404, {
          error: "vps directory not published",
          releasePublicKey: releaseStore.getPublicKeyHex(),
        });
      }
      return sendJson(res, 200, doc);
    }

    if (req.method === "GET" && p === "/v1/broadcasts") {
      if (!enforceRateLimit(req, res, "broadcasts", 60)) return;
      const address = String(url.searchParams.get("address") || "");
      const isFounder = address === broadcastStore.FOUNDER_ADDRESS;
      return sendJson(res, 200, {
        posts: broadcastStore.listForViewer(address, isFounder),
      });
    }

    if (req.method === "GET" && p.startsWith("/v1/broadcasts/")) {
      if (!enforceRateLimit(req, res, "broadcasts", 60)) return;
      const parts = p.split("/").filter(Boolean); // v1, broadcasts, id, maybe vote|reply|close
      if (parts.length === 3) {
        const id = parts[2];
        const address = String(url.searchParams.get("address") || "");
        const isFounder = address === broadcastStore.FOUNDER_ADDRESS;
        const post = broadcastStore.getForViewer(id, address, isFounder);
        if (!post) return sendJson(res, 404, { error: "not found" });
        return sendJson(res, 200, { post });
      }
    }

    if (req.method === "POST" && p === "/v1/broadcasts") {
      if (!enforceRateLimit(req, res, "broadcast-write", 20)) return;
      const body = await readBody(req);
      const result = await broadcastStore.publish(body);
      if (!result.ok) return sendJson(res, result.code || 400, { error: result.error });
      return sendJson(res, 200, result);
    }

    if (req.method === "POST" && p.startsWith("/v1/broadcasts/") && p.endsWith("/close")) {
      if (!enforceRateLimit(req, res, "broadcast-write", 20)) return;
      const id = p.slice("/v1/broadcasts/".length, -"/close".length);
      const body = await readBody(req);
      const result = await broadcastStore.closePost(id, body);
      if (!result.ok) return sendJson(res, result.code || 400, { error: result.error });
      return sendJson(res, 200, result);
    }

    if (req.method === "POST" && p.startsWith("/v1/broadcasts/") && p.endsWith("/vote")) {
      if (!enforceRateLimit(req, res, "broadcast-write", 30)) return;
      const id = p.slice("/v1/broadcasts/".length, -"/vote".length);
      const body = await readBody(req);
      const result = await broadcastStore.vote(id, body);
      if (!result.ok) return sendJson(res, result.code || 400, { error: result.error });
      return sendJson(res, 200, result);
    }

    if (req.method === "POST" && p.startsWith("/v1/broadcasts/") && p.endsWith("/reply")) {
      if (!enforceRateLimit(req, res, "broadcast-write", 30)) return;
      const id = p.slice("/v1/broadcasts/".length, -"/reply".length);
      const body = await readBody(req);
      const result = await broadcastStore.reply(id, body);
      if (!result.ok) return sendJson(res, result.code || 400, { error: result.error });
      return sendJson(res, 200, result);
    }

    if (req.method === "GET" && p === "/v1/snapshot/latest") {
      if (!enforceRateLimit(req, res, "snapshot", 20)) return;
      const pack = snapshotStore.readLatestSnapshotBuffer();
      if (!pack) return sendJson(res, 404, { error: "no snapshot yet" });
      res.writeHead(200, {
        "Content-Type": "application/gzip",
        "Content-Length": pack.buffer.length,
        "Content-Disposition": `attachment; filename="${pack.filename}"`,
        "X-Snapshot-Height": String(pack.meta && pack.meta.height != null ? pack.meta.height : ""),
        "X-Snapshot-Bytes": String(pack.buffer.length),
      });
      return res.end(pack.buffer);
    }

    if (req.method === "GET" && p === "/v1/mempool") {
      if (!enforceRateLimit(req, res, "mempool", 60)) return;
      return sendJson(res, 200, { txs: relay.getMempool() });
    }

    if (req.method === "POST" && p === "/v1/tx") {
      if (!enforceRateLimit(req, res, "tx", 20)) return;
      const body = await readBody(req);
      if (await maybeForwardFollowerWrite(req, res, "/v1/tx", body)) return;
      const tx = body.tx || body;
      const result = relay.submitTx(tx);
      if (!result.ok) return sendJson(res, 400, result);
      if (!result.duplicate) fanoutTx(tx);
      return sendJson(res, 200, result);
    }

    if (req.method === "POST" && p === "/v1/hub/tx") {
      // Ceiling is far above real mesh gossip (a handful of hubs) but bounds a flood.
      if (!enforceRateLimit(req, res, "hub-tx", HUB_RATE_LIMIT_PER_MIN)) return;
      const body = await readBody(req);
      const tx = body.tx || body;
      const result = relay.submitTx(tx);
      if (!result.ok) return sendJson(res, 400, result);
      // Re-fanout so mesh reaches hubs this node knows that the sender may not.
      if (!result.duplicate) fanoutTx(tx);
      return sendJson(res, 200, result);
    }

    if (req.method === "POST" && p === "/v1/heartbeat") {
      if (!enforceRateLimit(req, res, "heartbeat", 20)) return;
      const body = await readBody(req);
      const applied = await applyPhoneHeartbeat(body, req);
      if (!applied.ok) return sendJson(res, 403, applied);
      fanoutHeartbeat(body);
      if (!vpsCluster.isLeader()) {
        vpsCluster.forwardIfFollower("/v1/heartbeat", body).catch(() => {});
      }
      const payload = await authoritativeCyclePayload();
      payload.wsUrl = wsUrlHintForRequest(req);
      payload.tip = relay.getTip();
      if (applied.producingEnabled === false) payload.producingEnabled = false;
      if (applied.admission) payload.admission = applied.admission;
      return sendJson(res, 200, payload);
    }

    if (req.method === "POST" && p === "/v1/hub/heartbeat") {
      const body = await readBody(req);
      const applied = await applyPhoneHeartbeat(body, req);
      if (!applied.ok) return sendJson(res, 403, applied);
      fanoutHeartbeat(body); // throttled; fills mesh without endless loops
      const payload = cyclePayload();
      payload.tip = relay.getTip();
      return sendJson(res, 200, payload);
    }

    if (req.method === "POST" && p === "/v1/cluster/merge-nodes") {
      if (!vpsCluster.enabled()) {
        return sendJson(res, 404, { error: "cluster disabled" });
      }
      if (!isClusterPeerRequest(req)) {
        return sendJson(res, 403, { error: "cluster peers only" });
      }
      const body = await readBody(req);
      const merged = mergeClusterNodes(body.nodes || body.activeNodes || [], body.parkedProducers);
      return sendJson(res, 200, {
        ok: true,
        merged,
        activeCount: liveNodes().length,
        tip: relay.getTip(),
      });
    }

    if (req.method === "POST" && p === "/v1/cluster/handoff-ack") {
      if (!vpsCluster.isLeader()) {
        return sendJson(res, 403, { error: "not cluster leader" });
      }
      const body = await readBody(req);
      const pendingHandoffId = body.pendingHandoffId || body.pendingHandoffID || "";
      const pendingHandoffEpoch = Number(body.pendingHandoffEpoch || 0);
      const fromUrl = body.fromUrl || body.from || "";
      const tipHash = body.tipHash || body.tipHashHex || "";
      const ok = vpsCluster.recordHandoffAck({
        pendingHandoffId,
        pendingHandoffEpoch,
        fromUrl,
        tipHash,
      });
      if (!ok) return sendJson(res, 400, { ok: false, error: "ack rejected" });
      return sendJson(res, 200, { ok: true, ...vpsCluster.statusFields(0) });
    }

    if (req.method === "POST" && p === "/v1/block-report") {
      if (!enforceRateLimit(req, res, "block-report", 30)) return;
      const body = await readBody(req);
      if (await maybeForwardFollowerWrite(req, res, "/v1/block-report", body)) return;
      const stored = ingestReportedBlock(body, {
        skipProducerTurn: isClusterForward(req),
      });
      if (stored && stored.ok === false) {
        return sendJson(res, 400, {
          ...cyclePayload(),
          blockStored: false,
          error: stored.error || "invalid block",
        });
      }
      if (stored && stored.ok && !stored.duplicate && body.block) {
        fanoutBlock(body);
      }
      return sendJson(res, 200, {
        ...cyclePayload(),
        blockStored: stored ? stored.ok : null,
        duplicate: Boolean(stored && stored.duplicate),
      });
    }

    if (req.method === "POST" && p === "/v1/hub/block") {
      // High enough for bulk backfill during catch-up, low enough to stop a flood.
      if (!enforceRateLimit(req, res, "hub-block", HUB_RATE_LIMIT_PER_MIN)) return;
      const body = await readBody(req);
      const stored = ingestReportedBlock(body, { skipProducerTurn: true });
      if (stored && stored.ok === false) {
        return sendJson(res, 400, {
          ...cyclePayload(),
          blockStored: false,
          error: stored.error || "invalid block",
        });
      }
      if (stored && stored.ok && !stored.duplicate && body.block) {
        fanoutBlock(body);
      }
      return sendJson(res, 200, {
        ...cyclePayload(),
        blockStored: stored ? stored.ok : null,
        duplicate: Boolean(stored && stored.duplicate),
      });
    }

    // Publish the network-wide genesis timestamp (dev reset). Same token as /v1/reset.
    if (req.method === "POST" && p === "/v1/genesis") {
      const auth = authorizeReset(req);
      if (!auth.ok) return sendJson(res, auth.code, { error: auth.error });
      let body = {};
      try {
        body = await readBody(req);
      } catch (_) {}
      const requested = body.genesisTimestamp;
      if (requested == null || !Number.isFinite(Number(requested))) {
        return sendJson(res, 400, { error: "genesisTimestamp required" });
      }
      // First publisher wins: overwriting would split the network into two chains.
      // Cluster fanout after reset may overwrite a stale follower genesis.
      if (genesisTimestamp != null && genesisTimestamp !== Math.floor(Number(requested))) {
        if (!isClusterForward(req)) {
          return sendJson(res, 409, {
            error: "genesis already published",
            genesisTimestamp,
          });
        }
      }
      setGenesisTimestamp(requested);
      if (!isClusterForward(req)) {
        await fanoutClusterAuthPost(
          "/v1/genesis",
          { genesisTimestamp: genesisTimestamp },
          getResetToken()
        );
      }
      return sendJson(res, 200, { ok: true, genesisTimestamp, genesisEpochSeconds });
    }

    if (req.method === "POST" && p === "/v1/reset") {
      const auth = authorizeReset(req);
      if (!auth.ok) return sendJson(res, auth.code, { error: auth.error });
      try {
        await readBody(req);
      } catch (_) {}
      resetCoordinatorState();
      if (!isClusterForward(req)) {
        await fanoutClusterAuthPost("/v1/reset", {}, getResetToken());
      }
      console.log("[sarmad-coordinator] network reset: registry cleared, cycle at genesis defaults");
      return sendJson(res, 200, {
        ok: true,
        chainHeight: cycle.chainHeight,
        activeCount: 0,
        cycleId: cycle.cycleId,
        genesisEpochSeconds,
        genesisTimestamp,
      });
    }

    if (req.method === "GET" && p === "/v1/coordinators") {
      if (!enforceRateLimit(req, res, "coordinators", 60)) return;
      // Single-VPS mode: no edge EXE hubs. Clients always use this directory.
      return sendJson(res, 200, {
        coordinators: [],
        directory: !DIRECTORY_URL,
        singleCoordinator: true,
        genesisTimestamp,
        genesisEpochSeconds,
      });
    }

    if (req.method === "POST" && p === "/v1/coordinator-announce") {
      if (!enforceRateLimit(req, res, "coordinator-announce", 30)) return;
      // Accept but do not register — old EXEs may still announce; phones ignore the list.
      try {
        await readBody(req);
      } catch (_) {}
      return sendJson(res, 200, {
        ok: true,
        ttlMs: 0,
        accepted: false,
        reason: "single-directory mode: edge hubs disabled",
      });
    }

    if (req.method === "GET" && p === "/v1/hub/peers") {
      return sendJson(res, 200, {
        self: selfHubBaseUrl() || null,
        peers: hubGossip.listPeers(selfHubBaseUrl(), selfHubAddress()),
        learned: hubGossip.peersSnapshot(),
      });
    }

    if (req.method === "GET" && p === "/v1/exe/status") {
      let reachMode = null;
      let walletCreated = null;
      try {
        const sessionFile = process.env.SARMAD_EXE_SESSION_FILE;
        if (sessionFile && fs.existsSync(sessionFile)) {
          const sess = JSON.parse(fs.readFileSync(sessionFile, "utf8"));
          reachMode = sess.reachMode || null;
          walletCreated = sess.walletCreated === true;
        }
      } catch (_) {}
      return sendJson(res, 200, {
        directoryUrl: DIRECTORY_URL || null,
        directoryLabel: DIRECTORY_URL ? DIRECTORY_LABEL : "self",
        directoryUrls: DIRECTORY_URL ? DIRECTORY_URLS : [],
        publicBaseUrl: exeConfig.publicBaseUrl || null,
        address: exeConfig.address || null,
        inviterAddress: exeConfig.inviterAddress || null,
        linked: Boolean(DIRECTORY_URL && exeConfig.address && exeConfig.publicBaseUrl),
        activeCount: liveNodes().length,
        chainHeight: Math.max(cycle.chainHeight, relay.getTip().height),
        version: COORD_VERSION,
        reachMode,
        walletCreated,
      });
    }

    if (req.method === "GET" && p === "/v1/exe/session") {
      // Local desktop bootstrap only — seed is for this machine's dashboard heartbeats.
      // The comment was the only thing enforcing that: this server binds 0.0.0.0, so a
      // coordinator with SARMAD_EXE_SESSION_FILE set would hand the wallet seed to
      // anyone who asked. Require a direct loopback peer and no proxy hop.
      if (!isDirectLoopbackRequest(req)) {
        return sendJson(res, 403, { error: "session is available on loopback only" });
      }
      try {
        const sessionFile = process.env.SARMAD_EXE_SESSION_FILE;
        if (!sessionFile || !fs.existsSync(sessionFile)) {
          return sendJson(res, 404, { error: "no session" });
        }
        const sess = JSON.parse(fs.readFileSync(sessionFile, "utf8"));
        return sendJson(res, 200, sess);
      } catch (e) {
        return sendJson(res, 500, { error: String(e && e.message ? e.message : e) });
      }
    }

    if (req.method === "GET" && p === "/v1/exe/rewards") {
      const address = String(url.searchParams.get("address") || "").trim();
      if (!address) return sendJson(res, 400, { error: "address required" });
      const rewards = coordRewards.getRewards(address);
      return sendJson(res, 200, {
        ...rewards,
        inviterAddress: exeConfig.address === address ? exeConfig.inviterAddress || null : null,
        inviterMinClaim: 10,
        inviterDailyCap: 333,
        inviterClaimWindowHours: 7 * 24,
      });
    }

    if (req.method === "POST" && p === "/v1/exe/claim-arsc") {
      const fn = global.__sarmadSubmitCoordClaim;
      if (typeof fn !== "function") {
        return sendJson(res, 501, {
          error: "ARSC claim available only in desktop EXE (COORD:CLAIM)",
        });
      }
      try {
        const r = await fn();
        if (!r || !r.ok) {
          return sendJson(res, 400, {
            error: (r && r.json && r.json.error) || "claim rejected",
            status: r && r.status,
          });
        }
        return sendJson(res, 200, { ok: true, ...(r.json || {}) });
      } catch (e) {
        return sendJson(res, 500, { error: String(e && e.message ? e.message : e) });
      }
    }

    if (req.method === "POST" && p === "/v1/exe/reset-local") {
      // Never wipe the directory coordinator from an unauthenticated EXE path.
      // Local UI clear is handled by the desktop UI server (127.0.0.1) only.
      return sendJson(res, 403, {
        error: "use founder Reset Network (token) or clear local UI on the desktop app",
      });
    }

    if (req.method === "POST" && p === "/v1/exe/config") {
      if (!DIRECTORY_URL) {
        return sendJson(res, 400, { error: "not an EXE coordinator (SARMAD_DIRECTORY_URL unset)" });
      }
      const body = await readBody(req);
      const address = String(body.address || "").trim();
      const publicKey = String(body.publicKey || "");
      const signature = String(body.signature || "");
      const publicBaseUrl = String(body.publicBaseUrl || body.baseUrl || "").replace(/\/$/, "");
      const inviterAddress = String(body.inviterAddress || "").trim();
      if (!address.startsWith("SRMD") || !publicKey || !signature || !isValidCoordBaseUrl(publicBaseUrl)) {
        return sendJson(res, 400, { error: "address, publicKey, signature, publicBaseUrl required" });
      }
      exeConfig = {
        address,
        publicKey,
        publicBaseUrl,
        signature,
        inviterAddress: inviterAddress.startsWith("SRMD") ? inviterAddress : "",
      };
      return sendJson(res, 200, { ok: true, ...exeConfig, directoryUrl: DIRECTORY_URL });
    }

    if (req.method === "GET" && p === "/srmd.html") {
      return sendFile(res, safeJoin(publicDir, "srmd.html"));
    }

    if (req.method === "GET" && p === "/coordinator.html") {
      res.writeHead(302, { Location: "/srmd.html" });
      return res.end();
    }

    sendJson(res, 404, { error: "not found" });
  } catch (e) {
    sendJson(res, 500, { error: String(e && e.message ? e.message : e) });
  }
});

const wss = new WebSocketServer({ noServer: true });
server.on("upgrade", (req, socket, head) => {
  try {
    const url = new URL(req.url || "/", `http://${req.headers.host || "localhost"}`);
    if (url.pathname !== "/v1/ws") {
      socket.destroy();
      return;
    }
    wss.handleUpgrade(req, socket, head, (ws) => {
      relay.attachSocket(ws);
    });
  } catch (_) {
    socket.destroy();
  }
});

function announceToDirectory() {
  if (!DIRECTORY_URL && DIRECTORY_URLS.length === 0) return;
  if (!exeConfig.address || !exeConfig.publicBaseUrl || !exeConfig.signature) return;
  const tip = relay.getTip();
  const body = JSON.stringify({
    address: exeConfig.address,
    publicKey: exeConfig.publicKey,
    baseUrl: exeConfig.publicBaseUrl,
    activeCount: liveNodes().length,
    relayHeight: tip.height,
    version: COORD_VERSION,
    signature: exeConfig.signature,
  });

  const candidates = [];
  if (DIRECTORY_URL) candidates.push(DIRECTORY_URL);
  for (const u of DIRECTORY_URLS) {
    if (u && !candidates.includes(u)) candidates.push(u);
  }
  if (candidates.length === 0) return;

  const tryOne = (index) => {
    if (index >= candidates.length) {
      console.warn("[sarmad-coordinator] directory announce failed on all URLs");
      return;
    }
    const base = candidates[index];
    let url;
    try {
      url = new URL("/v1/coordinator-announce", base);
    } catch (_) {
      tryOne(index + 1);
      return;
    }
    const lib = url.protocol === "https:" ? require("https") : require("http");
    const req = lib.request(
      {
        hostname: url.hostname,
        port: url.port || (url.protocol === "https:" ? 443 : 80),
        path: url.pathname,
        method: "POST",
        headers: {
          "Content-Type": "application/json",
          "Content-Length": Buffer.byteLength(body),
        },
        timeout: 10_000,
      },
      (res) => {
        res.resume();
        if (res.statusCode && res.statusCode >= 200 && res.statusCode < 300) {
          if (DIRECTORY_URL !== base) {
            DIRECTORY_URL = base;
            process.env.SARMAD_DIRECTORY_URL = base;
            DIRECTORY_LABEL = index === 0 ? "primary" : "backup-" + index;
            console.log(`[sarmad-coordinator] directory switched → ${base} (${DIRECTORY_LABEL})`);
          }
          hubGossip.rememberPeer(base, { source: "directory" });
          hubGossip.refreshFromDirectory(base).catch(() => {});
          return;
        }
        console.warn(
          `[sarmad-coordinator] directory announce HTTP ${res.statusCode} @ ${base}`
        );
        tryOne(index + 1);
      }
    );
    req.on("error", (e) => {
      console.warn(
        "[sarmad-coordinator] directory announce failed:",
        e && e.message,
        "@",
        base
      );
      tryOne(index + 1);
    });
    req.on("timeout", () => {
      req.destroy();
      tryOne(index + 1);
    });
    req.write(body);
    req.end();
  };

  tryOne(0);
}

/** Background intervals, tracked so a second start() cannot double them up. */
const backgroundTimers = [];
let started = false;

function addInterval(fn, ms) {
  const t = setInterval(fn, ms);
  backgroundTimers.push(t);
  return t;
}

function stopBackgroundTimers() {
  while (backgroundTimers.length) clearInterval(backgroundTimers.pop());
}

function start() {
  if (started) return;
  started = true;
  ensureGenesisEpoch();
  syncCycleTipFromRelay();
  const seedUrls =
    DIRECTORY_URLS && DIRECTORY_URLS.length
      ? DIRECTORY_URLS
      : require("./directory_urls").directoryUrls();
  directoryStore
    .ensureSigned(seedUrls)
    .then((doc) => {
      if (doc && doc.urls) {
        console.log(
          `[sarmad-coordinator] signed VPS directory: ${doc.urls.length} url(s) → GET /v1/directory`
        );
      }
    })
    .catch(() => {});
  try {
    coordRewards.sanitizeAgainstTip(relay.getTip().timestamp);
  } catch (_) {}
  snapshotStore.init(relay).catch((e) => {
    console.error("[sarmad-coordinator] snapshot init:", e && e.message);
  });
  server.listen(PORT, HOST, () => {
    console.log(`[sarmad-coordinator] listening on http://${HOST}:${PORT}`);
    console.log(`[sarmad-coordinator] dataDir: ${DATA_DIR}`);
    console.log(
      `[sarmad-coordinator] genesis epoch: ${genesisEpochSeconds} (${new Date(
        genesisEpochSeconds * 1000
      ).toISOString()})`
    );
    console.log(`[sarmad-coordinator] role: ${DIRECTORY_URL ? "exe → " + DIRECTORY_URL : "directory"}`);
    console.log(`[sarmad-coordinator] srmd.exe UI: http://<host>:${PORT}/srmd.html`);
    console.log(`[sarmad-coordinator] websocket relay: ws://<host>:${PORT}/v1/ws`);
    console.log(`[sarmad-coordinator] status page: http://<host>:${PORT}/`);
    console.log(`[sarmad-coordinator] safe to run beside other bot services (different port)`);
    console.log(
      `[sarmad-coordinator] reset endpoint: ${
        getResetToken() ? "enabled (token configured)" : "disabled (set SARMAD_RESET_TOKEN)"
      }`
    );
    console.log(
      `[sarmad-coordinator] hub peers: ${hubGossip.listPeers(selfHubBaseUrl(), selfHubAddress()).length} known`
    );
    const clusterSnap = vpsCluster.statusFields(0);
    console.log(
      `[sarmad-coordinator] cluster: enabled=${clusterSnap.clusterEnabled} ` +
        `leader=${clusterSnap.isLeader} peers=${clusterSnap.clusterPeers.length} ` +
        `(set SARMAD_VPS_CLUSTER for multi-VPS; default = single-VPS)`
    );
    if (DIRECTORY_URL) {
      addInterval(announceToDirectory, Math.max(15_000, Math.floor(COORDINATOR_TTL_MS / 3)));
      setTimeout(announceToDirectory, 2000);
      // Pull the live chain so local WebNode / phones on this EXE see real height.
      syncChainFromDirectory().catch(() => {});
      addInterval(() => syncChainFromDirectory().catch(() => {}), 20_000);
      hubGossip.refreshFromDirectory(DIRECTORY_URL).catch(() => {});
      addInterval(() => {
        if (DIRECTORY_URL) hubGossip.refreshFromDirectory(DIRECTORY_URL).catch(() => {});
      }, 45_000);
      addInterval(() => {
        for (const p of hubGossip.listPeers(selfHubBaseUrl(), selfHubAddress())) {
          syncChainFromHub(p).catch(() => {});
        }
      }, 30_000);
    } else {
      // Directory hub: keep peer file warm from live announces.
      addInterval(() => {
        for (const c of liveCoordinators()) {
          hubGossip.rememberPeer(c.baseUrl, { address: c.address, source: "live" });
        }
      }, 30_000);
      // Slot advance used to happen only when a client polled /v1/cycle, so a quiet
      // moment could hold a burned slot open past its deadline. Drive it on a timer.
      addInterval(() => {
        try {
          advanceSkipsIfNeeded();
        } catch (_) {}
      }, 2_000);
      // Multi-VPS scaffold (no-ops when clusterEnabled=false).
      addInterval(() => {
        vpsCluster.onLeaderWatchdogTick();
      }, 5_000);
      addInterval(() => {
        vpsCluster.maybeSyncFromLeader().catch(() => {});
      }, 15_000);
      addInterval(() => {
        vpsCluster
          .reconcileWithPeers(async (peerUrl) => {
            const st = await httpGetJson(peerUrl, "/v1/status");
            if (!st || st.status < 200 || st.status >= 300) return null;
            return st.json || null;
          }, attachedNodes().length)
          .catch(() => {});
      }, 5_000);
      // Keep leader/follower miner registries aligned (fixes activeCount split).
      addInterval(() => {
        syncActiveSetWithCluster().catch(() => {});
      }, 12_000);
      setTimeout(() => {
        syncActiveSetWithCluster().catch(() => {});
      }, 3_000);
    }
  });
}

if (require.main === module) {
  start();
}

module.exports = {
  start,
  stopBackgroundTimers,
  authorizeReset,
  createGenesisCycle,
  resetCoordinatorState,
  getResetToken,
  liveCoordinators,
  /** @internal test helpers */
  _test: {
    nodes,
    coordinators,
    getCycle: () => cycle,
    seedNode: (address, extra = {}) => {
      nodes.set(address, {
        address,
        publicKey: "",
        peerHost: null,
        peerPort: null,
        chainHeight: 0,
        tipHash: "",
        tipTimestamp: 0,
        producingEnabled: true,
        lastSeenMs: Date.now(),
        ...extra,
      });
    },
  },
};
