//! Seed-join boot (m11p5 §3.4–§3.6): a node that is NOT declared in the local
//! topology joins an existing cluster by contacting a `--seed`.
//!
//! # The boot sequence (§2.7 / §3.4)
//!
//! For a FRESH data dir + `--seed`, on a dedicated boot thread:
//!
//! 1. **Discover a leader** over `(seeds ∪ MembershipStore cache)` HTTP status
//!    endpoints (the cache wins precedence for discovery once it exists — a
//!    restart-of-a-joiner knows its peers without the seed).
//! 2. **Join** via the standalone `JoinCluster` gRPC verb. Idempotent by name: a
//!    re-join on retry returns the existing id/role, appends nothing. A
//!    non-leader seed answers `accepted=false` with a leader hint → re-target.
//! 3. **Persist before act**: write the returned roster to the `MembershipStore`
//!    cache AND the `ElectionStore {term}` (the joiner stamps the snapshot fetch
//!    and stream pulls with this term).
//! 4. **FetchSnapshot** `from_seqno=1`: `needed=true` → staged install (reuse the
//!    §2.3 staging/verify/swap machinery; a fresh dir makes the swap trivial);
//!    `needed=false` → proceed empty (the small-cluster case where the stream
//!    serves from 1).
//! 5. **Synthesize a topology** from the JOIN-RESPONSE roster (this node's
//!    `--region` = its assigned id), carrying the local file's behavioral knob
//!    blocks (§3.5). `ShardReplica::new` builds the transport/ship/election
//!    peer tables from THIS, not the local topology's `regions:` list.
//!
//! A **restart of a joiner** (the data dir already has a WAL + a membership
//! cache) skips the seed loop: it boots from the cache roster (no leader
//! discovery, no fetch) and the runtime catch-up self-heal converges it. The
//! authoritative roster once the WAL opens is the recovered `ClusterMembership`
//! cell (§3.6 precedence, resolved inside `ShardReplica::new`); the cache
//! and the synthesized topology only feed the pre-open peer-table construction,
//! and the cache is rewritten from the cell after every applied record.
//!
//! # A joiner is a LEARNER
//!
//! The join response role is `Learner`. The synthesized topology marks the
//! joiner's role so `ShardReplica`'s era-0 view (when no cell exists yet)
//! does not mistake it for a voter — but in practice a snapshot install always
//! carries the leader's WAL with the kind-4 Learner record, so the view boots
//! `from_record` with the correct role. `ElectionState` boots with `self` NOT in
//! the voter set (campaign-gated) until a `Voter` record applies (the leader's
//! auto-promotion duty appends one once the joiner catches up).

use std::path::Path;
use std::time::{Duration, Instant};

use tidaldb::replication::{ElectionStore, HardState, MembershipSnapshot, MembershipStore};
use tidaldb::wal::format::{MemberEntry, MemberRole};

use super::reseed::{self, SeedInstallError};
use super::topology::{
    ElectionSpec, GrpcTlsSpec, RegionSpec, ReplicationSpec, TimeoutsSpec, TopologySpec, WalSpec,
};
use crate::error::{Result, ServerError};

/// Per-status-poll HTTP timeout (the discovery loop iterates). Generous because
/// a COLD poll pays a full TLS handshake (rustls/aws-lc-rs) on top of DNS + TCP,
/// and on a real (TLS) cluster under CPU contention that handshake alone can blow
/// a sub-second budget — the joiner then times out every poll and burns the whole
/// 120 s discovery window despite the peer being reachable (TCP connects fine).
/// The in-process tests never saw this: plaintext loopback has no handshake cost.
/// Env-overridable via `TIDAL_SEED_STATUS_TIMEOUT_MS`.
fn status_poll_timeout() -> Duration {
    std::env::var("TIDAL_SEED_STATUS_TIMEOUT_MS")
        .ok()
        .and_then(|v| v.trim().parse::<u64>().ok())
        .filter(|&ms| ms > 0)
        .map_or(Duration::from_secs(5), Duration::from_millis)
}
/// Per-join RPC timeout (the leader's bounded same-term commit wait is itself
/// bounded by `quorum_timeout`, so allow generous headroom).
const JOIN_RPC_TIMEOUT: Duration = Duration::from_secs(10);
/// Backoff bounds for the join loop.
const BACKOFF_MIN: Duration = Duration::from_millis(250);
const BACKOFF_MAX: Duration = Duration::from_secs(5);
/// Default join window before the boot gives up (env-overridable for tests via
/// `TIDAL_SEED_JOIN_MS`). A fresh joiner has nothing to fall back to, so on
/// timeout it returns an error and the process exits — the orchestrator restarts
/// it and the loop runs again (crash-loop is the orchestration's retry, not the
/// in-process mechanism). 120 s gives a slow loopback discover+join+fetch room.
const DEFAULT_JOIN_WINDOW_MS: u64 = 120_000;

/// The synthesized topology a seed-join boot constructs for
/// `ShardReplica::new` (§3.4).
///
/// The join-response roster becomes the `regions:` list, the local file's knob
/// blocks are carried verbatim, and this node is included at its assigned id.
pub struct SeedJoinBoot {
    /// The synthesized topology (roster from the join, knobs from the local file).
    pub topology: TopologySpec,
    /// This node's region name (echoed for `ShardReplica::new`).
    pub region: String,
}

