openraft 0.10.0-alpha.18

use std::cmp::Ordering;
use std::fmt::Debug;
use std::fmt::Display;

use openraft_macros::since;

use crate::base::OptionalFeatures;
use crate::type_config::alias::LeaderCommitted;
use crate::type_config::alias::LeaderNodeId;
use crate::type_config::alias::LeaderTerm;
use crate::vote::RaftLeaderId;
use crate::vote::Vote;
use crate::vote::committed::CommittedVote;
use crate::vote::leader_id::raft_leader_id::RaftLeaderIdExt;
use crate::vote::non_committed::UncommittedVote;
use crate::vote::ref_vote::RefVote;
use crate::vote::vote_status::VoteStatus;

/// Represents a vote in Raft consensus, including both votes for leader candidates
/// and committed leader (a leader granted by a quorum).
///
/// The associated type `LeaderId` identifies the leader this vote is for.
/// It is decoupled from [`RaftTypeConfig`](crate::RaftTypeConfig) to break
/// circular type dependencies.
#[since(
    version = "0.10.0",
    change = "from `RaftVote<C>` to `RaftVote` with associated type `LeaderId`"
)]
#[since(version = "0.10.0")]
pub trait RaftVote
where Self: OptionalFeatures + Eq + Clone + Debug + Display + 'static
{
    /// The leader ID type that this vote wraps.
    #[since(version = "0.10.0")]
    type LeaderId: RaftLeaderId;

    /// Create a new vote for the specified leader with optional quorum commitment.
    #[since(version = "0.10.0")]
    fn from_leader_id(leader_id: Self::LeaderId, committed: bool) -> Self;

    /// Get a reference to this vote's LeaderId([`RaftLeaderId`] implementation).
    #[since(version = "0.10.0", change = "non-Option return")]
    #[since(version = "0.10.0")]
    fn leader_id(&self) -> &Self::LeaderId;

    /// Whether this vote has been committed by a quorum.
    #[since(version = "0.10.0")]
    fn is_committed(&self) -> bool;

    /// Convert to the openraft-provided [`Vote`] struct.
    ///
    /// This creates an owned [`Vote<LID>`] from any vote implementation.
    /// [`Vote`] is openraft's standard vote representation, which differs from
    /// user-defined vote types that may be optimized for storage.
    #[since(version = "0.10.0")]
    fn to_owned_vote(&self) -> Vote<Self::LeaderId> {
        Vote::from_leader_id(self.leader_id().clone(), self.is_committed())
    }

    /// Compare this vote with another vote by partial order.
    ///
    /// Returns `Some(Ordering)` if the votes are comparable, `None` if incomparable.
    /// Votes are compared first by leader_id, then by committed status.
    /// When leader IDs are incomparable, committed votes are considered greater
    /// than uncommitted ones to minimize election conflicts.
    #[since(version = "0.10.0")]
    fn partial_cmp<V>(&self, other: &V) -> Option<Ordering>
    where V: RaftVote<LeaderId = Self::LeaderId> {
        let self_ref: RefVote<'_, Self::LeaderId> = RefVote::new(self.leader_id(), self.is_committed());
        let other_ref: RefVote<'_, Self::LeaderId> = RefVote::new(other.leader_id(), other.is_committed());
        PartialOrd::partial_cmp(&self_ref, &other_ref)
    }
}

pub(crate) trait RaftVoteExt: RaftVote {
    #[allow(dead_code)]
    fn new_with_default_term(node_id: LeaderNodeId<Self::LeaderId>) -> Self {
        let leader_id = Self::LeaderId::new_with_default_term(node_id);
        Self::from_leader_id(leader_id, false)
    }

    /// Creates a new vote for a node in a specific term, with uncommitted status.
    fn from_term_node_id(term: LeaderTerm<Self::LeaderId>, node_id: LeaderNodeId<Self::LeaderId>) -> Self {
        let leader_id = Self::LeaderId::new(term, node_id);
        Self::from_leader_id(leader_id, false)
    }

