//! Remake records: `_continuum/remakes/<id>` in the replicated document.
//!
//! A record has up to three parts, each with a single kind of writer, so concurrent
//! writes never fight over the same key:
//!
//! * `proposal`: written once, by the coordinator.
//! * `decision`: written once, by the coordinator only (committed or aborted).
//! * `abortRequested`: written by any node on an operator's request. It is an input to
//!   the coordinator's decision, not a decision itself.
//!
//! Once a remake has expired, its coordinator no longer commits it, and any node may
//! abort it on an operator's request (`api`). Should clocks disagree by more than the
//! grace margin, a commit and such an abort could still be written concurrently. Such
//! a record is *contested*: it reads as committed, and nodes don't switch to it on
//! their own.
//!
//! Votes are kept apart, in the votes register `_continuum/votes/<pk>`: one entry per
//! member for the whole epoch, written by that member only (`vote`). Reading a record
//! fills in the votes that name it.
//!
//! A proposal whose coordinator never decides (crash, partition) is reported as
//! `expired` once its settle period plus a grace margin has passed. Expiry is derived,
//! never written, so a late but valid commit can't collide with someone's abort.
//!
//! Records edited into an unreadable shape through the store API are skipped with a
//! warning.

use std::collections::BTreeMap;

use automerge::{AutoCommit, ChangeHash};
use serde::{Deserialize, Serialize};
use serde_json::Value as JsonValue;
use tracing::warn;

use super::kind::DEFAULT_KIND;
use super::snapshot::{META_ROOT, REMAKES, SnapshotRequest, VOTES};
use crate::document::replica::Epoch;
use crate::document::store::{self, StoreError};

/// One day, for retention defaults.
const DAY_SECS: u64 = 24 * 60 * 60;

pub(crate) fn unix_millis() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}

// --- record types ------------------------------------------------------------

/// An epoch in a record: generation plus hex-encoded snapshot hash.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct EpochRecord {
    pub(crate) generation: u64,
    pub(crate) id: String,
}

impl From<Epoch> for EpochRecord {
    fn from(epoch: Epoch) -> Self {
        Self {
            generation: epoch.generation,
            id: epoch.id_hex(),
        }
    }
}

impl EpochRecord {
    pub(crate) fn to_epoch(&self) -> Option<Epoch> {
        Some(Epoch {
            generation: self.generation,
            id: decode_hex32(&self.id)?,
        })
    }
}

/// How long something is kept. A value is kept until every given limit has passed.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct Retention {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(crate) epochs: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(crate) seconds: Option<u64>,
}

impl Retention {
    /// True while at least one given limit has not passed: `epochs` epochs and
    /// `age_ms` milliseconds after the fact.
    pub(crate) fn covers(&self, epochs: u64, age_ms: u64) -> bool {
        self.epochs.is_some_and(|limit| epochs <= limit)
            || self
                .seconds
                .is_some_and(|limit| age_ms <= limit.saturating_mul(1000))
    }
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct Policy {
    /// How stale a returning node's changes may be and still be replayed automatically;
    /// beyond it they are quarantined.
    #[serde(default = "Policy::default_replay_window")]
    pub(crate) replay_window: Retention,
    /// How long nodes keep the old epoch's document.
    #[serde(default = "Policy::default_archive_retention")]
    pub(crate) archive_retention: Retention,
}

impl Policy {
    pub(crate) fn default_replay_window() -> Retention {
        Retention {
            epochs: Some(1),
            seconds: Some(7 * DAY_SECS),
        }
    }

    pub(crate) fn default_archive_retention() -> Retention {
        Retention {
            epochs: Some(2),
            seconds: Some(30 * DAY_SECS),
        }
    }
}

#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct Proposal {
    pub(crate) coordinator: String,
    pub(crate) from_epoch: EpochRecord,
    /// Hex change hashes, sorted. The proposal record itself is not part of the base.
    pub(crate) base_heads: Vec<String>,
    pub(crate) policy: Policy,
    pub(crate) settle_ms: u64,
    pub(crate) proposed_at: u64,
    pub(crate) settle_until: u64,
    /// How long nodes wait after seeing the commit before switching (ms); one keepalive
    /// interval if absent.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(crate) switch_delay_ms: Option<u64>,
    /// Members whose votes count (sorted), fixed at proposal time (`vote`).
    pub(crate) electorate: Vec<String>,
    /// Votes needed to commit: a majority of the electorate.
    pub(crate) quorum: usize,
    /// The remake kind (`compact` for records written before kinds existed).
    #[serde(default = "default_kind")]
    pub(crate) kind: String,
    /// The kind's build input, if it has one: held by the coordinator and fetched by
    /// every voter (`content`), not stored in the record.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(crate) content: Option<ContentRef>,
    /// The kind's translation input.
    #[serde(default, skip_serializing_if = "JsonValue::is_null")]
    pub(crate) translation: JsonValue,
}

