//! Typed objects and their canonical encodings (spec §6, §7, §13.2).
//!
//! Encoding conventions not fixed by the spec text are defined here:
//! - Enum-like items (`TreeOp`, `Edit`) are maps whose key 0 is the variant tag.
//! - An `OpRef`/atom reference is `[bytes32, uint]`; `AtomRef::Start` is `0`
//!   and `AtomRef::End` is `1`.
//! - A `Mode` is an unsigned integer: bit 0 = executable, bit 1 = opaque.
//! - `NodeKind`: 0 = file, 1 = dir, 2 = symlink.

use crate::cbor::{self, Value};
use crate::hash::{hash_object, ChangeHash, KeyId, ObjectRef, ObjectType};
use crate::ids::{ChangeId, NodeId};
use std::collections::BTreeMap;

pub const FORMAT_VERSION: u64 = 1;

/// The change format writers produce (spec §6.1). Version 2 stores each
/// large insert's lines as a text blob; readers accept 1 and 2, and a
/// change keeps the version it was made in, so its hash never moves.
pub const CHANGE_VERSION: u64 = 2;

/// In a version 2 change, an insert whose lines encode (as a CBOR array of
/// byte strings) to at least this many bytes is stored as a text blob
/// holding exactly that encoding; a shorter one is inline (spec §6.1).
pub const TEXT_BLOB_MIN: usize = 4096;

#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum ObjectError {
    #[error("cbor: {0}")]
    Cbor(#[from] cbor::CborError),
    #[error("schema: {0}")]
    Schema(String),
}

pub(crate) fn schema<T>(msg: impl Into<String>) -> Result<T, ObjectError> {
    Err(ObjectError::Schema(msg.into()))
}

pub(crate) type Map = BTreeMap<u64, Value>;

pub(crate) fn field<'a>(m: &'a Map, k: u64, what: &str) -> Result<&'a Value, ObjectError> {
    m.get(&k)
        .ok_or_else(|| ObjectError::Schema(format!("missing field {k} ({what})")))
}
pub(crate) fn uint(v: &Value, what: &str) -> Result<u64, ObjectError> {
    v.as_uint()
        .ok_or_else(|| ObjectError::Schema(format!("{what}: expected unsigned integer")))
}
pub(crate) fn u32_(v: &Value, what: &str) -> Result<u32, ObjectError> {
    u32::try_from(uint(v, what)?).map_err(|_| ObjectError::Schema(format!("{what}: exceeds u32")))
}
pub(crate) fn text<'a>(v: &'a Value, what: &str) -> Result<&'a str, ObjectError> {
    v.as_text()
        .ok_or_else(|| ObjectError::Schema(format!("{what}: expected text")))
}
pub(crate) fn array<'a>(v: &'a Value, what: &str) -> Result<&'a [Value], ObjectError> {
    v.as_array()
        .ok_or_else(|| ObjectError::Schema(format!("{what}: expected array")))
}
pub(crate) fn map<'a>(v: &'a Value, what: &str) -> Result<&'a Map, ObjectError> {
    v.as_map()
        .ok_or_else(|| ObjectError::Schema(format!("{what}: expected map")))
}
pub(crate) fn bytes<'a>(v: &'a Value, what: &str) -> Result<&'a [u8], ObjectError> {
    v.as_bytes()
        .ok_or_else(|| ObjectError::Schema(format!("{what}: expected bytes")))
}
pub(crate) fn oref(v: &Value, what: &str) -> Result<ObjectRef, ObjectError> {
    ObjectRef::from_slice(bytes(v, what)?)
        .ok_or_else(|| ObjectError::Schema(format!("{what}: expected 32 bytes")))
}
pub(crate) fn node_id(v: &Value, what: &str) -> Result<NodeId, ObjectError> {
    NodeId::from_slice(bytes(v, what)?)
        .ok_or_else(|| ObjectError::Schema(format!("{what}: expected 16 bytes")))
}
pub(crate) fn check_version(m: &Map) -> Result<(), ObjectError> {
    let v = uint(field(m, 0, "version")?, "version")?;
    if v != FORMAT_VERSION {
        return schema(format!("unsupported format version {v}"));
    }
    Ok(())
}

// ---------------------------------------------------------------- Hlc

/// Hybrid logical clock (spec §6.4).
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Default)]
pub struct Hlc {
    pub wall_ms: u64,
    pub logical: u32,
}

impl Hlc {
    pub fn now() -> Hlc {
        let wall_ms = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_millis() as u64)
            .unwrap_or(0);
        Hlc {
            wall_ms,
            logical: 0,
        }
    }
    pub(crate) fn to_value(self) -> Value {
        Value::map([
            (0, Value::Uint(self.wall_ms)),
            (1, Value::Uint(self.logical as u64)),
        ])
    }
    pub(crate) fn from_value(v: &Value) -> Result<Hlc, ObjectError> {
        let m = map(v, "hlc")?;
        Ok(Hlc {
            wall_ms: uint(field(m, 0, "hlc.wall_ms")?, "hlc.wall_ms")?,
            logical: u32_(field(m, 1, "hlc.logical")?, "hlc.logical")?,
        })
    }
}

// ---------------------------------------------------------------- tree

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum NodeKind {
    File,
    Dir,
    Symlink,
    /// A nested repository pinned to a state (spec §7.8); content is a pin.
    Subrepo,
}

impl NodeKind {
    pub(crate) fn tag(self) -> u64 {
        match self {
            NodeKind::File => 0,
            NodeKind::Dir => 1,
            NodeKind::Symlink => 2,
            NodeKind::Subrepo => 3,
        }
    }
    pub(crate) fn from_tag(t: u64) -> Result<NodeKind, ObjectError> {
        Ok(match t {
            0 => NodeKind::File,
            1 => NodeKind::Dir,
            2 => NodeKind::Symlink,
            3 => NodeKind::Subrepo,
            _ => return schema(format!("unknown node kind {t}")),
        })
    }
}

