//! 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 records,
//! votes, the policy, bans, whether a map it makes is there) is what its writer saw,
//! unless one of those ancestors changed it. So a change that does (`relevant`), and
//! those before a change making a map where they wrote, are 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.
//! One that builds on a change refused meanwhile is refused then, not left waiting.
//! With signatures required (`access::policy`), so does one no signature by its
//! committer vouches for yet, and whatever builds on it, until one comes, but an epoch's
//! root, which no node signs: a signature vouches for its change and the changes of its
//! actor among those received it builds on (`Signatures::vouch_batch`). One made under
//! an actor at a seq whose change the document holds is forged (`Signatures::forks`):
//! refused, by whoever brought it, not by the node the actor is.

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

use anyhow::Result;

use super::rules::{self, Refusal};
use crate::document::inspect::{Action, 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};

/// Changes waiting for their dependencies, kept at most: the oldest go.
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>,
    /// Changes this node holds that peers say the swarm's rules refuse, and its own do
    /// too, whose drop would take much: each with the peers that said so (`hold_drop`).
    held_drops: HashMap<ChangeHash, BTreeSet<String>>,
    /// Which change an empty document of the first epoch takes as its root
    /// (`swarm::roots`); kept from one document to the next.
    pub(crate) root_pin: RootPin,
}

/// Which change an empty document of the first epoch takes as its root: any (nothing
/// heard of it, nor given), the one given (`--genesis-root`) or that its peers name, or
/// none while they name different ones.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub(crate) enum RootPin {
    #[default]
    Any,
    Only(ChangeHash),
    Disputed,
}

/// 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, each with the one
/// that broke a rule (itself, or the one it builds on).
#[derive(Debug, Default)]
pub(crate) struct Refusals {
    known: HashMap<ChangeHash, ChangeHash>,
    order: std::collections::VecDeque<ChangeHash>,
}

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

    /// The change that broke a rule, of the refused change `hash`.
    fn root(&self, hash: &ChangeHash) -> Option<ChangeHash> {
        self.known.get(hash).copied()
    }

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

impl Screening {
    /// Empty, with `root_pin`.
    pub(crate) fn pinned(root_pin: RootPin) -> Self {
        Screening {
            root_pin,
            ..Screening::default()
        }
    }

    /// 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,
                root_pin: std::mem::take(&mut self.root_pin),
                ..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();
            }
        }
    }

    /// `peer` says `root`, held here, is refused by the swarm's rules, its drop taking
    /// much (`AppState::drop_taken`): how many peers said so of it, this one with them.
    pub(crate) fn hold_drop(&mut self, root: ChangeHash, peer: &str) -> usize {
        let named = self.held_drops.entry(root).or_default();
        named.insert(peer.to_owned());
        named.len()
    }

    /// Drops held, for another peer to say so too (`hold_drop`).
    pub(crate) fn drops_held(&self) -> usize {
        self.held_drops.len()
    }

    /// `dropped`, changes this node took unjudged and dropped (`AppState::drop_taken`),
    /// kept with those refused, the oldest going past `MAX_REFUSED`; logged once, there.
    pub(crate) fn keep_dropped(&mut self, dropped: Vec<Refused>) {
        for dropped in dropped {
            self.refused.push_back(dropped);
            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)
    }

    /// `node` refused as a whole no more: it has another document since, without what
    /// was refused, or it would build on it (`swarm::sync`).
    pub(crate) fn forgive(&mut self, node: &str) {
        if self.refusing.remove(node).is_some() {
            tracing::info!(node = %node, "A member refused as a whole has another document since: refused no longer");
        }
    }

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

    /// The newest `MAX_WAITING`: the oldest go, so changes a peer sent that will never
    /// be taken (whose dependencies or signatures never come) make way for others.
    fn keep_newest(&mut self) {
        let excess = self.0.len().saturating_sub(MAX_WAITING);
        self.0.drain(..excess);
    }
}