/// Content of a remake, by reference.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct ContentRef {
    /// BLAKE3 of the bytes, hex.
    pub(crate) blake3: String,
    pub(crate) size: u64,
}

impl ContentRef {
    pub(crate) fn of(bytes: &[u8]) -> Self {
        Self {
            blake3: blake3::hash(bytes).to_hex().to_string(),
            size: bytes.len() as u64,
        }
    }

    pub(crate) fn hash(&self) -> Option<[u8; 32]> {
        decode_hex32(&self.blake3)
    }
}

fn default_kind() -> String {
    DEFAULT_KIND.to_owned()
}

impl Proposal {
    pub(crate) fn base_change_hashes(&self) -> Option<Vec<ChangeHash>> {
        self.base_heads
            .iter()
            .map(|hex| decode_hex32(hex).map(ChangeHash))
            .collect()
    }
}

#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(tag = "outcome", rename_all = "camelCase")]
pub(crate) enum Decision {
    #[serde(rename_all = "camelCase")]
    Committed {
        to_epoch: EpochRecord,
        snapshot_bytes: u64,
        decided_at: u64,
    },
    #[serde(rename_all = "camelCase")]
    Aborted { reason: String, decided_at: u64 },
}

#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct AbortRequest {
    pub(crate) by: String,
    pub(crate) at: u64,
}

/// A member's vote: its entry in the votes register.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct Vote {
    pub(crate) remake: String,
    pub(crate) at: u64,
}

/// The votes register: every concurrent value of each member's entry, by member public
/// key. Two values mean the member voted again without seeing its first vote (`vote`).
pub(crate) type Register = BTreeMap<String, Vec<Vote>>;

#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct Record {
    pub(crate) proposal: Proposal,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(crate) decision: Option<Decision>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(crate) abort_requested: Option<AbortRequest>,
    /// Decisions written concurrently disagree; `decision` is then the commit. Not
    /// stored: found when the record is read.
    #[serde(
        default,
        skip_deserializing,
        skip_serializing_if = "std::ops::Not::not"
    )]
    pub(crate) contested: bool,
    /// The register's votes naming this remake, by member public key. Not stored in
    /// the record: filled in when it is read (`load`).
    #[serde(default, skip_deserializing)]
    pub(crate) votes: BTreeMap<String, Vote>,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) enum Phase {
    /// Settling; the coordinator has not decided yet.
    Proposed,
    /// No decision long after the settle period: the coordinator is gone.
    Expired,
    Committed,
    Aborted,
}

impl Phase {
    /// Still able to become (or already) the next epoch.
    pub(crate) fn is_open(self) -> bool {
        matches!(self, Phase::Proposed | Phase::Committed)
    }

    pub(crate) fn is_final(self) -> bool {
        !matches!(self, Phase::Proposed)
    }
}

impl Record {
    /// The epoch this remake committed to, if committed and readable.
    pub(crate) fn committed_epoch(&self) -> Option<Epoch> {
        match &self.decision {
            Some(Decision::Committed { to_epoch, .. }) => to_epoch.to_epoch(),
            _ => None,
        }
    }

    /// Everything needed to (re)build this remake's snapshot but the content (which the
    /// caller adds, `content::ensure`); None if the proposal is unreadable.
    pub(crate) fn snapshot_request(&self, id: &str) -> Option<SnapshotRequest> {
        Some(SnapshotRequest {
            remake_id: id.to_owned(),
            previous: self.proposal.from_epoch.to_epoch()?,
            base_heads: self.proposal.base_change_hashes()?,
            policy: serde_json::to_value(&self.proposal.policy).ok()?,
            proposed_at: self.proposal.proposed_at,
            kind: self.proposal.kind.clone(),
            content: None,
            translation: self.proposal.translation.clone(),
        })
    }