/// Executable bit plus content mode (spec §7.1, §7.5).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default, PartialOrd, Ord)]
pub struct Mode {
    pub exec: bool,
    pub opaque: bool,
}

impl Mode {
    pub const TEXT: Mode = Mode {
        exec: false,
        opaque: false,
    };
    pub const OPAQUE: Mode = Mode {
        exec: false,
        opaque: true,
    };
    pub(crate) fn bits(self) -> u64 {
        (self.exec as u64) | ((self.opaque as u64) << 1)
    }
    pub(crate) fn from_bits(b: u64) -> Result<Mode, ObjectError> {
        if b > 3 {
            return schema(format!("unknown mode bits {b}"));
        }
        Ok(Mode {
            exec: b & 1 != 0,
            opaque: b & 2 != 0,
        })
    }
}

/// Reference to a tree operation of some change: the placement or mode atom it
/// introduced. `change == ObjectRef::ZERO` means this change.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct OpRef {
    pub change: ChangeHash,
    pub index: u32,
}

impl OpRef {
    fn to_value(self) -> Value {
        Value::Array(vec![
            Value::Bytes(self.change.0.to_vec()),
            Value::Uint(self.index as u64),
        ])
    }
    fn from_value(v: &Value, what: &str) -> Result<OpRef, ObjectError> {
        let a = array(v, what)?;
        if a.len() != 2 {
            return schema(format!("{what}: expected [hash, index]"));
        }
        Ok(OpRef {
            change: oref(&a[0], what)?,
            index: u32_(&a[1], what)?,
        })
    }
    /// Replace a self reference with the concrete change hash.
    pub fn resolve(self, me: ChangeHash) -> OpRef {
        if self.change.is_zero() {
            OpRef { change: me, ..self }
        } else {
            self
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TreeOp {
    AddNode {
        node: NodeId,
        kind: NodeKind,
        mode: Mode,
        parent: NodeId,
        name: String,
    },
    RemoveNode {
        node: NodeId,
        placement: OpRef,
    },
    Move {
        node: NodeId,
        old: OpRef,
        parent: NodeId,
        name: String,
    },
    SetMode {
        node: NodeId,
        old: OpRef,
        mode: Mode,
    },
}

/// Names must be valid, non-empty, not `.`/`..`, and free of `/` and NUL (spec §7.1).
pub fn validate_name(name: &str) -> Result<(), ObjectError> {
    if name.is_empty() || name == "." || name == ".." || name.contains('/') || name.contains('\0') {
        return schema(format!("invalid placement name {name:?}"));
    }
    Ok(())
}

impl TreeOp {
    pub fn node(&self) -> NodeId {
        match self {
            TreeOp::AddNode { node, .. }
            | TreeOp::RemoveNode { node, .. }
            | TreeOp::Move { node, .. }
            | TreeOp::SetMode { node, .. } => *node,
        }
    }

    fn to_value(&self) -> Value {
        let nid = |n: &NodeId| Value::Bytes(n.0.to_vec());
        match self {
            TreeOp::AddNode {
                node,
                kind,
                mode,
                parent,
                name,
            } => Value::map([
                (0, Value::Uint(0)),
                (1, nid(node)),
                (2, Value::Uint(kind.tag())),
                (3, Value::Uint(mode.bits())),
                (4, nid(parent)),
                (5, Value::Text(name.clone())),
            ]),
            TreeOp::RemoveNode { node, placement } => Value::map([
                (0, Value::Uint(1)),
                (1, nid(node)),
                (2, placement.to_value()),
            ]),
            TreeOp::Move {
                node,
                old,
                parent,
                name,
            } => Value::map([
                (0, Value::Uint(2)),
                (1, nid(node)),
                (2, old.to_value()),
                (4, nid(parent)),
                (5, Value::Text(name.clone())),
            ]),
            TreeOp::SetMode { node, old, mode } => Value::map([
                (0, Value::Uint(3)),
                (1, nid(node)),
                (2, old.to_value()),
                (3, Value::Uint(mode.bits())),
            ]),
        }
    }

    fn from_value(v: &Value) -> Result<TreeOp, ObjectError> {
        let m = map(v, "tree_op")?;
        let tag = uint(field(m, 0, "tree_op tag")?, "tree_op tag")?;
        let node = node_id(field(m, 1, "tree_op.node")?, "tree_op.node")?;
        let op = match tag {
            0 => {
                let name = text(field(m, 5, "name")?, "name")?.to_owned();
                validate_name(&name)?;
                TreeOp::AddNode {
                    node,
                    kind: NodeKind::from_tag(uint(field(m, 2, "kind")?, "kind")?)?,
                    mode: Mode::from_bits(uint(field(m, 3, "mode")?, "mode")?)?,
                    parent: node_id(field(m, 4, "parent")?, "parent")?,
                    name,
                }
            }
            1 => TreeOp::RemoveNode {
                node,
                placement: OpRef::from_value(field(m, 2, "placement")?, "placement")?,
            },
            2 => {
                let name = text(field(m, 5, "name")?, "name")?.to_owned();
                validate_name(&name)?;
                TreeOp::Move {
                    node,
                    old: OpRef::from_value(field(m, 2, "old")?, "old")?,
                    parent: node_id(field(m, 4, "parent")?, "parent")?,
                    name,
                }
            }
            3 => TreeOp::SetMode {
                node,
                old: OpRef::from_value(field(m, 2, "old")?, "old")?,
                mode: Mode::from_bits(uint(field(m, 3, "mode")?, "mode")?)?,
            },
            _ => return schema(format!("unknown tree op tag {tag}")),
        };
        if node.is_root() {
            return schema("tree op on the root node");
        }
        Ok(op)
    }
}

// ---------------------------------------------------------------- edits

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum AtomRef {
    Start,
    End,
    Atom { change: ChangeHash, index: u32 },
}

impl AtomRef {
    fn to_value(self) -> Value {
        match self {
            AtomRef::Start => Value::Uint(0),
            AtomRef::End => Value::Uint(1),
            AtomRef::Atom { change, index } => Value::Array(vec![
                Value::Bytes(change.0.to_vec()),
                Value::Uint(index as u64),
            ]),
        }
    }
    fn from_value(v: &Value, what: &str) -> Result<AtomRef, ObjectError> {
        match v {
            Value::Uint(0) => Ok(AtomRef::Start),
            Value::Uint(1) => Ok(AtomRef::End),
            Value::Array(a) if a.len() == 2 => Ok(AtomRef::Atom {
                change: oref(&a[0], what)?,
                index: u32_(&a[1], what)?,
            }),
            _ => schema(format!("{what}: expected atom reference")),
        }
    }
    pub fn resolve(self, me: ChangeHash) -> AtomRef {
        match self {
            AtomRef::Atom { change, index } if change.is_zero() => {
                AtomRef::Atom { change: me, index }
            }
            other => other,
        }
    }
}

/// Half-open range of atoms introduced by one change.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct AtomRange {
    pub change: ChangeHash,
    pub from: u32,
    pub to: u32,
}

impl AtomRange {
    fn to_value(self) -> Value {
        Value::Array(vec![
            Value::Bytes(self.change.0.to_vec()),
            Value::Uint(self.from as u64),
            Value::Uint(self.to as u64),
        ])
    }
    fn from_value(v: &Value) -> Result<AtomRange, ObjectError> {
        let a = array(v, "atom range")?;
        if a.len() != 3 {
            return schema("atom range: expected [hash, from, to]");
        }
        let r = AtomRange {
            change: oref(&a[0], "range.change")?,
            from: u32_(&a[1], "range.from")?,
            to: u32_(&a[2], "range.to")?,
        };
        if r.from >= r.to {
            return schema("atom range: empty or inverted");
        }
        Ok(r)
    }
    pub fn resolve(self, me: ChangeHash) -> AtomRange {
        if self.change.is_zero() {
            AtomRange { change: me, ..self }
        } else {
            self
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Edit {
    Insert {
        node: NodeId,
        after: AtomRef,
        before: AtomRef,
        lines: Vec<Vec<u8>>,
    },
    Delete {
        node: NodeId,
        atoms: Vec<AtomRange>,
    },
    Replace {
        node: NodeId,
        old: Option<ObjectRef>,
        new: Option<ObjectRef>,
    },
}

impl Edit {
    pub fn node(&self) -> NodeId {
        match self {
            Edit::Insert { node, .. } | Edit::Delete { node, .. } | Edit::Replace { node, .. } => {
                *node
            }
        }
    }

    /// Whether this insert is stored as a text blob in a change of `version`.
    fn stored_text(&self, version: u64) -> bool {
        matches!(self, Edit::Insert { lines, .. }
            if version >= 2 && cbor::byte_strings_len(lines) >= TEXT_BLOB_MIN)
    }

    fn to_value(&self, version: u64, texts: &mut Option<&mut Vec<StoredText>>) -> Value {
        let nid = |n: &NodeId| Value::Bytes(n.0.to_vec());
        let opt = |o: &Option<ObjectRef>| match o {
            Some(r) => Value::Bytes(r.0.to_vec()),
            None => Value::Null,
        };
        match self {
            Edit::Insert {
                node,
                after,
                before,
                lines,
            } if self.stored_text(version) => {
                let content = cbor::encode_byte_strings(lines);
                let blob = Blob::of(&content);
                let r = blob.hash();
                if let Some(t) = texts {
                    t.push(StoredText { blob, content });
                }
                Value::map([
                    (0, Value::Uint(0)),
                    (1, nid(node)),
                    (2, after.to_value()),
                    (3, before.to_value()),
                    (5, Value::Bytes(r.0.to_vec())),
                    (6, Value::Uint(lines.len() as u64)),
                ])
            }
            Edit::Insert {
                node,
                after,
                before,
                lines,
            } => Value::map([
                (0, Value::Uint(0)),
                (1, nid(node)),
                (2, after.to_value()),
                (3, before.to_value()),
                (
                    4,
                    Value::Array(lines.iter().map(|l| Value::Bytes(l.clone())).collect()),
                ),
            ]),
            Edit::Delete { node, atoms } => Value::map([
                (0, Value::Uint(1)),
                (1, nid(node)),
                (
                    2,
                    Value::Array(atoms.iter().map(|r| r.to_value()).collect()),
                ),
            ]),
            Edit::Replace { node, old, new } => Value::map([
                (0, Value::Uint(2)),
                (1, nid(node)),
                (2, opt(old)),
                (3, opt(new)),
            ]),
        }
    }

    fn from_value(v: &Value, version: u64, resolve: &mut Resolver) -> Result<Edit, ObjectError> {
        let m = map(v, "edit")?;
        let tag = uint(field(m, 0, "edit tag")?, "edit tag")?;
        let node = node_id(field(m, 1, "edit.node")?, "edit.node")?;
        let opt = |v: &Value, what: &str| -> Result<Option<ObjectRef>, ObjectError> {
            if v.is_null() {
                Ok(None)
            } else {
                Ok(Some(oref(v, what)?))
            }
        };
        Ok(match tag {
            0 => {
                let lines = match (m.get(&4), m.get(&5)) {
                    (Some(inline), None) => {
                        let lines = array(inline, "lines")?
                            .iter()
                            .map(|l| bytes(l, "line").map(|b| b.to_vec()))
                            .collect::<Result<Vec<_>, _>>()?;
                        if version >= 2 && cbor::byte_strings_len(&lines) >= TEXT_BLOB_MIN {
                            return schema("an insert this large is stored as a text blob in a version 2 change");
                        }
                        if lines.is_empty() {
                            return schema("insert with no lines");
                        }
                        lines
                    }
                    (None, Some(text)) if version >= 2 => {
                        let text = oref(text, "insert text")?;
                        let count = u32_(field(m, 6, "insert line count")?, "insert line count")?;
                        if count == 0 {
                            return schema("insert with no lines");
                        }
                        // The resolver checks the count (`text_lines`).
                        resolve(&text, count)?
                    }
                    _ => {
                        return schema(
                            "an insert has either its lines or, in a version 2 change, a text blob",
                        )
                    }
                };
                Edit::Insert {
                    node,
                    after: AtomRef::from_value(field(m, 2, "after")?, "after")?,
                    before: AtomRef::from_value(field(m, 3, "before")?, "before")?,
                    lines,
                }
            }
            1 => {
                let atoms = array(field(m, 2, "atoms")?, "atoms")?
                    .iter()
                    .map(AtomRange::from_value)
                    .collect::<Result<Vec<_>, _>>()?;
                if atoms.is_empty() {
                    return schema("delete with no atoms");
                }
                Edit::Delete { node, atoms }
            }
            2 => Edit::Replace {
                node,
                old: opt(field(m, 2, "old")?, "old")?,
                new: opt(field(m, 3, "new")?, "new")?,
            },
            _ => return schema(format!("unknown edit tag {tag}")),
        })
    }
}

// ---------------------------------------------------------------- change

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum ChangeKind {
    Normal,
    Resolution,
    WorkingCopy,
}

impl ChangeKind {
    pub(crate) fn tag(self) -> u64 {
        match self {
            ChangeKind::Normal => 0,
            ChangeKind::Resolution => 1,
            ChangeKind::WorkingCopy => 2,
        }
    }
    pub(crate) fn from_tag(t: u64) -> Result<ChangeKind, ObjectError> {
        Ok(match t {
            0 => ChangeKind::Normal,
            1 => ChangeKind::Resolution,
            2 => ChangeKind::WorkingCopy,
            _ => return schema(format!("unknown change kind {t}")),
        })
    }
}

/// A change (spec §6.1). `deps` is kept sorted and unique by `normalize`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Change {
    /// The format this change is encoded in: [`CHANGE_VERSION`] for a new
    /// one, whatever it was made in for one read back.
    pub version: u64,
    pub change_id: ChangeId,
    pub deps: Vec<ChangeHash>,
    pub author: KeyId,
    pub hlc: Hlc,
    pub message: String,
    pub tree_ops: Vec<TreeOp>,
    pub edits: Vec<Edit>,
    pub kind: ChangeKind,
}

impl Change {
    pub fn normalize(&mut self) {
        self.deps.sort();
        self.deps.dedup();
    }

    pub fn to_value(&self) -> Value {
        self.value_with(&mut None)
    }

    fn value_with(&self, texts: &mut Option<&mut Vec<StoredText>>) -> Value {
        Value::map([
            (0, Value::Uint(self.version)),
            (1, Value::Bytes(self.change_id.0.to_vec())),
            (
                2,
                Value::Array(
                    self.deps
                        .iter()
                        .map(|d| Value::Bytes(d.0.to_vec()))
                        .collect(),
                ),
            ),
            (3, Value::Bytes(self.author.0.to_vec())),
            (4, self.hlc.to_value()),
            (5, Value::Text(self.message.clone())),
            (
                6,
                Value::Array(self.tree_ops.iter().map(|t| t.to_value()).collect()),
            ),
            (
                7,
                Value::Array(
                    self.edits
                        .iter()
                        .map(|e| e.to_value(self.version, texts))
                        .collect(),
                ),
            ),
            (8, Value::Uint(self.kind.tag())),
            (9, Value::Map(BTreeMap::new())),
        ])
    }

    pub fn encode(&self) -> Vec<u8> {
        debug_assert!(
            self.deps.windows(2).all(|w| w[0] < w[1]),
            "deps must be normalized"
        );
        cbor::encode(&self.to_value())
    }

    /// The encoding and the text blobs its large inserts are stored as,
    /// which must be stored with it.
    pub fn encode_parts(&self) -> (Vec<u8>, Vec<StoredText>) {
        debug_assert!(
            self.deps.windows(2).all(|w| w[0] < w[1]),
            "deps must be normalized"
        );
        let mut texts = Vec::new();
        let bytes = cbor::encode(&self.value_with(&mut Some(&mut texts)));
        (bytes, texts)
    }

    /// Whether any insert is stored as a text blob.
    pub fn has_stored_text(&self) -> bool {
        self.edits.iter().any(|e| e.stored_text(self.version))
    }

    pub fn hash(&self) -> ChangeHash {
        hash_object(ObjectType::Change, &self.encode())
    }

    /// Decode a change that stores no text blob; one that does needs
    /// [`Change::decode_with`] and something to read them from.
    pub fn decode(bytes: &[u8]) -> Result<Change, ObjectError> {
        Self::decode_with(bytes, &mut |r, _| {
            schema(format!(
                "text blob {} needs a store to read it from",
                r.short(12)
            ))
        })
    }

    /// Decode a change, reading each stored insert's lines with `resolve`
    /// (given the text blob and the line count the insert claims).
    pub fn decode_with(bytes: &[u8], resolve: &mut Resolver) -> Result<Change, ObjectError> {
        let v = cbor::decode(bytes)?;
        Self::from_value_with(&v, resolve)
    }

    pub fn from_value(v: &Value) -> Result<Change, ObjectError> {
        Self::from_value_with(v, &mut |r, _| {
            schema(format!(
                "text blob {} needs a store to read it from",
                r.short(12)
            ))
        })
    }

    pub(crate) fn from_value_with(
        v: &Value,
        resolve: &mut Resolver,
    ) -> Result<Change, ObjectError> {
        let m = map(v, "change")?;
        let version = uint(field(m, 0, "version")?, "version")?;
        if !(1..=CHANGE_VERSION).contains(&version) {
            return schema(format!("unsupported change format version {version}"));
        }
        let change_id = ChangeId::from_slice(bytes(field(m, 1, "change_id")?, "change_id")?)
            .ok_or_else(|| ObjectError::Schema("change_id: expected 16 bytes".into()))?;
        let deps = array(field(m, 2, "deps")?, "deps")?
            .iter()
            .map(|d| oref(d, "dep"))
            .collect::<Result<Vec<_>, _>>()?;
        if !deps.windows(2).all(|w| w[0] < w[1]) {
            return schema("deps must be sorted and unique");
        }
        if deps.iter().any(|d| d.is_zero()) {
            return schema("deps may not contain the self reference");
        }
        let extra = map(field(m, 9, "extra")?, "extra")?;
        if !extra.is_empty() {
            return schema("extra must be empty in format version 1");
        }
        Ok(Change {
            version,
            change_id,
            deps,
            author: oref(field(m, 3, "author")?, "author")?,
            hlc: Hlc::from_value(field(m, 4, "hlc")?)?,
            message: text(field(m, 5, "message")?, "message")?.to_owned(),
            tree_ops: array(field(m, 6, "tree_ops")?, "tree_ops")?
                .iter()
                .map(TreeOp::from_value)
                .collect::<Result<Vec<_>, _>>()?,
            edits: array(field(m, 7, "edits")?, "edits")?
                .iter()
                .map(|e| Edit::from_value(e, version, resolve))
                .collect::<Result<Vec<_>, _>>()?,
            kind: ChangeKind::from_tag(uint(field(m, 8, "kind")?, "kind")?)?,
        })
    }
}

/// Reads a stored insert's lines, given the text blob and the line count
/// the insert claims; [`text_lines`] reads and checks a blob's content.
/// A resolver may also return no lines, leaving a placeholder for a reader
/// that fills it in later (history-only states, spec §11).
pub type Resolver<'a> = dyn FnMut(&ObjectRef, u32) -> Result<Vec<Vec<u8>>, ObjectError> + 'a;

/// Where a history-only load finds an insert's lines when it builds that
/// file's graph (spec §11): it keeps no line until then.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TextRef {
    /// A stored insert: its text blob and line count.
    Blob(ObjectRef, u32),
    /// An inline insert: the byte range of its lines' array in the
    /// encoding of `change`.
    Inline {
        change: ChangeHash,
        start: u32,
        end: u32,
    },
}

/// The lines of an inline insert from its array's bytes (see
/// [`TextRef::Inline`]).
pub fn inline_lines(array: &[u8]) -> Result<Vec<Vec<u8>>, ObjectError> {
    let v = cbor::decode(array)?;
    let lines = array_lines(&v)?;
    if lines.is_empty() {
        return schema("insert with no lines");
    }
    Ok(lines)
}

fn array_lines(v: &Value) -> Result<Vec<Vec<u8>>, ObjectError> {
    array(v, "lines")?
        .iter()
        .map(|l| bytes(l, "line").map(|b| b.to_vec()))
        .collect()
}

impl Change {
    /// Decode a change for a history-only state (spec §11): every insert is
    /// left without its lines (an empty placeholder), and the result says
    /// where each one's lines are, in order. The inline lines are stepped
    /// over, not decoded: for a change that adds a whole tree, most of its
    /// bytes. `me` is the change's hash.
    pub fn decode_skeleton(
        bytes: &[u8],
        me: ChangeHash,
    ) -> Result<(Change, Vec<TextRef>), ObjectError> {
        let mut c = cbor::Cursor::new(bytes);
        let (major, n) = c.head()?;
        if major != 5 {
            return schema("change: expected map");
        }
        let mut fields = Map::new();
        let mut edits = Vec::new();
        let mut texts = Vec::new();
        let mut version = None;
        let mut last: Option<u64> = None;
        for _ in 0..n {
            let k = map_key(&mut c, &mut last)?;
            if k == 7 {
                let Some(v) = version else {
                    return schema("change: version must come first");
                };
                let (am, an) = c.head()?;
                if am != 4 {
                    return schema("edits: expected array");
                }
                for _ in 0..an {
                    edits.push(edit_skeleton(&mut c, v, me, &mut texts)?);
                }
                fields.insert(7, Value::Array(Vec::new()));
            } else {
                let val = c.value()?;
                if k == 0 {
                    version = Some(uint(&val, "version")?);
                }
                fields.insert(k, val);
            }
        }
        if !c.at_end() {
            return Err(ObjectError::Cbor(cbor::CborError::TrailingBytes));
        }
        let mut change = Change::from_value(&Value::Map(fields))?;
        change.edits = edits;
        Ok((change, texts))
    }
}

impl Change {
    /// A change's metadata alone: its tree operations and edits are
    /// stepped over, not decoded. For log, signature checks and walks of
    /// the dependency graph, which read nothing else.
    pub fn decode_meta(bytes: &[u8]) -> Result<crate::state::ChangeMeta, ObjectError> {
        let mut c = cbor::Cursor::new(bytes);
        let (major, n) = c.head()?;
        if major != 5 {
            return schema("change: expected map");
        }
        let mut fields = Map::new();
        let mut last: Option<u64> = None;
        for _ in 0..n {
            let k = map_key(&mut c, &mut last)?;
            if k == 6 || k == 7 {
                c.skip()?;
                fields.insert(k, Value::Array(Vec::new()));
            } else {
                fields.insert(k, c.value()?);
            }
        }
        if !c.at_end() {
            return Err(ObjectError::Cbor(cbor::CborError::TrailingBytes));
        }
        Ok(crate::state::ChangeMeta::of(&Change::from_value(
            &Value::Map(fields),
        )?))
    }
}

/// A canonical map key: an unsigned integer above the previous one.
fn map_key(c: &mut cbor::Cursor, last: &mut Option<u64>) -> Result<u64, ObjectError> {
    let (km, k) = c.head()?;
    if km != 0 || last.is_some_and(|l| k <= l) {
        return Err(ObjectError::Cbor(cbor::CborError::BadMapKey));
    }
    *last = Some(k);
    Ok(k)
}

/// One edit of [`Change::decode_skeleton`]: an insert's lines are left out.
fn edit_skeleton(
    c: &mut cbor::Cursor,
    version: u64,
    me: ChangeHash,
    texts: &mut Vec<TextRef>,
) -> Result<Edit, ObjectError> {
    let (major, n) = c.head()?;
    if major != 5 {
        return schema("edit: expected map");
    }
    let mut fields = Map::new();
    let mut inline: Option<(usize, usize, u64)> = None;
    let mut last: Option<u64> = None;
    for _ in 0..n {
        let k = map_key(c, &mut last)?;
        if k == 4 {
            let start = c.pos();
            let (am, count) = c.head()?;
            if am != 4 {
                return schema("lines: expected array");
            }
            for _ in 0..count {
                c.skip()?;
            }
            inline = Some((start, c.pos(), count));
            continue;
        }
        fields.insert(k, c.value()?);
    }
    let Some((start, end, count)) = inline else {
        return Edit::from_value(&Value::Map(fields), version, &mut |text, count| {
            texts.push(TextRef::Blob(*text, count));
            Ok(Vec::new())
        });
    };
    if uint(field(&fields, 0, "edit tag")?, "edit tag")? != 0 || fields.contains_key(&5) {
        return schema("only an insert has inline lines, and not with a text blob");
    }
    if count == 0 {
        return schema("insert with no lines");
    }
    if version >= 2 && end - start >= TEXT_BLOB_MIN {
        return schema("an insert this large is stored as a text blob in a version 2 change");
    }
    let (Ok(start), Ok(end)) = (u32::try_from(start), u32::try_from(end)) else {
        return schema("change too large");
    };
    texts.push(TextRef::Inline {
        change: me,
        start,
        end,
    });
    Ok(Edit::Insert {
        node: node_id(field(&fields, 1, "edit.node")?, "edit.node")?,
        after: AtomRef::from_value(field(&fields, 2, "after")?, "after")?,
        before: AtomRef::from_value(field(&fields, 3, "before")?, "before")?,
        lines: Vec::new(),
    })
}

/// A large insert's lines as stored (spec §6.1): the blob and its content,
/// the canonical CBOR array of the lines' byte strings.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StoredText {
    pub blob: Blob,
    pub content: Vec<u8>,
}

/// The lines a text blob's `content` holds, checked against the rules for
/// one: canonical, as many as the insert claims, and not short enough to
/// have been inline.
pub fn text_lines(content: &[u8], count: u32) -> Result<Vec<Vec<u8>>, ObjectError> {
    if content.len() < TEXT_BLOB_MIN {
        return schema(format!(
            "a text blob of {} bytes is short enough to be inline",
            content.len()
        ));
    }
    let v = cbor::decode(content)?;
    let lines = array(&v, "text blob")?
        .iter()
        .map(|l| bytes(l, "line").map(|b| b.to_vec()))
        .collect::<Result<Vec<_>, _>>()?;
    if lines.len() != count as usize {
        return schema(format!(
            "a text blob holds {} lines, the insert says {count}",
            lines.len()
        ));
    }
    Ok(lines)
}

// ---------------------------------------------------------------- blob

/// File content as an ordered list of chunks (spec §13.2).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Blob {
    pub length: u64,
    pub chunks: Vec<ObjectRef>,
}

impl Blob {
    /// The blob for `content`: its chunks as content-defined chunking cuts
    /// them (spec §13.2).
    pub fn of(content: &[u8]) -> Blob {
        Blob {
            length: content.len() as u64,
            chunks: crate::chunk::chunks(content)
                .into_iter()
                .map(|c| hash_object(ObjectType::Chunk, c))
                .collect(),
        }
    }

