//! Changes signed by the node that committed them: an Ed25519 signature by its node
//! key over the change's hash (`message`), kept beside the document, not in it (a
//! change's hash covers its message, so it can't carry its own signature).
//!
//! One signature vouches for more than its change: a change's hash covers its
//! dependencies, and a node's change builds on its previous one, so a signed change of
//! an actor vouches for all that actor's changes up to it (`check`: the actor and seq
//! it verifies up to). So not every change needs one: a change is signed when it
//! touches protocol data (`_continuum`) or a path the swarm lists as signed
//! (`_continuum/access/signed`, `listed`), or when its write asks for it (`Sign`); a
//! change left unsigned is vouched for by the node's next signed one, which, as
//! `--sign-every` says, is its latest change, signed a while after it if nothing signed
//! it before (`SignEvery`), or every change at once.
//!
//! Signatures travel with sync messages (`SyncEnvelope::signatures`): each actor's
//! latest one a node verified, those a peer wasn't sent yet (`newer_than`). A received
//! signature is checked once its change is there (`offer`), and kept if it vouches for
//! more of its actor's changes than any before; one of a change not there yet waits
//! (`pending`, a bounded number).

use std::collections::HashMap;

use super::{ChangeHash, Document};

/// Where the swarm lists the paths whose writes are signed: a list of paths, each a
/// list of segments, as capabilities name them (`access.md`).
pub(crate) const SIGNED_PATHS: [&str; 3] = [super::meta::ROOT, super::meta::ACCESS, "signed"];

/// What a signature is made over: a domain, then the change's hash.
pub(crate) fn message(hash: &ChangeHash) -> Vec<u8> {
    let mut message = b"continuum/change/1\0".to_vec();
    message.extend_from_slice(&hash.0);
    message
}

/// Whether a write is signed (`wanted`).
#[derive(Debug, Clone)]
pub(crate) enum Sign {
    /// The node's own work: protocol data, always signed.
    Protocol,
    /// A write through the API at `path`, signed if `asked` (`?signed=true`) or if the
    /// path is protocol data or listed as signed.
    Write { path: Vec<String>, asked: bool },
}

impl Sign {
    /// Whether a change made so, in `doc`, is to be signed.
    pub(crate) fn wanted(&self, doc: &Document) -> bool {
        match self {
            Sign::Protocol => true,
            Sign::Write { path, asked } => {
                *asked
                    || super::meta::is_protocol(path)
                    || listed(doc).iter().any(|signed| related(signed, path))
            }
        }
    }
}

/// Whether a write at `path` touches what is at `signed`: within it, or above it
/// (replacing or deleting what holds it).
fn related(signed: &[String], path: &[String]) -> bool {
    path.starts_with(signed) || signed.starts_with(path)
}

/// The paths `doc` lists as signed (`SIGNED_PATHS`); an entry that isn't a list of
/// strings is left out.
pub(crate) fn listed(doc: &Document) -> Vec<Vec<String>> {
    let at: Vec<String> = SIGNED_PATHS.iter().map(|s| (*s).to_owned()).collect();
    let Ok(Some(serde_json::Value::Array(entries))) = super::store::read_path(doc, &at) else {
        return Vec::new();
    };
    entries
        .iter()
        .filter_map(|entry| {
            entry
                .as_array()?
                .iter()
                .map(|segment| segment.as_str().map(str::to_owned))
                .collect()
        })
        .collect()
}

/// Which of this node's changes it signs besides those `Sign` says (`--sign-every`).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum SignEvery {
    /// Every change, as it is made (`all`, `true`).
    Change,
    /// No others (`none`, `false`): one left unsigned waits for the node's next signed
    /// change.
    Never,
    /// Its latest change, if unsigned, this often: every change is vouched for within it.
    Interval(std::time::Duration),
}

impl Default for SignEvery {
    fn default() -> Self {
        SignEvery::Interval(std::time::Duration::from_secs(1))
    }
}

impl std::fmt::Display for SignEvery {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            SignEvery::Change => f.write_str("all"),
            SignEvery::Never => f.write_str("none"),
            SignEvery::Interval(every) => f.write_str(&crate::units::format_duration(*every)),
        }
    }
}

impl std::str::FromStr for SignEvery {
    type Err = String;