    fn from_term_node_id_committed(
        term: LeaderTerm<Self::LeaderId>,
        node_id: LeaderNodeId<Self::LeaderId>,
        committed: bool,
    ) -> Self {
        let leader_id = Self::LeaderId::new(term, node_id);
        Self::from_leader_id(leader_id, committed)
    }

    fn term(&self) -> LeaderTerm<Self::LeaderId> {
        self.leader_id().term()
    }

    /// Gets the node ID of the leader this vote is for.
    fn to_leader_node_id(&self) -> LeaderNodeId<Self::LeaderId> {
        self.leader_node_id().clone()
    }

    /// Gets a reference to the node ID of the leader this vote is for.
    fn leader_node_id(&self) -> &LeaderNodeId<Self::LeaderId> {
        self.leader_id().node_id()
    }

    /// Gets the leader ID this vote is associated with.
    fn to_leader_id(&self) -> Self::LeaderId {
        self.leader_id().clone()
    }

    /// Creates a reference view of this vote.
    ///
    /// Returns a lightweight `RefVote` that borrows the data from this vote.
    fn as_ref_vote(&self) -> RefVote<'_, Self::LeaderId> {
        RefVote::new(self.leader_id(), self.is_committed())
    }

    /// Create a [`CommittedVote`] with the same leader id.
    fn to_committed(&self) -> CommittedVote<Self::LeaderId> {
        CommittedVote::new(self.to_leader_id())
    }

    /// Create a [`UncommittedVote`] with the same leader id.
    fn to_non_committed(&self) -> UncommittedVote<Self::LeaderId> {
        UncommittedVote::new(self.to_leader_id())
    }

    /// Convert this vote into a [`CommittedVote`]
    fn into_committed(self) -> CommittedVote<Self::LeaderId> {
        CommittedVote::new(self.to_leader_id())
    }

    /// Convert this vote into a [`UncommittedVote`]
    fn into_non_committed(self) -> UncommittedVote<Self::LeaderId> {
        UncommittedVote::new(self.to_leader_id())
    }

    /// Converts this vote into a [`VoteStatus`] enum based on its commitment state.
    fn into_vote_status(self) -> VoteStatus<Self::LeaderId> {
        if self.is_committed() {
            VoteStatus::Committed(self.into_committed())
        } else {
            VoteStatus::Pending(self.into_non_committed())
        }
    }

    /// Converts this vote to a [`CommittedVote`] if it is committed.
    ///
    /// Returns `Some(CommittedVote)` if the vote is committed, otherwise returns `None`.
    #[allow(dead_code)]
    fn try_to_committed(&self) -> Option<CommittedVote<Self::LeaderId>> {
        if self.is_committed() {
            Some(self.to_committed())
        } else {
            None
        }
    }

    /// Returns the leader ID as `CommittedLeaderId` if this vote is committed.
    ///
    /// Returns `Some(CommittedLeaderId)` if the vote is committed and has a leader ID.
    /// Returns `None` if the vote is not committed or has no leader ID.
    #[allow(dead_code)]
    fn try_to_committed_leader_id(&self) -> Option<LeaderCommitted<Self::LeaderId>> {
        if self.is_committed() {
            Some(self.leader_id().to_committed())
        } else {
            None
        }
    }

    /// Checks if this vote is for the same leader as specified by the given committed leader ID.
    fn is_same_leader(&self, leader_id: &LeaderCommitted<Self::LeaderId>) -> bool {
        self.leader_id().to_committed() == *leader_id
    }
}

impl<T> RaftVoteExt for T where T: RaftVote {}

#[cfg(test)]
mod tests {
    use std::cmp::Ordering;
    use std::fmt;

    use crate::Vote;
    use crate::engine::testing::UTLeaderId;
    use crate::vote::RaftVote;
    use crate::vote::raft_vote::RaftVoteExt;

    #[derive(Debug, Clone, PartialEq, Eq)]
    #[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize))]
    struct UserVote {
        leader_id: UTLeaderId,
        committed: bool,
    }

