datum-cluster 0.10.2

use std::{collections::BTreeMap, net::SocketAddr, time::SystemTime};

/// Akka-style member lifecycle state used by Datum membership.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum MemberState {
    /// The member has been discovered but is not yet considered available.
    Joining,
    /// The member is available for future control-plane decisions.
    Up,
    /// The member requested a graceful leave.
    Leaving,
    /// The member is exiting after a graceful leave was acknowledged.
    Exiting,
    /// The member was downed after failure detection/downing.
    Down,
    /// The member has been removed from the active membership view.
    Removed,
}

/// Datum-owned member record. No foca types appear in this model.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Member {
    /// Stable logical node id.
    pub node_id: String,
    /// Gossip address peers use for this member.
    pub address: SocketAddr,
    /// DCP agent address peers use for node-to-node control, if advertised.
    pub agent_addr: Option<SocketAddr>,
    /// Member roles advertised by the node.
    pub roles: Vec<String>,
    /// Current membership state.
    pub state: MemberState,
    /// True after foca suspicion and before reachability is restored or the
    /// member is downed/removed.
    pub unreachable: bool,
    /// Monotonic incarnation used to distinguish restarts with the same node id
    /// and address.
    pub incarnation: u64,
    /// Last state update time.
    pub seen_at: SystemTime,
    /// First time this member was marked unreachable, if currently unreachable.
    pub unreachable_since: Option<SystemTime>,
}

impl Member {
    #[must_use]
    pub(crate) fn new(
        node_id: String,
        address: SocketAddr,
        agent_addr: Option<SocketAddr>,
        roles: Vec<String>,
        state: MemberState,
        incarnation: u64,
        now: SystemTime,
    ) -> Self {
        Self {
            node_id,
            address,
            agent_addr,
            roles,
            state,
            unreachable: false,
            incarnation,
            seen_at: now,
            unreachable_since: None,
        }
    }

    /// Return `true` when the member has `role`.
    #[must_use]
    pub fn has_role(&self, role: &str) -> bool {
        self.roles.iter().any(|candidate| candidate == role)
    }
}

/// Immutable current membership snapshot.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ClusterState {
    /// Local node id for the node that produced this view.
    pub self_node: String,
    /// Members keyed by logical node id.
    pub members: BTreeMap<String, Member>,
}

impl ClusterState {
    #[must_use]
    pub(crate) fn new(self_member: Member) -> Self {
        let self_node = self_member.node_id.clone();
        let mut members = BTreeMap::new();
        members.insert(self_member.node_id.clone(), self_member);
        Self { self_node, members }
    }

    /// Return a member by node id.
    #[must_use]
    pub fn member(&self, node_id: &str) -> Option<&Member> {
        self.members.get(node_id)
    }

    /// Return all members currently in `Up` state and reachable.
    #[must_use]
    pub fn up_members(&self) -> Vec<&Member> {
        self.members
            .values()
            .filter(|member| member.state == MemberState::Up && !member.unreachable)
            .collect()
    }

    /// Return the deterministic placement coordinator for this membership view.
    ///
    /// Datum v0.10 mirrors Akka's "oldest member" coordinator convention without
    /// adding a consensus protocol. C1 does not assign Akka-style up numbers, so
    /// the ordering key is the propagated member incarnation, then node id. Every
    /// node computes the same answer from the same C1 view. Under a partition and
    /// timeout downing, different views can transiently produce different
    /// coordinators; quorum or lease fencing is a v0.11+ follow-up.
    #[must_use]
    pub fn placement_coordinator(&self) -> Option<&Member> {
        self.up_members().into_iter().min_by(|left, right| {
            left.incarnation
                .cmp(&right.incarnation)
                .then_with(|| left.node_id.cmp(&right.node_id))
        })
    }

    /// Return `true` when `node_id` is the coordinator in this view.
    #[must_use]
    pub fn is_placement_coordinator(&self, node_id: &str) -> bool {
        self.placement_coordinator()
            .is_some_and(|member| member.node_id == node_id)
    }
}

/// Ordered member event.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemberEvent {
    /// Monotonic sequence assigned by the membership actor.
    pub sequence: u64,
    /// Event kind.
    pub kind: MemberEventKind,
    /// Member snapshot after the event.
    pub member: Member,
    /// Event publication time.
    pub at: SystemTime,
}

impl MemberEvent {
    #[must_use]
    pub(crate) const fn new(
        sequence: u64,
        kind: MemberEventKind,
        member: Member,
        at: SystemTime,
    ) -> Self {
        Self {
            sequence,
            kind,
            member,
            at,
        }
    }
}

/// Kind of [`MemberEvent`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum MemberEventKind {
    /// Initial event held by the lossless subscription before the actor starts.
    Initialized,
    /// Member entered `Joining`.
    MemberJoining,
    /// Member entered `Up`.
    MemberUp,
    /// Member roles or advertised address changed without a state change.
    MemberUpdated,
    /// Member was marked unreachable by foca suspicion.
    MemberUnreachable,
    /// Member became reachable again.
    MemberReachable,
    /// Member entered `Leaving`.
    MemberLeaving,
    /// Member entered `Exiting`.
    MemberExiting,
    /// Member entered `Down`.
    MemberDown,
    /// Member entered `Removed`.
    MemberRemoved,
    /// A down/removed logical node returned with a newer incarnation.
    MemberRejoined,
}