//! The membership runtime (m11p5 §3.1–§3.3): the effective roster, the one
//! fenced apply path, the conf-change gates, and the leader-side join logic.
//!
//! # The effective roster
//!
//! [`MembershipView`] is the single source of roster truth for a running node.
//! It is derived at boot from the WAL-recovered [`ClusterMembership`] cell when
//! that cell is non-`None` (the **membership era** — ids are the RECORD ids,
//! permanent), else from the topology's positional ids (**era 0** — byte-for-byte
//! today's behavior). Every roster lookup — name↔id, peer HTTP address, the
//! voter / learner / member sets, the status surfaces — reads through the view,
//! so a conf-change mutates one place and the whole node agrees.
//!
//! # The one fenced apply path
//!
//! [`apply_record`](MembershipView::apply_record) swaps the view's roster under
//! one write lock. The caller (the node's membership-apply critical section)
//! then drives the FOUR other surfaces — `PeerPool`, `ShipQueue`, `CommitIndex`,
//! `ElectionState` — from the diff this method returns, so the five copies can
//! never disagree about the roster. The view never reaches across crates itself
//! (it owns no transport, no engine) — it computes the plan; the node executes
//! it. That keeps this module pure and exhaustively unit-testable.
//!
//! # The conf-change gates (§3.1, §3.2/§3.3)
//!
//! - [`capability_gate`]: before the FIRST conf-change (era begin) and before
//!   every join/remove, every current VOTER's last-reported capabilities must
//!   carry [`CAP_KIND4_MEMBERSHIP`]. The leader's own binary is capable by
//!   definition. Refusal names the incapable voters.
//! - The one-at-a-time gate is the caller's [`CommitIndex::committed_in_term`]
//!   check on the previous record's seq — never the activation-reset
//!   `committed()`. This module surfaces the *plan* (what the next record's
//!   roster is); the node enforces the gate against the live commit index.
//!
//! [`ClusterMembership`]: tidaldb::wal::ClusterMembership
//! [`CommitIndex::committed_in_term`]: tidaldb::replication::commit::CommitIndex::committed_in_term
//! [`CAP_KIND4_MEMBERSHIP`]: tidal_net::CAP_KIND4_MEMBERSHIP

use std::collections::HashMap;
use std::sync::RwLock;

use tidaldb::replication::shard::{RegionId, ShardId};
use tidaldb::wal::format::{MemberEntry, MemberRole, MembershipRecord};

use super::topology::{ResolvedShardGroup, TopologySpec};

/// A coherent snapshot of the roster the view derived (era 0 or a kind-4
/// record). All the lookup maps are built from this so they can never drift.
#[derive(Debug, Clone)]
pub struct Roster {
    /// The conf version (0 in era 0; the record's version in the membership era).
    pub version: u64,
    /// The leadership term that authored the roster (0 in era 0).
    pub term: u64,
    /// Whether this roster came from a kind-4 record (the membership era) vs the
    /// topology file (era 0). The capability gate's "era has begun" predicate.
    pub from_record: bool,
    /// The full member roster (Voter / Learner / Removed), id-keyed.
    pub members: Vec<MemberEntry>,
}

impl Roster {
    /// All non-removed members' region ids (Voters + Learners). Removed
    /// tombstones are excluded — their ids are burned, not addressable.
    fn live_ids(&self) -> impl Iterator<Item = (RegionId, &MemberEntry)> {
        self.members
            .iter()
            .filter(|m| m.role != MemberRole::Removed)
            .map(|m| (RegionId(m.id), m))
    }

    /// Voter region ids in id order.
    #[must_use]
    pub fn voter_ids(&self) -> Vec<RegionId> {
        let mut v: Vec<RegionId> = self
            .members
            .iter()
            .filter(|m| m.role == MemberRole::Voter)
            .map(|m| RegionId(m.id))
            .collect();
        v.sort_unstable_by_key(|r| r.0);
        v
    }

    /// Learner region ids in id order.
    #[must_use]
    pub fn learner_ids(&self) -> Vec<RegionId> {
        let mut v: Vec<RegionId> = self
            .members
            .iter()
            .filter(|m| m.role == MemberRole::Learner)
            .map(|m| RegionId(m.id))
            .collect();
        v.sort_unstable_by_key(|r| r.0);
        v
    }

    /// The role of a region, or `None` if it is not in this roster at all (an
    /// id that was never assigned). A `Removed` tombstone returns
    /// `Some(Removed)`.
    #[must_use]
    pub fn role_of(&self, id: RegionId) -> Option<MemberRole> {
        self.members.iter().find(|m| m.id == id.0).map(|m| m.role)
    }
}

/// The effective roster + the lookup tables derived from it, behind one
/// `RwLock`. A conf-change takes the write lock for the whole swap; every
/// reader takes the read lock briefly.
pub struct MembershipView {
    /// This node's own region id (permanent across the node's life — it is the
    /// node's identity, never renumbered).
    self_region: RegionId,
    inner: RwLock<ViewInner>,
}

struct ViewInner {
    roster: Roster,
    /// name → id (live members + tombstones; a removed id keeps its name so a
    /// stale forward resolves to "removed", not "unknown").
    name_to_id: HashMap<String, RegionId>,
    /// id → name (inverse).
    id_to_name: HashMap<RegionId, String>,
    /// Peer (NOT self) region id → advertised HTTP address, live members only.
    peer_http: HashMap<RegionId, String>,
    /// Peer (NOT self) shard id → advertised gRPC address, live members only.
    peer_grpc: HashMap<ShardId, String>,
}

