//! Kinds of remake: what makes one remake differ from another.
//!
//! Everything else about a remake (proposal, settle period, decision, switch, rejoin,
//! replay window, quarantine, archive) is the same for every kind. A kind decides two
//! things only:
//!
//! * the user content of the next epoch's initial document ([`RemakeKind::build`]),
//!   which must be deterministic: every node builds it on its own and must produce the
//!   same bytes, since its hash names the epoch;
//! * where a write made in the old epoch after the base goes in the new one
//!   ([`RemakeKind::translate`]).
//!
//! Kinds: [`Compact`] (same content, no history), [`Replace`] (an operator's document
//! in store form, with rules for where later writes go) and [`Import`] (the same with
//! an Automerge document, e.g. one exported with `GET /admin/document`).
//!
//! Protocol data under `_continuum` is not a kind's business: the generic part carries
//! it forward and replays it unchanged.

use automerge::AutoCommit;
use base64::engine::{Engine, general_purpose::STANDARD as BASE64};
use serde::{Deserialize, Serialize};
use serde_json::Value as JsonValue;

use super::snapshot::{META_ROOT, canonical};
use crate::document::store::{self, StoreError};

/// Name of the kind used when a request doesn't name one.
pub(crate) const DEFAULT_KIND: &str = "compact";

/// What the coordinator prepares for every node.
#[derive(Debug, Default)]
pub(crate) struct Prepared {
    /// Input to [`RemakeKind::build`], if the kind takes any. The proposal refers to it
    /// by hash; voters fetch it from the coordinator and each other (`content`).
    pub(crate) content: Option<Vec<u8>>,
    /// Input to [`RemakeKind::translate`]; also kept in the new epoch's metadata, so
    /// nodes that rejoin later can translate their writes.
    pub(crate) translation: JsonValue,
}

/// Where a path written in the old epoch after the base goes.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum Translation {
    /// To the same path (compaction).
    Same,
    /// To another path.
    To(Vec<String>),
    /// Nowhere automatically: quarantined for an operator, with the reason.
    Quarantine(String),
    /// Nowhere: the remake declared the path obsolete.
    Drop,
}

pub(crate) trait RemakeKind: Send + Sync + 'static {
    /// Stable name, stored in proposals and epoch metadata.
    fn name(&self) -> &'static str;

    /// Coordinator only: validates the kind-specific part of the request.
    fn propose(&self, spec: &JsonValue) -> Result<Prepared, String>;

    /// Every node: writes the user content of the next epoch into `target`, a fresh
    /// document. `old_at_base` is the old epoch at the base. Must be deterministic and
    /// must not write under `_continuum` (anything there is replaced).
    fn build(
        &self,
        old_at_base: &AutoCommit,
        content: Option<&[u8]>,
        target: &mut AutoCommit,
    ) -> Result<(), StoreError>;

    /// Every node: where a user path written after the base goes.
    fn translate(&self, path: &[String], translation: &JsonValue) -> Translation;
}

/// The kind with this name.
pub(crate) fn by_name(name: &str) -> Option<&'static dyn RemakeKind> {
    match name {
        "compact" => Some(&Compact),
        "replace" => Some(&Replace),
        "import" => Some(&Import),
        _ => None,
    }
}

/// Paths owned by the protocol rather than by any kind.
pub(crate) fn is_reserved(path: &[String]) -> bool {
    path.first().is_some_and(|root| root == META_ROOT)
}

/// Compaction: the new document has the old one's content at the base, without its
/// history. Writes keep their paths.
pub(crate) struct Compact;

impl RemakeKind for Compact {
    fn name(&self) -> &'static str {
        "compact"
    }

    fn propose(&self, spec: &JsonValue) -> Result<Prepared, String> {
        match spec {
            JsonValue::Null => Ok(Prepared::default()),
            JsonValue::Object(fields) if fields.is_empty() => Ok(Prepared::default()),
            _ => Err("compaction takes no spec".to_owned()),
        }
    }

    fn build(
        &self,
        old_at_base: &AutoCommit,
        _content: Option<&[u8]>,
        target: &mut AutoCommit,
    ) -> Result<(), StoreError> {
        store::materialize_into(old_at_base, target, &is_reserved)
    }

    fn translate(&self, _path: &[String], _translation: &JsonValue) -> Translation {
        Translation::Same
    }
}