/// 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>,
    /// The changes that broke a rule, of those refused (now, or again).
    pub(crate) roots: BTreeSet<ChangeHash>,
    /// Received changes forged under another node's actor (`Signatures::forks`).
    pub(crate) forged: usize,
}

/// 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();
    let from_received: HashSet<ChangeHash> = received.iter().map(Change::hash).collect();
    // 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| !crate::document::changes::holds(doc, &change.hash()))
        .filter(|change| match refusals.root(&change.hash()) {
            Some(root) => {
                screened.refused_again += 1;
                screened.roots.insert(root);
                false
            }
            None => true,
        })
        .collect();
    // Building on one refused before: refused too, without waiting for it (it never
    // comes) or judging it.
    let candidates = refuse_building(candidates, refusals, &mut screened);
    let (ready, unready) = in_order(doc, candidates);
    waiting.0 = unready;
    if ready.is_empty() {
        waiting.keep_newest();
        return Ok(screened);
    }

    let decoded: Vec<crate::document::ExpandedChange> =
        ready.iter().map(|change| change.decode()).collect();
    // Without a policy here, nor one among these: none judged but an epoch's root, and
    // nothing placed. A change writing the policy writes at its key.
    let judging = super::policy::of(&*doc).enforced()
        || crate::document::inspect::touch_key(&decoded, "policy");
    let ops: Vec<Vec<Op>> = if judging {
        Inspector::new(doc).ops(&decoded)
    } else {
        vec![Vec::new(); decoded.len()]
    };
    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);
    }
    // Those a signature vouches for: a signed one, and the changes of its actor among
    // these it builds on (`ready` is in order: one's descendants come after it).
    let mut vouched: HashSet<ChangeHash> = HashSet::new();
    if let Some(signatures) = signatures.as_deref() {
        let in_batch: HashSet<ChangeHash> = ready.iter().map(Change::hash).collect();
        let mut signed_after: HashMap<ChangeHash, HashSet<crate::document::ActorId>> =
            HashMap::new();
        for change in ready.iter().rev() {
            let hash = change.hash();
            let mut actors = signed_after.remove(&hash).unwrap_or_default();
            if signatures.signed(&hash) {
                actors.insert(change.actor_id().clone());
            }
            if actors.contains(change.actor_id()) {
                vouched.insert(hash);
            }
            for dep in change.deps().iter().filter(|dep| in_batch.contains(dep)) {
                signed_after
                    .entry(*dep)
                    .or_default()
                    .extend(actors.iter().cloned());
            }
        }
    }

    // The changes of the document each accepted one, not applied yet, builds on.
    let mut frontier: HashMap<ChangeHash, Vec<ChangeHash>> = HashMap::new();
    // Refused in this batch, each with the change that broke a rule.
    let mut refused: HashMap<ChangeHash, ChangeHash> = HashMap::new();
    let mut unsigned: HashSet<ChangeHash> = HashSet::new();
    let mut forged: HashSet<ChangeHash> = HashSet::new();
    let mut accepted: Vec<Change> = Vec::new();
    // What the accepted, not applied yet, touch: a change making a map at one of these
    // paths is judged with them in (`rules`' maps made where there are none).
    let mut unapplied: BTreeSet<Path> = BTreeSet::new();
    // What the rules read at the heads judged at, kept while they stay the same.
    let mut reads = rules::Reads::default();
    // The values their users sign, under a strict policy, as the document holds them:
    // read again once the document takes the accepted (`reaches_signed`).
    let mut signed: Option<Option<Vec<Vec<String>>>> = None;
    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_key(dep) || refusals.contains(dep))
        {
            let root = refused
                .get(dep)
                .copied()
                .or_else(|| refusals.root(dep))
                .unwrap_or(*dep);
            refused.insert(hash, root);
            screened
                .refused
                .push(refuse(format!("it builds on refused change {dep}")));
            continue;
        }
        // Made under an actor whose change at its seq the document holds: forged, by
        // whoever brought it, not by the node the actor is.
        if signatures
            .as_deref()
            .is_some_and(|signatures| signatures.forks(doc, change))
        {
            refused.insert(hash, hash);
            forged.insert(hash);
            if from_received.contains(&hash) {
                screened.forged += 1;
            }
            screened.refused.push(Refused {
                committer: None,
                ..refuse(format!(
                    "it is forged: the document holds the change of its actor ({}) at its seq ({})",
                    committer.unwrap_or("unknown"),
                    change.seq()
                ))
            });
            continue;
        }
        // No signature vouches for it yet, or for what it builds on: it waits.
        if signatures.is_some() && !change.deps().is_empty() && !vouched.contains(&hash)
            || change.deps().iter().any(|dep| unsigned.contains(dep))
        {
            unsigned.insert(hash);
            waiting.0.push(change.clone());
            continue;
        }
        // It makes a map where one accepted before may have: the rules read whether one
        // is there, so those go in first.
        if !accepted.is_empty()
            && ops[index]
                .iter()
                .any(|op| op.action == Action::MakeMap && unapplied.contains(&op.path))
        {
            screened.applied += accepted.len();
            apply_taken(doc, std::mem::take(&mut accepted), log)?;
            frontier.clear();
            unapplied.clear();
            signed = None;
        }
        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 root = change.deps().is_empty();
        let judged = if judging || root {
            rules::judge_reading(
                doc,
                epoch,
                &rules::Change {
                    hash,
                    deps: change.deps(),
                    seen: &seen_at,
                    committer,
                    user: user.as_deref(),
                    paths: &paths[index],
                    ops: &ops[index],
                },
                &mut reads,
            )
        } else {
            Ok(())
        };
        match judged {
            Ok(()) => {
                accepted.push(change.clone());
                if let Some(committer) = committer {
                    screened.taken_from.insert(committer.to_owned());
                }
                // A root too: whether another may be one is read in the document.
                if root
                    || relevant(&paths[index])
                    || reaches_signed(doc, &mut signed, &paths[index])
                {
                    // What rules read changed: in before the next is judged.
                    screened.applied += accepted.len();
                    apply_taken(doc, std::mem::take(&mut accepted), log)?;
                    frontier.clear();
                    unapplied.clear();
                    signed = None;
                } else {
                    frontier.insert(hash, seen_at);
                    unapplied.extend(paths[index].iter().cloned());
                }
            }
            Err(Refusal::Unready) => {
                // Can't be: its dependencies are in the document or among these.
                refused.insert(hash, hash);
                screened.refused.push(refuse(Refusal::Unready.to_string()));
            }
            Err(Refusal::Rule(why)) => {
                refused.insert(hash, 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() {
        apply_taken(doc, accepted, log)?;
    }
    for (hash, root) in refused {
        refusals.insert(hash, root);
        // A forged one broke no rule its sender would judge it by.
        if !forged.contains(&root) {
            screened.roots.insert(root);
        }
    }
    // Waiting for one refused since (or building on such): refused now, not left to
    // age out.
    waiting.0 = refuse_building(std::mem::take(&mut waiting.0), refusals, &mut screened);
    screened.roots.retain(|root| !forged.contains(root));
    waiting.keep_newest();
    Ok(screened)
}

/// Of `changes`, those building on a change `refusals` holds, or on one of these that
/// does, and so on: refused, into `refusals` and `screened`, without being judged. The
/// rest, as they were.
fn refuse_building(
    mut changes: Vec<Change>,
    refusals: &mut Refusals,
    screened: &mut Screened,
) -> Vec<Change> {
    loop {
        let (building, rest): (Vec<Change>, Vec<Change>) = changes
            .into_iter()
            .partition(|change| change.deps().iter().any(|dep| refusals.contains(dep)));
        changes = rest;
        if building.is_empty() {
            return changes;
        }
        for change in building {
            let dep = change
                .deps()
                .iter()
                .find(|dep| refusals.contains(dep))
                .copied()
                .expect("partitioned so");
            let root = refusals.root(&dep).unwrap_or(dep);
            refusals.insert(change.hash(), root);
            screened.roots.insert(root);
            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,
            });
        }
    }
}

