//! Received changes judged by rules by path before they are applied (`rules::judge`),
//! and applied if allowed (`apply`): a refused change stays out, and so does whatever
//! builds on it; it is kept aside for an operator to look at (`Refused`).
//!
//! A change is judged at the heads its writer saw. Those of a batch aren't in the
//! document yet, so one is judged at the changes of the document its batch ancestors
//! build on: what rules read there (the members, the user registry, remake proposals,
//! the policy, bans) is what its writer saw, unless one of those ancestors changed it. So a change that
//! does (`relevant`) is applied before the next is judged: the rest go in at once.
//!
//! A member whose own change broke a rule, as its signature proves (signatures
//! required), is refused as a whole (`Screening::refusing`): the node sends it nothing
//! more, as all it commits builds on what was refused, until a change of its is taken
//! again (it rejoined) or the epoch changes. One that only passed a refused change on
//! isn't: the committer is, whoever brought it.
//!
//! A change refused is remembered by its hash (`Refusals`): when it comes again, as its
//! writer keeps offering what the node never takes, it is refused again without being
//! decoded or logged, and the sync session with whoever sent only such changes backs off
//! (`Outcome::refused_again`, `SyncSession::refused_again`).
//!
//! A change whose dependencies are neither in the document nor among those received
//! waits (`Waiting`) until they come, a bounded number; Automerge never sees it before.
//! With signatures required (`access::policy`), so does one no signature by its
//! committer vouches for yet (`Signatures::vouch_batch`), and whatever builds on it,
//! until one comes, but an epoch's root, which no node signs.

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

use anyhow::Result;

use super::rules::{self, Refusal};
use crate::document::inspect::{Inspector, Op, Path};
use crate::document::meta;
use crate::document::replica::{Epoch, user_of};
use crate::document::signatures::Signatures;
use crate::document::{Change, ChangeHash, Document, ReadDoc};

/// Changes waiting for their dependencies, kept at most.
const MAX_WAITING: usize = 10_000;
/// Refused changes kept for operators, at most: the oldest go.
pub(crate) const MAX_REFUSED: usize = 1_000;
/// Hashes of refused changes remembered, at most: the oldest go (and are judged again,
/// and refused again, should they come).
const MAX_REFUSALS: usize = 100_000;

/// A refused change, and why.
#[derive(Debug, Clone)]
pub(crate) struct Refused {
    pub(crate) change: Change,
    pub(crate) committer: Option<String>,
    pub(crate) user: Option<String>,
    pub(crate) paths: BTreeSet<Path>,
    pub(crate) why: String,
}

/// What screening keeps between batches, for one document (`AppState::screening`): the
/// changes waiting for their dependencies, and those refused, for operators
/// (`GET /admin/refused`).
#[derive(Debug, Default)]
pub(crate) struct Screening {
    /// The document's (`Replica::epoch_seq`): another's start empty (`of`).
    epoch_seq: u32,
    pub(crate) waiting: Waiting,
    refused: std::collections::VecDeque<Refused>,
    refusals: Refusals,
    /// The members refused as a whole, by node ID (`refusing`).
    refusing: std::collections::BTreeMap<String, Refusing>,
}

/// A member refused as a whole: since when, for which change of its, and why.
#[derive(Debug, Clone)]
pub(crate) struct Refusing {
    pub(crate) since: tokio::time::Instant,
    pub(crate) change: ChangeHash,
    pub(crate) why: String,
}

/// The hashes of the changes refused, to refuse them again by hash.
#[derive(Debug, Default)]
pub(crate) struct Refusals {
    known: HashSet<ChangeHash>,
    order: std::collections::VecDeque<ChangeHash>,
}

impl Refusals {
    fn contains(&self, hash: &ChangeHash) -> bool {
        self.known.contains(hash)
    }

    fn insert(&mut self, hash: ChangeHash) {
        if self.known.insert(hash) {
            self.order.push_back(hash);
            if self.order.len() > MAX_REFUSALS
                && let Some(oldest) = self.order.pop_front()
            {
                self.known.remove(&oldest);
            }
        }
    }
}