    pub fn encode(&self) -> Vec<u8> {
        cbor::encode(&Value::map([
            (0, Value::Uint(FORMAT_VERSION)),
            (1, Value::Uint(self.length)),
            (
                2,
                Value::Array(
                    self.chunks
                        .iter()
                        .map(|c| Value::Bytes(c.0.to_vec()))
                        .collect(),
                ),
            ),
        ]))
    }
    pub fn hash(&self) -> ObjectRef {
        hash_object(ObjectType::Blob, &self.encode())
    }
    pub fn decode(bytes: &[u8]) -> Result<Blob, ObjectError> {
        let v = cbor::decode(bytes)?;
        let m = map(&v, "blob")?;
        check_version(m)?;
        Ok(Blob {
            length: uint(field(m, 1, "length")?, "length")?,
            chunks: array(field(m, 2, "chunks")?, "chunks")?
                .iter()
                .map(|c| oref(c, "chunk"))
                .collect::<Result<Vec<_>, _>>()?,
        })
    }
}

// ---------------------------------------------------------------- checkpoint

/// A materialized state (spec §11): the heads it stands for, how many changes
/// it holds, and a blob with the serialized graphs (`State::checkpoint_payload`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Checkpoint {
    pub heads: Vec<ChangeHash>,
    pub count: u64,
    pub payload: ObjectRef,
}