/// Inputs to [`seed_join_boot`].
pub struct SeedJoinInput<'a> {
    /// The `--seed` HTTP base URLs.
    pub seeds: &'a [String],
    /// The local topology file, parsed — the knob source (§3.5). Its `regions:`
    /// list is IGNORED for the roster.
    pub knobs: &'a TopologySpec,
    /// This node's `--region` name.
    pub region: &'a str,
    /// This node's advertised gRPC `host:port`.
    pub advertise_grpc: &'a str,
    /// This node's advertised HTTP `host:port`.
    pub advertise_http: &'a str,
    /// This node's `--metrics` bind (seed-join's only metrics source, §3.5).
    pub metrics: Option<&'a str>,
    /// The data dir (already validated as a mount subdirectory for the swap).
    pub data_dir: &'a Path,
    /// The API key for the seed status/join HTTP/gRPC calls (forwarded verbatim).
    pub api_key: Option<String>,
}

/// Run the seed-join boot (§3.4). See the module docs for the sequence.
///
/// # Errors
///
/// [`ServerError::Cluster`] when the join window elapses without a successful
/// join+install (a fresh joiner has nothing to fall back to), or for any
/// unrecoverable durable-persist / swap fault.
// One linear boot sequence (cache fast-path → discover → join → persist → fetch
// → synthesize); splitting it would scatter the §2.7 ordering it documents.
#[allow(clippy::too_many_lines)]
pub fn seed_join_boot(input: &SeedJoinInput<'_>) -> Result<SeedJoinBoot> {
    let store = MembershipStore::new(input.data_dir);

    // §3.6 precedence for the PRE-OPEN loop: a restart of a joiner whose data dir
    // already carries a WAL boots from the cache WITHOUT the seed. The
    // authoritative roster once the WAL opens is the recovered cell (resolved
    // inside `ShardReplica::new`); this cache only seeds the peer tables.
    let wal_exists = input
        .data_dir
        .join("wal")
        .read_dir()
        .map(|mut it| it.next().is_some())
        .unwrap_or(false);
    if wal_exists {
        match store.load() {
            Ok(Some(snap)) => {
                tracing::info!(
                    region = input.region,
                    version = snap.version,
                    members = snap.members.len(),
                    "seed-join restart: a WAL + membership cache exist — booting from the cache \
                     roster without contacting a seed (§3.6; the WAL-recovered cell wins once \
                     opened, the runtime catch-up self-heal converges)"
                );
                // The restart path has no fresh join to name the current leader;
                // the WAL cell wins at open and heartbeats correct the leader hint.
                let topology = synthesize_topology(input, &snap.members, None)?;
                return Ok(SeedJoinBoot {
                    topology,
                    region: input.region.to_string(),
                });
            }
            Ok(None) => {
                tracing::warn!(
                    region = input.region,
                    "seed-join restart: a WAL exists but the membership cache is missing — \
                     re-running the seed loop to recover the roster before opening"
                );
            }
            Err(e) => {
                // A corrupt cache must be DETECTED, never guessed around — but a
                // re-join recovers the roster, so do not refuse boot: log loudly
                // and fall through to the seed loop.
                tracing::error!(
                    region = input.region,
                    error = %e,
                    "seed-join restart: the membership cache is corrupt — re-running the seed \
                     loop to recover the roster (the corrupt cache is overwritten on success)"
                );
            }
        }
    }

    // The TLS posture the joiner advertises (from the knob file's own region
    // block, when one names this region; else plaintext — the loopback/VPC
    // default every shipped topology uses).
    let tls = own_grpc_tls(input.knobs, input.region);
    // m11p7: when the cluster runs inter-node TLS, the seed-join HTTP dials go
    // `https://` (peer_url) and the blocking clients trust the cluster CA.
    super::forward::set_inter_node_https(tls.is_some());
    let join_window = join_window();
    let deadline = Instant::now() + join_window;
    let mut backoff = BACKOFF_MIN;

    let status_client = build_join_client(status_poll_timeout(), tls.as_ref())?;

    while Instant::now() < deadline {
        // (1) Discover a leader over (seeds ∪ cache). The cache (when present)
        // contributes its members' HTTP addresses so a restart-of-a-joiner that
        // lost its cache-only fast path still finds the leader.
        let candidates = discovery_candidates(input.seeds, &store);
        let Some(leader) = discover_leader(&status_client, &candidates, input.api_key.as_deref())
        else {
            std::thread::sleep(backoff);
            backoff = (backoff * 2).min(BACKOFF_MAX);
            continue;
        };

        // (2) Join via the standalone gRPC verb (idempotent by name).
        let join = match join_via_leader(
            &leader,
            input.region,
            input.advertise_grpc,
            input.advertise_http,
            tls.as_ref(),
        ) {
            Ok(JoinAttempt::Joined(resp)) => resp,
            Ok(JoinAttempt::RetargetOrRetry(reason)) => {
                tracing::info!(reason = %reason, "seed-join: join not accepted; re-discovering");
                std::thread::sleep(backoff);
                backoff = (backoff * 2).min(BACKOFF_MAX);
                continue;
            }
            Err(e) => {
                tracing::warn!(error = %e, "seed-join: join RPC failed; re-discovering");
                std::thread::sleep(backoff);
                backoff = (backoff * 2).min(BACKOFF_MAX);
                continue;
            }
        };

        // (3) Persist-before-act: the cache, then the term, BEFORE any fetch.
        let snapshot = MembershipSnapshot {
            version: join.membership_version,
            term: join.term,
            members: join.members.clone(),
        };
        if let Err(e) = store.persist(&snapshot) {
            // A failed cache persist is non-fatal to the join itself (the cell
            // installed via snapshot is the source of truth); log loudly and
            // continue — the cache is rewritten from the cell after open.
            tracing::error!(error = %e, "seed-join: persist membership cache failed (non-fatal)");
        }
        adopt_term(input.data_dir, join.term)?;

        // (4) FetchSnapshot from_seqno=1: needed → install, else proceed empty.
        match reseed::install_snapshot_from_leader(
            input.data_dir,
            &join.leader_grpc_addr,
            join.leader_region_id,
            join.term,
            1,
            tls.as_ref(),
        ) {
            Ok(Some(seq)) => {
                tracing::info!(
                    region = input.region,
                    assigned_id = join.assigned_id,
                    snapshot_seq = seq,
                    term = join.term,
                    "seed-join: snapshot installed; constructing the node from the join roster"
                );
            }
            Ok(None) => {
                tracing::info!(
                    region = input.region,
                    assigned_id = join.assigned_id,
                    term = join.term,
                    "seed-join: leader reports needed=false (the stream serves from 1) — \
                     proceeding empty; the post-open catch-up pull converges the joiner"
                );
            }
            Err(SeedInstallError::Retry) => {
                tracing::warn!("seed-join: snapshot fetch retryable; re-discovering the leader");
                std::thread::sleep(backoff);
                backoff = (backoff * 2).min(BACKOFF_MAX);
                continue;
            }
            Err(SeedInstallError::Fatal(e)) => return Err(e),
        }

        // (5) Synthesize the topology from the join roster + the local knobs,
        // naming the CURRENT leader (so the `needed=false` boot self-heal pulls
        // from the live stream).
        let topology = synthesize_topology(input, &join.members, Some(join.leader_region_id))?;
        return Ok(SeedJoinBoot {
            topology,
            region: input.region.to_string(),
        });
    }

    Err(ServerError::Cluster(format!(
        "seed-join: could not join via any seed within {join_window:?} (a fresh joiner has \
         nothing to fall back to). The orchestrator restarts this process and the loop runs \
         again — crash-loop is the retry mechanism, never an in-process spin."
    )))
}