    pub(crate) fn phase(&self, now_ms: u64, grace_ms: u64) -> Phase {
        match &self.decision {
            Some(Decision::Committed { .. }) => Phase::Committed,
            Some(Decision::Aborted { .. }) => Phase::Aborted,
            None if now_ms > self.proposal.settle_until.saturating_add(grace_ms) => Phase::Expired,
            None => Phase::Proposed,
        }
    }
}

// --- document access ---------------------------------------------------------

fn records_dir() -> Vec<String> {
    vec![META_ROOT.to_owned(), REMAKES.to_owned()]
}

pub(crate) fn record_path(id: &str, part: &str) -> Vec<String> {
    let mut path = records_dir();
    path.push(id.to_owned());
    if !part.is_empty() {
        path.push(part.to_owned());
    }
    path
}

/// The record with this ID, if it exists and is readable, with its votes.
pub(crate) fn load(doc: &AutoCommit, id: &str) -> Option<Record> {
    let record = load_record(doc, id)?;
    Some(with_votes(id, record, &load_register(doc)))
}

/// All readable records, with their votes.
pub(crate) fn load_all(doc: &AutoCommit) -> Vec<(String, Record)> {
    let register = load_register(doc);
    store::keys_at(doc, &records_dir())
        .unwrap_or_default()
        .into_iter()
        .filter_map(|id| {
            let record = load_record(doc, &id)?;
            Some((id.clone(), with_votes(&id, record, &register)))
        })
        .collect()
}

fn with_votes(id: &str, mut record: Record, register: &Register) -> Record {
    record.votes = register
        .iter()
        .filter_map(|(voter, votes)| {
            let vote = votes.iter().find(|vote| vote.remake == id)?;
            Some((voter.clone(), vote.clone()))
        })
        .collect();
    record
}

fn load_record(doc: &AutoCommit, id: &str) -> Option<Record> {
    let value = match store::read_path(doc, &record_path(id, "")) {
        Ok(Some(value)) => value,
        Ok(None) => return None,
        Err(error) => {
            warn!(remake = %id, error = %error, "Failed to read remake record");
            return None;
        }
    };
    let mut record: Record = match serde_json::from_value(value) {
        Ok(record) => record,
        Err(error) => {
            warn!(remake = %id, error = %error, "Skipping unreadable remake record");
            return None;
        }
    };
    if let Some(commit) = contested_decision(doc, id) {
        record.decision = Some(commit);
        record.contested = true;
    }
    Some(record)
}

/// A commit among concurrent decisions of remake `id` that disagree, if they do.
fn contested_decision(doc: &AutoCommit, id: &str) -> Option<Decision> {
    let decisions: Vec<Decision> = store::read_candidates(doc, &record_path(id, "decision"))
        .ok()?
        .into_iter()
        .filter_map(|value| serde_json::from_value(value).ok())
        .collect();
    let committed: Vec<&EpochRecord> = decisions
        .iter()
        .filter_map(|decision| match decision {
            Decision::Committed { to_epoch, .. } => Some(to_epoch),
            Decision::Aborted { .. } => None,
        })
        .collect();
    let agree = committed.is_empty()
        || (committed.len() == decisions.len() && committed.iter().all(|e| *e == committed[0]));
    if agree {
        return None;
    }
    decisions
        .into_iter()
        .find(|decision| matches!(decision, Decision::Committed { .. }))
}

fn register_path(voter: &str) -> Vec<String> {
    vec![META_ROOT.to_owned(), VOTES.to_owned(), voter.to_owned()]
}

/// The votes register; unreadable entries are skipped with a warning.
pub(crate) fn load_register(doc: &AutoCommit) -> Register {
    store::keys_at(doc, &[META_ROOT.to_owned(), VOTES.to_owned()])
        .unwrap_or_default()
        .into_iter()
        .filter_map(|voter| {
            let values = match store::read_candidates(doc, &register_path(&voter)) {
                Ok(values) => values,
                Err(error) => {
                    warn!(voter = %voter, error = %error, "Failed to read vote");
                    return None;
                }
            };
            let votes: Vec<Vote> = values
                .into_iter()
                .filter_map(|value| match serde_json::from_value(value) {
                    Ok(vote) => Some(vote),
                    Err(error) => {
                        warn!(voter = %voter, error = %error, "Skipping unreadable vote");
                        None
                    }
                })
                .collect();
            (!votes.is_empty()).then_some((voter, votes))
        })
        .collect()
}