impl Checkpoint {
    pub fn encode(&self) -> Vec<u8> {
        cbor::encode(&Value::map([
            (0, Value::Uint(FORMAT_VERSION)),
            (
                1,
                Value::Array(
                    self.heads
                        .iter()
                        .map(|h| Value::Bytes(h.0.to_vec()))
                        .collect(),
                ),
            ),
            (2, Value::Uint(self.count)),
            (3, Value::Bytes(self.payload.0.to_vec())),
        ]))
    }
    pub fn hash(&self) -> ObjectRef {
        hash_object(ObjectType::Checkpoint, &self.encode())
    }
    pub fn decode(bytes: &[u8]) -> Result<Checkpoint, ObjectError> {
        let v = cbor::decode(bytes)?;
        let m = map(&v, "checkpoint")?;
        check_version(m)?;
        Ok(Checkpoint {
            heads: array(field(m, 1, "heads")?, "heads")?
                .iter()
                .map(|h| oref(h, "head"))
                .collect::<Result<Vec<_>, _>>()?,
            count: uint(field(m, 2, "count")?, "count")?,
            payload: oref(field(m, 3, "payload")?, "payload")?,
        })
    }
}

// ---------------------------------------------------------------- state

/// A state named by its heads (spec §10.1): what a bookmark or tag points at.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct StateObject {
    pub heads: Vec<ChangeHash>,
}