    /// `all` or `true`, `none` or `false`, or a duration (`units::parse_duration`; 0 is
    /// `none`).
    fn from_str(text: &str) -> Result<Self, String> {
        match text.trim() {
            "all" | "true" => Ok(SignEvery::Change),
            "none" | "false" => Ok(SignEvery::Never),
            other => match crate::units::parse_duration(other) {
                Ok(every) if every.is_zero() => Ok(SignEvery::Never),
                Ok(every) => Ok(SignEvery::Interval(every)),
                Err(error) => Err(format!(
                    "{other:?} is neither all, none nor a duration ({error})"
                )),
            },
        }
    }
}

impl serde::Serialize for SignEvery {
    fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
        s.serialize_str(&self.to_string())
    }
}

impl<'de> serde::Deserialize<'de> for SignEvery {
    fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
        use serde::de::Error as _;
        match serde_json::Value::deserialize(d)? {
            serde_json::Value::Bool(all) => Ok(if all {
                SignEvery::Change
            } else {
                SignEvery::Never
            }),
            serde_json::Value::String(text) => text.parse().map_err(D::Error::custom),
            serde_json::Value::Number(number) => {
                number.to_string().parse().map_err(D::Error::custom)
            }
            other => Err(D::Error::custom(format!(
                "{other} is neither all, none nor a duration"
            ))),
        }
    }
}

/// Signatures a sync message carries at most (`newer_than`).
pub(crate) const MAX_PER_MESSAGE: usize = 256;
/// A signature as messages and the document file carry it (`pack`): the change's
/// hash (32 bytes), then the signature (64).
pub(crate) const PACKED: usize = 32 + 64;

/// `signatures` as bytes, one after another (`PACKED`).
pub(crate) fn pack(signatures: &[(ChangeHash, [u8; 64])]) -> Vec<u8> {
    let mut bytes = Vec::with_capacity(signatures.len() * PACKED);
    for (hash, signature) in signatures {
        bytes.extend_from_slice(&hash.0);
        bytes.extend_from_slice(signature);
    }
    bytes
}

/// What `pack` made; None if `bytes` aren't a whole number of signatures.
pub(crate) fn unpack(bytes: &[u8]) -> Option<Vec<(ChangeHash, [u8; 64])>> {
    if !bytes.len().is_multiple_of(PACKED) {
        return None;
    }
    Some(
        bytes
            .chunks_exact(PACKED)
            .map(|entry| {
                let (hash, signature) = entry.split_at(32);
                (
                    ChangeHash(hash.try_into().expect("32 bytes")),
                    signature.try_into().expect("64 bytes"),
                )
            })
            .collect(),
    )
}
/// Received signatures of changes not there yet, kept at most (`offer`).
const MAX_PENDING: usize = 4096;

/// A document's signatures, by change.
#[derive(Debug, Default)]
pub(crate) struct Signatures {
    by_change: HashMap<ChangeHash, [u8; 64]>,
    /// The changes signed, in the order their signatures came: the document file
    /// appends those after the ones it wrote (`since`).
    order: Vec<ChangeHash>,
    /// This node's latest change, if no signature vouches for it yet (`SignEvery`).
    unsigned: Option<ChangeHash>,
    /// By actor: its latest change a signature verified, so its changes up to it.
    latest: HashMap<super::ActorId, Latest>,
    /// Counts each change of `latest`: a peer is sent those after the last it was
    /// (`newer_than`).
    stamp: u64,
    /// Received signatures of changes the document doesn't hold yet.
    pending: HashMap<ChangeHash, [u8; 64]>,
}

/// An actor's latest change a signature verified.
#[derive(Debug, Clone, Copy)]
struct Latest {
    seq: u64,
    hash: ChangeHash,
    /// `Signatures::stamp` when it became the latest.
    stamp: u64,
}

impl Signatures {
    /// Signatures kept before, as `since` gave them (the document file), of `doc`'s
    /// changes: each actor's latest checked (`check`), as it vouches for the others.
    pub(crate) fn from_kept(doc: &Document, kept: Vec<(ChangeHash, [u8; 64])>) -> Self {
        let mut signatures = Self::default();
        let mut by_actor: HashMap<super::ActorId, (u64, ChangeHash)> = HashMap::new();
        for (hash, signature) in kept {
            if let Some(meta) = doc.get_change_meta_by_hash(&hash) {
                let entry = by_actor
                    .entry(meta.actor.into_owned())
                    .or_insert((meta.seq, hash));
                if meta.seq > entry.0 {
                    *entry = (meta.seq, hash);
                }
            }
            signatures.insert(hash, signature);
        }
        for (hash, signature) in by_actor
            .into_values()
            .map(|(_, hash)| (hash, signatures.by_change[&hash]))
            .collect::<Vec<_>>()
        {
            signatures.vouch(doc, hash, &signature);
        }
        signatures
    }

