//! 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) and [`Replace`] (an operator's
//! document, with rules for where later writes go).
//!
//! Protocol data under `_continuum` is not a kind's business: the generic part carries
//! it forward and replays it unchanged.

use automerge::AutoCommit;
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 stores in the proposal for every node.
#[derive(Debug, Default)]
pub(crate) struct Prepared {
    /// Input to [`RemakeKind::build`]; replicated with the proposal (old epoch only).
    pub(crate) content: JsonValue,
    /// 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: &JsonValue,
        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),
        _ => 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: &JsonValue,
        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 mut rules = Vec::with_capacity(spec.rules.len());
        for rule in spec.rules {
            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 });
        }

        Ok(Prepared {
            content: canonical(spec.document),
            translation: serde_json::to_value(Rules { rules })
                .map_err(|error| error.to_string())?,
        })
    }

    fn build(
        &self,
        _old_at_base: &AutoCommit,
        content: &JsonValue,
        target: &mut AutoCommit,
    ) -> Result<(), StoreError> {
        // Canonical again: the content went through the old epoch's document, and the
        // operations (hence the epoch ID) must not depend on any key order.
        let JsonValue::Object(document) = canonical(content.clone()) 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 {
        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)
                }
            }
        }
    }
}

/// 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: &JsonValue,
            _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 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, &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 } }));
    }
}