/// The join window (env-overridable for tests).
fn join_window() -> Duration {
    let ms = std::env::var("TIDAL_SEED_JOIN_MS")
        .ok()
        .and_then(|v| v.trim().parse::<u64>().ok())
        .unwrap_or(DEFAULT_JOIN_WINDOW_MS);
    Duration::from_millis(ms)
}

/// This region's advertised gRPC TLS material, when the knob file names this
/// Build a seed-join blocking HTTP client (m11p7): trusts the cluster CA when
/// `tls` is `Some` so an `https://` dial to a TLS-serving seed/leader verifies.
///
/// # Errors
///
/// Returns [`ServerError::Cluster`] if the CA cert cannot be read/parsed or the
/// client cannot be built.
fn build_join_client(
    timeout: std::time::Duration,
    tls: Option<&tidal_net::config::TlsConfig>,
) -> Result<reqwest::blocking::Client> {
    // Separate connect timeout from the (generous) total timeout: a genuinely
    // dead seed (SYN blackhole) fails fast at the connect stage instead of
    // blocking the full `timeout` before the discovery loop tries the next seed,
    // while a slow-but-live TLS handshake (the reason `timeout` is generous —
    // see `status_poll_timeout`) still gets the full budget to complete.
    let mut builder = reqwest::blocking::Client::builder()
        .connect_timeout(Duration::from_secs(2))
        .timeout(timeout);
    if let Some(t) = tls {
        let pem = std::fs::read(&t.ca_cert)
            .map_err(|e| ServerError::Cluster(format!("seed-join: read CA cert: {e}")))?;
        let ca = reqwest::Certificate::from_pem(&pem)
            .map_err(|e| ServerError::Cluster(format!("seed-join: parse CA cert: {e}")))?;
        builder = builder.add_root_certificate(ca);
    }
    builder
        .build()
        .map_err(|e| ServerError::Cluster(format!("seed-join: build HTTP client: {e}")))
}

/// region with a `grpc_tls` block (the seed dial uses the same posture).
fn own_grpc_tls(knobs: &TopologySpec, region: &str) -> Option<tidal_net::config::TlsConfig> {
    grpc_tls_for(knobs, region).map(GrpcTlsSpec::to_tls_config)
}