/// The diff the node executes against the four other surfaces after the view
/// swaps (`PeerPool`, `ShipQueue`, `CommitIndex`, `ElectionState`). Computed
/// inside the apply lock so it reflects exactly the roster that just became
/// effective.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ApplyPlan {
    /// Peer shards (NOT self) that are newly addressable: `(shard, grpc_addr)`.
    /// `PeerPool::add_peer` + `ShipQueue::add_peer` for each.
    pub added_peers: Vec<(ShardId, String)>,
    /// Peer shards (NOT self) no longer in the live roster (removed tombstones
    /// or that left): `PeerPool::remove_peer` + `ShipQueue::remove_peer`.
    pub removed_peers: Vec<ShardId>,
    /// The new voter peer shards (NOT self) for `CommitIndex::reconfigure`.
    pub commit_voters: Vec<ShardId>,
    /// The new learner peer shards (NOT self) for `CommitIndex::reconfigure`.
    pub commit_learners: Vec<ShardId>,
    /// The new election voter set (every OTHER voter region) for
    /// `ElectionState::reconfigure`.
    pub election_voters: Vec<RegionId>,
    /// The new election learner set (every OTHER learner region). Learners are
    /// NOT voters (they never count toward a quorum) but ARE in the heartbeat
    /// fan-out so a learner joins the term and reports — the auto-promotion input.
    pub election_learners: Vec<RegionId>,
    /// Whether THIS node is a voter in the new roster (drives
    /// `ElectionState::set_voter` — the campaign gate).
    pub self_is_voter: bool,
    /// The new total voter count, for the `ControlPlane` topology-count update.
    pub voter_count: usize,
}

impl MembershipView {
    /// Build the era-0 view for one shard `group` from its resolved replicas.
    /// `name_to_id`/`id_to_name` stay CLUSTER-WIDE (the node's declaration-order
    /// maps) so a stale cross-shard forward still resolves a non-replica region's
    /// NAME; `peer_http`/`peer_grpc` are the group-scoped `build_group_peer_tables`
    /// outputs. The roster lists exactly the GROUP's replicas (every one a Voter,
    /// id = positional `RegionId`, gRPC from the resolved replica address, HTTP
    /// from the hosting node's spec) so `voter_ids`/`voter_count`/`live_ids` and
    /// the status surfaces agree with the (group-scoped) peer/quorum/election sets
    /// that the rest of `ShardReplica::new` built. For the legacy single group
    /// (`resolve_shard_groups` synthesizes every region as a replica, addresses
    /// verbatim) this is byte-for-byte the pre-m11p6 all-regions-Voter roster.
    #[must_use]
    pub fn era0(
        self_region: RegionId,
        group: &ResolvedShardGroup,
        topology: &TopologySpec,
        name_to_id: &HashMap<String, RegionId>,
        id_to_name: &HashMap<RegionId, String>,
        peer_http: &HashMap<RegionId, String>,
        peer_grpc: &HashMap<ShardId, String>,
    ) -> Self {
        // Reconstruct the era-0 roster from THIS GROUP's replicas (NOT every
        // topology region): each replica is a Voter, id = positional RegionId,
        // gRPC from the resolved replica address, HTTP from the hosting node's
        // RegionSpec (ResolvedReplica carries no HTTP — one HTTP server per node).
        let mut members: Vec<MemberEntry> = group
            .replicas
            .iter()
            .map(|r| MemberEntry {
                id: r.region.0,
                name: r.name.clone(),
                grpc_addr: r.grpc_addr.clone(),
                http_addr: topology
                    .regions
                    .get(usize::from(r.region.0))
                    .and_then(|n| n.http_addr.clone())
                    .unwrap_or_default(),
                role: MemberRole::Voter,
            })
            .collect();
        members.sort_unstable_by_key(|m| m.id);

        let roster = Roster {
            version: 0,
            term: 0,
            from_record: false,
            members,
        };
        let inner = ViewInner {
            roster,
            name_to_id: name_to_id.clone(),
            id_to_name: id_to_name.clone(),
            peer_http: peer_http.clone(),
            peer_grpc: peer_grpc.clone(),
        };
        Self {
            self_region,
            inner: RwLock::new(inner),
        }
    }

    /// Build the membership-era view from a WAL-recovered roster (the
    /// `ClusterMembership` cell was non-`None` at boot — ids are RECORD ids).
    /// All lookup tables are derived from the roster, so the view is internally
    /// consistent from the first read.
    #[must_use]
    pub fn from_record(
        self_region: RegionId,
        version: u64,
        term: u64,
        members: Vec<MemberEntry>,
    ) -> Self {
        let roster = Roster {
            version,
            term,
            from_record: true,
            members,
        };
        let inner = Self::tables_from_roster(self_region, roster);
        Self {
            self_region,
            inner: RwLock::new(inner),
        }
    }

    /// Build the lookup tables from a roster. `name_to_id`/`id_to_name` include
    /// tombstones (a removed id resolves by name to "removed", never reused);
    /// `peer_http`/`peer_grpc` include only LIVE peers (not self, not removed).
    fn tables_from_roster(self_region: RegionId, roster: Roster) -> ViewInner {
        let mut name_to_id = HashMap::new();
        let mut id_to_name = HashMap::new();
        let mut peer_http = HashMap::new();
        let mut peer_grpc = HashMap::new();
        for m in &roster.members {
            let rid = RegionId(m.id);
            name_to_id.insert(m.name.clone(), rid);
            id_to_name.insert(rid, m.name.clone());
            if m.role != MemberRole::Removed && rid != self_region {
                peer_http.insert(rid, m.http_addr.clone());
                peer_grpc.insert(ShardId(m.id), m.grpc_addr.clone());
            }
        }
        ViewInner {
            roster,
            name_to_id,
            id_to_name,
            peer_http,
            peer_grpc,
        }
    }