impl Screening {
    /// Itself for the document `epoch_seq`: emptied if it was another's.
    pub(crate) fn of(&mut self, epoch_seq: u32) -> &mut Self {
        if self.epoch_seq != epoch_seq {
            *self = Screening {
                epoch_seq,
                ..Screening::default()
            };
        }
        self
    }

    /// Keeps `refused`, logged, the oldest going past `MAX_REFUSED`.
    pub(crate) fn record(&mut self, refused: Vec<Refused>) {
        for refused in refused {
            tracing::warn!(
                change = %refused.change.hash(),
                committer = refused.committer.as_deref().unwrap_or("none"),
                user = refused.user.as_deref().unwrap_or("none"),
                why = %refused.why,
                "Refused a change (_continuum/access/policy)"
            );
            self.refused.push_back(refused);
            if self.refused.len() > MAX_REFUSED {
                self.refused.pop_front();
            }
        }
    }

    /// Those refused, oldest first.
    pub(crate) fn refused(&self) -> impl DoubleEndedIterator<Item = &Refused> {
        self.refused.iter()
    }

    /// Whether `node` is refused as a whole, and since when, for what.
    pub(crate) fn refusing(&self, node: &str) -> Option<&Refusing> {
        self.refusing.get(node)
    }

    /// The members refused as a whole and taken again of `screened`, logged.
    fn judge_members(&mut self, screened: &Screened) {
        for node in &screened.taken_from {
            if self.refusing.remove(node).is_some() {
                tracing::info!(node = %node, "A member refused as a whole wrote a change taken: refused no longer");
            }
        }
        for (node, change, why) in &screened.broke_rules {
            if !self.refusing.contains_key(node) {
                tracing::warn!(node = %node, change = %change, why = %why, "A member's own change broke the rules: refusing it as a whole, sending it nothing");
                self.refusing.insert(
                    node.clone(),
                    Refusing {
                        since: tokio::time::Instant::now(),
                        change: *change,
                        why: why.clone(),
                    },
                );
            }
        }
    }
}

/// Received changes waiting for their dependencies.
#[derive(Debug, Default)]
pub(crate) struct Waiting(Vec<Change>);

impl Waiting {
    pub(crate) fn len(&self) -> usize {
        self.0.len()
    }
}

/// What `apply` did: how many changes it applied, and those it refused.
#[derive(Debug, Default)]
pub(crate) struct Screened {
    pub(crate) applied: usize,
    pub(crate) refused: Vec<Refused>,
    /// Received changes refused before (`Refusals`), refused again by their hash.
    pub(crate) refused_again: usize,
    /// Members whose own changes broke a rule, a signature proving it: with the change,
    /// and why.
    pub(crate) broke_rules: Vec<(String, ChangeHash, String)>,
    /// The committers of changes taken.
    pub(crate) taken_from: BTreeSet<String>,
}