impl StateObject {
    pub fn new(mut heads: Vec<ChangeHash>) -> StateObject {
        heads.sort();
        heads.dedup();
        StateObject { heads }
    }
    pub fn encode(&self) -> Vec<u8> {
        debug_assert!(self.heads.windows(2).all(|w| w[0] < w[1]));
        cbor::encode(&Value::map([
            (0, Value::Uint(FORMAT_VERSION)),
            (
                1,
                Value::Array(
                    self.heads
                        .iter()
                        .map(|h| Value::Bytes(h.0.to_vec()))
                        .collect(),
                ),
            ),
        ]))
    }
    pub fn hash(&self) -> ObjectRef {
        hash_object(ObjectType::State, &self.encode())
    }
    pub fn decode(bytes: &[u8]) -> Result<StateObject, ObjectError> {
        let v = cbor::decode(bytes)?;
        let m = map(&v, "state")?;
        check_version(m)?;
        let heads = array(field(m, 1, "heads")?, "heads")?
            .iter()
            .map(|h| oref(h, "head"))
            .collect::<Result<Vec<_>, _>>()?;
        if !heads.windows(2).all(|w| w[0] < w[1]) {
            return schema("state heads must be sorted and unique");
        }
        Ok(StateObject { heads })
    }
}

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

    pub(crate) fn sample_change() -> Change {
        let n = NodeId([7; 16]);
        let dep = ObjectRef([1; 32]);
        let mut c = Change {
            version: crate::objects::CHANGE_VERSION,
            change_id: ChangeId([9; 16]),
            deps: vec![ObjectRef([2; 32]), dep],
            author: ObjectRef([3; 32]),
            hlc: Hlc {
                wall_ms: 1_700_000_000_000,
                logical: 4,
            },
            message: "Add greeting".into(),
            tree_ops: vec![
                TreeOp::AddNode {
                    node: n,
                    kind: NodeKind::File,
                    mode: Mode::TEXT,
                    parent: NodeId::ROOT,
                    name: "hello.txt".into(),
                },
                TreeOp::SetMode {
                    node: n,
                    old: OpRef {
                        change: ObjectRef::ZERO,
                        index: 0,
                    },
                    mode: Mode {
                        exec: true,
                        opaque: false,
                    },
                },
            ],
            edits: vec![
                Edit::Insert {
                    node: n,
                    after: AtomRef::Start,
                    before: AtomRef::End,
                    lines: vec![b"hello\n".to_vec(), b"world\n".to_vec()],
                },
                Edit::Delete {
                    node: n,
                    atoms: vec![AtomRange {
                        change: dep,
                        from: 0,
                        to: 2,
                    }],
                },
                Edit::Replace {
                    node: n,
                    old: None,
                    new: Some(ObjectRef([5; 32])),
                },
            ],
            kind: ChangeKind::Normal,
        };
        c.normalize();
        c
    }

    #[test]
    fn change_round_trips_and_hash_is_stable() {
        let c = sample_change();
        let bytes = c.encode();
        let d = Change::decode(&bytes).unwrap();
        assert_eq!(c, d);
        assert_eq!(c.hash(), d.hash());
        assert_eq!(d.encode(), bytes);
        // Any semantic difference changes the hash.
        let mut e = c.clone();
        e.message.push('!');
        assert_ne!(e.hash(), c.hash());
    }

    #[test]
    fn decode_rejects_bad_changes() {
        let c = sample_change();
        let mut v = c.to_value();
        if let Value::Map(m) = &mut v {
            m.insert(0, Value::Uint(3));
        }
        assert!(
            matches!(Change::from_value(&v), Err(ObjectError::Schema(s)) if s.contains("version"))
        );

        let mut v = c.to_value();
        if let Value::Map(m) = &mut v {
            m.insert(
                2,
                Value::Array(vec![Value::Bytes(vec![2; 32]), Value::Bytes(vec![1; 32])]),
            );
        }
        assert!(
            matches!(Change::from_value(&v), Err(ObjectError::Schema(s)) if s.contains("sorted"))
        );

        let mut v = c.to_value();
        if let Value::Map(m) = &mut v {
            m.insert(9, Value::map([(0, Value::Null)]));
        }
        assert!(
            matches!(Change::from_value(&v), Err(ObjectError::Schema(s)) if s.contains("extra"))
        );
    }

    fn big_lines(n: usize) -> Vec<Vec<u8>> {
        (0..n)
            .map(|i| format!("line {i} of a large insert\n").into_bytes())
            .collect()
    }

    fn with_insert(mut c: Change, lines: Vec<Vec<u8>>) -> Change {
        c.edits[0] = Edit::Insert {
            node: NodeId([7; 16]),
            after: AtomRef::Start,
            before: AtomRef::End,
            lines,
        };
        c
    }

    #[test]
    fn large_inserts_are_stored_as_text_blobs() {
        assert!(
            !sample_change().has_stored_text(),
            "a small insert stays inline"
        );
        let lines = big_lines(500);
        let c = with_insert(sample_change(), lines.clone());
        assert!(c.has_stored_text());
        let (bytes, texts) = c.encode_parts();
        assert_eq!(bytes, c.encode());
        assert!(bytes.len() < 1024, "{}", bytes.len());
        assert_eq!(texts.len(), 1);
        assert_eq!(texts[0].content, cbor::encode_byte_strings(&lines));
        assert_eq!(texts[0].blob, Blob::of(&texts[0].content));
        assert!(Change::decode(&bytes).is_err(), "reading it needs the text");
        let d = Change::decode_with(&bytes, &mut |r, n| {
            assert_eq!(*r, texts[0].blob.hash());
            text_lines(&texts[0].content, n)
        })
        .unwrap();
        assert_eq!(d, c);
        assert_eq!(d.hash(), c.hash());
    }

    #[test]
    fn version_1_changes_keep_their_inline_inserts_and_hashes() {
        let mut c = with_insert(sample_change(), big_lines(500));
        c.version = 1;
        let bytes = c.encode();
        assert!(bytes.len() > TEXT_BLOB_MIN);
        let (again, texts) = c.encode_parts();
        assert!(texts.is_empty());
        assert_eq!(again, bytes);
        let d = Change::decode(&bytes).unwrap();
        assert_eq!(d.version, 1);
        assert_eq!(d.encode(), bytes);
        // The same inline insert in a version 2 change is not canonical.
        let mut v = c.to_value();
        if let Value::Map(m) = &mut v {
            m.insert(0, Value::Uint(2));
        }
        assert!(
            matches!(Change::from_value(&v), Err(ObjectError::Schema(s)) if s.contains("text blob"))
        );
    }

    #[test]
    fn a_skeleton_leaves_every_insert_out_and_says_where_its_lines_are() {
        let mut c = with_insert(sample_change(), big_lines(500));
        c.edits.push(Edit::Insert {
            node: NodeId([8; 16]),
            after: AtomRef::Start,
            before: AtomRef::End,
            lines: vec![b"small\n".to_vec(), b"insert\n".to_vec()],
        });
        let (bytes, stored) = c.encode_parts();
        let me = ObjectRef([4; 32]);
        let (skel, texts) = Change::decode_skeleton(&bytes, me).unwrap();
        assert_eq!(texts.len(), 2);
        let mut filled = skel.clone();
        let mut next = texts.iter();
        for e in filled.edits.iter_mut() {
            if let Edit::Insert { lines, .. } = e {
                assert!(lines.is_empty(), "no line kept");
                *lines = match next.next().unwrap() {
                    TextRef::Blob(r, n) => {
                        assert_eq!(*r, stored[0].blob.hash());
                        text_lines(&stored[0].content, *n).unwrap()
                    }
                    TextRef::Inline { change, start, end } => {
                        assert_eq!(*change, me);
                        inline_lines(&bytes[*start as usize..*end as usize]).unwrap()
                    }
                };
            }
        }
        assert_eq!(filled, c, "filled in, it is the change");
        // A version 1 change's large inline insert is fine; in version 2 it is not.
        let mut v1 = with_insert(sample_change(), big_lines(500));
        v1.version = 1;
        assert!(Change::decode_skeleton(&v1.encode(), me).is_ok());
        let mut v = v1.to_value();
        if let Value::Map(m) = &mut v {
            m.insert(0, Value::Uint(2));
        }
        assert!(Change::decode_skeleton(&cbor::encode(&v), me).is_err());
    }

    #[test]
    fn metadata_reads_without_the_operations() {
        let c = with_insert(sample_change(), big_lines(500));
        let (bytes, _) = c.encode_parts();
        let m = Change::decode_meta(&bytes).unwrap();
        assert_eq!(m, crate::state::ChangeMeta::of(&c));
    }

    #[test]
    fn text_blobs_are_checked() {
        let lines = big_lines(500);
        let content = cbor::encode_byte_strings(&lines);
        assert_eq!(content.len(), cbor::byte_strings_len(&lines));
        assert_eq!(
            content,
            cbor::encode(&Value::Array(
                lines.iter().map(|l| Value::Bytes(l.clone())).collect()
            ))
        );
        assert_eq!(text_lines(&content, 500).unwrap(), lines);
        assert!(text_lines(&content, 499).is_err());
        let small = cbor::encode_byte_strings(&[b"x\n".to_vec()]);
        assert!(text_lines(&small, 1).is_err(), "short enough to be inline");
    }

    #[test]
    fn names_are_validated() {
        for bad in ["", ".", "..", "a/b", "a\0b"] {
            assert!(validate_name(bad).is_err(), "{bad:?}");
        }
        assert!(validate_name("Foo.c").is_ok());
        assert!(validate_name("...").is_ok());
    }

    #[test]
    fn mode_and_kind_tags() {
        for m in [
            Mode::TEXT,
            Mode::OPAQUE,
            Mode {
                exec: true,
                opaque: true,
            },
        ] {
            assert_eq!(Mode::from_bits(m.bits()).unwrap(), m);
        }
        assert!(Mode::from_bits(4).is_err());
        for k in [
            NodeKind::File,
            NodeKind::Dir,
            NodeKind::Symlink,
            NodeKind::Subrepo,
        ] {
            assert_eq!(NodeKind::from_tag(k.tag()).unwrap(), k);
        }
    }

    #[test]
    fn self_references_resolve() {
        let me = ObjectRef([8; 32]);
        assert_eq!(
            AtomRef::Atom {
                change: ObjectRef::ZERO,
                index: 3
            }
            .resolve(me),
            AtomRef::Atom {
                change: me,
                index: 3
            }
        );
        assert_eq!(AtomRef::Start.resolve(me), AtomRef::Start);
        let other = ObjectRef([1; 32]);
        assert_eq!(
            OpRef {
                change: other,
                index: 1
            }
            .resolve(me)
            .change,
            other
        );
    }

    #[test]
    fn state_object_round_trip() {
        let st = StateObject::new(vec![
            ObjectRef([2; 32]),
            ObjectRef([1; 32]),
            ObjectRef([2; 32]),
        ]);
        assert_eq!(st.heads, vec![ObjectRef([1; 32]), ObjectRef([2; 32])]);
        assert_eq!(StateObject::decode(&st.encode()).unwrap(), st);
        assert_eq!(StateObject::new(vec![]).heads, vec![]);
        assert_ne!(StateObject::new(vec![]).hash(), st.hash());
    }

    #[test]
    fn blob_round_trip() {
        let b = Blob {
            length: 10,
            chunks: vec![ObjectRef([1; 32]), ObjectRef([2; 32])],
        };
        assert_eq!(Blob::decode(&b.encode()).unwrap(), b);
    }
}