//! Votes on remakes: a remake commits only once a majority of its electorate voted for
//! it.
//!
//! The electorate is fixed in the proposal: the members that renewed their alive stamp
//! within the coordinator's `--electorate-window-days` (`swarm::members::electorate`),
//! the coordinator always included.
//!
//! Votes live in the votes register (`_continuum/votes/<pk>`, written only by that
//! member): one entry per member for the whole epoch (the snapshot leaves it behind).
//! A member votes for at most one remake of its epoch, the first it sees (the oldest
//! proposal, if it sees several at once), and votes again only once that remake is
//! aborted. Neither expiry nor a deleted record frees a vote. Two competing remakes,
//! even in different partitions, can't both reach a majority, so at most one commits.
//!
//! A node votes once it is ready, on its current epoch, only if it is in the electorate,
//! and only once it holds the remake's content (`content`), so a committed remake has
//! its content on a majority.
//!
//! A vote can be lost: a node whose vote reached only the coordinator restarts and
//! resyncs from nodes that never saw it. Should it vote again for a competitor, its
//! entry ends up with two concurrent values, and it counts for no remake. To avoid
//! that, a node doesn't vote while more than one undecided remake it could have voted
//! for is older than its start, and with one, votes only for that one.

use std::collections::HashSet;

use tracing::{info, warn};

use super::Host;
use super::content;
use super::records::{self, Decision, EpochRecord, Phase, Record, Register, Vote, unix_millis};
use crate::document::store::StoreError;

/// A majority of `electorate` members.
pub(crate) fn quorum(electorate: usize) -> usize {
    electorate / 2 + 1
}

/// Where a remake stands with its electorate.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct Tally {
    pub(crate) electorate: usize,
    pub(crate) quorum: usize,
    /// Voted for it.
    pub(crate) yes: usize,
    /// Voted for a competing remake that isn't aborted.
    pub(crate) elsewhere: usize,
    /// Voted for two remakes: counts for neither.
    pub(crate) twice: usize,
    /// Not voted (yet).
    pub(crate) pending: usize,
}

impl Tally {
    pub(crate) fn reached(&self) -> bool {
        self.yes >= self.quorum
    }

    /// Even if every pending member votes for it, it falls short.
    pub(crate) fn out_of_reach(&self) -> bool {
        self.yes + self.pending < self.quorum
    }
}

impl std::fmt::Display for Tally {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "{} of {} voted for it, {} for a competing remake, ",
            self.yes, self.electorate, self.elsewhere
        )?;
        if self.twice > 0 {
            write!(f, "{} for two, ", self.twice)?;
        }
        write!(f, "{} not yet; {} needed", self.pending, self.quorum)
    }
}

/// What a member's entry in the register amounts to.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Held<'a> {
    /// No vote, or only for aborted remakes.
    Free,
    For(&'a str),
    /// For two remakes, neither aborted.
    Twice,
}

/// What `votes` (one member's entry) amount to among `all` records.
pub(crate) fn held<'a>(votes: &'a [Vote], all: &[(String, Record)]) -> Held<'a> {
    let aborted = |remake: &str| {
        all.iter().any(|(id, record)| {
            id == remake && matches!(record.decision, Some(Decision::Aborted { .. }))
        })
    };
    let mut kept = votes
        .iter()
        .map(|vote| vote.remake.as_str())
        .filter(|remake| !aborted(remake));
    let Some(first) = kept.next() else {
        return Held::Free;
    };
    if kept.all(|other| other == first) {
        Held::For(first)
    } else {
        Held::Twice
    }
}

fn held_by<'a>(member: &str, register: &'a Register, all: &[(String, Record)]) -> Held<'a> {
    register
        .get(member)
        .map_or(Held::Free, |votes| held(votes, all))
}

/// The tally of remake `id` among `all` records.
pub(crate) fn tally(
    id: &str,
    record: &Record,
    all: &[(String, Record)],
    register: &Register,
) -> Tally {
    let electorate: HashSet<&str> = record
        .proposal
        .electorate
        .iter()
        .map(String::as_str)
        .collect();
    let (mut yes, mut elsewhere, mut twice) = (0, 0, 0);
    for member in &electorate {
        match held_by(member, register, all) {
            Held::For(remake) if remake == id => yes += 1,
            Held::For(_) => elsewhere += 1,
            Held::Twice => twice += 1,
            Held::Free => {}
        }
    }
    Tally {
        electorate: electorate.len(),
        quorum: record.proposal.quorum,
        yes,
        elsewhere,
        twice,
        pending: electorate.len() - yes - elsewhere - twice,
    }
}