/// Judges `received` (and those `waiting` before) and applies to `doc` (of `epoch`)
/// those rules allow, what they change recorded in `log`. Those already in `doc` are
/// left out; those whose dependencies are missing wait; those `refusals` holds are
/// refused again by their hash, and those building on them, and those refused now go
/// into it. With `signatures` (required), those no signature vouches for wait too.
pub(crate) fn apply(
    doc: &mut Document,
    epoch: Epoch,
    received: Vec<Change>,
    waiting: &mut Waiting,
    refusals: &mut Refusals,
    mut signatures: Option<&mut Signatures>,
    log: &mut crate::document::PatchLog,
) -> Result<Screened> {
    let mut screened = Screened::default();
    // Each once, not in the document yet, not refused before.
    let mut seen = HashSet::new();
    let candidates: Vec<Change> = std::mem::take(&mut waiting.0)
        .into_iter()
        .chain(received)
        .filter(|change| seen.insert(change.hash()))
        .filter(|change| doc.get_change_by_hash(&change.hash()).is_none())
        .filter(|change| {
            let again = refusals.contains(&change.hash());
            screened.refused_again += usize::from(again);
            !again
        })
        .collect();
    // Building on one refused before: refused too, without waiting for it (it never
    // comes) or judging it.
    let (building, candidates): (Vec<Change>, Vec<Change>) = candidates
        .into_iter()
        .partition(|change| change.deps().iter().any(|dep| refusals.contains(dep)));
    for change in building {
        let dep = change
            .deps()
            .iter()
            .find(|dep| refusals.contains(dep))
            .copied()
            .expect("partitioned so");
        refusals.insert(change.hash());
        screened.refused.push(Refused {
            committer: change.author().and_then(|author| {
                std::str::from_utf8(author.as_bytes())
                    .ok()
                    .map(str::to_owned)
            }),
            user: change.message().and_then(user_of).map(str::to_owned),
            paths: BTreeSet::new(),
            why: format!("it builds on refused change {dep}"),
            change,
        });
    }
    let (ready, unready) = in_order(doc, candidates);
    waiting.0 = unready;
    if ready.is_empty() {
        waiting.0.truncate(MAX_WAITING);
        return Ok(screened);
    }

    let decoded: Vec<crate::document::ExpandedChange> =
        ready.iter().map(|change| change.decode()).collect();
    let ops: Vec<Vec<Op>> = Inspector::new(doc).ops(&decoded);
    let paths: Vec<BTreeSet<Path>> = ops
        .iter()
        .map(|ops| ops.iter().map(|op| op.path.clone()).collect())
        .collect();
    // Who made each: its author, or that of its actor's first change (in the document,
    // or among these).
    let mut authors: HashMap<crate::document::ActorId, String> = HashMap::new();
    let committers: Vec<Option<String>> = ready
        .iter()
        .map(|change| {
            let actor = change.actor_id();
            let named = change
                .author()
                .or_else(|| doc.get_author_for_actor(actor))
                .and_then(|author| {
                    std::str::from_utf8(author.as_bytes())
                        .ok()
                        .map(str::to_owned)
                })
                .or_else(|| authors.get(actor).cloned());
            if let Some(named) = &named {
                authors.insert(actor.clone(), named.clone());
            }
            named
        })
        .collect();
    if let Some(signatures) = signatures.as_deref_mut() {
        signatures.vouch_batch(&ready, &committers);
    }

    // The changes of the document each accepted one, not applied yet, builds on.
    let mut frontier: HashMap<ChangeHash, Vec<ChangeHash>> = HashMap::new();
    let mut refused: HashSet<ChangeHash> = HashSet::new();
    let mut unsigned: HashSet<ChangeHash> = HashSet::new();
    let mut accepted: Vec<Change> = Vec::new();
    for (index, change) in ready.iter().enumerate() {
        let hash = change.hash();
        let committer = committers[index].as_deref();
        let user = change.message().and_then(user_of).map(str::to_owned);
        let refuse = |why: String| Refused {
            change: change.clone(),
            committer: committer.map(str::to_owned),
            user: user.clone(),
            paths: paths[index].clone(),
            why,
        };
        if let Some(dep) = change
            .deps()
            .iter()
            .find(|dep| refused.contains(dep) || refusals.contains(dep))
        {
            refused.insert(hash);
            screened
                .refused
                .push(refuse(format!("it builds on refused change {dep}")));
            continue;
        }
        // No signature vouches for it yet, or for what it builds on: it waits.
        let vouched = |signatures: &Signatures| {
            change.deps().is_empty() || signatures.verified(change.actor_id(), change.seq())
        };
        if signatures
            .as_deref()
            .is_some_and(|signatures| !vouched(signatures))
            || change.deps().iter().any(|dep| unsigned.contains(dep))
        {
            unsigned.insert(hash);
            waiting.0.push(change.clone());
            continue;
        }
        let mut seen_at: BTreeSet<ChangeHash> = BTreeSet::new();
        for dep in change.deps() {
            match frontier.get(dep) {
                Some(heads) => seen_at.extend(heads.iter().copied()),
                None => {
                    seen_at.insert(*dep);
                }
            }
        }
        let seen_at: Vec<ChangeHash> = seen_at.into_iter().collect();
        let judged = rules::judge(
            doc,
            epoch,
            &rules::Change {
                hash,
                deps: change.deps(),
                seen: &seen_at,
                committer,
                user: user.as_deref(),
                paths: &paths[index],
                ops: &ops[index],
            },
        );
        match judged {
            Ok(()) => {
                accepted.push(change.clone());
                if let Some(committer) = committer {
                    screened.taken_from.insert(committer.to_owned());
                }
                if relevant(&paths[index]) {
                    // What rules read changed: in before the next is judged.
                    screened.applied += accepted.len();
                    doc.apply_changes_log_patches(std::mem::take(&mut accepted), log)?;
                    frontier.clear();
                } else {
                    frontier.insert(hash, seen_at);
                }
            }
            Err(Refusal::Unready) => {
                // Can't be: its dependencies are in the document or among these.
                refused.insert(hash);
                screened.refused.push(refuse(Refusal::Unready.to_string()));
            }
            Err(Refusal::Rule(why)) => {
                refused.insert(hash);
                // Its committer's own, as a signature by it vouches.
                if signatures.is_some()
                    && let Some(committer) = committer
                {
                    screened
                        .broke_rules
                        .push((committer.to_owned(), hash, why.clone()));
                }
                screened.refused.push(refuse(why));
            }
        }
    }
    screened.applied += accepted.len();
    if !accepted.is_empty() {
        doc.apply_changes_log_patches(accepted, log)?;
    }
    for hash in refused {
        refusals.insert(hash);
    }
    waiting.0.truncate(MAX_WAITING);
    Ok(screened)
}