/// Replacement: the new document is supplied by an operator, in store form (the JSON
/// `GET /store` returns), and may be structured entirely differently. Writes made after
/// the base go where the remake's rules say; a path no rule covers is quarantined.
///
/// Spec: `{ "document": {…}, "rules": [ {"from": "a/b", "to": "c"},
/// {"from": "old", "drop": true} ] }`. Paths are slash-separated strings (`""` is the
/// root, matching everything) or arrays of segments (for keys containing `/`). The
/// longest matching `from` wins; the rest of the path is appended to `to`.
pub(crate) struct Replace;

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct ReplaceSpec {
    document: JsonValue,
    #[serde(default)]
    rules: Vec<RuleSpec>,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RuleSpec {
    from: PathSpec,
    #[serde(default)]
    to: Option<PathSpec>,
    #[serde(default)]
    drop: bool,
}

#[derive(Deserialize)]
#[serde(untagged)]
enum PathSpec {
    Text(String),
    Segments(Vec<String>),
}

impl PathSpec {
    fn segments(self) -> Result<Vec<String>, String> {
        let segments: Vec<String> = match self {
            PathSpec::Text(text) if text.is_empty() => Vec::new(),
            PathSpec::Text(text) => text.split('/').map(str::to_owned).collect(),
            PathSpec::Segments(segments) => segments,
        };
        if segments.iter().any(String::is_empty) {
            return Err(format!("empty segment in path {segments:?}"));
        }
        Ok(segments)
    }
}

/// A rule as stored in the translation (and the epoch metadata).
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
struct Rule {
    from: Vec<String>,
    /// None: drop.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    to: Option<Vec<String>>,
}

#[derive(Debug, Default, Serialize, Deserialize)]
struct Rules {
    rules: Vec<Rule>,
}

impl RemakeKind for Replace {
    fn name(&self) -> &'static str {
        "replace"
    }

    fn propose(&self, spec: &JsonValue) -> Result<Prepared, String> {
        let spec: ReplaceSpec =
            serde_json::from_value(spec.clone()).map_err(|error| error.to_string())?;

        let JsonValue::Object(document) = &spec.document else {
            return Err("document must be a JSON object (the store root)".to_owned());
        };
        if document.contains_key(META_ROOT) {
            return Err(format!(
                "document must not contain the reserved key {META_ROOT:?}"
            ));
        }

        let content =
            serde_json::to_vec(&canonical(spec.document)).map_err(|error| error.to_string())?;
        Ok(Prepared {
            content: Some(content),
            translation: prepare_rules(spec.rules)?,
        })
    }

    fn build(
        &self,
        _old_at_base: &AutoCommit,
        content: Option<&[u8]>,
        target: &mut AutoCommit,
    ) -> Result<(), StoreError> {
        let content = content.ok_or_else(|| anyhow::anyhow!("replace without content"))?;
        let document: JsonValue = serde_json::from_slice(content)
            .map_err(|error| anyhow::anyhow!("replacement document: {error}"))?;
        // Canonical again: the operations (hence the epoch ID) must not depend on any
        // key order.
        let JsonValue::Object(document) = canonical(document) else {
            return Ok(()); // propose only accepts objects
        };
        for (key, value) in document {
            if key != META_ROOT {
                store::write_path(target, &[key], &value)?;
            }
        }
        Ok(())
    }

    fn translate(&self, path: &[String], translation: &JsonValue) -> Translation {
        translate_by_rules(path, translation)
    }
}

/// Validates rules from a request; the translation to store.
fn prepare_rules(specs: Vec<RuleSpec>) -> Result<JsonValue, String> {
    let mut rules = Vec::with_capacity(specs.len());
    for rule in specs {
        let from = rule.from.segments()?;
        let to = match (rule.to, rule.drop) {
            (Some(to), false) => Some(to.segments()?),
            (None, true) => None,
            _ => {
                return Err(format!(
                    "rule for {from:?} needs exactly one of \"to\" and \"drop\""
                ));
            }
        };
        if is_reserved(&from) || to.as_deref().is_some_and(is_reserved) {
            return Err(format!(
                "rules must not touch the reserved key {META_ROOT:?}"
            ));
        }
        if rules.iter().any(|existing: &Rule| existing.from == from) {
            return Err(format!("two rules for {from:?}"));
        }
        rules.push(Rule { from, to });
    }
    serde_json::to_value(Rules { rules }).map_err(|error| error.to_string())
}

