/**
 * Multi-VPS cluster scaffold (single-VPS default).
 *
 * Spec (agreed):
 * - One leader at a time; followers sync tip + forward txs/tasks to leader.
 * - Handoff when local produce-queue finishes OR leader silent (timeout).
 * - Full-queue rotate among ordered peers.
 * - Phones: sticky home VPS (shortest queue at first pick / after offline).
 *
 * Env:
 *   SARMAD_VPS_SELF_URL     = this node's public base URL (e.g. http://78.47.21.219:9334)
 *   SARMAD_VPS_CLUSTER      = comma-separated peer URLs (include self). Empty / self-only = today's behaviour.
 *   SARMAD_VPS_LEADER_TTL_MS = leader silence before failover (default 45000)
 *   SARMAD_VPS_ID           = optional stable id (defaults to self URL)
 *   SARMAD_VPS_LEADER_PRIORITY = comma-separated preferred-leader order. Must be
 *     identical on every peer. Unlisted peers rank after listed ones. Cold-start
 *     leadership is unaffected (still lexicographic index 0); a preferred peer only
 *     takes over through the planned-handoff barrier once it proves itself healthy.
 *   SARMAD_VPS_RECLAIM_AFTER_MS = how long a higher-priority peer must stay
 *     continuously healthy before the leader hands back (default max(3×TTL, 120000)).
 *     This is the anti-flap hysteresis window.
 *   SARMAD_VPS_RECLAIM_MAX_LAG_BLOCKS = max height lag tolerated in a reclaim
 *     candidate (default 2). Stops handing the chain to a peer that is behind.
 *
 * With 0–1 peers configured, isLeader() is always true and handoff is a no-op.
 */

function normalizeUrl(u) {
  return String(u || "")
    .trim()
    .replace(/\/$/, "");
}

function parseClusterList(raw, selfUrl) {
  const parts = String(raw || "")
    .split(",")
    .map(normalizeUrl)
    .filter(Boolean);
  const self = normalizeUrl(selfUrl);
  const seen = new Set();
  const out = [];
  for (const p of parts) {
    if (seen.has(p)) continue;
    seen.add(p);
    out.push(p);
  }
  if (self && !seen.has(self)) {
    out.unshift(self);
  }
  // Stable order: lexicographic so every node agrees without election chatter.
  out.sort((a, b) => a.localeCompare(b));
  return out;
}