    pub(crate) fn insert(&mut self, hash: ChangeHash, signature: [u8; 64]) {
        if self.by_change.insert(hash, signature).is_none() {
            self.order.push(hash);
        }
    }

    /// `signature` of `hash`, if valid (`check`), as its actor's latest verified change
    /// where it is later than the one before; whether it is.
    fn vouch(&mut self, doc: &Document, hash: ChangeHash, signature: &[u8; 64]) -> bool {
        // Not checked where it couldn't vouch for more: a signature costs about 30 µs.
        let Some(meta) = doc.get_change_meta_by_hash(&hash) else {
            return false;
        };
        if self
            .latest
            .get(meta.actor.as_ref())
            .is_some_and(|latest| latest.seq >= meta.seq)
        {
            return false;
        }
        let Some((actor, seq)) = check(doc, &hash, signature) else {
            return false;
        };
        if self
            .latest
            .get(&actor)
            .is_some_and(|latest| latest.seq >= seq)
        {
            return false;
        }
        self.stamp += 1;
        self.latest.insert(
            actor,
            Latest {
                seq,
                hash,
                stamp: self.stamp,
            },
        );
        true
    }

    /// Received signatures, from a peer: each checked once its change is in `doc`, kept
    /// if it vouches for more of its actor's changes than any before; those of changes
    /// not there yet wait, a bounded number. Then those waiting whose changes came
    /// meanwhile. How many were kept.
    pub(crate) fn offer(&mut self, doc: &Document, received: Vec<(ChangeHash, [u8; 64])>) -> usize {
        let mut kept = 0;
        for (hash, signature) in received {
            if doc.get_change_meta_by_hash(&hash).is_none() {
                if self.pending.len() < MAX_PENDING {
                    self.pending.insert(hash, signature);
                }
            } else if self.vouch(doc, hash, &signature) {
                self.insert(hash, signature);
                kept += 1;
            }
        }
        let arrived: Vec<ChangeHash> = self
            .pending
            .keys()
            .filter(|hash| doc.get_change_meta_by_hash(hash).is_some())
            .copied()
            .collect();
        for hash in arrived {
            if let Some(signature) = self.pending.remove(&hash)
                && self.vouch(doc, hash, &signature)
            {
                self.insert(hash, signature);
                kept += 1;
            }
        }
        kept
    }

    /// Each actor's latest verified signature that became so after `stamp`, oldest first,
    /// `MAX_PER_MESSAGE` at most; with the stamp to take up from next time.
    pub(crate) fn newer_than(&self, stamp: u64) -> (Vec<(ChangeHash, [u8; 64])>, u64) {
        let mut newer: Vec<&Latest> = self
            .latest
            .values()
            .filter(|latest| latest.stamp > stamp)
            .collect();
        newer.sort_unstable_by_key(|latest| latest.stamp);
        newer.truncate(MAX_PER_MESSAGE);
        let next = newer.last().map_or(stamp, |latest| latest.stamp);
        let signatures = newer
            .into_iter()
            .map(|latest| (latest.hash, self.by_change[&latest.hash]))
            .collect();
        (signatures, next)
    }

    /// Whether a signature vouches for `actor`'s change `seq`.
    pub(crate) fn verified(&self, actor: &super::ActorId, seq: u64) -> bool {
        self.latest
            .get(actor)
            .is_some_and(|latest| latest.seq >= seq)
    }

    /// How many signatures there are.
    pub(crate) fn count(&self) -> usize {
        self.order.len()
    }

    /// The signatures after the first `count` (`count`), in the order they came.
    pub(crate) fn since(&self, count: usize) -> Vec<(ChangeHash, [u8; 64])> {
        self.order
            .get(count..)
            .unwrap_or_default()
            .iter()
            .map(|hash| (*hash, self.by_change[hash]))
            .collect()
    }

    /// Signs `hash`, this node's latest change in `doc`, with `key` and keeps the
    /// signature: it vouches for every change of the node's before it too.
    pub(crate) fn sign(
        &mut self,
        doc: &Document,
        hash: ChangeHash,
        key: &crate::identity::NodeKey,
    ) {
        let signature = key.sign(&message(&hash));
        self.insert(hash, signature);
        self.vouch(doc, hash, &signature);
        self.unsigned = None;
    }

