//! 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. Each member votes for at most one open remake of
//! its epoch: the first it sees (the oldest proposal, if it sees several at once). Two
//! competing remakes, even in different partitions, can't both reach a majority, so at
//! most one commits. A vote for a remake that aborted or expired is void; the member
//! may vote again.
//!
//! Votes live in the record (`votes/<pk>`, written only by that member). A node votes
//! once it is ready, on its current epoch, and only if it is in the electorate.
//!
//! Gap: votes are not persisted locally. A node whose vote reached only the coordinator
//! and that then restarts and resyncs from nodes that never saw the vote may vote again
//! for a competitor.

use std::collections::HashSet;

use tracing::{info, warn};

use super::Host;
use super::records::{self, EpochRecord, Phase, Record, 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 open remake.
    pub(crate) elsewhere: 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, {} not yet; {} needed",
            self.yes, self.electorate, self.elsewhere, self.pending, self.quorum
        )
    }
}

fn open(record: &Record, now: u64, grace: u64) -> bool {
    record.phase(now, grace).is_open()
}

/// The tally of remake `id` among `all` records.
pub(crate) fn tally(
    id: &str,
    record: &Record,
    all: &[(String, Record)],
    now: u64,
    grace: u64,
) -> Tally {
    let electorate: HashSet<&str> = record
        .proposal
        .electorate
        .iter()
        .map(String::as_str)
        .collect();
    let yes: HashSet<&str> = record
        .votes
        .keys()
        .map(String::as_str)
        .filter(|voter| electorate.contains(voter))
        .collect();
    let elsewhere: HashSet<&str> = all
        .iter()
        .filter(|(other, candidate)| {
            other != id
                && candidate.proposal.from_epoch == record.proposal.from_epoch
                && open(candidate, now, grace)
        })
        .flat_map(|(_, candidate)| candidate.votes.keys().map(String::as_str))
        .filter(|voter| electorate.contains(voter) && !yes.contains(voter))
        .collect();
    Tally {
        electorate: electorate.len(),
        quorum: record.proposal.quorum,
        yes: yes.len(),
        elsewhere: elsewhere.len(),
        pending: electorate.len() - yes.len() - elsewhere.len(),
    }
}

/// The remake `me` should vote for now, if any: none if it already voted for an open
/// remake of `epoch`; otherwise the oldest proposed one whose electorate includes it.
pub(crate) fn choice(
    me: &str,
    epoch: &EpochRecord,
    all: &[(String, Record)],
    now: u64,
    grace: u64,
) -> Option<String> {
    let of_epoch = || {
        all.iter()
            .filter(move |(_, record)| &record.proposal.from_epoch == epoch)
    };
    if of_epoch().any(|(_, record)| open(record, now, grace) && record.votes.contains_key(me)) {
        return None;
    }
    of_epoch()
        .filter(|(_, record)| {
            record.phase(now, grace) == Phase::Proposed
                && record.proposal.electorate.iter().any(|member| member == me)
        })
        .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;

    // 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| {
            choice(
                &voter,
                &wanted,
                &records::load_all(doc),
                unix_millis(),
                grace,
            )
            .is_some()
        })
        .await;
    if !due {
        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 Some(id) = choice(&voter, &epoch, &all, now, grace) else {
                return Ok(None);
            };
            records::write_vote(doc, &id, &voter, &Vote { at: now })?;
            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::{Decision, Policy, Proposal};
    use serde_json::Value as JsonValue;
    use std::collections::BTreeMap;

    fn record(proposed_at: u64, electorate: &[&str], votes: &[&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: JsonValue::Null,
                translation: JsonValue::Null,
            },
            decision: None,
            abort_requested: None,
            votes: votes
                .iter()
                .map(|v| ((*v).to_owned(), Vote { at: proposed_at }))
                .collect::<BTreeMap<_, _>>(),
        }
    }

    const ALL: [&str; 5] = ["a", "b", "c", "d", "e"];

    #[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, &["a", "b"])),
            ("y".to_owned(), record(200, &ALL, &["c", "z"])), // z isn't in the electorate
        ];
        let x = tally("x", &all[0].1, &all, 150, 0);
        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, 150, 0);
        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 all = vec![
            ("x".to_owned(), record(100, &ALL, &["a", "b", "d"])),
            ("y".to_owned(), record(200, &ALL, &["c"])),
        ];
        assert!(tally("x", &all[0].1, &all, 150, 0).reached());
        assert!(tally("y", &all[1].1, &all, 150, 0).out_of_reach());
    }

    #[test]
    fn votes_for_aborted_remakes_are_void() {
        let mut aborted = record(100, &ALL, &["a", "b", "c"]);
        aborted.decision = Some(Decision::Aborted {
            reason: "test".into(),
            decided_at: 150,
        });
        let all = vec![
            ("x".to_owned(), aborted),
            ("y".to_owned(), record(200, &ALL, &["d"])),
        ];
        let y = tally("y", &all[1].1, &all, 250, 0);
        assert_eq!((y.yes, y.elsewhere, y.pending), (1, 0, 4));
        // c may vote again.
        assert_eq!(
            choice("c", &Epoch::GENESIS.into(), &all, 250, 0).as_deref(),
            Some("y")
        );
    }

    #[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 epoch: EpochRecord = Epoch::GENESIS.into();
        assert_eq!(choice("a", &epoch, &all, 150, 0).as_deref(), Some("x"));
        assert_eq!(choice("outsider", &epoch, &all, 150, 0), None);

        let voted = vec![
            ("y".to_owned(), record(200, &ALL, &["a"])),
            ("x".to_owned(), record(100, &ALL, &[])),
        ];
        assert_eq!(choice("a", &epoch, &voted, 150, 0), None);
    }
}