/// Where `path` goes under the stored rules: the longest matching `from` wins.
fn translate_by_rules(path: &[String], translation: &JsonValue) -> Translation {
    let Ok(Rules { rules }) = serde_json::from_value::<Rules>(translation.clone()) else {
        return Translation::Quarantine("unreadable replacement rules".to_owned());
    };
    let Some(rule) = rules
        .iter()
        .filter(|rule| path.starts_with(&rule.from))
        .max_by_key(|rule| rule.from.len())
    else {
        return Translation::Quarantine("no replacement rule covers this path".to_owned());
    };
    match &rule.to {
        None => Translation::Drop,
        Some(to) => {
            let mut target = to.clone();
            target.extend_from_slice(&path[rule.from.len()..]);
            if target.as_slice() == path {
                Translation::Same
            } else {
                Translation::To(target)
            }
        }
    }
}

/// Import: like [`Replace`], but the new document is an Automerge document (bytes, as
/// saved), so Automerge types the store's JSON can't express survive: text, counters,
/// bytes, timestamps, floats. Its `_continuum` (protocol data of whatever swarm it came
/// from) is left out, as is its history: the content is the merged view.
///
/// Spec: `{ "document": "<base64 of the saved document>", "rules": […] }`, rules as for
/// [`Replace`]. To restore a backup of this swarm with every later write kept where it
/// was: `"rules": [{"from": "", "to": ""}]`.
pub(crate) struct Import;

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct ImportSpec {
    document: String,
    #[serde(default)]
    rules: Vec<RuleSpec>,
}

/// The document in a request (base64 of a saved Automerge document).
fn load_encoded(encoded: &str) -> Result<AutoCommit, String> {
    let bytes = BASE64
        .decode(encoded.trim())
        .map_err(|error| format!("the document is not valid base64: {error}"))?;
    AutoCommit::load(&bytes)
        .map_err(|error| format!("the document is not an Automerge document: {error}"))
}

impl RemakeKind for Import {
    fn name(&self) -> &'static str {
        "import"
    }

    fn propose(&self, spec: &JsonValue) -> Result<Prepared, String> {
        let spec: ImportSpec =
            serde_json::from_value(spec.clone()).map_err(|error| error.to_string())?;
        let imported = load_encoded(&spec.document)?;

        // Keep only what goes into the new epoch: the merged view without protocol data
        // and history. Usually much smaller than what was uploaded, and it is what every
        // voter downloads.
        let mut content = AutoCommit::new();
        store::materialize_into(&imported, &mut content, &is_reserved)
            .map_err(|error| format!("the document can't be read: {error}"))?;

        Ok(Prepared {
            content: Some(content.save()),
            translation: prepare_rules(spec.rules)?,
        })
    }

    fn build(
        &self,
        _old_at_base: &AutoCommit,
        content: Option<&[u8]>,
        target: &mut AutoCommit,
    ) -> Result<(), StoreError> {
        let content = content.ok_or_else(|| anyhow::anyhow!("import without content"))?;
        let imported = AutoCommit::load(content)?;
        // Deterministic: the same bytes give the same view, copied in a fixed order.
        store::materialize_into(&imported, target, &is_reserved)
    }

    fn translate(&self, path: &[String], translation: &JsonValue) -> Translation {
        translate_by_rules(path, translation)
    }
}

/// The translations of several consecutive remakes, applied in order: a node that
/// rejoins several epochs later carries its writes through every remake in between.
#[derive(Default, Clone)]
pub(crate) struct Translator {
    steps: Vec<(&'static dyn RemakeKind, JsonValue)>,
}

impl Translator {
    /// One remake.
    pub(crate) fn single(kind: &'static dyn RemakeKind, translation: JsonValue) -> Self {
        Self {
            steps: vec![(kind, translation)],
        }
    }

    pub(crate) fn push(&mut self, kind: &'static dyn RemakeKind, translation: JsonValue) {
        self.steps.push((kind, translation));
    }