/// The `grpc_tls` posture for `region`: its OWN block when the knob file names
/// it, ELSE ANY region's block.
///
/// A seed-joiner is NEVER in the shared-ConfigMap knob file — it learns its
/// identity from the join, so `regions:` lists only the bootstrap voters. But
/// every pod mounts the SAME cert Secret at the SAME paths, so any region's
/// `grpc_tls` block carries the correct CA/cert/key paths for the joiner too.
/// Returning `None` here (the prior behavior, premised on "the joiner's region
/// isn't named ⇒ plaintext loopback") silently dropped a TLS-cluster joiner to
/// PLAINTEXT: it then failed CA verification on the seed-discover HTTPS dial
/// (`invalid peer certificate: UnknownIssuer`, burning the whole 120 s window)
/// and could not mTLS-replicate with its TLS peers. The fallback is safe: with
/// no `grpc_tls` anywhere it is still `None` (the genuine plaintext case).
fn grpc_tls_for<'a>(knobs: &'a TopologySpec, region: &str) -> Option<&'a GrpcTlsSpec> {
    knobs
        .regions
        .iter()
        .find(|r| r.name == region)
        .and_then(|r| r.grpc_tls.as_ref())
        .or_else(|| knobs.regions.iter().find_map(|r| r.grpc_tls.as_ref()))
}

/// One discovery candidate: an HTTP status base + (when known) the gRPC address
/// the same node advertises.
#[derive(Debug, Clone)]
struct Candidate {
    http_addr: String,
}

/// Build the discovery candidate set: the `--seed` URLs plus the cache's
/// members' HTTP addresses (deduplicated). Seeds first (the operator-named
/// bootstrap), then the cache (a restart-of-a-joiner's known peers).
fn discovery_candidates(seeds: &[String], store: &MembershipStore) -> Vec<Candidate> {
    let mut out: Vec<Candidate> = Vec::new();
    let mut seen = std::collections::HashSet::new();
    for s in seeds {
        if seen.insert(s.clone()) {
            out.push(Candidate {
                http_addr: s.clone(),
            });
        }
    }
    if let Ok(Some(snap)) = store.load() {
        for m in &snap.members {
            if m.role != MemberRole::Removed
                && !m.http_addr.is_empty()
                && seen.insert(m.http_addr.clone())
            {
                out.push(Candidate {
                    http_addr: m.http_addr.clone(),
                });
            }
        }
    }
    out
}

/// What the discovery loop learned (§2.7): the current leader's advertised gRPC
/// address + its term + region id, so a join can dial it directly.
#[derive(Debug, Clone)]
struct LeaderInfo {
    http_addr: String,
}

/// Poll every candidate's `/cluster/status/local`; return the first that either
/// reports ITSELF the leader OR names a leader (whose HTTP address the roster
/// resolves). The join RPC is forwarded by a non-leader anyway, so any reachable
/// member is a usable join target — but preferring a leader avoids an extra hop.
fn discover_leader(
    client: &reqwest::blocking::Client,
    candidates: &[Candidate],
    api_key: Option<&str>,
) -> Option<LeaderInfo> {
    // First pass: a candidate that reports itself the leader (direct target).
    let mut any_reachable: Option<LeaderInfo> = None;
    for cand in candidates {
        let url = super::forward::peer_url(&cand.http_addr, "/cluster/status/local");
        let mut req = client.get(&url);
        if let Some(key) = api_key {
            req = req.bearer_auth(key);
        }
        // Log every failure mode at debug — a silent discovery loop turned a
        // too-tight TLS poll timeout into an opaque "could not join within 120s".
        let resp = match req.send() {
            Ok(resp) => resp,
            Err(e) => {
                // Walk the std::error source chain so the ROOT transport/TLS
                // cause (hidden behind reqwest's terse Display) is visible.
                let mut chain = format!("{e}");
                let mut src = std::error::Error::source(&e);
                while let Some(s) = src {
                    chain.push_str(" -> ");
                    chain.push_str(&s.to_string());
                    src = s.source();
                }
                tracing::warn!(%url, error = %chain, "seed-join discover: status poll failed (transport/TLS/timeout)");
                continue;
            }
        };
        if !resp.status().is_success() {
            tracing::debug!(%url, status = %resp.status(), "seed-join discover: status poll non-2xx");
            continue;
        }
        let Ok(json) = resp.json::<serde_json::Value>() else {
            tracing::debug!(%url, "seed-join discover: status body was not JSON");
            continue;
        };
        // Any reachable member is a fallback join target (it forwards to the
        // leader); a self-reported leader is preferred (one fewer hop).
        any_reachable.get_or_insert_with(|| LeaderInfo {
            http_addr: cand.http_addr.clone(),
        });
        if json.get("is_leader").and_then(serde_json::Value::as_bool) == Some(true) {
            return Some(LeaderInfo {
                http_addr: cand.http_addr.clone(),
            });
        }
    }
    any_reachable
}

/// A successful join: the assigned id/role + the roster + the leader's
/// dial-able addresses + term.
struct Joined {
    assigned_id: u16,
    term: u64,
    membership_version: u64,
    leader_grpc_addr: String,
    leader_region_id: u16,
    members: Vec<MemberEntry>,
}