    fn read(&self) -> std::sync::RwLockReadGuard<'_, ViewInner> {
        self.inner
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }

    /// The current roster snapshot (clones — used by status surfaces and the
    /// activation re-append, not the hot path).
    #[must_use]
    pub fn roster(&self) -> Roster {
        self.read().roster.clone()
    }

    /// The current applied membership version.
    #[must_use]
    pub fn version(&self) -> u64 {
        self.read().roster.version
    }

    /// The leadership term that authored the current roster (0 in era 0).
    #[must_use]
    pub fn term(&self) -> u64 {
        self.read().roster.term
    }

    /// Whether the membership era has begun (the cell is backed by a kind-4
    /// record). The capability gate's "first conf-change" predicate reads this:
    /// `false` = era 0 (topology), so the next conf-change BEGINS the era.
    #[must_use]
    pub fn era_begun(&self) -> bool {
        self.read().roster.from_record
    }

    /// Resolve a region name → id (live members + tombstones).
    #[must_use]
    pub fn name_to_id(&self, name: &str) -> Option<RegionId> {
        self.read().name_to_id.get(name).copied()
    }

    /// A peer's advertised HTTP address (live peers only; `None` for self, a
    /// removed member, or an unknown id).
    #[must_use]
    pub fn peer_http(&self, id: RegionId) -> Option<String> {
        self.read().peer_http.get(&id).cloned()
    }

    /// Every live peer's `(name, http_addr)` (NOT self), in id order — the
    /// `peers_named` surface.
    #[must_use]
    pub fn peers_named(&self) -> Vec<(String, String)> {
        let inner = self.read();
        let mut out: Vec<(RegionId, String, String)> = inner
            .peer_http
            .iter()
            .map(|(&rid, http)| {
                let name = inner
                    .id_to_name
                    .get(&rid)
                    .cloned()
                    .unwrap_or_else(|| "unknown".to_string());
                (rid, name, http.clone())
            })
            .collect();
        drop(inner);
        out.sort_unstable_by_key(|(rid, ..)| rid.0);
        out.into_iter().map(|(_, n, h)| (n, h)).collect()
    }

    /// Every live region (self + peers) as `(id, name, http_or_none)` in id
    /// order — the `all_regions_for_status` surface. Self's HTTP entry is
    /// `None` (a node never forwards to itself), peers carry their address.
    #[must_use]
    pub fn all_regions_for_status(&self) -> Vec<(RegionId, String, Option<String>)> {
        let inner = self.read();
        let mut out: Vec<(RegionId, String, Option<String>)> = inner
            .roster
            .live_ids()
            .map(|(rid, m)| {
                let http = if rid == self.self_region {
                    None
                } else {
                    inner.peer_http.get(&rid).cloned()
                };
                (rid, m.name.clone(), http)
            })
            .collect();
        drop(inner);
        out.sort_unstable_by_key(|(rid, ..)| rid.0);
        out
    }

    /// This node's role in the current roster (`None` = not in the roster — the
    /// era-0 view always has self as a Voter; a removed self returns
    /// `Some(Removed)`).
    #[must_use]
    pub fn self_role(&self) -> Option<MemberRole> {
        self.read().roster.role_of(self.self_region)
    }

    /// Apply a kind-4 record's roster, swapping the view under one write lock,
    /// and return the [`ApplyPlan`] the node executes against the other four
    /// surfaces. Idempotent-by-version: a record at-or-below the current version
    /// is a replay (recovery re-stream); the view does NOT regress and the plan
    /// is a no-op diff against the unchanged roster.
    ///
    /// The diff is computed from the LIVE peer sets before and after, so a
    /// promote (learner→voter, same id) yields no add/remove peer churn — only a
    /// `commit_voters`/`commit_learners`/`election_voters` change.
    pub fn apply_record(&self, record: &MembershipRecord) -> ApplyPlan {
        let mut inner = self
            .inner
            .write()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        // Replay / regression guard: never move the view backwards (recovery can
        // re-deliver an older record on a re-stream). An equal version is the
        // same record re-seen — idempotent, the plan is a no-op.
        if record.version <= inner.roster.version && inner.roster.from_record {
            return self.plan_from(&inner, &inner.roster.clone());
        }

        // The live peer sets BEFORE the swap (for the add/remove diff).
        let before_peers: HashMap<ShardId, String> = inner.peer_grpc.clone();

        let new_roster = Roster {
            version: record.version,
            term: record.term,
            from_record: true,
            members: record.members.clone(),
        };
        let new_inner = Self::tables_from_roster(self.self_region, new_roster.clone());

        // Compute the peer diff from the before/after live gRPC peer maps.
        let mut added_peers: Vec<(ShardId, String)> = Vec::new();
        for (&shard, addr) in &new_inner.peer_grpc {
            if !before_peers.contains_key(&shard) {
                added_peers.push((shard, addr.clone()));
            }
        }
        added_peers.sort_unstable_by_key(|(s, _)| s.0);
        let mut removed_peers: Vec<ShardId> = before_peers
            .keys()
            .filter(|s| !new_inner.peer_grpc.contains_key(s))
            .copied()
            .collect();
        removed_peers.sort_unstable_by_key(|s| s.0);

        let plan = self.plan_diff(&new_roster, added_peers, removed_peers);
        *inner = new_inner;
        plan
    }

    /// A no-op plan against an unchanged roster (the replay path): no peer
    /// churn, the quorum sets are the current roster's.
    fn plan_from(&self, _inner: &ViewInner, roster: &Roster) -> ApplyPlan {
        self.plan_diff(roster, Vec::new(), Vec::new())
    }

    /// Build the [`ApplyPlan`] for a roster + a precomputed peer add/remove diff.
    fn plan_diff(
        &self,
        roster: &Roster,
        added_peers: Vec<(ShardId, String)>,
        removed_peers: Vec<ShardId>,
    ) -> ApplyPlan {
        let commit_voters: Vec<ShardId> = roster
            .voter_ids()
            .into_iter()
            .filter(|&r| r != self.self_region)
            .map(|r| ShardId(r.0))
            .collect();
        let commit_learners: Vec<ShardId> = roster
            .learner_ids()
            .into_iter()
            .filter(|&r| r != self.self_region)
            .map(|r| ShardId(r.0))
            .collect();
        let election_voters: Vec<RegionId> = roster
            .voter_ids()
            .into_iter()
            .filter(|&r| r != self.self_region)
            .collect();
        let election_learners: Vec<RegionId> = roster
            .learner_ids()
            .into_iter()
            .filter(|&r| r != self.self_region)
            .collect();
        let self_is_voter = roster.role_of(self.self_region) == Some(MemberRole::Voter);
        let voter_count = roster.voter_ids().len();
        ApplyPlan {
            added_peers,
            removed_peers,
            commit_voters,
            commit_learners,
            election_voters,
            election_learners,
            self_is_voter,
            voter_count,
        }
    }
}