/// `accepted` applied to `doc`. Automerge takes them all or none: one it won't take (a
/// fork of an actor's changes, under a policy without signatures) fails the lot, so the
/// error names them (the first few), with the node each says made it, for the log of
/// whoever brought them.
fn apply_taken(
    doc: &mut Document,
    accepted: Vec<Change>,
    log: &mut crate::document::PatchLog,
) -> Result<()> {
    const NAMED: usize = 8;
    let count = accepted.len();
    let mut named: Vec<String> = accepted
        .iter()
        .take(NAMED)
        .map(|change| {
            let author = change
                .author()
                .and_then(|author| {
                    std::str::from_utf8(author.as_bytes())
                        .ok()
                        .map(str::to_owned)
                })
                .unwrap_or_else(|| "its actor's node".to_owned());
            format!("{} (seq {} by {author})", change.hash(), change.seq())
        })
        .collect();
    if count > NAMED {
        named.push(format!("{} more", count - NAMED));
    }
    doc.apply_changes_log_patches(accepted, log)
        .map_err(|error| {
            anyhow::anyhow!(
                "Automerge took none of the {count} changes judged: {error} ({})",
                named.join(", ")
            )
        })?;
    Ok(())
}

/// What `apply_bytes` did.
#[derive(Debug, Default, Clone)]
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,
    /// The changes that broke a rule, of those refused (`swarm::refusals`).
    pub(crate) roots: Vec<ChangeHash>,
    /// Changes the bytes carried forged under another node's actor: their sender's.
    pub(crate) forged: usize,
}