/// What `apply_bytes` did.
#[derive(Debug, Default, Clone, Copy)]
pub(crate) struct Outcome {
    /// Changes applied (received, or waiting before).
    pub(crate) applied: usize,
    /// Whether the bytes carried changes, and all of them refused before: their sender
    /// offers again what the node never takes, and its sync session backs off.
    pub(crate) refused_again: bool,
}

/// [`apply`] of the changes in `bytes`, as sync messages and downloads carry them, with
/// what `screening` keeps for document `epoch_seq` (of `epoch`). Bytes that don't load
/// as changes, or with a misplaced author (`changes::misplaced_author`), are refused
/// whole.
/// With `signatures` (required), those no signature vouches for wait too; and with no
/// bytes, those waiting are judged again (signatures came since).
pub(crate) fn apply_bytes(
    doc: &mut Document,
    epoch: Epoch,
    epoch_seq: u32,
    bytes: &[u8],
    screening: &std::sync::Mutex<Screening>,
    signatures: Option<&mut Signatures>,
    log: &mut crate::document::PatchLog,
) -> Result<Outcome> {
    use crate::locks::MutexExt;
    if bytes.is_empty() && signatures.is_none() {
        return Ok(Outcome::default());
    }
    if let Some(misplaced) = crate::document::changes::misplaced_author(bytes) {
        anyhow::bail!("Refused changes: {misplaced}");
    }
    let changes = crate::document::changes::changes_in(bytes)
        .ok_or_else(|| anyhow::anyhow!("Refused changes that don't load"))?;
    let mut screening = screening.lock_or_recover();
    let screening = screening.of(epoch_seq);
    if changes.is_empty() && screening.waiting.len() == 0 {
        return Ok(Outcome::default());
    }
    let received = changes.len();
    let screened = apply(
        doc,
        epoch,
        changes,
        &mut screening.waiting,
        &mut screening.refusals,
        signatures,
        log,
    )?;
    let outcome = Outcome {
        applied: screened.applied,
        refused_again: received > 0 && screened.refused_again == received,
    };
    screening.judge_members(&screened);
    screening.record(screened.refused);
    Ok(outcome)
}

