//! Compaction records: `_continuum/compactions/<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.
//!
//! 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 automerge::{AutoCommit, ChangeHash};
use serde::{Deserialize, Serialize};
use tracing::warn;

use super::snapshot::{SnapshotRequest, COMPACTIONS, META_ROOT};
use crate::replica::Epoch;
use crate::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,
    /// Peers that must stay active during the settle period.
    pub(crate) peers_considered: Vec<String>,
}

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,
}

#[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>,
}

#[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 compaction 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 compaction's snapshot; None if the proposal
    /// is unreadable.
    pub(crate) fn snapshot_request(&self, id: &str) -> Option<SnapshotRequest> {
        Some(SnapshotRequest {
            compaction_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,
        })
    }

    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(), COMPACTIONS.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.
pub(crate) fn load(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!(compaction = %id, error = %error, "Failed to read compaction record");
            return None;
        }
    };
    match serde_json::from_value(value) {
        Ok(record) => Some(record),
        Err(error) => {
            warn!(compaction = %id, error = %error, "Skipping unreadable compaction record");
            None
        }
    }
}

/// All readable records.
pub(crate) fn load_all(doc: &AutoCommit) -> Vec<(String, Record)> {
    store::keys_at(doc, &records_dir())
        .unwrap_or_default()
        .into_iter()
        .filter_map(|id| load(doc, &id).map(|record| (id, record)))
        .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)
}

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,
            peers_considered: vec!["node-b".into()],
        }
    }

    #[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();
        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.phase(0, 0), Phase::Committed);
    }

    #[test]
    fn phases() {
        let mut record = Record {
            proposal: proposal(10_000),
            decision: None,
            abort_requested: None,
        };
        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 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
        );
    }
}