    impl fmt::Display for UserVote {
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
            write!(f, "<{}:{}>", self.leader_id, if self.committed { "Q" } else { "-" })
        }
    }

    impl RaftVote for UserVote {
        type LeaderId = UTLeaderId;

        fn from_leader_id(leader_id: Self::LeaderId, committed: bool) -> Self {
            Self { leader_id, committed }
        }

        fn leader_id(&self) -> &Self::LeaderId {
            &self.leader_id
        }

        fn is_committed(&self) -> bool {
            self.committed
        }
    }

    #[test]
    fn test_partial_cmp() {
        // Same vote: Equal
        let v1 = Vote::<UTLeaderId>::new(1, 2);
        let v2 = Vote::<UTLeaderId>::new(1, 2);
        assert_eq!(RaftVote::partial_cmp(&v1, &v2), Some(Ordering::Equal));

        // Different terms: comparable
        let v_lower = Vote::<UTLeaderId>::new(1, 2);
        let v_higher = Vote::<UTLeaderId>::new(2, 2);
        assert_eq!(RaftVote::partial_cmp(&v_lower, &v_higher), Some(Ordering::Less));
        assert_eq!(RaftVote::partial_cmp(&v_higher, &v_lower), Some(Ordering::Greater));

        // Same leader_id, committed > uncommitted
        let uncommitted = Vote::<UTLeaderId>::new(1, 2);
        let committed = Vote::<UTLeaderId>::new_committed(1, 2);
        assert_eq!(RaftVote::partial_cmp(&uncommitted, &committed), Some(Ordering::Less));
        assert_eq!(RaftVote::partial_cmp(&committed, &uncommitted), Some(Ordering::Greater));

        // Compare Vote with CommittedVote type
        let vote = Vote::<UTLeaderId>::new(1, 2);
        let committed_vote = vote.to_committed();
        assert_eq!(RaftVote::partial_cmp(&vote, &committed_vote), Some(Ordering::Less));
        assert_eq!(RaftVote::partial_cmp(&committed_vote, &vote), Some(Ordering::Greater));

        // Compare Vote with UncommittedVote type
        let vote = Vote::<UTLeaderId>::new(1, 2);
        let uncommitted_vote = vote.to_non_committed();
        assert_eq!(RaftVote::partial_cmp(&vote, &uncommitted_vote), Some(Ordering::Equal));
        assert_eq!(RaftVote::partial_cmp(&uncommitted_vote, &vote), Some(Ordering::Equal));
    }

    #[test]
    fn test_to_owned_vote() {
        // From Vote
        let vote = Vote::<UTLeaderId>::new(1, 2);
        let owned = vote.to_owned_vote();
        assert_eq!(owned, vote);

        let committed_vote = Vote::<UTLeaderId>::new_committed(3, 4);
        let owned = committed_vote.to_owned_vote();
        assert_eq!(owned, committed_vote);

        // From CommittedVote
        let committed = vote.to_committed();
        let owned = committed.to_owned_vote();
        assert_eq!(owned.leader_id, vote.leader_id);
        assert!(owned.is_committed());

        // From UncommittedVote
        let uncommitted = committed_vote.to_non_committed();
        let owned = uncommitted.to_owned_vote();
        assert_eq!(owned.leader_id, committed_vote.leader_id);
        assert!(!owned.is_committed());
    }

    #[test]
    fn test_into_vote_generic() {
        let committed = Vote::<UTLeaderId>::new_committed(7, 9).to_committed();
        let committed_user: UserVote = committed.into_vote();
        assert!(committed_user.committed);
        assert_eq!(committed_user.leader_id, Vote::<UTLeaderId>::new(7, 9).leader_id);

        let uncommitted = Vote::<UTLeaderId>::new(8, 10).to_non_committed();
        let uncommitted_user: UserVote = uncommitted.into_vote();
        assert!(!uncommitted_user.committed);
        assert_eq!(uncommitted_user.leader_id, Vote::<UTLeaderId>::new(8, 10).leader_id);
    }
}