    /// Whether this node's latest change waits for a signature (`sign_latest`).
    pub(crate) fn pending(&self) -> bool {
        self.unsigned.is_some()
    }

    /// `hash`, this node's latest change, is left unsigned for now.
    pub(crate) fn left_unsigned(&mut self, hash: ChangeHash) {
        self.unsigned = Some(hash);
    }

    /// Signs this node's latest change in `doc` if no signature vouches for it yet;
    /// whether it did.
    pub(crate) fn sign_latest(&mut self, doc: &Document, key: &crate::identity::NodeKey) -> bool {
        match self.unsigned {
            Some(hash) => {
                self.sign(doc, hash, key);
                true
            }
            None => false,
        }
    }
}

/// What a valid signature of a change in `doc` verifies: its actor's changes up to its
/// seq, committed by the node its author names (its node ID, whose key it is). None if
/// `doc` doesn't hold the change, its author isn't a node ID, or the signature isn't
/// that node's over it.
pub(crate) fn check(
    doc: &Document,
    hash: &ChangeHash,
    signature: &[u8; 64],
) -> Option<(super::ActorId, u64)> {
    let meta = doc.get_change_meta_by_hash(hash)?;
    let author = std::str::from_utf8(meta.author.as_ref()?.as_bytes()).ok()?;
    let key = crate::identity::public_of(author)?;
    crate::identity::verify(&key, &message(hash), signature)
        .then(|| (meta.actor.into_owned(), meta.seq))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::clock::UnixMillis;
    use crate::document::replica::{Writer, transact};
    use crate::document::store;
    use crate::identity::NodeKey;
    use serde_json::json;

    fn write(doc: &mut Document, node: &str, path: &[&str], value: serde_json::Value) {
        let path: Vec<String> = path.iter().map(|s| (*s).to_owned()).collect();
        let writer = Writer { node, user: None };
        transact(doc, UnixMillis(1), &writer, |doc| {
            store::write_path(doc, &path, &value)
        })
        .unwrap();
    }

    /// A signature verifies its actor's changes up to it, by the node its author names;
    /// another node's key, or another change, doesn't.
    #[test]
    fn a_signature_vouches_for_its_actors_changes_up_to_it() {
        let key = NodeKey::from_seed([1; 32]);
        let other = NodeKey::from_seed([2; 32]);
        let node = key.id();
        let mut doc = Document::new();
        write(&mut doc, &node, &["a"], json!(1));
        write(&mut doc, &node, &["b"], json!(2));
        let last = doc.get_heads()[0];
        let (actor, seq) = check(&doc, &last, &key.sign(&message(&last))).unwrap();
        assert_eq!(&actor, doc.get_actor());
        assert_eq!(seq, 2);
        assert!(check(&doc, &last, &other.sign(&message(&last))).is_none());
        let first = doc.get_changes(&[])[0].hash();
        assert!(check(&doc, &last, &key.sign(&message(&first))).is_none());
        // A change whose author isn't a node ID: nobody's to check.
        let mut named = Document::new();
        write(&mut named, "node", &["a"], json!(1));
        let hash = named.get_heads()[0];
        assert!(check(&named, &hash, &key.sign(&message(&hash))).is_none());
    }

    /// `--sign-every`: `all`, `none`, a duration; 0 is `none`.
    #[test]
    fn sign_every_reads() {
        use std::time::Duration;
        let read = |text: &str| text.parse::<SignEvery>();
        assert_eq!(read("all"), Ok(SignEvery::Change));
        assert_eq!(read("true"), Ok(SignEvery::Change));
        assert_eq!(read("none"), Ok(SignEvery::Never));
        assert_eq!(read("0"), Ok(SignEvery::Never));
        assert_eq!(
            read("500ms"),
            Ok(SignEvery::Interval(Duration::from_millis(500)))
        );
        assert!(read("sometimes").is_err());
        let toml: SignEvery = serde_json::from_value(json!(false)).unwrap();
        assert_eq!(toml, SignEvery::Never);
        assert_eq!(SignEvery::default().to_string(), "1s");
    }

    /// The latest change left unsigned is signed later, once; a signature since covers it.
    #[test]
    fn the_latest_change_is_signed_later() {
        let key = NodeKey::from_seed([1; 32]);
        let doc = Document::new();
        let mut signatures = Signatures::default();
        assert!(!signatures.sign_latest(&doc, &key));
        let a = ChangeHash([1; 32]);
        let b = ChangeHash([2; 32]);
        signatures.left_unsigned(a);
        assert!(signatures.sign_latest(&doc, &key));
        assert!(signatures.by_change.contains_key(&a));
        assert!(!signatures.sign_latest(&doc, &key));
        signatures.left_unsigned(a);
        signatures.sign(&doc, b, &key);
        assert!(!signatures.sign_latest(&doc, &key));
        // Kept in the order they came, each once.
        assert_eq!(signatures.count(), 2);
        assert_eq!(signatures.since(1)[0].0, b);
        assert!(signatures.since(5).is_empty());
        let again = Signatures::from_kept(&doc, signatures.since(0));
        assert_eq!(again.since(0), signatures.since(0));
    }

    /// Received signatures: one vouches for its actor's changes up to it, an earlier one
    /// adds nothing, one of a change not there yet waits for it, a forged one is
    /// dropped; each actor's latest goes on to peers once (`newer_than`).
    #[test]
    fn received_signatures_vouch_for_their_actors_changes() {
        let key = NodeKey::from_seed([1; 32]);
        let other = NodeKey::from_seed([2; 32]);
        let mut theirs = Document::new();
        write(&mut theirs, &key.id(), &["a"], json!(1));
        let first = theirs.get_heads()[0];
        write(&mut theirs, &key.id(), &["b"], json!(2));
        let second = theirs.get_heads()[0];
        let actor = theirs.get_actor().clone();
        let signed = |hash: ChangeHash| (hash, key.sign(&message(&hash)));

        let mut ours = Document::new();
        ours.load_incremental(&theirs.save_after(&[])).unwrap();
        let mut signatures = Signatures::default();
        assert!(!signatures.verified(&actor, 1));
        assert_eq!(signatures.offer(&ours, vec![signed(second)]), 1);
        assert!(signatures.verified(&actor, 1) && signatures.verified(&actor, 2));
        // An earlier one vouches for nothing more; a forged one for nothing.
        assert_eq!(signatures.offer(&ours, vec![signed(first)]), 0);
        assert_eq!(
            signatures.offer(&ours, vec![(second, other.sign(&message(&second)))]),
            0
        );
        let (newer, stamp) = signatures.newer_than(0);
        assert_eq!(newer, vec![signed(second)]);
        assert!(signatures.newer_than(stamp).0.is_empty());

        // A change not there yet: its signature waits for it.
        write(&mut theirs, &key.id(), &["c"], json!(3));
        let third = theirs.get_heads()[0];
        assert_eq!(signatures.offer(&ours, vec![signed(third)]), 0);
        assert!(!signatures.verified(&actor, 3));
        ours.load_incremental(&theirs.save_after(&[second]))
            .unwrap();
        assert_eq!(signatures.offer(&ours, Vec::new()), 1);
        assert!(signatures.verified(&actor, 3));
        assert_eq!(signatures.newer_than(stamp).0, vec![signed(third)]);

        // Kept in a file and read again: verified as before.
        let again = Signatures::from_kept(&ours, signatures.since(0));
        assert!(again.verified(&actor, 3));
    }

    /// Protocol data, listed paths (within or above them) and asked-for writes are
    /// signed; others aren't.
    #[test]
    fn which_writes_are_signed() {
        let mut doc = Document::new();
        let write_at = |path: &[&str], asked| Sign::Write {
            path: path.iter().map(|s| (*s).to_owned()).collect(),
            asked,
        };
        assert!(Sign::Protocol.wanted(&doc));
        assert!(write_at(&["_continuum", "users", "a"], false).wanted(&doc));
        assert!(!write_at(&["metrics", "cpu"], false).wanted(&doc));
        assert!(write_at(&["metrics", "cpu"], true).wanted(&doc));
        write(
            &mut doc,
            "node",
            &SIGNED_PATHS,
            json!([["config"], ["servers", "eu"], "junk", [1]]),
        );
        assert_eq!(
            listed(&doc),
            vec![
                vec!["config".to_owned()],
                vec!["servers".to_owned(), "eu".to_owned()]
            ]
        );
        assert!(write_at(&["config", "x"], false).wanted(&doc));
        assert!(write_at(&["servers"], false).wanted(&doc));
        assert!(!write_at(&["servers", "us"], false).wanted(&doc));
        assert!(!write_at(&["configs"], false).wanted(&doc));
    }
}