/// The outcome of one join attempt.
enum JoinAttempt {
    Joined(Joined),
    /// Not accepted (a non-leader seed, a held conf-change): re-discover.
    RetargetOrRetry(String),
}

/// Join via the candidate over the HTTP `/cluster/join` verb — which wraps the
/// same leader-side logic the gRPC `JoinCluster` uses and is FORWARDED by a
/// non-leader, so the joiner never needs a peer's gRPC address for the join
/// itself (it learns the snapshot-source gRPC address from the returned roster).
/// Idempotent by name.
///
/// gRPC TLS is NOT used here (the HTTP gateway terminates its own TLS); the
/// snapshot fetch that follows dials the returned `leader_grpc_addr` with the
/// joiner's gRPC TLS posture.
fn join_via_leader(
    leader: &LeaderInfo,
    region: &str,
    advertise_grpc: &str,
    advertise_http: &str,
    tls: Option<&tidal_net::config::TlsConfig>,
) -> Result<JoinAttempt> {
    let client = build_join_client(JOIN_RPC_TIMEOUT, tls)?;

    let url = super::forward::peer_url(&leader.http_addr, "/cluster/join");
    let body = serde_json::json!({
        "name": region,
        "grpc_addr": advertise_grpc,
        "http_addr": advertise_http,
    });
    let mut req = client.post(&url).json(&body);
    if let Ok(key) = std::env::var("TIDAL_API_KEY") {
        req = req.bearer_auth(key);
    }
    let resp = req
        .send()
        .map_err(|e| ServerError::Cluster(format!("seed-join: POST /cluster/join: {e}")))?;
    let status = resp.status();
    if !status.is_success() {
        // A 503 (not leader / conf-change held) is RETRYABLE; surface the body.
        let body = resp.text().unwrap_or_default();
        return Ok(JoinAttempt::RetargetOrRetry(format!(
            "/cluster/join returned {status}: {body}"
        )));
    }
    let json: serde_json::Value = resp
        .json()
        .map_err(|e| ServerError::Cluster(format!("seed-join: /cluster/join body: {e}")))?;
    let assigned_id = u16::try_from(json["assigned_id"].as_u64().unwrap_or(0)).unwrap_or(0);
    let term = json["term"].as_u64().unwrap_or(0);
    let membership_version = json["membership_version"].as_u64().unwrap_or(0);

    // The HTTP /cluster/join body is intentionally lean (it does not echo the
    // full roster). Fetch the full roster from /cluster/members on the same
    // gateway (it serves the effective roster from any node) so the joiner can
    // build its peer tables.
    let members = fetch_roster(&client, &leader.http_addr)?;
    // The snapshot SOURCE must be the CURRENT leader — not just any voter — so
    // the post-install catch-up resumes from the live stream (a re-election may
    // have moved leadership since the topology declared region 0 the leader). The
    // gateway's /cluster/status/local names the current leader; resolve its gRPC
    // address + region from the roster. Fall back to the lowest-id voter only if
    // the leader is momentarily unknown (mid-election) — the install retries.
    let (leader_grpc_addr, leader_region_id) =
        current_leader_dial_target(&client, &leader.http_addr, &members)
            .unwrap_or_else(|| leader_dial_target(&members));

    Ok(JoinAttempt::Joined(Joined {
        assigned_id,
        term,
        membership_version,
        leader_grpc_addr,
        leader_region_id,
        members,
    }))
}

/// Resolve the CURRENT leader's gRPC dial target from the gateway's
/// `/cluster/status/local` (the `leader` field names it) + the roster (resolves
/// its advertised gRPC address + id). `None` when the leader is unknown
/// (mid-election) or not in the roster — the caller falls back to a voter and the
/// install retries.
fn current_leader_dial_target(
    client: &reqwest::blocking::Client,
    gateway_http: &str,
    members: &[MemberEntry],
) -> Option<(String, u16)> {
    let url = super::forward::peer_url(gateway_http, "/cluster/status/local");
    let mut req = client.get(&url);
    if let Ok(key) = std::env::var("TIDAL_API_KEY") {
        req = req.bearer_auth(key);
    }
    let json: serde_json::Value = req.send().ok()?.json().ok()?;
    let leader_name = json.get("leader").and_then(serde_json::Value::as_str)?;
    members
        .iter()
        .find(|m| m.name == leader_name && m.role != MemberRole::Removed)
        .map(|m| (m.grpc_addr.clone(), m.id))
}