/// Why a conf-change was refused (the leader-side gate verdict). Every variant
/// is RETRYABLE — the operator/joiner re-issues once the precondition clears.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ConfChangeRefusal {
    /// This node is not the leader; the join/remove forwards to the hint.
    NotLeader {
        /// The leader's region name, if known.
        leader: Option<String>,
    },
    /// One or more current voters do not report kind-4 capability (§3.1). The
    /// conf-change is held until every voter's binary is upgraded.
    IncapableVoters {
        /// The incapable voters' region names.
        voters: Vec<String>,
    },
    /// The previous conf-change record is not yet same-term quorum-committed
    /// (§3.2/§3.3 one-at-a-time). Bounded retry.
    PriorChangeUncommitted {
        /// The seq the gate is waiting on.
        awaiting_seq: u64,
    },
}

impl std::fmt::Display for ConfChangeRefusal {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::NotLeader { leader } => match leader {
                Some(l) => write!(f, "not the leader; re-target {l}"),
                None => write!(f, "not the leader (no leader currently known); retry"),
            },
            Self::IncapableVoters { voters } => write!(
                f,
                "conf-change held: these voters do not report kind-4 capability \
                 (upgrade their binaries first): {voters:?}"
            ),
            Self::PriorChangeUncommitted { awaiting_seq } => write!(
                f,
                "conf-change held: the prior membership record (seq {awaiting_seq}) is not yet \
                 same-term quorum-committed; retry"
            ),
        }
    }
}

/// The capability gate (§3.1): every current VOTER (except self, the leader,
/// which is capable by definition) must report [`CAP_KIND4_MEMBERSHIP`].
///
/// `peer_capability` returns a voter's last-reported capability bit-field, or
/// `None` when no report has been seen yet (a fresh voter, or a momentary
/// election) — treated conservatively as INCAPABLE (proto3 zero-default is the
/// pre-p5 story; an unknown voter blocks the change until it reports).
///
/// `self_region` is the leader's own id — always capable, never gated.
///
/// Returns `Ok(())` when every voter is capable, else the incapable voter ids.
///
/// # Errors
///
/// [`ConfChangeRefusal::IncapableVoters`] naming the incapable voters.
pub fn capability_gate(
    roster: &Roster,
    self_region: RegionId,
    mut peer_capability: impl FnMut(RegionId) -> Option<u64>,
) -> Result<(), Vec<RegionId>> {
    let mut incapable: Vec<RegionId> = Vec::new();
    for rid in roster.voter_ids() {
        if rid == self_region {
            continue; // the leader's own binary is capable by definition.
        }
        let cap = peer_capability(rid).unwrap_or(0);
        if cap & tidal_net::CAP_KIND4_MEMBERSHIP == 0 {
            incapable.push(rid);
        }
    }
    if incapable.is_empty() {
        Ok(())
    } else {
        incapable.sort_unstable_by_key(|r| r.0);
        Err(incapable)
    }
}

/// Assign the next member id for a join (§3.3): `max(all ids EVER) + 1`,
/// including `Removed` tombstones, so a burned id is never reused even after a
/// remove. Returns `None` only on the (structurally impossible) all-`u16::MAX`
/// roster — the caller maps that to a hard error.
#[must_use]
pub fn next_member_id(roster: &Roster) -> Option<u16> {
    let max_ever = roster.members.iter().map(|m| m.id).max()?;
    max_ever.checked_add(1)
}

/// Build the kind-4 record for a join (§3.3). Idempotent by name: a known member
/// (any role, including a tombstone) returns its EXISTING entry and `None` (the
/// leader appends nothing). An unknown name appends a fresh `Learner` at
/// `next_member_id`.
///
/// `version`/`term` are the new record's version (`current + 1`) and the leader's
/// term. The returned record carries the FULL roster (every existing member
/// unchanged plus the new learner) — records are snapshots, not deltas.
pub fn plan_join(
    roster: &Roster,
    ask_name: &str,
    ask_grpc: &str,
    ask_http: &str,
    next_version: u64,
    term: u64,
) -> JoinPlan {
    // Idempotent by name: a known member returns its current entry, no append.
    if let Some(existing) = roster.members.iter().find(|m| m.name == ask_name) {
        return JoinPlan::Existing {
            id: existing.id,
            role: existing.role,
        };
    }
    let Some(id) = next_member_id(roster) else {
        return JoinPlan::IdSpaceExhausted;
    };
    let mut members = roster.members.clone();
    members.push(MemberEntry {
        id,
        name: ask_name.to_string(),
        grpc_addr: ask_grpc.to_string(),
        http_addr: ask_http.to_string(),
        role: MemberRole::Learner,
    });
    members.sort_unstable_by_key(|m| m.id);
    JoinPlan::Append {
        id,
        record: MembershipRecord {
            version: next_version,
            term,
            members,
        },
    }
}