/// What `me` may vote for in `epoch`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum Freedom {
    Any,
    /// Only this remake: the one it may have voted for before its start.
    Only(String),
    /// Nothing, for this reason.
    Held(String),
}

/// What `me` may vote for in `epoch`. Remakes proposed before `started` (this node's
/// start, plus a margin for clocks) may hold a vote of ours that we lost in a restart.
pub(crate) fn freedom(
    me: &str,
    epoch: &EpochRecord,
    all: &[(String, Record)],
    register: &Register,
    started: u64,
) -> Freedom {
    match held_by(me, register, all) {
        Held::Free => {}
        Held::For(remake) => {
            return Freedom::Held(format!("this node voted for remake {remake}"));
        }
        Held::Twice => {
            return Freedom::Held("this node voted for two remakes".to_owned());
        }
    }
    let mut older = all
        .iter()
        .filter(|(_, record)| {
            &record.proposal.from_epoch == epoch
                && record.decision.is_none()
                && record.proposal.proposed_at <= started
                && record.proposal.electorate.iter().any(|member| member == me)
        })
        .map(|(id, _)| id);
    match (older.next(), older.next()) {
        (None, _) => Freedom::Any,
        (Some(id), None) => Freedom::Only(id.clone()),
        (Some(_), Some(_)) => Freedom::Held(
            "several undecided remakes are older than this node's start; it may have voted \
             for one before"
                .to_owned(),
        ),
    }
}

/// The remake `me` should vote for now, if any: the oldest proposed one of `epoch`
/// whose electorate includes it, as far as it is free to vote (`freedom`).
pub(crate) fn choice(
    me: &str,
    epoch: &EpochRecord,
    all: &[(String, Record)],
    register: &Register,
    started: u64,
    now: u64,
    grace: u64,
) -> Option<String> {
    let only = match freedom(me, epoch, all, register, started) {
        Freedom::Any => None,
        Freedom::Only(id) => Some(id),
        Freedom::Held(_) => return None,
    };
    all.iter()
        .filter(|(id, record)| {
            &record.proposal.from_epoch == epoch
                && record.phase(now, grace) == Phase::Proposed
                && record.proposal.electorate.iter().any(|member| member == me)
                && only.as_ref().is_none_or(|only| only == id)
        })
        .min_by(|(a_id, a), (b_id, b)| {
            (a.proposal.proposed_at, a_id).cmp(&(b.proposal.proposed_at, b_id))
        })
        .map(|(id, _)| id.clone())
}