    /// Where `path` ends up after all steps. Reserved paths always stay put.
    pub(crate) fn translate(&self, path: &[String]) -> Translation {
        if is_reserved(path) {
            return Translation::Same;
        }
        let mut current = path.to_vec();
        let mut moved = false;
        for (kind, translation) in &self.steps {
            match kind.translate(&current, translation) {
                Translation::Same => {}
                Translation::To(next) => {
                    current = next;
                    moved = true;
                }
                stop @ (Translation::Quarantine(_) | Translation::Drop) => return stop,
            }
        }
        if moved {
            Translation::To(current)
        } else {
            Translation::Same
        }
    }
}

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

    /// A test kind that moves `a/...` to `b/...`, drops `x`, quarantines the rest.
    struct Rename;

    impl RemakeKind for Rename {
        fn name(&self) -> &'static str {
            "rename-test"
        }
        fn propose(&self, _spec: &JsonValue) -> Result<Prepared, String> {
            Ok(Prepared::default())
        }
        fn build(
            &self,
            _old: &AutoCommit,
            _content: Option<&[u8]>,
            _target: &mut AutoCommit,
        ) -> Result<(), StoreError> {
            Ok(())
        }
        fn translate(&self, path: &[String], _translation: &JsonValue) -> Translation {
            match path.split_first() {
                Some((first, rest)) if first == "a" => {
                    let mut to = vec!["b".to_owned()];
                    to.extend_from_slice(rest);
                    Translation::To(to)
                }
                Some((first, _)) if first == "b" => Translation::Same,
                Some((first, _)) if first == "x" => Translation::Drop,
                _ => Translation::Quarantine("no rule".to_owned()),
            }
        }
    }

    fn path(parts: &[&str]) -> Vec<String> {
        parts.iter().map(|part| (*part).to_owned()).collect()
    }

    #[test]
    fn compaction_keeps_paths() {
        assert_eq!(
            Translator::single(&Compact, JsonValue::Null).translate(&path(&["a", "b"])),
            Translation::Same
        );
        assert!(Compact.propose(&JsonValue::Null).is_ok());
        assert!(Compact.propose(&serde_json::json!({ "x": 1 })).is_err());
    }

    #[test]
    fn steps_compose() {
        static RENAME: Rename = Rename;
        let mut translator = Translator::default();
        translator.push(&RENAME, JsonValue::Null);
        translator.push(&Compact, JsonValue::Null);
        translator.push(&RENAME, JsonValue::Null); // b stays b

        assert_eq!(
            translator.translate(&path(&["a", "k"])),
            Translation::To(path(&["b", "k"]))
        );
        assert_eq!(translator.translate(&path(&["x"])), Translation::Drop);
        assert!(matches!(
            translator.translate(&path(&["other"])),
            Translation::Quarantine(_)
        ));
        // Protocol data is never translated.
        assert_eq!(
            translator.translate(&path(&[META_ROOT, "forgotten", "pk"])),
            Translation::Same
        );
    }

    fn replace(spec: serde_json::Value) -> Result<Prepared, String> {
        Replace.propose(&spec)
    }

    #[test]
    fn replace_validates_the_spec() {
        use serde_json::json;
        assert!(replace(json!({ "document": [] })).is_err());
        assert!(replace(json!({ "document": { META_ROOT: {} } })).is_err());
        assert!(replace(json!({ "document": {}, "rules": [{ "from": "a" }] })).is_err());
        assert!(
            replace(json!({ "document": {}, "rules": [{ "from": "a", "to": "b", "drop": true }] }))
                .is_err()
        );
        assert!(
            replace(json!({ "document": {}, "rules": [{ "from": "a//b", "to": "c" }] })).is_err()
        );
        assert!(
            replace(json!({ "document": {}, "rules": [{ "from": "a", "to": META_ROOT }] }))
                .is_err()
        );
        assert!(replace(json!({ "document": {}, "rules": [{ "from": "a", "drop": true }, { "from": ["a"], "to": "b" }] })).is_err());
        assert!(replace(json!({ "document": {}, "extra": 1 })).is_err());
        assert!(replace(json!({ "document": { "k": 1 } })).is_ok());
    }

    #[test]
    fn replace_rules_translate_by_longest_prefix() {
        use serde_json::json;
        let prepared = replace(json!({
            "document": {},
            "rules": [
                { "from": "users", "to": "accounts" },
                { "from": "users/admin", "drop": true },
                { "from": ["keep", "a/b"], "to": ["keep", "a/b"] },
            ],
        }))
        .unwrap();
        let translator = Translator::single(&Replace, prepared.translation);

        assert_eq!(
            translator.translate(&path(&["users", "alice", "name"])),
            Translation::To(path(&["accounts", "alice", "name"]))
        );
        assert_eq!(
            translator.translate(&path(&["users", "admin", "x"])),
            Translation::Drop
        );
        assert_eq!(
            translator.translate(&path(&["keep", "a/b", "c"])),
            Translation::Same
        );
        assert!(matches!(
            translator.translate(&path(&["settings"])),
            Translation::Quarantine(_)
        ));
    }

    #[test]
    fn import_keeps_types_and_drops_protocol_data() {
        use automerge::{ObjType, ROOT, ReadDoc, transaction::Transactable};
        use serde_json::json;

        let mut source = AutoCommit::new();
        let text = source.put_object(ROOT, "notes", ObjType::Text).unwrap();
        source.splice_text(&text, 0, 0, "hello").unwrap();
        source
            .put(ROOT, "hits", automerge::ScalarValue::counter(3))
            .unwrap();
        let meta = source.put_object(ROOT, META_ROOT, ObjType::Map).unwrap();
        source.put(&meta, "members", "elsewhere").unwrap();
        let encoded = BASE64.encode(source.save());

        assert!(
            Import
                .propose(&json!({ "document": "not base64!" }))
                .is_err()
        );
        assert!(
            Import
                .propose(&json!({ "document": BASE64.encode(b"junk") }))
                .is_err()
        );

        let prepared = Import
            .propose(&json!({ "document": encoded, "rules": [{ "from": "", "to": "" }] }))
            .unwrap();
        let build = || {
            let mut target = AutoCommit::new().with_actor(automerge::ActorId::from(&b"fixed"[..]));
            Import
                .build(&AutoCommit::new(), prepared.content.as_deref(), &mut target)
                .unwrap();
            target.commit_with(automerge::transaction::CommitOptions::default().with_time(0));
            target
        };
        let mut a = build();
        let mut b = build();
        assert_eq!(a.get_heads(), b.get_heads());

        assert!(a.get(ROOT, META_ROOT).unwrap().is_none());
        let (notes, notes_id) = a.get(ROOT, "notes").unwrap().unwrap();
        assert!(matches!(notes, automerge::Value::Object(ObjType::Text)));
        assert_eq!(a.text(&notes_id).unwrap(), "hello");
        assert!(matches!(
            a.get(ROOT, "hits").unwrap().unwrap().0,
            automerge::Value::Scalar(ref s) if matches!(s.as_ref(), automerge::ScalarValue::Counter(_))
        ));

        // Writes after the base keep their paths.
        assert_eq!(
            Translator::single(&Import, prepared.translation).translate(&path(&["x", "y"])),
            Translation::Same
        );
    }

    #[test]
    fn replace_builds_the_document_deterministically() {
        use serde_json::json;
        let build = |document: serde_json::Value| {
            let prepared = replace(json!({ "document": document })).unwrap();
            let mut target = AutoCommit::new().with_actor(automerge::ActorId::from(&b"fixed"[..]));
            Replace
                .build(&AutoCommit::new(), prepared.content.as_deref(), &mut target)
                .unwrap();
            target.commit_with(automerge::transaction::CommitOptions::default().with_time(0));
            (
                target.get_heads(),
                store::read_path(&target, &[]).unwrap().unwrap(),
            )
        };
        // Same content, different key order.
        let (heads_a, view_a) = build(json!({ "b": { "y": 2, "x": 1 }, "a": [1, 2] }));
        let (heads_b, view_b) = build(json!({ "a": [1, 2], "b": { "x": 1, "y": 2 } }));
        assert_eq!(heads_a, heads_b);
        assert_eq!(view_a, view_b);
        assert_eq!(view_a, json!({ "a": [1, 2], "b": { "x": 1, "y": 2 } }));
    }
}