fn write_part<T: Serialize>(
    doc: &mut AutoCommit,
    id: &str,
    part: &str,
    value: &T,
) -> Result<(), StoreError> {
    let json = serde_json::to_value(value).map_err(anyhow::Error::from)?;
    store::write_path(doc, &record_path(id, part), &json)
}

pub(crate) fn write_proposal(
    doc: &mut AutoCommit,
    id: &str,
    proposal: &Proposal,
) -> Result<(), StoreError> {
    write_part(doc, id, "proposal", proposal)
}

pub(crate) fn write_decision(
    doc: &mut AutoCommit,
    id: &str,
    decision: &Decision,
) -> Result<(), StoreError> {
    write_part(doc, id, "decision", decision)
}

pub(crate) fn write_abort_request(
    doc: &mut AutoCommit,
    id: &str,
    request: &AbortRequest,
) -> Result<(), StoreError> {
    write_part(doc, id, "abortRequested", request)
}

/// Writes `voter`'s entry in the votes register (replacing its earlier one).
pub(crate) fn write_vote(doc: &mut AutoCommit, voter: &str, vote: &Vote) -> Result<(), StoreError> {
    let json = serde_json::to_value(vote).map_err(anyhow::Error::from)?;
    store::write_path(doc, &register_path(voter), &json)
}