function createCluster(opts = {}) {
  const selfUrl = normalizeUrl(
    opts.selfUrl || process.env.SARMAD_VPS_SELF_URL || process.env.SARMAD_PUBLIC_BASE_URL || ""
  );
  const peers = parseClusterList(
    opts.clusterCsv != null ? opts.clusterCsv : process.env.SARMAD_VPS_CLUSTER,
    selfUrl
  );
  const leaderTtlMs = Number(
    opts.leaderTtlMs != null
      ? opts.leaderTtlMs
      : process.env.SARMAD_VPS_LEADER_TTL_MS || 45_000
  );
  /** Max time to wait for follower ACKs before activating a planned handoff. */
  const handoffAckTimeoutMs = Number(
    opts.handoffAckTimeoutMs != null
      ? opts.handoffAckTimeoutMs
      : process.env.SARMAD_VPS_HANDOFF_ACK_TIMEOUT_MS ||
        Math.max(15_000, Math.floor(leaderTtlMs * 1.2))
  );
  const vpsId = String(opts.vpsId || process.env.SARMAD_VPS_ID || selfUrl || "solo");

  /**
   * Preferred-leader order. Shared config, so every peer computes the same ranking.
   * Empty list = no preference, which keeps the original lexicographic behaviour.
   */
  const leaderPriority = String(
    opts.leaderPriority != null ? opts.leaderPriority : process.env.SARMAD_VPS_LEADER_PRIORITY || ""
  )
    .split(",")
    .map(normalizeUrl)
    .filter(Boolean);
  /** Anti-flap: a candidate must be healthy this long before we hand leadership back. */
  const reclaimAfterMs = Number(
    opts.reclaimAfterMs != null
      ? opts.reclaimAfterMs
      : process.env.SARMAD_VPS_RECLAIM_AFTER_MS || Math.max(3 * leaderTtlMs, 120_000)
  );
  const reclaimMaxLagBlocks = Number(
    opts.reclaimMaxLagBlocks != null
      ? opts.reclaimMaxLagBlocks
      : process.env.SARMAD_VPS_RECLAIM_MAX_LAG_BLOCKS || 2
  );
  /** url -> epoch ms when the peer's current healthy streak began. Deleted on any miss. */
  const healthySinceMs = new Map();
  /**
   * Opt-in: a leader that cannot reach any peer steps down instead of producing
   * blocks nobody can see. Intended for leaders on unreliable links (home
   * connection, SSH reverse tunnel), where the surviving peers will have already
   * promoted someone else. Off by default so a VPS that outlives its peers keeps
   * leading.
   */
  const stepDownWhenIsolated =
    opts.stepDownWhenIsolated != null
      ? Boolean(opts.stepDownWhenIsolated)
      : process.env.SARMAD_VPS_STEP_DOWN_WHEN_ISOLATED === "1";
  let lastPeerContactMs = Date.now();

  /** Lower rank wins. Unlisted peers sort after every listed one. */
  function priorityRank(url) {
    const idx = leaderPriority.indexOf(normalizeUrl(url));
    return idx < 0 ? Number.MAX_SAFE_INTEGER : idx;
  }

  function preferredLeaderUrl() {
    for (const p of leaderPriority) {
      if (peers.includes(p)) return p;
    }
    return null;
  }

  /** Index into peers[] of who we believe is leader. */
  let leaderIndex = 0;
  /** Monotonic epoch; bumps on every handoff. */
  let leaderEpoch = 1;
  /** Last time we heard a healthy status from the current leader. */
  let leaderLastSeenMs = Date.now();
  /**
   * Pending handoff (anti-fork barrier).
   * The leader stays leader until followers ACK that they reached leaderTipHash.
   */
  let pendingHandoff = null;
  /** Set of peer URLs that have ACKed current pendingHandoff.id. */
  let pendingAcks = new Set();
  /** Optional hooks for server.js */
  let hooks = {
    /** Called when this node becomes follower — should start sync-from-leader. */
    onBecameFollower: null,
    /** Called when this node becomes leader — should stop follower-only mode. */
    onBecameLeader: null,
    /** Pull tip from URL (async). */
    syncFromLeader: null,
    /** Forward a JSON body to leader path (async). */
    forwardToLeader: null,
    /** Provide local canonical tip hash (string) for ACK barrier. */
    getLocalTipHash: null,
    /** Provide local canonical chain height (number) for reclaim lag checks. */
    getLocalHeight: null,
    /** Provide local producer-queue size for reclaim agreement checks. */
    getLocalProducerCount: null,
    log: console.log,
  };

  function enabled() {
    return peers.length >= 2 && Boolean(selfUrl);
  }

  function selfIndex() {
    if (!selfUrl) return -1;
    return peers.indexOf(selfUrl);
  }

  function currentLeaderUrl() {
    if (!enabled()) return selfUrl || null;
    return peers[leaderIndex] || peers[0] || null;
  }

  function isLeader() {
    if (!enabled()) return true;
    const me = selfIndex();
    return me >= 0 && me === leaderIndex;
  }

  function setHooks(next) {
    hooks = { ...hooks, ...next };
  }

  function bumpLeader(nextIndex, reason) {
    if (!enabled()) return;
    const n = peers.length;
    if (n === 0) return;
    const idx = ((nextIndex % n) + n) % n;
    // Clear anti-fork barrier on activation; a new handoff will create a new pending record.
    pendingHandoff = null;
    pendingAcks = new Set();
    const prev = leaderIndex;
    const wasLeader = isLeader();
    leaderIndex = idx;
    leaderEpoch += 1;
    leaderLastSeenMs = Date.now();
    const nowLeader = isLeader();
    hooks.log(
      `[vps-cluster] handoff epoch=${leaderEpoch} leader=${currentLeaderUrl()} ` +
        `(was=${peers[prev] || "?"}) reason=${reason}`
    );
    if (wasLeader && !nowLeader && typeof hooks.onBecameFollower === "function") {
      try {
        hooks.onBecameFollower(currentLeaderUrl(), leaderEpoch);
      } catch (_) {}
    }
    if (!wasLeader && nowLeader && typeof hooks.onBecameLeader === "function") {
      try {
        hooks.onBecameLeader(leaderEpoch);
      } catch (_) {}
    }
  }

  /** Full-queue rotate: advance to next peer in the ordered list. */
  function onLocalQueueFinished() {
    // Same-host A/B rotation after a skip-drain forked the tip and made EXE
    // sync fail. Keep leadership until watchdog failover.
    return false;
  }

  /**
   * Hot-standby model: do not rotate leadership for load/attachment.
   * Leadership moves only on leader silence (watchdog) or explicit planned handoff.
   */
  function onAttachmentQueueHandoff(_queueByUrl) {
    return false;
  }

  /** Leader silence failover. */
  function onLeaderWatchdogTick() {
    if (!enabled()) return false;
    if (isLeader()) {
      const isolatedFor = Date.now() - lastPeerContactMs;
      if (stepDownWhenIsolated && isolatedFor >= leaderTtlMs) {
        hooks.log(
          `[vps-cluster] isolated leader for ${isolatedFor}ms (ttl=${leaderTtlMs}ms) — stepping down`
        );
        bumpLeader(leaderIndex + 1, `isolated_${isolatedFor}ms`);
        // Do not re-arm immediately; the next reconcile adopts whoever really leads.
        lastPeerContactMs = Date.now();
        return true;
      }
      leaderLastSeenMs = Date.now();
      return false;
    }
    const silentFor = Date.now() - leaderLastSeenMs;
    if (silentFor < leaderTtlMs) return false;
    // Skip dead leader; try next after current.
    bumpLeader(leaderIndex + 1, `leader_timeout_${silentFor}ms`);
    return true;
  }

  /**
   * Hand leadership back to a higher-priority peer once it has proven itself.
   *
   * Only the current leader runs this, and only through startHandoff(), so the
   * anti-fork ACK barrier still applies and followers never fight over the role.
   * Three gates must all pass, which is what stops a flaky preferred peer from
   * bouncing leadership every time it reappears:
   *   1. it outranks us in SARMAD_VPS_LEADER_PRIORITY,
   *   2. it has answered every status probe for reclaimAfterMs,
   *   3. its height is within reclaimMaxLagBlocks of ours,
   *   4. it computes the same producer queue we do.
   *
   * Gate 4 matters because the producer queue is chain state minus each node's
   * local parked list, and the parked list only propagates while clients are
   * attached. A candidate that has not received it yet would seat offline
   * producers and burn slots, so it waits until its view agrees with ours.
   */
  function maybePriorityReclaim(heightByUrl, producerCountByUrl) {
    if (!enabled() || !isLeader()) return false;
    if (!leaderPriority.length || pendingHandoff) return false;
    const myRank = priorityRank(selfUrl);
    if (myRank === 0) return false;

    const localHeight =
      typeof hooks.getLocalHeight === "function" ? Number(hooks.getLocalHeight()) || 0 : 0;
    const localProducers =
      typeof hooks.getLocalProducerCount === "function"
        ? Number(hooks.getLocalProducerCount())
        : null;
    const now = Date.now();

    let best = null;
    for (const peer of peers) {
      if (peer === selfUrl) continue;
      const rank = priorityRank(peer);
      if (rank >= myRank) continue;
      const since = healthySinceMs.get(peer);
      if (!since || now - since < reclaimAfterMs) continue;
      const peerHeight = Number((heightByUrl || {})[peer]) || 0;
      if (localHeight > 0 && peerHeight < localHeight - reclaimMaxLagBlocks) continue;
      const peerProducers = (producerCountByUrl || {})[peer];
      if (
        localProducers != null &&
        Number.isFinite(Number(peerProducers)) &&
        Number(peerProducers) !== localProducers
      ) {
        continue;
      }
      if (!best || rank < best.rank) best = { peer, rank, since, height: peerHeight };
    }
    if (!best) return false;

    const idx = peers.indexOf(best.peer);
    if (idx < 0) return false;
    hooks.log(
      `[vps-cluster] priority reclaim → ${best.peer} (rank=${best.rank} < self=${myRank}) ` +
        `healthy=${now - best.since}ms/${reclaimAfterMs}ms height=${best.height}/${localHeight}`
    );
    return startHandoff(idx, "priority_reclaim");
  }

  /**
   * Start a coordinated handoff with anti-fork ACK barrier.
   * Only used by the current leader for planned rotations.
   */
  function startHandoff(nextIndex, reason) {
    if (!enabled() || !isLeader()) return false;
    const n = peers.length;
    if (n === 0) return false;
    const idx = ((nextIndex % n) + n) % n;
    if (idx === leaderIndex) return false;

    // If we already have a pending barrier, don't stack another one.
    if (pendingHandoff) {
      if (pendingHandoff.targetIndex === idx) return true;
      return false;
    }

    const getTip = typeof hooks.getLocalTipHash === "function" ? hooks.getLocalTipHash : null;
    const tipHash = getTip ? String(getTip() || "") : "";
    const pendingHandoffEpoch = leaderEpoch + 1;
    const pendingId = `handoff_${pendingHandoffEpoch}_${Date.now()}`;

    pendingHandoff = {
      id: pendingId,
      handoffEpoch: pendingHandoffEpoch,
      targetIndex: idx,
      tipHash,
      reason,
      startedAtMs: Date.now(),
    };
    pendingAcks = new Set();

    hooks.log(
      `[vps-cluster] pending handoff id=${pendingHandoff.id} ` +
        `leaderTip=${(pendingHandoff.tipHash || "").slice(0, 10)}… ` +
        `target=${peers[pendingHandoff.targetIndex]} reason=${reason}`
    );

    maybeActivateHandoff();
    return true;
  }

  function recordHandoffAck({
    pendingHandoffId,
    pendingHandoffEpoch,
    fromUrl,
    tipHash,
  }) {
    if (!enabled() || !isLeader()) return false;
    if (!pendingHandoff) return false;
    const id = String(pendingHandoffId || "");
    if (!id || id !== pendingHandoff.id) return false;
    const epoch = Number(pendingHandoffEpoch) || 0;
    if (!epoch || epoch !== pendingHandoff.handoffEpoch) return false;
    const from = normalizeUrl(fromUrl);
    if (!from || !peers.includes(from)) return false;
    if (pendingHandoff.tipHash && tipHash && String(tipHash) !== pendingHandoff.tipHash) return false;
    pendingAcks.add(from);
    hooks.log(
      `[vps-cluster] handoff ACK id=${pendingHandoff.id} from=${from} ` +
        `(${pendingAcks.size}/${peers.filter((p) => p !== selfUrl).length})`
    );
    return maybeActivateHandoff();
  }

  function maybeActivateHandoff() {
    if (!enabled() || !isLeader()) return false;
    if (!pendingHandoff) return false;

    const requiredPeers = peers.filter((p) => p !== selfUrl);
    if (requiredPeers.length === 0) {
      const target = pendingHandoff.targetIndex;
      const reason = pendingHandoff.reason;
      pendingHandoff = null;
      pendingAcks = new Set();
      bumpLeader(target, `${reason}_no_peers`);
      return true;
    }

    const elapsed = Date.now() - pendingHandoff.startedAtMs;
    const allAcked = requiredPeers.every((p) => pendingAcks.has(normalizeUrl(p)));

    if (!allAcked && elapsed < handoffAckTimeoutMs) return false;

    const reason = allAcked
      ? `${pendingHandoff.reason}_acked`
      : `${pendingHandoff.reason}_ack_timeout_${elapsed}ms`;
    hooks.log(
      `[vps-cluster] activating handoff id=${pendingHandoff.id} reason=${reason} ` +
        `acks=${pendingAcks.size}/${requiredPeers.length}`
    );

    const target = pendingHandoff.targetIndex;
    pendingHandoff = null;
    pendingAcks = new Set();
    bumpLeader(target, reason);
    return true;
  }

  function noteLeaderSeen() {
    leaderLastSeenMs = Date.now();
  }

  /**
   * Adopt remote cluster view when gossip/status says a higher epoch (scaffold).
   * Prevents fighting when peers announce handoff.
   */
  function adoptRemoteLeader(remoteUrl, remoteEpoch) {
    if (!enabled()) return false;
    const url = normalizeUrl(remoteUrl);
    const epoch = Number(remoteEpoch) || 0;
    if (!url || epoch <= leaderEpoch) return false;
    const idx = peers.indexOf(url);
    if (idx < 0) return false;
    const wasLeader = isLeader();
    leaderIndex = idx;
    leaderEpoch = epoch;
    leaderLastSeenMs = Date.now();
    const nowLeader = isLeader();
    hooks.log(`[vps-cluster] adopted remote leader=${url} epoch=${epoch}`);
    if (wasLeader && !nowLeader && typeof hooks.onBecameFollower === "function") {
      try {
        hooks.onBecameFollower(url, epoch);
      } catch (_) {}
    }
    if (!wasLeader && nowLeader && typeof hooks.onBecameLeader === "function") {
      try {
        hooks.onBecameLeader(epoch);
      } catch (_) {}
    }
    return true;
  }

  async function maybeSyncFromLeader() {
    if (!enabled() || isLeader()) return;
    const leader = currentLeaderUrl();
    if (!leader || typeof hooks.syncFromLeader !== "function") return;
    try {
      await hooks.syncFromLeader(leader);
      noteLeaderSeen();
    } catch (e) {
      hooks.log(
        `[vps-cluster] sync-from-leader failed: ${e && e.message ? e.message : e}`
      );
    }
  }

  /**
   * Probe peers and adopt a higher leaderEpoch (heals restart split-brain).
   * Safe to call on leader and follower.
   */
  async function reconcileWithPeers(fetchStatus, selfQueueLength = 0) {
    if (!enabled() || typeof fetchStatus !== "function") return false;
    let adopted = false;
    const queueByUrl = {};
    const heightByUrl = {};
    const producerCountByUrl = {};
    if (selfUrl) queueByUrl[selfUrl] = Number(selfQueueLength) || 0;
    for (const peer of peers) {
      if (peer === selfUrl) continue;
      let st = null;
      try {
        st = await fetchStatus(peer);
      } catch (_) {
        st = null;
      }
      if (!st) {
        // Any miss restarts the healthy streak, so an intermittent peer can never
        // accumulate enough uptime to reclaim leadership.
        healthySinceMs.delete(peer);
        continue;
      }
      if (!healthySinceMs.has(peer)) healthySinceMs.set(peer, Date.now());
      lastPeerContactMs = Date.now();
      queueByUrl[peer] = Number(st.queueLength) || Number(st.activeCount) || 0;
      heightByUrl[peer] = Number(st.chainHeight) || 0;
      if (st.chainProducerCount != null) {
        producerCountByUrl[peer] = Number(st.chainProducerCount);
      }
      const url = normalizeUrl(st.leaderUrl || (st.isLeader ? st.selfUrl || peer : ""));
      const epoch = Number(st.leaderEpoch) || 0;
      if (url && epoch > leaderEpoch) {
        if (adoptRemoteLeader(url, epoch)) adopted = true;
      }
      // Only the actual leader process counts as "seen". A follower advertising
      // leaderUrl must not refresh the watchdog — that blocked failover forever.
      if (normalizeUrl(peer) === currentLeaderUrl()) noteLeaderSeen();
    }
    if (isLeader()) {
      onAttachmentQueueHandoff(queueByUrl);
      maybeActivateHandoff();
      maybePriorityReclaim(heightByUrl, producerCountByUrl);
    }
    return adopted;
  }

  async function forwardIfFollower(pathname, body, headers = {}) {
    if (!enabled() || isLeader()) return { forwarded: false };
    const leader = currentLeaderUrl();
    if (!leader || typeof hooks.forwardToLeader !== "function") {
      return { forwarded: false, error: "no forward hook" };
    }
    const result = await hooks.forwardToLeader(leader, pathname, body, headers);
    return { forwarded: true, result };
  }

  function statusFields(queueLength) {
    return {
      clusterEnabled: enabled(),
      clusterPeers: peers.slice(),
      vpsId,
      selfUrl: selfUrl || "",
      isLeader: isLeader(),
      leaderUrl: currentLeaderUrl() || "",
      leaderEpoch,
      leaderTtlMs,
      leaderPriority: leaderPriority.slice(),
      preferredLeaderUrl: preferredLeaderUrl() || "",
      stepDownWhenIsolated,
      /** Attached clients (any heartbeat) — hint for shortest-queue home pick. */
      queueLength: Number(queueLength) || 0,
      pendingHandoffId: isLeader() && pendingHandoff ? pendingHandoff.id : "",
      pendingHandoffEpoch: isLeader() && pendingHandoff ? pendingHandoff.handoffEpoch : 0,
      pendingHandoffTipHash: isLeader() && pendingHandoff ? pendingHandoff.tipHash : "",
      pendingHandoffTargetUrl:
        isLeader() && pendingHandoff ? peers[pendingHandoff.targetIndex] : "",
    };
  }

  return {
    enabled,
    peers: () => peers.slice(),
    selfUrl: () => selfUrl || null,
    vpsId: () => vpsId,
    isLeader,
    currentLeaderUrl,
    leaderEpoch: () => leaderEpoch,
    preferredLeaderUrl,
    maybePriorityReclaim,
    setHooks,
    onLocalQueueFinished,
    onAttachmentQueueHandoff,
    onLeaderWatchdogTick,
    noteLeaderSeen,
    adoptRemoteLeader,
    maybeSyncFromLeader,
    recordHandoffAck,
    reconcileWithPeers,
    forwardIfFollower,
    statusFields,
  };
}

module.exports = {
  createCluster,
  parseClusterList,
  normalizeUrl,
};