/// Fetch the effective roster from `/cluster/members` (served from any node).
fn fetch_roster(client: &reqwest::blocking::Client, http_addr: &str) -> Result<Vec<MemberEntry>> {
    let url = super::forward::peer_url(http_addr, "/cluster/members");
    let mut req = client.get(&url);
    if let Ok(key) = std::env::var("TIDAL_API_KEY") {
        req = req.bearer_auth(key);
    }
    let resp = req
        .send()
        .map_err(|e| ServerError::Cluster(format!("seed-join: GET /cluster/members: {e}")))?;
    if !resp.status().is_success() {
        return Err(ServerError::Cluster(format!(
            "seed-join: /cluster/members returned {}",
            resp.status()
        )));
    }
    let json: serde_json::Value = resp
        .json()
        .map_err(|e| ServerError::Cluster(format!("seed-join: /cluster/members body: {e}")))?;
    let arr = json["members"].as_array().ok_or_else(|| {
        ServerError::Cluster("seed-join: /cluster/members has no members array".into())
    })?;
    let mut members = Vec::with_capacity(arr.len());
    for m in arr {
        let id = u16::try_from(m["id"].as_u64().unwrap_or(0)).unwrap_or(0);
        let name = m["name"].as_str().unwrap_or_default().to_string();
        let grpc_addr = m["grpc_addr"].as_str().unwrap_or_default().to_string();
        let http_addr_m = m["http_addr"].as_str().unwrap_or_default().to_string();
        let role = match m["role"].as_str() {
            Some("learner") => MemberRole::Learner,
            Some("removed") => MemberRole::Removed,
            _ => MemberRole::Voter,
        };
        members.push(MemberEntry {
            id,
            name,
            grpc_addr,
            http_addr: http_addr_m,
            role,
        });
    }
    if members.is_empty() {
        return Err(ServerError::Cluster(
            "seed-join: /cluster/members returned an empty roster".into(),
        ));
    }
    Ok(members)
}

/// The snapshot-fetch target: a current VOTER's advertised gRPC address + its
/// region id (the snapshot source is the leader; any voter serves it, and the
/// `FetchSnapshot` handler resolves its own shard regardless of the requested
/// id). Prefers the lowest-id voter for determinism.
fn leader_dial_target(members: &[MemberEntry]) -> (String, u16) {
    members
        .iter()
        .filter(|m| m.role == MemberRole::Voter)
        .min_by_key(|m| m.id)
        .map(|m| (m.grpc_addr.clone(), m.id))
        .unwrap_or_default()
}

/// Adopt the discovered leadership term durably BEFORE pulling (§2.7
/// persist-before-act). A fresh joiner has no `election_state`; a restart-of-a-
/// joiner adopts only a HIGHER term (never regresses).
fn adopt_term(data_dir: &Path, term: u64) -> Result<()> {
    let store = ElectionStore::new(data_dir);
    let wal_exists = data_dir
        .join("wal")
        .read_dir()
        .map(|mut it| it.next().is_some())
        .unwrap_or(false);
    let durable = match store.load(wal_exists) {
        Ok(tidaldb::replication::BootState::Rejoin(h)) => h.current_term,
        Ok(_) => 0,
        Err(e) => {
            return Err(ServerError::Cluster(format!(
                "seed-join: refusing to adopt a term — corrupt election_state ({e}); a join \
                 does not restore term knowledge (the m11p4 corrupt-state rule stands, §2.5)"
            )));
        }
    };
    if term > durable {
        store
            .persist(HardState {
                current_term: term,
                voted_for: None,
            })
            .map_err(|e| {
                ServerError::Cluster(format!(
                    "seed-join: persist discovered term {term} (persist-before-act): {e}"
                ))
            })?;
    }
    Ok(())
}