/// [`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 mut 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);
    // Into an empty document of the first epoch, a root only as pinned: one given or
    // named by the peers, none while they disagree (left for later). What builds on a
    // root refused is refused with it; on one left, it waits.
    let mut pinned_out = Vec::new();
    if epoch.generation == 0 && doc.get_heads().is_empty() && screening.root_pin != RootPin::Any {
        changes.retain(|change| {
            if !change.deps().is_empty() {
                return true;
            }
            match &screening.root_pin {
                RootPin::Only(root) if *root == change.hash() => true,
                RootPin::Only(root) => {
                    pinned_out.push((
                        change.clone(),
                        format!("a root other than the swarm's ({root})"),
                    ));
                    false
                }
                _ => false,
            }
        });
    }
    let mut refused_roots = Vec::new();
    for (change, why) in pinned_out {
        screening.refusals.insert(change.hash(), change.hash());
        refused_roots.push(Refused {
            committer: change.author().and_then(|author| {
                std::str::from_utf8(author.as_bytes())
                    .ok()
                    .map(str::to_owned)
            }),
            user: None,
            paths: BTreeSet::new(),
            why,
            change,
        });
    }
    screening.record(refused_roots);
    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,
        roots: screened.roots.iter().copied().collect(),
        forged: screened.forged,
    };
    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, votes, the policy and the rest of `_continuum/access`,
/// bans, the protocol data as a whole; under a strict policy, values their users sign
/// (`reaches_signed`); and, for a change making a map, whether one is there, which
/// `apply` checks against what the accepted before it touch. A rule that reads anything
/// else (`rules::judge`) must add it here, or a batch judged at its frontier is judged
/// at heads it doesn't hold.
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::VOTES,
            meta::ACCESS,
            meta::BANNED,
        ]
        .contains(&part.as_str()),
    })
}

