//! 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 the changes of that actor it builds on: in the document, all
//! up to it (`check`: the actor and seq it verifies up to); received, not in it yet,
//! those among them it builds on (`access::screen`). Only those: anyone can make a
//! change under any actor, and one at a seq whose change the document holds is a fork
//! (`forks`). 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 (`SignEvery`), is its latest change, signed as it leaves in a
//! sync message (`sent`), or every change at once (`all`).
//!
//! Signatures travel with the changes they are of, in the sync messages that carry them
//! (`SyncEnvelope::signatures`, `for_changes`). 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: the oldest go).

use std::collections::{HashMap, HashSet};

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`, a pattern as capabilities and
/// signed values take them (`*` for any one key): within it, or above it (replacing or
/// deleting what holds it).
fn related(signed: &[String], path: &[String]) -> bool {
    crate::access::agree(signed, 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_plain(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, Default)]
pub(crate) enum SignEvery {
    /// Every change, as it is made (`all`, `true`).
    Change,
    /// Its latest change, if unsigned, as a sync message takes it (`sent`): one
    /// signature per message, however many changes it carries, each vouching for those
    /// before it.
    #[default]
    Sent,
    /// No others (`none`, `false`): one left unsigned waits for the node's next signed
    /// change; unless the policy requires signatures, then as `sent`.
    Never,
}

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

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

    /// `all` or `true`, `sent`, `none` or `false`.
    fn from_str(text: &str) -> Result<Self, String> {
        match text.trim() {
            "all" | "true" => Ok(SignEvery::Change),
            "sent" => Ok(SignEvery::Sent),
            "none" | "false" => Ok(SignEvery::Never),
            other => Err(format!("{other:?} is neither all, sent nor none")),
        }
    }
}

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),
            other => Err(D::Error::custom(format!(
                "{other} is neither all, sent nor none"
            ))),
        }
    }
}

/// Signatures a sync message carries at most, and as many wait for their changes
/// (`MAX_PENDING`): 6 MiB of them. A message carries one per writer it holds changes of
/// at most (`for_changes`, `for_delta`), and a writer is a node's run (each start, rejoin
/// or drop makes a fresh actor) since the epoch began: a transfer of an epoch with more
/// runs writing than this, a whole document pulled or a delta, leaves the rest unvouched
/// for, and a node taking it under a policy requiring signatures waits for them. A remake
/// starts the count again.
pub(crate) const MAX_PER_MESSAGE: usize = 1 << 16;
/// Past this many signed changes among those a message carries, it carries each
/// writer's latest among them instead (`for_changes`): they vouch for the rest.
const EACH_WRITER_PAST: 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`): the oldest go.
/// As many as a message carries, so a delta's (whose changes come after it) all wait.
const MAX_PENDING: usize = MAX_PER_MESSAGE;

/// Received signatures of changes the document doesn't hold yet, `MAX_PENDING` at most:
/// the oldest go, so junk a peer sent once (signatures of changes that never come)
/// makes way for those that keep coming.
#[derive(Debug, Default, Clone)]
struct Pending {
    by_change: HashMap<ChangeHash, [u8; 64]>,
    /// The changes, oldest first; some may have left `by_change` since.
    order: std::collections::VecDeque<ChangeHash>,
}

impl Pending {
    fn insert(&mut self, hash: ChangeHash, signature: [u8; 64]) {
        if self.by_change.insert(hash, signature).is_none() {
            self.order.push_back(hash);
        }
        while self.by_change.len() > MAX_PENDING {
            match self.order.pop_front() {
                Some(oldest) => {
                    self.by_change.remove(&oldest);
                }
                None => break,
            }
        }
        if self.order.len() > 2 * MAX_PENDING {
            let by_change = &self.by_change;
            self.order.retain(|hash| by_change.contains_key(hash));
        }
    }

    fn remove(&mut self, hash: &ChangeHash) -> Option<[u8; 64]> {
        self.by_change.remove(hash)
    }

    fn keys(&self) -> impl Iterator<Item = &ChangeHash> {
        self.by_change.keys()
    }
}