/// Synthesize the topology `ShardReplica::new` consumes (§3.4): the
/// join-response roster as `regions:` (in id order so the positional `RegionId`
/// MATCHES the assigned member id — the seam every durable id depends on), this
/// node's `--region` carried, the local file's knob blocks copied verbatim
/// (§3.5), this node's `metrics_addr` from `--metrics`.
///
/// Removed tombstones are EXCLUDED (they are not live peers) — but excluding a
/// tombstone whose id is below a live member's would shift positional ids. The
/// 1:1 region↔shard mapping (`shard_of_region`) requires the positional index to
/// equal the member id, so we keep the roster id-DENSE by emitting a placeholder
/// region for each burned id below the max live id. A placeholder declares the
/// tombstone's name + addresses but is never dialed (it is excluded from the
/// live peer set by the membership view's `from_record` once the WAL opens, and
/// the era-0 synthesized view treats it as present-but-addressless).
fn synthesize_topology(
    input: &SeedJoinInput<'_>,
    members: &[MemberEntry],
    current_leader_region: Option<u16>,
) -> Result<TopologySpec> {
    // The roster's id range. RegionId = positional index, and
    // `shard_of_region(RegionId(n)) = ShardId(n)`, so the regions list must be
    // dense from 0..=max_id with each member at its own id index.
    let max_id = members.iter().map(|m| m.id).max().unwrap_or(0);
    let mut by_id: std::collections::HashMap<u16, &MemberEntry> = std::collections::HashMap::new();
    for m in members {
        by_id.insert(m.id, m);
    }

    let mut regions: Vec<RegionSpec> = Vec::with_capacity(usize::from(max_id) + 1);
    for id in 0..=max_id {
        if let Some(m) = by_id.get(&id) {
            // This node's own entry carries its real advertised addresses (the
            // roster already holds them from the join, but be explicit for self).
            let is_self = m.name == input.region;
            let grpc = if is_self {
                input.advertise_grpc.to_string()
            } else {
                m.grpc_addr.clone()
            };
            let http = if is_self {
                input.advertise_http.to_string()
            } else {
                m.http_addr.clone()
            };
            regions.push(RegionSpec {
                name: m.name.clone(),
                grpc_addr: Some(grpc),
                grpc_bind: None,
                http_addr: Some(http),
                grpc_tls: self_tls_spec(input.knobs, &m.name),
                metrics_addr: if is_self {
                    input.metrics.map(str::to_string)
                } else {
                    None
                },
                zone: None,
            });
        } else {
            // A burned id with no member entry at all (a gap that should not
            // happen — ids are assigned densely as max+1). Emit a placeholder so
            // the positional index stays aligned; it is never a live peer.
            regions.push(RegionSpec {
                name: format!("__burned-{id}"),
                grpc_addr: Some(format!("0.0.0.0:{}", 1u16.wrapping_add(id))),
                grpc_bind: None,
                http_addr: Some(format!("http://0.0.0.0:{}", 1u16.wrapping_add(id))),
                grpc_tls: None,
                metrics_addr: None,
                zone: None,
            });
        }
    }

    // The leader for the synthesized topology: the CURRENT leader the join
    // resolved (so the `needed=false` boot self-heal pulls from the live stream,
    // not a stale lowest-voter — the catch-up targets `shard_of_region(leader)`).
    // Fall back to the lowest-id voter when the join could not resolve a leader
    // (mid-election). A joiner is never the term-0 leader, so it must not name
    // itself here; the durable election state + heartbeats correct the hint.
    let leader = current_leader_region
        .and_then(|id| {
            members
                .iter()
                .find(|m| m.id == id && m.role != MemberRole::Removed)
                .map(|m| m.name.clone())
        })
        .filter(|name| name != input.region)
        .or_else(|| {
            members
                .iter()
                .filter(|m| m.role == MemberRole::Voter)
                .min_by_key(|m| m.id)
                .map(|m| m.name.clone())
        })
        .ok_or_else(|| {
            ServerError::Cluster("seed-join: the join roster has no voter to name as leader".into())
        })?;

    // Carry the local file's knob blocks verbatim (§3.5).
    Ok(TopologySpec {
        regions,
        leader,
        write_workers: input.knobs.write_workers,
        timeouts: clone_timeouts(&input.knobs.timeouts),
        replication: clone_replication(&input.knobs.replication),
        wal: clone_wal(&input.knobs.wal),
        election: clone_election(&input.knobs.election),
        // Seed-join synthesizes the legacy single group (the joiner learns its
        // shard assignment from the membership log, not the synthesized file).
        shards: None,
    })
}

/// This region's TLS spec for the synthesized topology. The joiner's own region
/// is NOT in the shared-ConfigMap knob file, so this falls back to ANY region's
/// `grpc_tls` ([`grpc_tls_for`]) — every pod mounts the same cert files at the
/// same paths. Without the fallback a TLS-cluster joiner synthesized a PLAINTEXT
/// gRPC posture and could not mTLS-replicate with its peers.
fn self_tls_spec(knobs: &TopologySpec, name: &str) -> Option<GrpcTlsSpec> {
    grpc_tls_for(knobs, name).map(|t| GrpcTlsSpec {
        ca_cert: t.ca_cert.clone(),
        server_cert: t.server_cert.clone(),
        server_key: t.server_key.clone(),
        client_cert: t.client_cert.clone(),
        client_key: t.client_key.clone(),
    })
}

// The knob blocks have no `Clone`; rebuild them field-by-field from the loaded
// file so the synthesized topology carries the operator's exact knobs (§3.5).
const fn clone_timeouts(t: &TimeoutsSpec) -> TimeoutsSpec {
    TimeoutsSpec {
        broadcast_peer_secs: t.broadcast_peer_secs,
    }
}

fn clone_replication(r: &ReplicationSpec) -> ReplicationSpec {
    ReplicationSpec {
        batch_max_events: r.batch_max_events,
        window: r.window,
        retry_ms: r.retry_ms,
        ack: r.ack.clone(),
        quorum_timeout_ms: r.quorum_timeout_ms,
        catchup_retry_ms: r.catchup_retry_ms,
        snapshot_artifact_ttl_ms: r.snapshot_artifact_ttl_ms,
        reseed_self_restart: r.reseed_self_restart,
        learner_promote_lag: r.learner_promote_lag,
    }
}

fn clone_wal(w: &WalSpec) -> WalSpec {
    WalSpec {
        batch_size: w.batch_size,
        batch_timeout_ms: w.batch_timeout_ms,
        archive_dir: w.archive_dir.clone(),
    }
}