/// The outcome of [`plan_join`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum JoinPlan {
    /// A known member (idempotent): return its id/role, append nothing.
    Existing { id: u16, role: MemberRole },
    /// A new member: append this Learner record, then promote later.
    Append { id: u16, record: MembershipRecord },
    /// The id space is exhausted (65 536 ids ever assigned) — a hard error.
    IdSpaceExhausted,
}

/// Build the kind-4 record that flips `name` to `Removed` (§3.3). Returns `None`
/// when the name is unknown or already removed (idempotent — nothing to do).
/// The record carries the full roster with that one member's role set to
/// `Removed`; its id is BURNED (kept in the roster as a tombstone).
#[must_use]
pub fn plan_remove(
    roster: &Roster,
    name: &str,
    next_version: u64,
    term: u64,
) -> Option<MembershipRecord> {
    let target = roster.members.iter().find(|m| m.name == name)?;
    if target.role == MemberRole::Removed {
        return None; // already a tombstone; idempotent no-op.
    }
    let members: Vec<MemberEntry> = roster
        .members
        .iter()
        .map(|m| {
            if m.name == name {
                MemberEntry {
                    role: MemberRole::Removed,
                    ..m.clone()
                }
            } else {
                m.clone()
            }
        })
        .collect();
    Some(MembershipRecord {
        version: next_version,
        term,
        members,
    })
}

/// Build the kind-4 record that PROMOTES a learner to `Voter` (§3.3 auto-
/// promotion). Returns `None` when the id is unknown, already a Voter, or a
/// `Removed` tombstone (idempotent / not promotable — the duty raced another
/// apply). The record carries the full roster with that one member's role set to
/// `Voter`; its id is unchanged (a promote is the same id, so it causes no peer
/// add/remove churn — only the commit/election voter sets grow).
#[must_use]
pub fn plan_promote(
    roster: &Roster,
    id: u16,
    next_version: u64,
    term: u64,
) -> Option<MembershipRecord> {
    let target = roster.members.iter().find(|m| m.id == id)?;
    if target.role != MemberRole::Learner {
        return None; // only a Learner is promotable.
    }
    let members: Vec<MemberEntry> = roster
        .members
        .iter()
        .map(|m| {
            if m.id == id {
                MemberEntry {
                    role: MemberRole::Voter,
                    ..m.clone()
                }
            } else {
                m.clone()
            }
        })
        .collect();
    Some(MembershipRecord {
        version: next_version,
        term,
        members,
    })
}