pub(crate) fn decode_hex32(hex: &str) -> Option<[u8; 32]> {
    if hex.len() != 64 || !hex.is_ascii() {
        return None;
    }
    let mut out = [0u8; 32];
    for (i, byte) in out.iter_mut().enumerate() {
        *byte = u8::from_str_radix(&hex[2 * i..2 * i + 2], 16).ok()?;
    }
    Some(out)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn retention_covers_until_every_limit_passed() {
        let both = Retention {
            epochs: Some(1),
            seconds: Some(10),
        };
        assert!(both.covers(1, 60_000)); // epochs still within
        assert!(both.covers(5, 5_000)); // time still within
        assert!(!both.covers(2, 10_001));
        let none = Retention {
            epochs: None,
            seconds: None,
        };
        assert!(!none.covers(0, 0));
    }

    fn proposal(settle_until: u64) -> Proposal {
        Proposal {
            coordinator: "node-a".into(),
            from_epoch: Epoch::GENESIS.into(),
            base_heads: vec!["ab".repeat(32)],
            policy: Policy {
                replay_window: Policy::default_replay_window(),
                archive_retention: Policy::default_archive_retention(),
            },
            settle_ms: 1_000,
            proposed_at: settle_until.saturating_sub(1_000),
            settle_until,
            switch_delay_ms: None,
            electorate: vec!["node-a".into(), "node-b".into()],
            quorum: 2,
            kind: DEFAULT_KIND.to_owned(),
            content: None,
            translation: JsonValue::Null,
        }
    }

    #[test]
    fn round_trip_through_the_document() {
        let mut doc = AutoCommit::new();
        let p = proposal(10_000);
        write_proposal(&mut doc, "c-1", &p).unwrap();
        write_abort_request(
            &mut doc,
            "c-1",
            &AbortRequest {
                by: "x".into(),
                at: 5,
            },
        )
        .unwrap();
        let decision = Decision::Committed {
            to_epoch: EpochRecord {
                generation: 1,
                id: "cd".repeat(32),
            },
            snapshot_bytes: 123,
            decided_at: 11_000,
        };
        write_decision(&mut doc, "c-1", &decision).unwrap();
        let vote = |remake: &str| Vote {
            remake: remake.into(),
            at: 7,
        };
        write_vote(&mut doc, "node-b", &vote("c-1")).unwrap();
        write_vote(&mut doc, "node-c", &vote("other")).unwrap();
        doc.commit();

        let all = load_all(&doc);
        assert_eq!(all.len(), 1);
        let (id, record) = &all[0];
        assert_eq!(id, "c-1");
        assert_eq!(record.proposal.policy, p.policy);
        assert_eq!(record.proposal.base_change_hashes().unwrap().len(), 1);
        assert!(record.abort_requested.is_some());
        assert_eq!(record.votes.keys().collect::<Vec<_>>(), ["node-b"]);
        assert_eq!(record.proposal.quorum, 2);
        assert_eq!(record.phase(0, 0), Phase::Committed);
    }

    #[test]
    fn concurrent_votes_are_kept_apart() {
        let mut a = AutoCommit::new();
        let vote = |remake: &str, at| Vote {
            remake: remake.into(),
            at,
        };
        write_vote(&mut a, "node-a", &vote("x", 1)).unwrap();
        a.commit();
        let mut b = a.fork();
        // node-b votes in both, never seeing its other vote (a lost vote).
        write_vote(&mut a, "node-b", &vote("x", 2)).unwrap();
        write_vote(&mut b, "node-b", &vote("y", 3)).unwrap();
        a.commit();
        b.commit();
        a.merge(&mut b).unwrap();

        let register = load_register(&a);
        assert_eq!(register["node-a"], [vote("x", 1)]);
        let mut twice = register["node-b"].clone();
        twice.sort_by_key(|vote| vote.at);
        assert_eq!(twice, [vote("x", 2), vote("y", 3)]);

        // Seen, a vote is replaced.
        write_vote(&mut a, "node-b", &vote("z", 4)).unwrap();
        a.commit();
        assert_eq!(load_register(&a)["node-b"], [vote("z", 4)]);
    }

    #[test]
    fn phases() {
        let mut record = Record {
            proposal: proposal(10_000),
            decision: None,
            abort_requested: None,
            contested: false,
            votes: BTreeMap::new(),
        };
        assert_eq!(record.phase(10_000, 500), Phase::Proposed);
        assert_eq!(record.phase(10_500, 500), Phase::Proposed);
        assert_eq!(record.phase(10_501, 500), Phase::Expired);
        record.decision = Some(Decision::Aborted {
            reason: "test".into(),
            decided_at: 1,
        });
        assert_eq!(record.phase(0, 0), Phase::Aborted);
    }

    #[test]
    fn disagreeing_decisions_are_contested() {
        let mut base = AutoCommit::new();
        write_proposal(&mut base, "c-1", &proposal(10_000)).unwrap();
        base.commit();
        let (mut a, mut b, mut c) = (base.fork(), base.fork(), base.fork());
        let committed = Decision::Committed {
            to_epoch: EpochRecord {
                generation: 1,
                id: "cd".repeat(32),
            },
            snapshot_bytes: 1,
            decided_at: 2,
        };
        write_decision(&mut a, "c-1", &committed).unwrap();
        write_decision(
            &mut b,
            "c-1",
            &Decision::Aborted {
                reason: "test".into(),
                decided_at: 3,
            },
        )
        .unwrap();
        // The same decision, concurrently: they agree.
        write_decision(&mut c, "c-1", &committed).unwrap();
        a.commit();
        b.commit();
        c.commit();
        let mut agreeing = a.fork();
        agreeing.merge(&mut c).unwrap();
        assert!(!load(&agreeing, "c-1").unwrap().contested);

        a.merge(&mut b).unwrap();
        let record = load(&a, "c-1").unwrap();
        assert!(record.contested);
        assert_eq!(record.phase(0, 0), Phase::Committed);
    }

    #[test]
    fn unreadable_records_are_skipped() {
        let mut doc = AutoCommit::new();
        store::write_path(
            &mut doc,
            &record_path("broken", ""),
            &serde_json::json!({ "proposal": "not an object" }),
        )
        .unwrap();
        write_proposal(&mut doc, "fine", &proposal(1)).unwrap();
        doc.commit();
        let all = load_all(&doc);
        assert_eq!(all.len(), 1);
        assert_eq!(all[0].0, "fine");
    }

    #[test]
    fn epoch_record_round_trip() {
        let epoch = Epoch {
            generation: 3,
            id: [0x5a; 32],
        };
        assert_eq!(EpochRecord::from(epoch).to_epoch(), Some(epoch));
        assert_eq!(
            EpochRecord {
                generation: 1,
                id: "zz".into()
            }
            .to_epoch(),
            None
        );
    }
}