const fn clone_election(e: &ElectionSpec) -> ElectionSpec {
    ElectionSpec {
        heartbeat_interval_ms: e.heartbeat_interval_ms,
        election_timeout_min_ms: e.election_timeout_min_ms,
        election_timeout_max_ms: e.election_timeout_max_ms,
        leader_lease_ms: e.leader_lease_ms,
        auto_election: e.auto_election,
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    use super::*;

    fn knobs_with(ack: &str) -> TopologySpec {
        TopologySpec {
            // The knob file's regions are IGNORED for the roster — a single
            // declared region proves we do not read them.
            regions: vec![RegionSpec {
                name: "us-east".into(),
                grpc_addr: Some("127.0.0.1:9000".into()),
                grpc_bind: None,
                http_addr: Some("127.0.0.1:9001".into()),
                grpc_tls: None,
                metrics_addr: None,
                zone: None,
            }],
            leader: "us-east".into(),
            write_workers: None,
            timeouts: TimeoutsSpec::default(),
            replication: ReplicationSpec {
                ack: Some(ack.to_string()),
                quorum_timeout_ms: Some(1500),
                ..ReplicationSpec::default()
            },
            wal: WalSpec::default(),
            election: ElectionSpec::default(),
            shards: None,
        }
    }

    fn member(id: u16, name: &str, role: MemberRole) -> MemberEntry {
        MemberEntry {
            id,
            name: name.to_string(),
            grpc_addr: format!("{name}.svc:9500"),
            http_addr: format!("http://{name}.svc:9501"),
            role,
        }
    }

    /// The synthesized topology carries the join roster as `regions:` (id-dense,
    /// positional index == member id), this node's advertised addresses for self,
    /// the knob file's behavioral blocks (NOT its regions), and the metrics flag.
    #[test]
    fn synthesize_topology_uses_roster_and_local_knobs() {
        let tmp = tempfile::tempdir().unwrap();
        let knobs = knobs_with("quorum");
        let members = vec![
            member(0, "us-east", MemberRole::Voter),
            member(1, "eu-west", MemberRole::Voter),
            member(3, "joiner", MemberRole::Learner), // id 2 was burned
        ];
        let input = SeedJoinInput {
            seeds: &["http://us-east.svc:9501".to_string()],
            knobs: &knobs,
            region: "joiner",
            advertise_grpc: "joiner.real:9500",
            advertise_http: "http://joiner.real:9501",
            metrics: Some("0.0.0.0:9091"),
            data_dir: tmp.path(),
            api_key: None,
        };
        // Name the current leader region 0 (us-east) explicitly.
        let synth = synthesize_topology(&input, &members, Some(0)).unwrap();

        // The regions list is id-dense 0..=3 (a burned id 2 gets a placeholder),
        // so the positional RegionId equals the assigned member id.
        assert_eq!(synth.regions.len(), 4);
        assert_eq!(synth.regions[0].name, "us-east");
        assert_eq!(synth.regions[1].name, "eu-west");
        assert!(
            synth.regions[2].name.starts_with("__burned-"),
            "id 2 burned"
        );
        assert_eq!(synth.regions[3].name, "joiner");
        // Self carries its REAL advertised addresses + the --metrics bind.
        assert_eq!(
            synth.regions[3].grpc_addr.as_deref(),
            Some("joiner.real:9500")
        );
        assert_eq!(
            synth.regions[3].http_addr.as_deref(),
            Some("http://joiner.real:9501")
        );
        assert_eq!(
            synth.regions[3].metrics_addr.as_deref(),
            Some("0.0.0.0:9091")
        );
        // The knob file's behavioral block is carried, NOT its single region.
        assert_eq!(synth.replication.ack.as_deref(), Some("quorum"));
        assert_eq!(synth.replication.quorum_timeout_ms, Some(1500));
        // The leader is the lowest-id voter (a joiner never names itself).
        assert_eq!(synth.leader, "us-east");
    }

    /// `discovery_candidates` lists seeds first, then the cache's live members'
    /// HTTP addresses, deduplicated.
    #[test]
    fn discovery_candidates_merges_seeds_and_cache() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MembershipStore::new(tmp.path());
        store
            .persist(&MembershipSnapshot {
                version: 2,
                term: 1,
                members: vec![
                    member(0, "us-east", MemberRole::Voter),
                    member(1, "eu-west", MemberRole::Voter),
                    member(2, "gone", MemberRole::Removed),
                ],
            })
            .unwrap();
        let seeds = vec![
            "http://seed-a:9501".to_string(),
            "http://us-east.svc:9501".to_string(),
        ];
        let cands = discovery_candidates(&seeds, &store);
        let addrs: Vec<&str> = cands.iter().map(|c| c.http_addr.as_str()).collect();
        // Seeds first.
        assert_eq!(addrs[0], "http://seed-a:9501");
        // us-east is in both; it appears once (seed wins position).
        assert_eq!(
            addrs
                .iter()
                .filter(|a| **a == "http://us-east.svc:9501")
                .count(),
            1
        );
        // eu-west comes from the cache.
        assert!(addrs.contains(&"http://eu-west.svc:9501"));
        // The removed tombstone is excluded.
        assert!(!addrs.iter().any(|a| a.contains("gone")));
    }

    /// `leader_dial_target` is the lowest-id voter's gRPC address (the snapshot
    /// source); learners/tombstones are not dial targets.
    #[test]
    fn leader_dial_target_is_lowest_id_voter() {
        let members = vec![
            member(0, "us-east", MemberRole::Voter),
            member(1, "eu-west", MemberRole::Voter),
            member(3, "joiner", MemberRole::Learner),
        ];
        let (grpc, id) = leader_dial_target(&members);
        assert_eq!(grpc, "us-east.svc:9500");
        assert_eq!(id, 0);
    }
}