/// Whether a change touching `paths` changes what rules read: the members, the user
/// registry, remake records, the policy and the rest of `_continuum/access`, bans, the
/// protocol data as a whole.
fn relevant(paths: &BTreeSet<Path>) -> bool {
    paths.iter().any(|path| match path.as_slice() {
        [] => true,
        [root, ..] if root != meta::ROOT => false,
        [_] => true,
        [_, part, ..] => [
            meta::MEMBERS,
            meta::USERS,
            meta::REMAKES,
            meta::ACCESS,
            meta::BANNED,
        ]
        .contains(&part.as_str()),
    })
}

/// `candidates` in an order where every change comes after its dependencies among them
/// (ready), and those with a dependency neither in `doc` nor among them, nor ready
/// (unready).
fn in_order(doc: &Document, candidates: Vec<Change>) -> (Vec<Change>, Vec<Change>) {
    let mut by_hash: HashMap<ChangeHash, Change> = candidates
        .into_iter()
        .map(|change| (change.hash(), change))
        .collect();
    let mut ready = Vec::new();
    let mut placed: HashSet<ChangeHash> = HashSet::new();
    loop {
        let mut progress = false;
        let mut hashes: Vec<ChangeHash> = by_hash.keys().copied().collect();
        // The same order on every node, whatever order they came in.
        hashes.sort_unstable();
        for hash in hashes {
            let change = &by_hash[&hash];
            if change
                .deps()
                .iter()
                .all(|dep| placed.contains(dep) || doc.get_change_by_hash(dep).is_some())
            {
                placed.insert(hash);
                ready.push(by_hash.remove(&hash).expect("there"));
                progress = true;
            }
        }
        if !progress {
            break;
        }
    }
    (ready, by_hash.into_values().collect())
}

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

    fn path(text: &str) -> Path {
        text.split('/').map(str::to_owned).collect()
    }

    fn write(
        doc: &mut Document,
        node: &str,
        user: Option<&str>,
        at: &str,
        value: serde_json::Value,
    ) {
        let writer = Writer { node, user };
        let at = path(at);
        transact(doc, UnixMillis(1), &writer, |doc| {
            store::write_path(doc, &at, &value)
        })
        .unwrap();
    }

    /// The changes of `doc` after `since`, as received.
    fn received(doc: &Document, since: &[ChangeHash]) -> Vec<Change> {
        doc.get_changes(since).into_iter().collect()
    }

    fn screen(doc: &mut Document, changes: Vec<Change>, waiting: &mut Waiting) -> Screened {
        apply(
            doc,
            Epoch::GENESIS,
            changes,
            waiting,
            &mut Refusals::default(),
            None,
            &mut crate::document::PatchLog::inactive(),
        )
        .unwrap()
    }

    /// A swarm whose genesis registers member `a` and user `alice`, who may write
    /// `notes`; written on `theirs`, received by an empty document in one batch.
    fn genesis() -> Document {
        let mut theirs = Document::new();
        let alice = registry::entry("a", 1, &[(path("notes"), &[Right::Write][..])]);
        let writer = Writer {
            node: "a",
            user: None,
        };
        transact(&mut theirs, UnixMillis(1), &writer, |doc| {
            store::write_path(doc, &path("_continuum/swarm"), &json!("s"))?;
            store::write_path(doc, &path("_continuum/users/alice"), &alice)
        })
        .unwrap();
        write(&mut theirs, "a", None, "_continuum/members/a", json!({}));
        theirs
    }

    /// A batch from the genesis on: honest changes applied, a refused one left out with
    /// what builds on it, the rest in.
    #[test]
    fn a_batch_judged_change_by_change() {
        let mut theirs = genesis();
        write(&mut theirs, "a", Some("alice"), "notes/x", json!(1));
        let before_bad = theirs.get_heads();
        write(&mut theirs, "a", Some("alice"), "config/x", json!(2));
        write(&mut theirs, "a", Some("alice"), "notes/y", json!(3));
        // Concurrent with the refused one: in.
        let mut side = theirs.fork_at(&before_bad).unwrap();
        write(&mut side, "a", Some("alice"), "notes/z", json!(4));
        theirs.merge(&mut side).unwrap();

        let mut ours = Document::new();
        let mut waiting = Waiting::default();
        let screened = screen(&mut ours, received(&theirs, &[]), &mut waiting);
        assert_eq!(screened.applied, 4);
        let why: Vec<&str> = screened.refused.iter().map(|r| r.why.as_str()).collect();
        assert_eq!(why.len(), 2, "{why:?}");
        assert!(why[0].contains("may not write /config"), "{why:?}");
        assert!(why[1].contains("builds on refused change"), "{why:?}");
        assert_eq!(
            store::read_path(&ours, &path("notes/x")).unwrap(),
            Some(json!(1))
        );
        assert_eq!(
            store::read_path(&ours, &path("notes/z")).unwrap(),
            Some(json!(4))
        );
        assert_eq!(store::read_path(&ours, &path("config")).unwrap(), None);
        assert!(waiting.len() == 0);
    }

    /// A member whose own change broke a rule: refused as a whole, until a change of its
    /// is taken.
    #[test]
    fn members_refused_as_a_whole() {
        let mut screening = Screening::default();
        let broke = Screened {
            broke_rules: vec![("b".to_owned(), ChangeHash([1; 32]), "why".to_owned())],
            ..Screened::default()
        };
        screening.judge_members(&broke);
        assert_eq!(screening.refusing("b").unwrap().change, ChangeHash([1; 32]));
        assert!(screening.refusing("a").is_none());
        let taken = Screened {
            taken_from: ["b".to_owned()].into(),
            ..Screened::default()
        };
        screening.judge_members(&taken);
        assert!(screening.refusing("b").is_none());
    }

    /// A refused change offered again: refused by its hash, not judged or kept again,
    /// and so is a new one building on it.
    #[test]
    fn refused_again_by_hash() {
        let mut theirs = genesis();
        write(&mut theirs, "a", Some("alice"), "config/x", json!(1));
        let mut ours = Document::new();
        let mut waiting = Waiting::default();
        let mut refusals = Refusals::default();
        let mut screen_with = |doc: &mut Document, changes: Vec<Change>| {
            apply(
                doc,
                Epoch::GENESIS,
                changes,
                &mut waiting,
                &mut refusals,
                None,
                &mut crate::document::PatchLog::inactive(),
            )
            .unwrap()
        };
        let first = screen_with(&mut ours, received(&theirs, &[]));
        assert_eq!(
            (first.applied, first.refused.len(), first.refused_again),
            (2, 1, 0)
        );
        let heads = ours.get_heads();
        let again = screen_with(&mut ours, received(&theirs, &heads));
        assert_eq!(
            (again.applied, again.refused.len(), again.refused_again),
            (0, 0, 1)
        );
        let before = theirs.get_heads();
        write(&mut theirs, "a", Some("alice"), "notes/x", json!(2));
        let building = screen_with(&mut ours, received(&theirs, &before));
        assert_eq!((building.applied, building.refused.len()), (0, 1));
        assert!(building.refused[0].why.contains("builds on refused change"));
    }

    /// Judged as its writer saw the registry: a grant earlier in the same batch counts.
    #[test]
    fn a_grant_in_the_same_batch_counts() {
        let mut theirs = genesis();
        let bob = registry::entry("a", 1, &[(path("config"), &[Right::Write][..])]);
        write(&mut theirs, "a", Some("alice"), "_continuum/users/bob", bob);
        write(&mut theirs, "a", Some("bob"), "config/x", json!(1));
        let mut ours = Document::new();
        let screened = screen(&mut ours, received(&theirs, &[]), &mut Waiting::default());
        // Alice may not grant (no write on the registry): both refused.
        assert_eq!(screened.refused.len(), 2);

        let mut theirs = genesis();
        let root = registry::entry("a", 1, &registry::all_caps());
        transact(
            &mut theirs,
            UnixMillis(1),
            &Writer {
                node: "a",
                user: None,
            },
            |doc| store::write_path(doc, &path("_continuum/users/root"), &root),
        )
        .unwrap();
        // Written as a member without a user: refused, and so is what builds on it.
        let mut ours = Document::new();
        let screened = screen(&mut ours, received(&theirs, &[]), &mut Waiting::default());
        assert_eq!(screened.refused.len(), 1);
    }

    /// Changes whose dependencies haven't come wait, and are judged once they do.
    #[test]
    fn changes_wait_for_their_dependencies() {
        let mut theirs = genesis();
        write(&mut theirs, "a", Some("alice"), "notes/x", json!(1));
        let middle = theirs.get_heads();
        write(&mut theirs, "a", Some("alice"), "notes/y", json!(2));
        let mut ours = Document::new();
        let mut waiting = Waiting::default();
        // The last first: it waits.
        let last = received(&theirs, &middle);
        let screened = screen(&mut ours, last, &mut waiting);
        assert_eq!(screened.applied, 0);
        assert_eq!(waiting.len(), 1);
        // The rest: all in, the waiting one too.
        let screened = screen(&mut ours, received(&theirs, &[]), &mut waiting);
        assert_eq!(screened.applied, 4);
        assert!(screened.refused.is_empty());
        assert_eq!(waiting.len(), 0);
        assert_eq!(
            store::read_path(&ours, &path("notes/y")).unwrap(),
            Some(json!(2))
        );
    }

    /// With signatures required, changes no signature vouches for wait, with what
    /// builds on them, until one comes; one signature vouches for its actor's earlier
    /// changes too.
    #[test]
    fn unsigned_changes_wait_for_a_signature() {
        let key = crate::identity::NodeKey::from_seed([3; 32]);
        let node = key.id();
        let mut theirs = Document::new();
        let alice = registry::entry(&node, 1, &[(path("notes"), &[Right::Write][..])]);
        transact(
            &mut theirs,
            UnixMillis(1),
            &Writer {
                node: &node,
                user: None,
            },
            |doc| {
                store::write_path(doc, &path("_continuum/swarm"), &json!("s"))?;
                store::write_path(doc, &path("_continuum/users/alice"), &alice)
            },
        )
        .unwrap();
        let root = theirs.get_heads();
        write(
            &mut theirs,
            &node,
            None,
            &format!("_continuum/members/{node}"),
            json!({}),
        );
        write(&mut theirs, &node, Some("alice"), "notes/x", json!(1));
        let last = theirs.get_heads()[0];

        let mut ours = Document::new();
        let mut waiting = Waiting::default();
        let mut signatures = Signatures::default();
        let required =
            |doc: &mut Document, waiting: &mut Waiting, signatures: &mut Signatures, changes| {
                apply(
                    doc,
                    Epoch::GENESIS,
                    changes,
                    waiting,
                    &mut Refusals::default(),
                    Some(signatures),
                    &mut crate::document::PatchLog::inactive(),
                )
                .unwrap()
            };
        // Unsigned: only the root goes in (no node signs it).
        let screened = required(
            &mut ours,
            &mut waiting,
            &mut signatures,
            received(&theirs, &[]),
        );
        assert_eq!(screened.applied, 1);
        assert_eq!(ours.get_heads(), root);
        assert_eq!(waiting.len(), 2);
        // A forged signature vouches for nothing.
        let forged = crate::identity::NodeKey::from_seed([4; 32]);
        signatures.offer(
            &ours,
            vec![(
                last,
                forged.sign(&crate::document::signatures::message(&last)),
            )],
        );
        let screened = required(&mut ours, &mut waiting, &mut signatures, Vec::new());
        assert_eq!(screened.applied, 0);
        assert_eq!(waiting.len(), 2);
        // The last one's: both in.
        signatures.offer(
            &ours,
            vec![(last, key.sign(&crate::document::signatures::message(&last)))],
        );
        let screened = required(&mut ours, &mut waiting, &mut signatures, Vec::new());
        assert_eq!(screened.applied, 2);
        assert!(screened.refused.is_empty());
        assert_eq!(waiting.len(), 0);
        assert_eq!(
            store::read_path(&ours, &path("notes/x")).unwrap(),
            Some(json!(1))
        );
    }
}