/// Votes for a remake of the local epoch if this node should. Called by the remake
/// worker whenever the records change (and on its timer).
pub(super) async fn cast<H: Host>(host: &H) {
    if !host.is_node_ready().await {
        return;
    }
    let me = host.node_id();
    let epoch = EpochRecord::from(host.current_epoch());
    let grace = host.dormant_after().as_millis() as u64;
    let started = host.remake().started_at() + grace;

    // Usually there is nothing to vote on: look first, without writing.
    let (voter, wanted) = (me.clone(), epoch.clone());
    let due = host
        .read_doc(move |doc| {
            let all = records::load_all(doc);
            let register = records::load_register(doc);
            let id = choice(
                &voter,
                &wanted,
                &all,
                &register,
                started,
                unix_millis(),
                grace,
            )?;
            let content = all
                .into_iter()
                .find(|(other, _)| *other == id)
                .and_then(|(_, record)| record.proposal.content);
            Some((id, content))
        })
        .await;
    let Some((chosen, content)) = due else {
        return;
    };
    // Only with the content in hand: a committed remake must have it on a majority.
    if let Some(reference) = &content {
        if content::available_or_fetch(host, reference).is_none() {
            return;
        }
    }

    // Decided again under the write lock, so two passes can't both vote.
    let voter = me.clone();
    let result = host
        .commit_doc(move |doc| -> Result<Option<String>, StoreError> {
            let now = unix_millis();
            let all = records::load_all(doc);
            let register = records::load_register(doc);
            let Some(id) = choice(&voter, &epoch, &all, &register, started, now, grace)
                .filter(|id| *id == chosen)
            else {
                return Ok(None);
            };
            let vote = Vote {
                remake: id.clone(),
                at: now,
            };
            records::write_vote(doc, &voter, &vote)?;
            Ok(Some(id))
        })
        .await;
    match result {
        Ok(Some(id)) => info!(remake = %id, "Voted for remake"),
        Ok(None) => {}
        Err(error) => warn!(error = %format!("{error:#}"), "Failed to vote on a remake"),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::document::replica::Epoch;
    use crate::remake::records::{Policy, Proposal};
    use serde_json::Value as JsonValue;

    fn record(proposed_at: u64, electorate: &[&str]) -> Record {
        Record {
            proposal: Proposal {
                coordinator: electorate[0].to_owned(),
                from_epoch: Epoch::GENESIS.into(),
                base_heads: Vec::new(),
                policy: Policy {
                    replay_window: Policy::default_replay_window(),
                    archive_retention: Policy::default_archive_retention(),
                },
                settle_ms: 1_000,
                proposed_at,
                settle_until: proposed_at + 1_000,
                switch_delay_ms: None,
                electorate: electorate.iter().map(|m| (*m).to_owned()).collect(),
                quorum: quorum(electorate.len()),
                kind: "compact".to_owned(),
                content: None,
                translation: JsonValue::Null,
            },
            decision: None,
            abort_requested: None,
            contested: false,
            votes: Default::default(),
        }
    }

    /// A register from (voter, remakes voted for) pairs.
    fn register(votes: &[(&str, &[&str])]) -> Register {
        votes
            .iter()
            .map(|(voter, remakes)| {
                let votes = remakes
                    .iter()
                    .map(|remake| Vote {
                        remake: (*remake).to_owned(),
                        at: 0,
                    })
                    .collect();
                ((*voter).to_owned(), votes)
            })
            .collect()
    }

    fn aborted(mut record: Record) -> Record {
        record.decision = Some(Decision::Aborted {
            reason: "test".into(),
            decided_at: 150,
        });
        record
    }

    const ALL: [&str; 5] = ["a", "b", "c", "d", "e"];
    /// Long before any proposal: no restart concerns.
    const STARTED: u64 = 0;

    fn genesis() -> EpochRecord {
        Epoch::GENESIS.into()
    }

    #[test]
    fn majorities() {
        assert_eq!(quorum(1), 1);
        assert_eq!(quorum(2), 2);
        assert_eq!(quorum(3), 2);
        assert_eq!(quorum(4), 3);
        assert_eq!(quorum(250), 126);
    }

    #[test]
    fn competing_remakes_split_the_vote() {
        let all = vec![
            ("x".to_owned(), record(100, &ALL)),
            ("y".to_owned(), record(200, &ALL)),
        ];
        // z isn't in the electorate.
        let votes = register(&[("a", &["x"]), ("b", &["x"]), ("c", &["y"]), ("z", &["y"])]);
        let x = tally("x", &all[0].1, &all, &votes);
        assert_eq!((x.yes, x.elsewhere, x.pending, x.quorum), (2, 1, 2, 3));
        assert!(!x.reached() && !x.out_of_reach());

        let y = tally("y", &all[1].1, &all, &votes);
        assert_eq!((y.yes, y.elsewhere, y.pending), (1, 2, 2));
        assert!(!y.out_of_reach());

        // d votes for x: x has its majority, y can't get one any more.
        let votes = register(&[("a", &["x"]), ("b", &["x"]), ("c", &["y"]), ("d", &["x"])]);
        assert!(tally("x", &all[0].1, &all, &votes).reached());
        assert!(tally("y", &all[1].1, &all, &votes).out_of_reach());
    }

    #[test]
    fn votes_for_aborted_remakes_are_void() {
        let all = vec![
            ("x".to_owned(), aborted(record(100, &ALL))),
            ("y".to_owned(), record(200, &ALL)),
        ];
        let votes = register(&[("a", &["x"]), ("b", &["x"]), ("c", &["x"]), ("d", &["y"])]);
        let y = tally("y", &all[1].1, &all, &votes);
        assert_eq!((y.yes, y.elsewhere, y.pending), (1, 0, 4));
        // c may vote again.
        assert_eq!(
            choice("c", &genesis(), &all, &votes, STARTED, 250, 0).as_deref(),
            Some("y")
        );
    }

    #[test]
    fn expiry_frees_no_vote() {
        // x's coordinator is gone: x expired, but a, b and c still hold their votes.
        let all = vec![
            ("x".to_owned(), record(100, &ALL)),
            ("y".to_owned(), record(5_000, &ALL)),
        ];
        let votes = register(&[("a", &["x"]), ("b", &["x"]), ("c", &["x"])]);
        assert_eq!(all[0].1.phase(5_000, 0), Phase::Expired);
        assert_eq!(
            choice("a", &genesis(), &all, &votes, STARTED, 5_000, 0),
            None
        );
        let y = tally("y", &all[1].1, &all, &votes);
        assert_eq!((y.yes, y.elsewhere, y.pending), (0, 3, 2));
        assert!(y.out_of_reach());
    }

    #[test]
    fn voting_twice_counts_for_neither() {
        let all = vec![
            ("x".to_owned(), record(100, &ALL)),
            ("y".to_owned(), record(200, &ALL)),
        ];
        let votes = register(&[("a", &["x", "y"]), ("b", &["x"]), ("c", &["x", "x"])]);
        let x = tally("x", &all[0].1, &all, &votes);
        assert_eq!((x.yes, x.elsewhere, x.twice, x.pending), (2, 0, 1, 2));
        assert_eq!(
            x.to_string(),
            "2 of 5 voted for it, 0 for a competing remake, 1 for two, 2 not yet; 3 needed"
        );
        assert!(matches!(
            freedom("a", &genesis(), &all, &votes, STARTED),
            Freedom::Held(_)
        ));

        // Once y is aborted, a's vote for x stands.
        let all = vec![
            ("x".to_owned(), record(100, &ALL)),
            ("y".to_owned(), aborted(record(200, &ALL))),
        ];
        assert_eq!(tally("x", &all[0].1, &all, &votes).yes, 3);
    }

    #[test]
    fn one_vote_per_epoch_for_the_oldest() {
        let all = vec![
            ("y".to_owned(), record(200, &ALL)),
            ("x".to_owned(), record(100, &ALL)),
        ];
        let none = register(&[]);
        assert_eq!(
            choice("a", &genesis(), &all, &none, STARTED, 150, 0).as_deref(),
            Some("x")
        );
        assert_eq!(
            choice("outsider", &genesis(), &all, &none, STARTED, 150, 0),
            None
        );

        let voted = register(&[("a", &["y"])]);
        assert_eq!(choice("a", &genesis(), &all, &voted, STARTED, 150, 0), None);
    }

    #[test]
    fn after_a_start_only_the_remake_voted_for_before() {
        let started = 150;
        // Proposed before the start: a may have voted for x then.
        let all = vec![
            ("x".to_owned(), record(100, &ALL)),
            ("y".to_owned(), record(200, &ALL)),
        ];
        let none = register(&[]);
        assert_eq!(
            freedom("a", &genesis(), &all, &none, started),
            Freedom::Only("x".to_owned())
        );
        assert_eq!(
            choice("a", &genesis(), &all, &none, started, 250, 1_000).as_deref(),
            Some("x")
        );

        // Two of them: a may have voted for either, so for none.
        let all = vec![
            ("w".to_owned(), record(50, &ALL)),
            ("x".to_owned(), record(100, &ALL)),
            ("y".to_owned(), record(200, &ALL)),
        ];
        assert!(matches!(
            freedom("a", &genesis(), &all, &none, started),
            Freedom::Held(_)
        ));
        assert_eq!(
            choice("a", &genesis(), &all, &none, started, 250, 1_000),
            None
        );

        // Once they are decided, a is free again; it couldn't vote for an outsider's.
        let all = vec![
            ("w".to_owned(), aborted(record(50, &ALL))),
            ("x".to_owned(), aborted(record(100, &ALL))),
            ("y".to_owned(), record(200, &ALL)),
            ("z".to_owned(), record(120, &["b", "c"])),
        ];
        assert_eq!(freedom("a", &genesis(), &all, &none, started), Freedom::Any);
    }
}