/// Build the activation re-append record (§3.2): the leader's CURRENT roster at
/// `current_version + 1`, stamped with the new `term`. This is the no-op-entry
/// analogue that makes committed membership reach every joiner of the term. Only
/// called when the era has begun (the cell is non-`None`) — a pre-p5 follower
/// never sees a kind-4 record because the era only begins via the
/// capability-gated first conf-change.
#[must_use]
pub fn plan_activation_reappend(roster: &Roster, term: u64) -> MembershipRecord {
    MembershipRecord {
        version: roster.version + 1,
        term,
        members: roster.members.clone(),
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    use super::super::topology::{RegionSpec, ShardReplicaSpec, ShardSpec, TopologySpec};
    use super::*;

    fn topo(names: &[&str]) -> TopologySpec {
        TopologySpec {
            leader: names[0].to_string(),
            regions: names
                .iter()
                .map(|n| RegionSpec {
                    name: (*n).to_string(),
                    grpc_addr: Some(format!("{n}.svc:9500")),
                    grpc_bind: None,
                    http_addr: Some(format!("http://{n}.svc:9501")),
                    grpc_tls: None,
                    metrics_addr: None,
                    zone: None,
                })
                .collect(),
            write_workers: None,
            timeouts: super::super::topology::TimeoutsSpec::default(),
            replication: super::super::topology::ReplicationSpec::default(),
            wal: super::super::topology::WalSpec::default(),
            election: super::super::topology::ElectionSpec::default(),
            shards: None,
        }
    }

    fn era0_view(self_idx: u16, names: &[&str]) -> MembershipView {
        let t = topo(names);
        // Legacy single group: every region is a replica (the synthesized
        // group is byte-for-byte the pre-m11p6 all-regions roster).
        let groups = t.resolve_shard_groups().expect("legacy synthesis");
        let mut name_to_id = HashMap::new();
        let mut id_to_name = HashMap::new();
        let mut peer_http = HashMap::new();
        let mut peer_grpc = HashMap::new();
        for (i, n) in names.iter().enumerate() {
            let rid = RegionId(u16::try_from(i).unwrap());
            name_to_id.insert((*n).to_string(), rid);
            id_to_name.insert(rid, (*n).to_string());
            if rid != RegionId(self_idx) {
                peer_http.insert(rid, format!("http://{n}.svc:9501"));
                peer_grpc.insert(ShardId(rid.0), format!("{n}.svc:9500"));
            }
        }
        MembershipView::era0(
            RegionId(self_idx),
            &groups[0],
            &t,
            &name_to_id,
            &id_to_name,
            &peer_http,
            &peer_grpc,
        )
    }

    /// An RF < N group's era-0 roster lists EXACTLY the group's replicas, not
    /// every region — so `voter_ids`/`voter_count` and `all_regions_for_status`
    /// agree with the group-scoped peer/quorum/election sets (the FIX-A guard
    /// against an N-voter majority on an RF-replica group).
    #[test]
    fn era0_roster_scoped_to_group_replicas_for_rf_lt_n() {
        // Three regions, one shard of RF=2 (us-east + ap-south; eu-west is NOT a
        // replica of this group). name_to_id stays cluster-wide (all three).
        let mut t = topo(&["us", "eu", "ap"]);
        t.shards = Some(vec![ShardSpec {
            id: 0,
            leader: Some("us".into()),
            replicas: vec![
                ShardReplicaSpec {
                    node: "us".into(),
                    grpc_addr: None,
                    grpc_bind: None,
                },
                ShardReplicaSpec {
                    node: "ap".into(),
                    grpc_addr: None,
                    grpc_bind: None,
                },
            ],
        }]);
        let groups = t.resolve_shard_groups().expect("resolve RF=2 group");
        let mut name_to_id = HashMap::new();
        let mut id_to_name = HashMap::new();
        for (i, n) in ["us", "eu", "ap"].iter().enumerate() {
            let rid = RegionId(u16::try_from(i).unwrap());
            name_to_id.insert((*n).to_string(), rid);
            id_to_name.insert(rid, (*n).to_string());
        }
        // self = us (region 0); the only peer in THIS group is ap (region 2).
        let mut peer_http = HashMap::new();
        let mut peer_grpc = HashMap::new();
        peer_http.insert(RegionId(2), "http://ap.svc:9501".to_string());
        peer_grpc.insert(ShardId(2), "ap.svc:9500".to_string());
        let v = MembershipView::era0(
            RegionId(0),
            &groups[0],
            &t,
            &name_to_id,
            &id_to_name,
            &peer_http,
            &peer_grpc,
        );
        // The roster is the 2 replicas, NOT all 3 regions — eu-west is absent.
        assert_eq!(
            v.roster().voter_ids(),
            vec![RegionId(0), RegionId(2)],
            "era-0 roster must be the group's RF replicas, not every region"
        );
        assert_eq!(v.roster().members.len(), 2);
        assert!(
            v.roster().members.iter().all(|m| m.name != "eu"),
            "the non-replica region must not appear in the group roster"
        );
        // all_regions_for_status reflects the group, not the cluster.
        let all = v.all_regions_for_status();
        assert_eq!(all.len(), 2);
        assert!(all.iter().all(|(_, name, _)| name != "eu"));
        // name_to_id stays cluster-wide: a stale forward to eu still resolves.
        assert_eq!(v.name_to_id("eu"), Some(RegionId(1)));
    }

    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 era-0 view's roster, voter set, and lookup tables match the topology
    /// positional tables byte-for-byte.
    #[test]
    fn era0_view_matches_topology_tables() {
        let v = era0_view(0, &["us", "eu", "ap"]);
        assert!(!v.era_begun(), "no kind-4 record → era 0");
        assert_eq!(v.version(), 0);
        // All three are voters with positional ids.
        assert_eq!(
            v.roster().voter_ids(),
            vec![RegionId(0), RegionId(1), RegionId(2)]
        );
        // name→id round-trips positionally; the roster carries the inverse name.
        assert_eq!(v.name_to_id("eu"), Some(RegionId(1)));
        assert_eq!(v.roster().role_of(RegionId(2)), Some(MemberRole::Voter));
        assert_eq!(
            v.roster().members.iter().find(|m| m.id == 2).unwrap().name,
            "ap"
        );
        // peer_http excludes self (region 0), includes the two siblings.
        assert_eq!(v.peer_http(RegionId(0)), None, "self has no peer entry");
        assert_eq!(
            v.peer_http(RegionId(1)).as_deref(),
            Some("http://eu.svc:9501")
        );
        // peers_named is the two siblings in id order.
        assert_eq!(
            v.peers_named(),
            vec![
                ("eu".to_string(), "http://eu.svc:9501".to_string()),
                ("ap".to_string(), "http://ap.svc:9501".to_string()),
            ]
        );
        // all_regions_for_status: self carries None, peers carry their http.
        let all = v.all_regions_for_status();
        assert_eq!(all[0], (RegionId(0), "us".to_string(), None));
        assert_eq!(
            all[1],
            (
                RegionId(1),
                "eu".to_string(),
                Some("http://eu.svc:9501".to_string())
            )
        );
        // self is a voter in era 0.
        assert_eq!(v.self_role(), Some(MemberRole::Voter));
        // The apply plan's commit voters exclude self (region 0).
        let plan = v.apply_record(&MembershipRecord {
            version: 1,
            term: 0,
            members: vec![
                member(0, "us", MemberRole::Voter),
                member(1, "eu", MemberRole::Voter),
                member(2, "ap", MemberRole::Voter),
            ],
        });
        assert_eq!(plan.commit_voters, vec![ShardId(1), ShardId(2)]);
    }

    /// `apply_record` swaps the roster and the diff reflects exactly the five
    /// surfaces a node must reconfigure. A learner join adds a peer and a commit
    /// learner; a promote turns it into a commit voter with NO peer churn.
    #[test]
    fn apply_record_yields_all_five_surfaces() {
        let v = era0_view(0, &["us", "eu"]); // self = us (region 0)
        // First conf-change: begin the era at v1 with the two existing voters
        // plus a new learner (region 5, a burned-id-respecting join).
        let rec = MembershipRecord {
            version: 1,
            term: 3,
            members: vec![
                member(0, "us", MemberRole::Voter),
                member(1, "eu", MemberRole::Voter),
                member(5, "joiner", MemberRole::Learner),
            ],
        };
        let plan = v.apply_record(&rec);
        assert!(v.era_begun());
        assert_eq!(v.version(), 1);
        // The learner is a new peer (gRPC + ship), a commit learner, NOT a voter.
        assert_eq!(
            plan.added_peers,
            vec![(ShardId(5), "joiner.svc:9500".into())]
        );
        assert!(plan.removed_peers.is_empty());
        assert_eq!(plan.commit_voters, vec![ShardId(1)]); // eu (self us excluded)
        assert_eq!(plan.commit_learners, vec![ShardId(5)]);
        assert_eq!(plan.election_voters, vec![RegionId(1)]);
        assert!(plan.self_is_voter);
        assert_eq!(plan.voter_count, 2);

        // Promote the learner → voter at v2: NO peer churn (same id), but the
        // commit/election voter sets gain region 5.
        let promote = MembershipRecord {
            version: 2,
            term: 3,
            members: vec![
                member(0, "us", MemberRole::Voter),
                member(1, "eu", MemberRole::Voter),
                member(5, "joiner", MemberRole::Voter),
            ],
        };
        let plan = v.apply_record(&promote);
        assert!(plan.added_peers.is_empty(), "a promote moves no peers");
        assert!(plan.removed_peers.is_empty());
        assert_eq!(plan.commit_voters, vec![ShardId(1), ShardId(5)]);
        assert!(plan.commit_learners.is_empty());
        assert_eq!(plan.election_voters, vec![RegionId(1), RegionId(5)]);
        assert_eq!(plan.voter_count, 3);
    }

    /// A remove turns the member into a tombstone: it leaves the live peer set
    /// (`remove_peer`) and the quorum sets, but its name still resolves (to keep
    /// a stale forward from reading "unknown") and its id is burned.
    #[test]
    fn apply_record_remove_retires_the_peer_and_burns_the_id() {
        let v = era0_view(0, &["us", "eu", "ap"]); // self = us
        // Era begins with all three voters at v1.
        v.apply_record(&MembershipRecord {
            version: 1,
            term: 2,
            members: vec![
                member(0, "us", MemberRole::Voter),
                member(1, "eu", MemberRole::Voter),
                member(2, "ap", MemberRole::Voter),
            ],
        });
        // Remove ap (region 2) at v2.
        let plan = v.apply_record(&MembershipRecord {
            version: 2,
            term: 2,
            members: vec![
                member(0, "us", MemberRole::Voter),
                member(1, "eu", MemberRole::Voter),
                member(2, "ap", MemberRole::Removed),
            ],
        });
        assert_eq!(plan.removed_peers, vec![ShardId(2)]);
        assert!(plan.added_peers.is_empty());
        assert_eq!(plan.commit_voters, vec![ShardId(1)]);
        assert_eq!(plan.voter_count, 2);
        // The tombstone name still resolves (stale forwards see "removed", not
        // "unknown"), but it is no longer a live peer.
        assert_eq!(v.name_to_id("ap"), Some(RegionId(2)));
        assert_eq!(v.peer_http(RegionId(2)), None);
        assert_eq!(v.roster().role_of(RegionId(2)), Some(MemberRole::Removed));
    }

    /// `apply_record` is idempotent by version: a replayed older/equal record
    /// never regresses the view, and the plan is a no-op diff.
    #[test]
    fn apply_record_idempotent_by_version() {
        let v = era0_view(0, &["us", "eu"]);
        let rec = MembershipRecord {
            version: 5,
            term: 1,
            members: vec![
                member(0, "us", MemberRole::Voter),
                member(1, "eu", MemberRole::Voter),
            ],
        };
        v.apply_record(&rec);
        assert_eq!(v.version(), 5);
        // Replay an OLDER record (recovery re-stream): the view does not regress.
        let older = MembershipRecord {
            version: 3,
            term: 1,
            members: vec![member(0, "us", MemberRole::Voter)],
        };
        let plan = v.apply_record(&older);
        assert_eq!(v.version(), 5, "an older record never regresses the view");
        assert!(plan.added_peers.is_empty());
        assert!(plan.removed_peers.is_empty());
    }

    /// The capability gate refuses naming exactly the incapable voters; the
    /// leader's own id is never gated.
    #[test]
    fn capability_gate_names_incapable_voters() {
        let roster = Roster {
            version: 1,
            term: 1,
            from_record: true,
            members: vec![
                member(0, "us", MemberRole::Voter),        // leader (self)
                member(1, "eu", MemberRole::Voter),        // capable
                member(2, "ap", MemberRole::Voter),        // incapable
                member(3, "learner", MemberRole::Learner), // not gated (not a voter)
            ],
        };
        let caps = |rid: RegionId| match rid.0 {
            1 => Some(tidal_net::CAP_KIND4_MEMBERSHIP),
            2 => Some(0), // reported, but not capable
            _ => None,
        };
        // Self (region 0) is the leader: never gated. Region 2 is incapable.
        let err = capability_gate(&roster, RegionId(0), caps).unwrap_err();
        assert_eq!(err, vec![RegionId(2)]);

        // Make region 2 capable: the gate passes.
        let caps_ok = |rid: RegionId| match rid.0 {
            1 | 2 => Some(tidal_net::CAP_KIND4_MEMBERSHIP),
            _ => None,
        };
        assert!(capability_gate(&roster, RegionId(0), caps_ok).is_ok());
    }

    /// A voter that has never reported (None) is conservatively incapable.
    #[test]
    fn capability_gate_unreported_voter_is_incapable() {
        let roster = Roster {
            version: 1,
            term: 1,
            from_record: true,
            members: vec![
                member(0, "us", MemberRole::Voter),
                member(1, "eu", MemberRole::Voter),
            ],
        };
        let err = capability_gate(&roster, RegionId(0), |_| None).unwrap_err();
        assert_eq!(err, vec![RegionId(1)]);
    }

    /// Join is idempotent by name; a new join burns the next id past EVERY id
    /// ever assigned, including removed tombstones.
    #[test]
    fn join_id_assignment_burns_tombstones() {
        // Roster with a removed tombstone at the HIGHEST id (7).
        let roster = Roster {
            version: 3,
            term: 2,
            from_record: true,
            members: vec![
                member(0, "us", MemberRole::Voter),
                member(1, "eu", MemberRole::Voter),
                member(7, "gone", MemberRole::Removed),
            ],
        };
        // A re-join from a known member returns its id/role, appends nothing.
        match plan_join(&roster, "eu", "x", "http://x", 4, 2) {
            JoinPlan::Existing { id, role } => {
                assert_eq!(id, 1);
                assert_eq!(role, MemberRole::Voter);
            }
            other => panic!("expected Existing, got {other:?}"),
        }
        // A re-join from a REMOVED member returns the tombstone (idempotent), so
        // a removed node re-contacting the seed does not get a fresh id.
        match plan_join(&roster, "gone", "x", "http://x", 4, 2) {
            JoinPlan::Existing { id, role } => {
                assert_eq!(id, 7);
                assert_eq!(role, MemberRole::Removed);
            }
            other => panic!("expected Existing tombstone, got {other:?}"),
        }
        // A NEW name burns id 8 (past the tombstone at 7), Learner role.
        match plan_join(&roster, "newcomer", "n.svc:9500", "http://n.svc:9501", 4, 2) {
            JoinPlan::Append { id, record } => {
                assert_eq!(id, 8, "the next id is max-ever+1, burning the tombstone");
                assert_eq!(record.version, 4);
                assert_eq!(record.term, 2);
                let added = record.members.iter().find(|m| m.id == 8).unwrap();
                assert_eq!(added.name, "newcomer");
                assert_eq!(added.role, MemberRole::Learner);
                // Records are full snapshots: every prior member survives.
                assert_eq!(record.members.len(), 4);
            }
            other => panic!("expected Append, got {other:?}"),
        }
    }

    /// Remove is idempotent and produces a full-roster tombstone record.
    #[test]
    fn remove_is_idempotent_and_tombstones() {
        let roster = Roster {
            version: 2,
            term: 1,
            from_record: true,
            members: vec![
                member(0, "us", MemberRole::Voter),
                member(1, "eu", MemberRole::Voter),
            ],
        };
        let rec = plan_remove(&roster, "eu", 3, 1).unwrap();
        assert_eq!(rec.version, 3);
        let eu = rec.members.iter().find(|m| m.name == "eu").unwrap();
        assert_eq!(eu.role, MemberRole::Removed, "eu became a tombstone");
        // Removing an unknown name is a no-op.
        assert!(plan_remove(&roster, "ghost", 3, 1).is_none());
        // Removing an already-removed member is a no-op (idempotent).
        let after = Roster {
            members: rec.members.clone(),
            ..roster
        };
        assert!(plan_remove(&after, "eu", 4, 1).is_none());
    }

    /// `plan_promote` flips exactly the named learner to Voter (full-roster
    /// record), is idempotent on a voter/tombstone, and refuses an unknown id.
    #[test]
    fn promote_flips_one_learner_and_is_idempotent() {
        let roster = Roster {
            version: 4,
            term: 2,
            from_record: true,
            members: vec![
                member(0, "us", MemberRole::Voter),
                member(1, "eu", MemberRole::Voter),
                member(5, "joiner", MemberRole::Learner),
                member(7, "gone", MemberRole::Removed),
            ],
        };
        // Promote the learner (id 5) → exactly one role flips, version advances.
        let rec = plan_promote(&roster, 5, 5, 2).unwrap();
        assert_eq!(rec.version, 5);
        let joiner = rec.members.iter().find(|m| m.id == 5).unwrap();
        assert_eq!(joiner.role, MemberRole::Voter, "the learner became a voter");
        // Every other member is unchanged (full-roster snapshot).
        assert_eq!(
            rec.members.iter().find(|m| m.id == 0).unwrap().role,
            MemberRole::Voter
        );
        assert_eq!(
            rec.members.iter().find(|m| m.id == 7).unwrap().role,
            MemberRole::Removed
        );
        // Promoting an already-voter / a tombstone / an unknown id is a no-op.
        assert!(
            plan_promote(&roster, 0, 5, 2).is_none(),
            "voter not promotable"
        );
        assert!(
            plan_promote(&roster, 7, 5, 2).is_none(),
            "tombstone not promotable"
        );
        assert!(
            plan_promote(&roster, 99, 5, 2).is_none(),
            "unknown id not promotable"
        );
    }

    /// The activation re-append carries the CURRENT roster at version+1 stamped
    /// with the new term (§3.2 — the no-op-entry analogue).
    #[test]
    fn activation_reappend_is_current_roster_at_next_version() {
        let roster = Roster {
            version: 9,
            term: 4,
            from_record: true,
            members: vec![
                member(0, "us", MemberRole::Voter),
                member(1, "eu", MemberRole::Learner),
            ],
        };
        let rec = plan_activation_reappend(&roster, 5);
        assert_eq!(rec.version, 10, "version advances by one");
        assert_eq!(rec.term, 5, "stamped with the NEW term");
        assert_eq!(rec.members, roster.members, "the roster is unchanged");
    }

    /// A view built from a record (membership-era boot) derives consistent
    /// tables; a removed-self boot reports Some(Removed) (the readiness 503 path).
    #[test]
    fn from_record_membership_era_boot() {
        let v = MembershipView::from_record(
            RegionId(5),
            4,
            3,
            vec![
                member(0, "us", MemberRole::Voter),
                member(5, "me", MemberRole::Learner),
            ],
        );
        assert!(v.era_begun());
        assert_eq!(v.version(), 4);
        assert_eq!(v.self_role(), Some(MemberRole::Learner));
        // Region 0 is the lone voter (region 5, self, is a learner) and a peer.
        assert_eq!(v.roster().voter_ids(), vec![RegionId(0)]);
        assert_eq!(
            v.peer_http(RegionId(0)).as_deref(),
            Some("http://us.svc:9501")
        );
    }
}