/// 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>,
    /// Received signatures of changes the document doesn't hold yet.
    pending: Pending,
}

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

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
    }

    /// These for `doc`, replacing the document they are of (one without some of its
    /// changes, or another of its epoch): those of the changes it holds, each actor's
    /// latest checked again (`from_kept`); those waiting for their changes still wait.
    pub(crate) fn kept_for(&self, doc: &Document) -> Self {
        let kept = self
            .order
            .iter()
            .filter(|hash| super::changes::holds(doc, hash))
            .map(|hash| (*hash, self.by_change[hash]))
            .collect();
        Self {
            pending: self.pending.clone(),
            ..Self::from_kept(doc, kept)
        }
    }

    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.latest.insert(actor, Latest { seq, hash });
        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 !super::changes::holds(doc, &hash) {
                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| super::changes::holds(doc, hash))
            .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
    }

    /// The signatures to send with a delta from `from` (`SyncBody::Delta`): of each
    /// writer with changes in it, its latest verified, vouching for its changes before
    /// it; `MAX_PER_MESSAGE` at most. The delta's changes come after it, so these wait
    /// for them at the peer (`offer`).
    pub(crate) fn for_delta(
        &self,
        doc: &Document,
        from: &[ChangeHash],
    ) -> Vec<(ChangeHash, [u8; 64])> {
        let writers: HashSet<super::ActorId> = doc
            .get_changes_meta(from)
            .into_iter()
            .map(|meta| meta.actor.into_owned())
            .collect();
        writers
            .iter()
            .filter_map(|writer| self.latest.get(writer))
            .map(|latest| (latest.hash, self.by_change[&latest.hash]))
            .take(MAX_PER_MESSAGE)
            .collect()
    }

    /// The signatures to send with `changes`, those a message carries: of each of them
    /// signed, each vouching for its writer's changes before it among them. Whatever the
    /// peer was sent before, so changes sent again (a message lost on the way, then
    /// repaired) arrive vouched for as the first time. Past `EACH_WRITER_PAST` (a whole
    /// document pulled, a message carrying the batches of many writers), each writer's
    /// latest among them, which vouches for the rest: one per writer,
    /// `MAX_PER_MESSAGE` at most.
    pub(crate) fn for_changes(
        &self,
        doc: &Document,
        changes: &[ChangeHash],
    ) -> Vec<(ChangeHash, [u8; 64])> {
        let signed: Vec<(ChangeHash, [u8; 64])> = changes
            .iter()
            .filter_map(|hash| {
                self.by_change
                    .get(hash)
                    .map(|signature| (*hash, *signature))
            })
            .collect();
        if signed.len() <= EACH_WRITER_PAST {
            return signed;
        }
        let signed: HashMap<ChangeHash, [u8; 64]> = signed.into_iter().collect();
        // Each signed one's writer and place among its changes: the metadata of the
        // document's changes, read once (a message this large carries most of them).
        let mut latest: HashMap<super::ActorId, (u64, ChangeHash)> = HashMap::new();
        for meta in doc.get_changes_meta(&[]) {
            if signed.contains_key(&meta.hash) {
                let at = latest
                    .entry(meta.actor.into_owned())
                    .or_insert((meta.seq, meta.hash));
                if meta.seq > at.0 {
                    *at = (meta.seq, meta.hash);
                }
            }
        }
        latest
            .into_values()
            .map(|(_, hash)| (hash, signed[&hash]))
            .take(MAX_PER_MESSAGE)
            .collect()
    }

    /// Received signatures waiting for their changes (`offer`) checked against
    /// `changes`, not in the document yet, each made by its committer (`committers`, as
    /// `access::screen` finds them): one valid vouches for its actor's changes up to it,
    /// these among them.
    pub(crate) fn vouch_batch(&mut self, changes: &[super::Change], committers: &[Option<String>]) {
        for (change, committer) in changes.iter().zip(committers) {
            let hash = change.hash();
            let Some(signature) = self.pending.remove(&hash) else {
                continue;
            };
            let valid = committer
                .as_deref()
                .and_then(crate::identity::public_of)
                .is_some_and(|key| crate::identity::verify(&key, &message(&hash), &signature));
            if !valid {
                continue;
            }
            self.insert(hash, signature);
            let actor = change.actor_id();
            if self
                .latest
                .get(actor)
                .is_none_or(|latest| latest.seq < change.seq())
            {
                self.latest.insert(
                    actor.clone(),
                    Latest {
                        seq: change.seq(),
                        hash,
                    },
                );
            }
        }
    }

    /// The latest change of each of this node's `actors` in `doc` that no signature
    /// vouches for, signed: one it made and kept before it stopped, under an actor of an
    /// earlier run, never sent (it signs what it sends, as it sends it); one left
    /// unsigned below a signed change a drop took away (`drop_taken`, a fresh actor
    /// after it). Peers pulling such a change unsigned would wait for a signature for
    /// good under a policy requiring them, and for everything built on it. At start,
    /// and after the document is swapped. How many it signed.
    pub(crate) fn sign_own_actors(
        &mut self,
        doc: &mut Document,
        key: &crate::identity::NodeKey,
        actors: &[super::ActorId],
    ) -> usize {
        let current = doc.get_actor().clone();
        let mut signed = 0;
        for actor in actors.iter().chain(std::iter::once(&current)) {
            doc.set_actor(actor.clone());
            let Some(change) = doc.get_last_local_change() else {
                continue;
            };
            if !self.verified(change.actor_id(), change.seq()) {
                self.sign_change(doc, change.hash(), key);
                signed += 1;
            }
        }
        doc.set_actor(current.clone());
        if doc
            .get_last_local_change()
            .is_some_and(|change| self.verified(change.actor_id(), change.seq()))
        {
            self.unsigned = None;
        }
        signed
    }

    /// Whether a valid signature of `hash` is kept: it vouches for its change and the
    /// changes of its actor it builds on.
    pub(crate) fn signed(&self, hash: &ChangeHash) -> bool {
        self.by_change.contains_key(hash)
    }

    /// Whether `change`, not in `doc`, forks its actor's changes: `doc` holds a change
    /// a signature verified of that actor at a seq as late, so its real change at
    /// `change`'s seq too. Anyone can make a change under any actor; only its node can
    /// sign one.
    pub(crate) fn forks(&self, doc: &Document, change: &super::Change) -> bool {
        self.latest.get(change.actor_id()).is_some_and(|latest| {
            latest.seq >= change.seq() && super::changes::holds(doc, &latest.hash)
        })
    }

    /// 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,
    ) {
        self.sign_change(doc, hash, key);
        self.unsigned = None;
    }

    /// `hash` signed with `key` and kept, vouching for its actor's changes up to it.
    fn sign_change(&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);
    }

    /// Whether this node's latest change waits for a signature (`sign_latest`, as a sync
    /// message takes it).
    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::ActorId;
    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();
    }

    /// Past `EACH_WRITER_PAST` signed changes, a message carries each writer's latest
    /// among them; a delta, each writer's with changes in it.
    #[test]
    fn each_writers_latest_past_many() {
        let key = NodeKey::from_seed([1; 32]);
        let mut doc = Document::new();
        let mut signatures = Signatures::default();
        let mut hashes = Vec::new();
        for i in 0..=EACH_WRITER_PAST {
            write(&mut doc, &key.id(), &["n"], json!(i));
            let hash = doc.get_heads()[0];
            signatures.sign(&doc, hash, &key);
            hashes.push(hash);
        }
        let last = *hashes.last().unwrap();
        assert_eq!(signatures.for_changes(&doc, &hashes).len(), 1);
        assert_eq!(signatures.for_changes(&doc, &hashes)[0].0, last);
        // Fewer than that: each signed one.
        assert_eq!(signatures.for_changes(&doc, &hashes[..10]).len(), 10);
        // Another writer's changes alone in a delta: only its latest.
        let mut theirs = doc.fork().with_actor(ActorId::random());
        write(&mut theirs, &key.id(), &["m"], json!(1));
        let other = theirs.get_heads()[0];
        signatures.sign(&theirs, other, &key);
        let delta = signatures.for_delta(&theirs, &[last]);
        assert_eq!(delta.len(), 1);
        assert_eq!(delta[0].0, other);
    }

    /// A listed signed path is a pattern: `*` takes any one key, within and above.
    #[test]
    fn listed_paths_take_a_star() {
        let path = |text: &str| -> Vec<String> { text.split('/').map(str::to_owned).collect() };
        let signed = path("metrics/*/alerts");
        assert!(related(&signed, &path("metrics/eu/alerts/x")));
        assert!(related(&signed, &path("metrics/eu")));
        assert!(related(&signed, &path("metrics")));
        assert!(!related(&signed, &path("metrics/eu/rates")));
        assert!(!related(&signed, &path("other")));
    }

    /// The latest unsigned change of each of this node's actors is signed, below the
    /// heads too (an earlier actor's, built on under a later one); another node's, and
    /// those vouched for, left.
    #[test]
    fn own_actors_latest_signed() {
        let key = NodeKey::from_seed([1; 32]);
        let other = NodeKey::from_seed([2; 32]);
        let mut doc = Document::new();
        let earlier = doc.get_actor().clone();
        write(&mut doc, &key.id(), &["a"], json!(1));
        // Another node's, concurrent.
        let mut theirs = Document::new();
        write(&mut theirs, &other.id(), &["b"], json!(2));
        doc.merge(&mut theirs).unwrap();
        // A later actor of this node builds on the earlier one's change: not a head now.
        let mut doc = doc.with_actor(ActorId::random());
        write(&mut doc, &key.id(), &["c"], json!(3));
        let mut signatures = Signatures::default();
        assert_eq!(
            signatures.sign_own_actors(&mut doc, &key, std::slice::from_ref(&earlier)),
            2
        );
        let ours = doc.get_last_local_change().unwrap();
        assert!(signatures.verified(ours.actor_id(), ours.seq()));
        assert!(signatures.verified(&earlier, 1));
        assert_eq!(signatures.sign_own_actors(&mut doc, &key, &[earlier]), 0);
        assert!(signatures.unsigned.is_none());
    }

    /// 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`, `sent`, `none`.
    #[test]
    fn sign_every_reads() {
        let read = |text: &str| text.parse::<SignEvery>();
        assert_eq!(read("all"), Ok(SignEvery::Change));
        assert_eq!(read("true"), Ok(SignEvery::Change));
        assert_eq!(read("sent"), Ok(SignEvery::Sent));
        assert_eq!(read("none"), Ok(SignEvery::Never));
        assert!(read("1s").is_err());
        let toml: SignEvery = serde_json::from_value(json!(false)).unwrap();
        assert_eq!(toml, SignEvery::Never);
        assert_eq!(SignEvery::default().to_string(), "sent");
    }

    /// 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
        );
        assert_eq!(signatures.for_delta(&ours, &[]), vec![signed(second)]);
        assert!(signatures.for_delta(&ours, &[second]).is_empty());
        // Sent with the changes they are of: the signed one's alone.
        assert_eq!(
            signatures.for_changes(&ours, &[first, second]),
            vec![signed(second)]
        );
        assert!(signatures.for_changes(&ours, &[first]).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.for_delta(&ours, &[]), 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));

        // For a document without the last change: those of the changes it holds.
        let without = ours.fork_at(&[second]).unwrap();
        let kept = signatures.kept_for(&without);
        assert!(kept.verified(&actor, 2) && !kept.verified(&actor, 3));
        assert_eq!(kept.since(0), vec![signed(second)]);
    }

    /// Signatures waiting for their changes: past the bound, the oldest go.
    #[test]
    fn the_oldest_waiting_signatures_go() {
        let mut pending = Pending::default();
        let hash = |i: usize| {
            let mut bytes = [0; 32];
            bytes[..8].copy_from_slice(&(i as u64).to_le_bytes());
            ChangeHash(bytes)
        };
        for i in 0..=MAX_PENDING {
            pending.insert(hash(i), [0; 64]);
        }
        assert_eq!(pending.keys().count(), MAX_PENDING);
        assert!(pending.remove(&hash(0)).is_none());
        assert!(pending.remove(&hash(MAX_PENDING)).is_some());
    }

    /// 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));
    }
}