/// Whether a change touching `paths` changes values their users sign, under a strict
/// policy, at, within or above one (`envelope::reaches_signed`): the rules read the
/// value at a signed place (`prev`) and what holds one, so it goes in before the next
/// is judged. Where they are, None if the policy isn't strict, is read into `signed`
/// once: a read costs more than the rest of screening a change, and what it reads
/// changes only as the document takes changes (and one changing the policy, or what
/// users sign, is `relevant`, taken at once).
fn reaches_signed(
    doc: &Document,
    signed: &mut Option<Option<Vec<Vec<String>>>>,
    paths: &BTreeSet<Path>,
) -> bool {
    if paths.is_empty() {
        return false;
    }
    let signed = signed.get_or_insert_with(|| {
        super::policy::of(doc)
            .strict()
            .then(|| super::envelope::signed_paths(doc))
    });
    signed.as_ref().is_some_and(|signed| {
        paths
            .iter()
            .any(|path| super::envelope::reaches_signed(signed, path))
    })
}

/// `candidates` (each once) 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). By their places in `candidates`, each hash looked
/// up once: hashing them costs more than the rest of ordering them.
fn in_order(doc: &Document, candidates: Vec<Change>) -> (Vec<Change>, Vec<Change>) {
    let place: HashMap<ChangeHash, usize> = candidates
        .iter()
        .enumerate()
        .map(|(at, change)| (change.hash(), at))
        .collect();
    // Each one's dependencies among these not placed yet, and those building on each;
    // one with a dependency neither in the document nor among these is never placed,
    // nor anything building on it.
    let mut unplaced = vec![0usize; candidates.len()];
    let mut building: Vec<Vec<usize>> = vec![Vec::new(); candidates.len()];
    let mut missing = vec![false; candidates.len()];
    for (at, change) in candidates.iter().enumerate() {
        for dep in change.deps() {
            match place.get(dep) {
                Some(&of) => {
                    unplaced[at] += 1;
                    building[of].push(at);
                }
                None if !crate::document::changes::holds(doc, dep) => missing[at] = true,
                None => {}
            }
        }
    }
    // The smallest hash of those ready first: the same order on every node, whatever
    // order they came in.
    let mut next: BTreeSet<(ChangeHash, usize)> = (0..candidates.len())
        .filter(|&at| unplaced[at] == 0 && !missing[at])
        .map(|at| (candidates[at].hash(), at))
        .collect();
    let mut order = Vec::new();
    while let Some((_, at)) = next.pop_first() {
        order.push(at);
        for &child in &building[at] {
            unplaced[child] -= 1;
            if unplaced[child] == 0 && !missing[child] {
                next.insert((candidates[child].hash(), child));
            }
        }
    }
    let mut slots: Vec<Option<Change>> = candidates.into_iter().map(Some).collect();
    let ready = order
        .into_iter()
        .filter_map(|at| slots[at].take())
        .collect();
    (ready, slots.into_iter().flatten().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/access/policy"),
                &json!({"version": 1}),
            )?;
            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());
    }

    /// Under a strict policy, two signed values written one after the other at the same
    /// path, received in one batch: both taken, the second judged with the first in.
    #[test]
    fn successive_signed_values_in_one_batch() {
        use crate::access::{envelope, sshsig};
        let key = ed25519_dalek::SigningKey::from_bytes(&[7; 32]);
        let mut theirs = Document::new();
        let mut alice = registry::entry("a", 1, &[(path("config"), &[Right::Write][..])]);
        alice["keys"] = json!({ "k": sshsig::PublicKey(key.verifying_key().to_bytes()).line() });
        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/access/policy"),
                &json!({"version": 1, "values": "signed"}),
            )?;
            store::write_path(
                doc,
                &path("_continuum/access/userSigned"),
                &json!([["config"]]),
            )?;
            store::write_path(doc, &path("_continuum/users/alice"), &alice)
        })
        .unwrap();
        write(&mut theirs, "a", None, "_continuum/members/a", json!({}));
        let sealed = |prev: &str, value: serde_json::Value| {
            let payload = envelope::Payload {
                swarm: "s".to_owned(),
                path: path("config"),
                value,
                user: "alice".to_owned(),
                at: 1,
                prev: prev.to_owned(),
            };
            let bytes = serde_json::to_vec(&payload).unwrap();
            envelope::seal(&bytes, &sshsig::sign(&key, envelope::NAMESPACE, &bytes)).unwrap()
        };
        let first = sealed("", json!(1));
        let second = sealed(&envelope::digest(Some(&first)), json!(2));
        write(&mut theirs, "a", Some("alice"), "config", json!(first));
        write(&mut theirs, "a", Some("alice"), "config", json!(second));

        let mut ours = Document::new();
        let screened = screen(&mut ours, received(&theirs, &[]), &mut Waiting::default());
        let why: Vec<&str> = screened.refused.iter().map(|r| r.why.as_str()).collect();
        assert!(why.is_empty(), "{why:?}");
        assert_eq!(
            store::read_path(&ours, &path("config")).unwrap(),
            Some(json!(second))
        );
    }

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

    /// A change waiting for a dependency that hasn't come, building on another refused
    /// meanwhile: refused at once, not left to wait.
    #[test]
    fn waiting_on_a_refused_change_refused_at_once() {
        let mut theirs = genesis();
        let mut other = theirs.fork();
        write(&mut theirs, "a", Some("mallory"), "notes/x", json!(1));
        write(&mut other, "a", Some("alice"), "notes/w", json!(2));
        let missing = other.get_heads()[0];
        theirs.merge(&mut other).unwrap();
        write(&mut theirs, "a", Some("alice"), "notes/z", json!(3));
        let last = theirs.get_heads()[0];
        let mut ours = Document::new();
        let mut waiting = Waiting::default();
        let mut refusals = Refusals::default();
        let mut screen_with = |ours: &mut Document, changes, waiting: &mut Waiting| {
            apply(
                ours,
                Epoch::GENESIS,
                changes,
                waiting,
                &mut refusals,
                None,
                &mut crate::document::PatchLog::inactive(),
            )
            .unwrap()
        };
        let all = received(&theirs, &[]);
        let first: Vec<Change> = all.iter().filter(|c| c.hash() == last).cloned().collect();
        let screened = screen_with(&mut ours, first, &mut waiting);
        assert_eq!((screened.applied, waiting.len()), (0, 1));
        // All but one it builds on: mallory's refused, and the waiting one with it.
        let rest: Vec<Change> = all
            .into_iter()
            .filter(|c| c.hash() != missing && c.hash() != last)
            .collect();
        let screened = screen_with(&mut ours, rest, &mut waiting);
        let refused: Vec<ChangeHash> = screened.refused.iter().map(|r| r.change.hash()).collect();
        assert_eq!(refused.len(), 2);
        assert!(refused.contains(&last));
        assert_eq!(waiting.len(), 0);
    }

    /// 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!({}),
        );
        let before = theirs.get_heads();
        write(&mut theirs, &node, Some("alice"), "notes/x", json!(1));
        let last = theirs.get_heads()[0];
        // Another change under the node's actor at the seq of its last: a fork anyone
        // could make, which no signature of the node vouches for.
        let mut forger = theirs
            .fork_at(&before)
            .unwrap()
            .with_actor(crate::document::actor(&theirs));
        write(&mut forger, &node, Some("alice"), "notes/forged", json!(4));

        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))
        );
        // The fork: refused as forged, by whoever brought it, not as the node's.
        let screened = required(
            &mut ours,
            &mut waiting,
            &mut signatures,
            received(&forger, &before),
        );
        assert_eq!((screened.applied, screened.forged), (0, 1));
        assert!(screened.broke_rules.is_empty() && screened.roots.is_empty());
        assert!(screened.refused[0].committer.is_none());
        assert!(
            store::read_path(&ours, &path("notes/forged"))
                .unwrap()
                .is_none()
        );
    }
}