//! What a chain of dependent pull requests *is*.
//!
//! A stack is a run of `sh.tangled.repo.pull` records linked bottom-to-top by
//! `dependentOn`. Every rule about whether such a run is well formed lives
//! here and nowhere else: what counts as a chain, what counts as damage to
//! one, and what a batch of writes would leave behind.
//!
//! # Why it is a model module
//!
//! This is where the model layer was first arrived at, by accident and
//! before it had a name. These rules had been spread across the stack verbs,
//! each verb re-deciding what a well-formed chain was, and the fix — one
//! total walk, one damage type, one gate every writer passes through — is
//! what stopped chain rules from drifting apart the way the pull-level ones
//! did. It sat in `cmd::stack` only because that is where it was written.
//!
//! What is *not* here: reading a listing, choosing an exit status, deciding
//! whose stack a branch's is, or saying any of this to a person. Those are
//! questions about a command, and they stayed in [`crate::cmd::stack`].
//!
//! # Five ways in
//!
//! [`chain_containing`] to read a chain, [`refuse_new_damage`] to gate a
//! write, [`closed_uris`] and [`state_of`] to say what a listing settled,
//! and [`Chain`] to hold the answer. The walk itself, the damage
//! classification and the projection of pending ops are private: they had
//! been reachable from every stack verb, which is an invitation to a verb to
//! answer half a chain question for itself, and no verb ever needed them.
//!
//! # The closed-member rule
//!
//! Retiring a member leaves a record that still names its parent, so a
//! parent can genuinely have two dependents — the retired one and the live
//! one relinked onto it. That is the ordinary state of a stack somebody has
//! taken work out of, and it may not stand between a person and their stack.
//! So a closed pull loses every tie: where a live sibling depends on the same
//! pull, the walk follows the live one. Where there is no live sibling it is
//! still followed, because it is still the only thing on record above its
//! parent.

use anyhow::{Result, bail};

/// A stack, assembled from a repo's pull listing.
#[derive(Debug)]
pub(crate) struct Chain<'a> {
    /// Bottom first: the order the commits sit in history and the order
    /// Tangled merges them in.
    pub members: Vec<&'a serde_json::Value>,
    /// A `dependentOn` at-uri that the listing did not contain, meaning the
    /// chain continues below what is shown. Deep pagination and Bobbin lag
    /// can both cause it; pretending the stack starts here could not.
    pub missing_below: Option<String>,
}

impl<'a> Chain<'a> {
    /// The top of the stack: the pull nothing depends on. Returns the
    /// listing's own lifetime, so a caller can keep the item after the
    /// `Chain` goes away: `pr view` swaps its detail item for this.
    pub(crate) fn top(&self) -> &'a serde_json::Value {
        self.members
            .last()
            .copied()
            .expect("a chain is never empty")
    }

    /// How big this stack is, said as honestly as the walk allows.
    ///
    /// `members.len()` is the whole count only when nothing dangles. A
    /// `missing_below` means the walk stopped at a link pointing outside the
    /// listing, so what was read is a **floor**: the stack continues below,
    /// and printing the floor as an exact count states as fact something the
    /// walk did not establish. This is the same defect that was fixed in
    /// `dependents_above` and left standing in the two `pr` refusals that
    /// redirect to `stack`; the redirect was right either way, the number
    /// was not.
    pub(crate) fn size(&self) -> String {
        match self.missing_below {
            None => format!("{}", self.members.len()),
            Some(_) => format!("at least {}", self.members.len()),
        }
    }

    /// Where a member sits, counting from the bottom, said as honestly as
    /// the walk allows. `?` when the uri is not in this chain at all, and a
    /// floor when the chain dangles below — a member two above a bottom that
    /// could not be read is not the third of anything.
    pub(crate) fn position_of(&self, uri: &str) -> String {
        let Some(i) = self
            .members
            .iter()
            .position(|m| m["uri"].as_str() == Some(uri))
        else {
            return "?".to_string();
        };
        match self.missing_below {
            None => format!("{}", i + 1),
            Some(_) => format!("at least {}", i + 1),
        }
    }
}

/// What a chain is unreadable *for*, named rather than described.
///
/// **The rules of a well-formed chain live here and nowhere else.** They used
/// to live only in the reader's `bail!` messages, which meant the five places
/// that *write* a chain each re-derived whichever subset their author had in
/// mind — and the ones they missed are most of this epic's fix commits. A
/// writer can now ask the same question the reader asks, about the records it
/// is *about* to send, and refuse before any of them leaves the machine.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
enum Damage {
    /// Two or more live pulls depend on one. Tangled's own walk asks for
    /// *the* pull depending on a given one and takes whichever row the
    /// database returns, so a fork reads differently from one request to the
    /// next rather than failing.
    Fork {
        parent: String,
        dependents: Vec<String>,
    },
    /// A `dependentOn` loop, which orders nothing.
    Cycle { entry: String, at: String },
    /// A `dependentOn` naming a pull the listing does not hold. Not damage on
    /// its own — deep pagination and a contributor's records both cause it —
    /// but a write that *introduces* one has broken the chain it was editing.
    Dangling { entry: String, below: String },
}

/// What makes two reports of damage *the same* damage.
///
/// **Not the entry point.** The reader walks from one pull, so every variant
/// carries whichever member the walk started at — and that is an accident of
/// where you looked, not a property of the records. Comparing whole `Damage`
/// values made a chain that already dangles look freshly broken the moment a
/// member was added above it, because the walk from the new member reported
/// the same dangle under a different entry. Every stack that pages deep, or
/// that has a member in a contributor's PDS, would then have had its next
/// write refused.
///
/// So a fork is identified by the parent two pulls share, a dangle by the
/// record nobody holds, and a cycle by nothing at all: a loop makes its whole
/// component unreadable and the node the walk happens to notice it at varies
/// with the entry too, so "there is a cycle" is the whole of the fact.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
enum DamageKey {
    Fork(String),
    Cycle,
    Dangling(String),
}

impl Damage {
    fn key(&self) -> DamageKey {
        match self {
            Damage::Fork { parent, .. } => DamageKey::Fork(parent.clone()),
            Damage::Cycle { .. } => DamageKey::Cycle,
            Damage::Dangling { below, .. } => DamageKey::Dangling(below.clone()),
        }
    }

    /// The refusal this reads as, in the words the reader has always used.
    fn refusal(&self) -> String {
        match self {
            Damage::Fork { parent, dependents } => format!(
                "{} pulls depend on {parent}: {}\n\
                 a stack is a chain, not a fork; refusing to pick one branch of it",
                dependents.len(),
                dependents.join(", ")
            ),
            Damage::Cycle { entry, at } => format!(
                "the dependentOn chain through {entry} loops back to {at}; \
                 refusing to order it"
            ),
            Damage::Dangling { entry, below } => format!(
                "the chain through {entry} continues below {below}, which no listing row \
                 answers to"
            ),
        }
    }
}

/// What the walk from `uri` arrives at: a chain, a pull standing alone, or
/// the reason neither is orderable.
#[derive(Debug)]
enum Shape<'a> {
    Chain(Chain<'a>),
    /// Nothing depends on it and it depends on nothing.
    Alone,
    /// Unreadable, and why.
    Broken(Damage),
    /// No pull with this at-uri in the listing at all.
    Absent,
}

/// The stack containing `uri`, or `None` when that pull stands alone.
///
/// `items` is the merged listing shape the `pr` read commands share: each
/// element carries the record under `"value"` and its at-uri under `"uri"`.
/// The walk follows links, not timestamps: `createdAt` says nothing about
/// order once a stack has been reconciled, since a reorder keeps every
/// record's creation date.
///
/// Refusals, not guesses: two pulls depending on the same target is a fork,
/// and a loop can order nothing, so both are errors that name the records
/// rather than a chain with the problem smoothed over.
///
/// `closed` is the at-uris the listing says are closed, and they are read
/// past rather than refused on: a retired member keeps a record that may
/// still name the member below it, and that is a stack in good order, not a
/// fork. Pass an empty set to order the records exactly as they are, which
/// is what a caller with no state in hand is honestly able to ask for.
pub(crate) fn chain_containing<'a>(
    items: &[&'a serde_json::Value],
    uri: &str,
    closed: &std::collections::HashSet<&str>,
) -> Result<Option<Chain<'a>>> {
    match shape_containing(items, uri, closed) {
        Shape::Chain(chain) => Ok(Some(chain)),
        Shape::Alone | Shape::Absent => Ok(None),
        // `Dangling` is not a refusal here: a chain that continues below what
        // the listing holds is still a chain, and `Chain::missing_below`
        // is how this has always reported it. Only a *writer* treats it as
        // damage, and only when its own ops introduced it.
        Shape::Broken(damage) => match damage {
            Damage::Dangling { .. } => unreachable!("the walk returns dangling inside a chain"),
            other => bail!("{}", other.refusal()),
        },
    }
}

/// The walk itself, total: every outcome is a value, including the two the
/// reader turns back into refusals.
fn shape_containing<'a>(
    items: &[&'a serde_json::Value],
    uri: &str,
    closed: &std::collections::HashSet<&str>,
) -> Shape<'a> {
    use std::collections::{HashMap, HashSet};

    let mut by_uri: HashMap<&str, &'a serde_json::Value> = HashMap::new();
    let mut dependents: HashMap<&str, Vec<&str>> = HashMap::new();
    for item in items {
        let Some(item_uri) = item["uri"].as_str() else {
            continue;
        };
        by_uri.insert(item_uri, item);
        if let Some(target) = item["value"]["dependentOn"].as_str() {
            dependents.entry(target).or_default().push(item_uri);
        }
    }

    // What depends on a pull, with retired members losing every tie.
    //
    // A closed pull is not damage and not a second branch: retiring a member
    // leaves its record naming the pull below it, and the live member above
    // is relinked to that same pull, so the two share a parent for as long
    // as the record lives. Reading the pair as a fork is what made the
    // sanctioned way out of a stack produce a stack no stack command would
    // read.
    //
    // It loses the tie rather than disappearing. A closed pull with no live
    // sibling is still the only thing recorded above its parent, and a walk
    // that dropped it would report a stack shorter than the one on record —
    // which is the reconcile's business, since that record is the one whose
    // link wants clearing.
    let live_above = |uri: &str| -> Vec<&str> {
        let all = dependents.get(uri).map(Vec::as_slice).unwrap_or_default();
        let live: Vec<&str> = all
            .iter()
            .copied()
            .filter(|u| !closed.contains(u))
            .collect();
        match live.is_empty() {
            true => all.to_vec(),
            false => live,
        }
    };

    let Some(&start) = by_uri.get(uri) else {
        return Shape::Absent;
    };
    let mut seen: HashSet<&str> = HashSet::new();
    seen.insert(uri);

    // Downward: the pulls this one sits on. The fork check runs here too —
    // it used to live only in the upward walk, so a fork was refused when
    // entered from the parent's side and silently linearized when entered
    // from a child's, which is the exact wrong-ordering outcome the refusal
    // exists to prevent.
    let mut below: Vec<&'a serde_json::Value> = Vec::new();
    let mut missing_below = None;
    let mut cursor = start;
    while let Some(target) = cursor["value"]["dependentOn"].as_str() {
        if !seen.insert(target) {
            return Shape::Broken(Damage::Cycle {
                entry: uri.to_string(),
                at: target.to_string(),
            });
        }
        let ups = live_above(target);
        if ups.len() > 1 {
            return Shape::Broken(Damage::Fork {
                parent: target.to_string(),
                dependents: ups.iter().map(|u| u.to_string()).collect(),
            });
        }
        match by_uri.get(target) {
            Some(&item) => {
                below.push(item);
                cursor = item;
            }
            None => {
                missing_below = Some(target.to_string());
                break;
            }
        }
    }

    // Upward: the pulls sitting on this one. More than one is a fork, and
    // not a tie to break: Tangled's own walk (`appview/db/pulls.go`,
    // `GetStack`) asks for *the* pull depending on a given one and takes
    // whichever row comes back, so a fork resolves differently depending on
    // who is looking. Refusing is the only answer that cannot be silently
    // wrong.
    let mut above: Vec<&'a serde_json::Value> = Vec::new();
    let mut cursor_uri = uri;
    loop {
        let ups = live_above(cursor_uri);
        let up = match ups.as_slice() {
            [] => break,
            [one] => *one,
            many => {
                return Shape::Broken(Damage::Fork {
                    parent: cursor_uri.to_string(),
                    dependents: many.iter().map(|u| u.to_string()).collect(),
                });
            }
        };
        if !seen.insert(up) {
            return Shape::Broken(Damage::Cycle {
                entry: uri.to_string(),
                at: up.to_string(),
            });
        }
        above.push(by_uri[up]);
        cursor_uri = up;
    }

    if below.is_empty() && above.is_empty() && missing_below.is_none() {
        return Shape::Alone;
    }
    let mut members: Vec<&'a serde_json::Value> = below.into_iter().rev().collect();
    members.push(start);
    members.extend(above);
    Shape::Chain(Chain {
        members,
        missing_below,
    })
}

/// The listing these ops would leave behind.
///
/// Every stack write builds its complete op list in memory and sends it as
/// one atomic `applyWrites`, so the records that will exist afterwards are
/// knowable before the first byte goes out. This applies the ops to the
/// listing already in hand and hands back what the reader would then see.
///
/// Only pull records are projected. A status record or a blob changes no
/// chain, and carrying them through would mean teaching this function every
/// collection atgc writes.
fn project(
    items: &[&serde_json::Value],
    me: &str,
    ops: &[crate::clients::atproto::record::Op],
) -> Vec<serde_json::Value> {
    use crate::clients::atproto::record::Op;

    let nsid = crate::lexicon::tangled::PULL_NSID;
    let mut out: Vec<serde_json::Value> = items.iter().map(|i| (*i).clone()).collect();
    let uri_of = |rkey: &str| format!("at://{me}/{nsid}/{rkey}");
    for op in ops {
        match op {
            Op::Create {
                nsid: n,
                rkey,
                value,
            } if *n == nsid => {
                out.push(serde_json::json!({
                    "uri": uri_of(rkey.0.as_str()),
                    "value": value,
                }));
            }
            Op::Update {
                nsid: n,
                rkey,
                value,
            } if *n == nsid => {
                let uri = uri_of(rkey.0.as_str());
                match out.iter_mut().find(|i| i["uri"].as_str() == Some(&uri)) {
                    Some(item) => item["value"] = value.clone(),
                    // An update to a record the listing did not carry. The
                    // PDS will refuse it, but projecting it as a create keeps
                    // this function's answer about the *chain* right either
                    // way, and a chain check is not the place to duplicate
                    // the PDS's own preconditions.
                    None => out.push(serde_json::json!({ "uri": uri, "value": value })),
                }
            }
            Op::Delete { nsid: n, rkey } if *n == nsid => {
                let uri = uri_of(rkey.0.as_str());
                out.retain(|i| i["uri"].as_str() != Some(uri.as_str()));
            }
            _ => {}
        }
    }
    out
}

/// Every way the chains in `items` are unreadable, entered from every pull.
///
/// The reader walks from one pull; this walks from all of them, because a
/// fork is invisible from the branch you did not enter through and a write
/// can create one anywhere. Deduplicated and ordered, so two runs over the
/// same records compare equal.
fn damage(
    items: &[&serde_json::Value],
    closed: &std::collections::HashSet<&str>,
) -> std::collections::BTreeMap<DamageKey, Damage> {
    let mut found = std::collections::BTreeMap::new();
    for item in items {
        let Some(uri) = item["uri"].as_str() else {
            continue;
        };
        match shape_containing(items, uri, closed) {
            Shape::Broken(d) => {
                found.entry(d.key()).or_insert(d);
            }
            Shape::Chain(chain) => {
                if let Some(below) = chain.missing_below {
                    let d = Damage::Dangling {
                        entry: uri.to_string(),
                        below,
                    };
                    found.entry(d.key()).or_insert(d);
                }
            }
            Shape::Alone | Shape::Absent => {}
        }
    }
    found
}

/// **Refuse a write that would leave the chain worse than it found it.**
///
/// The one check every stack write shares, and the reason this module owns
/// the rules rather than each verb re-deriving them. `create`, `resubmit`,
/// `link` and `unlink` each build a complete op list before sending it as one
/// atomic batch; this runs the reader over what those ops would produce and
/// refuses if the reader would find damage that is not there already.
///
/// **Worse, not merely broken.** A stack can already be forked or dangling
/// when a command starts — that is exactly when somebody reaches for
/// `stack unlink` — and a repair must not be refused for the damage it is
/// repairing. So the comparison is against the damage the records carry now,
/// and only what the write *adds* is refused.
pub(crate) fn refuse_new_damage(
    items: &[&serde_json::Value],
    me: &str,
    ops: &[crate::clients::atproto::record::Op],
    closed: &std::collections::HashSet<&str>,
) -> Result<()> {
    let before = damage(items, closed);
    let after_items = project(items, me, ops);
    let after_refs: Vec<&serde_json::Value> = after_items.iter().collect();
    let after = damage(&after_refs, closed);
    let introduced: Vec<&Damage> = after
        .iter()
        .filter(|(key, _)| !before.contains_key(*key))
        .map(|(_, d)| d)
        .collect();
    let Some(first) = introduced.first() else {
        return Ok(());
    };
    bail!(
        "this would leave a stack the stack commands cannot read, so nothing was written\n\
         {}{}",
        first.refusal(),
        match introduced.len() {
            1 => String::new(),
            n => format!("\n(and {} more like it)", n - 1),
        }
    )
}

/// The at-uris the listing calls closed, in the shape [`chain_containing`]
/// asks for.
///
/// Only the word "closed": a `?` is a state nobody could settle, and reading
/// past a member on a guess is how the wrong half of a stack gets merged.
pub(crate) fn closed_uris(
    rows: &[crate::model::pull::StackRow],
) -> std::collections::HashSet<&str> {
    rows.iter()
        .filter(|r| r.state == "closed")
        .filter_map(|r| r.item["uri"].as_str())
        .collect()
}

/// The state label a listing settled for `uri`, `?` when no row answers.
///
/// One definition, because three copies of this closure had already begun to
/// drift. Its doc comment had come adrift too: it sat above [`closed_uris`],
/// which meant the function describing the closed set was documented as the
/// one that reads a single state, and this one had no doc at all.
pub(crate) fn state_of(rows: &[crate::model::pull::StackRow], uri: &str) -> String {
    rows.iter()
        .find(|r| r.item["uri"].as_str() == Some(uri))
        .map(|r| r.state.clone())
        .unwrap_or_else(|| "?".to_string())
}

#[cfg(test)]
mod tests {
    use super::chain_containing;
    use serde_json::{Value, json};
    use std::collections::HashSet;

    /// The listing every test but the closed-member ones is describing: one
    /// where nothing has been retired.
    fn nothing_closed() -> HashSet<&'static str> {
        HashSet::new()
    }

    fn item(uri: &str, dependent_on: Option<&str>, title: &str) -> Value {
        let mut value = json!({
            "title": title,
            "source": {"branch": "claude/stack"},
            "target": {"branch": "main"},
            "createdAt": "2026-08-09T00:00:00Z",
        });
        if let Some(dep) = dependent_on {
            value["dependentOn"] = json!(dep);
        }
        json!({"uri": uri, "value": value})
    }

    fn uris(members: &[&Value]) -> Vec<String> {
        members
            .iter()
            .map(|m| m["uri"].as_str().unwrap().to_string())
            .collect()
    }

    /// The order is the links', bottom first, whichever member the walk
    /// starts from: `createdAt` deliberately says nothing here, because a
    /// reconcile reorders records without recreating them.
    #[test]
    fn orders_a_chain_bottom_to_top_from_any_member() {
        let a = item("at://x/p/a", None, "bottom");
        let b = item("at://x/p/b", Some("at://x/p/a"), "middle");
        let c = item("at://x/p/c", Some("at://x/p/b"), "top");
        // Listing order scrambled on purpose.
        let items = vec![&c, &a, &b];

        for start in ["at://x/p/a", "at://x/p/b", "at://x/p/c"] {
            let chain = chain_containing(&items, start, &nothing_closed())
                .expect("linear")
                .expect("stacked");
            assert_eq!(
                uris(&chain.members),
                ["at://x/p/a", "at://x/p/b", "at://x/p/c"]
            );
            assert!(chain.missing_below.is_none());
            assert_eq!(chain.top()["uri"], "at://x/p/c");
        }
    }

    /// A pull in no chain is None: the signal `stack view` turns into its
    /// bail, and so is a pull the listing does not contain at all.
    #[test]
    fn a_lone_pull_is_not_a_stack() {
        let a = item("at://x/p/a", None, "alone");
        let items = vec![&a];
        assert!(
            chain_containing(&items, "at://x/p/a", &nothing_closed())
                .unwrap()
                .is_none()
        );
        assert!(
            chain_containing(&items, "at://x/p/zzz", &nothing_closed())
                .unwrap()
                .is_none()
        );
    }

    /// A dependentOn naming a record the listing lacks is reported, not
    /// papered over: the chain is real, the listing is short.
    #[test]
    fn a_link_below_the_listing_is_named() {
        let b = item("at://x/p/b", Some("at://x/p/gone"), "orphaned middle");
        let c = item("at://x/p/c", Some("at://x/p/b"), "top");
        let items = vec![&b, &c];

        let chain = chain_containing(&items, "at://x/p/c", &nothing_closed())
            .expect("linear")
            .expect("stacked");
        assert_eq!(uris(&chain.members), ["at://x/p/b", "at://x/p/c"]);
        assert_eq!(chain.missing_below.as_deref(), Some("at://x/p/gone"));
    }

    /// What a dangling chain is allowed to *say* about its size. The floor
    /// is the honest answer; printing it as an exact count states a fact the
    /// walk did not establish, which is what the two `pr` refusals that
    /// redirect to `stack` used to do.
    #[test]
    fn a_dangling_chain_counts_itself_as_a_floor() {
        let b = item("at://x/p/b", Some("at://x/p/gone"), "orphaned middle");
        let c = item("at://x/p/c", Some("at://x/p/b"), "top");
        let items = vec![&b, &c];

        let chain = chain_containing(&items, "at://x/p/c", &nothing_closed())
            .expect("linear")
            .expect("stacked");
        assert_eq!(chain.size(), "at least 2");
        assert_eq!(chain.position_of("at://x/p/c"), "at least 2");
        assert_eq!(chain.position_of("at://x/p/b"), "at least 1");
        assert_eq!(chain.position_of("at://x/p/nowhere"), "?");
    }

    /// And an intact one says the plain number, so the hedge is carried only
    /// by the case that earns it.
    #[test]
    fn an_intact_chain_counts_itself_exactly() {
        let a = item("at://x/p/a", None, "bottom");
        let b = item("at://x/p/b", Some("at://x/p/a"), "top");
        let items = vec![&a, &b];

        let chain = chain_containing(&items, "at://x/p/b", &nothing_closed())
            .expect("linear")
            .expect("stacked");
        assert_eq!(chain.size(), "2");
        assert_eq!(chain.position_of("at://x/p/a"), "1");
        assert_eq!(chain.position_of("at://x/p/b"), "2");
    }

    /// Two pulls on one parent is a fork, and meeting one in a listing means
    /// damaged data or a stale link. Both branches are named
    /// and nothing is ordered: from *every* entry point: entered from a
    /// child, the fork used to be silently linearized into that child's
    /// branch of it, which is the wrong-ordering outcome the refusal exists
    /// to prevent.
    #[test]
    fn a_fork_is_refused_with_both_branches_named() {
        let a = item("at://x/p/a", None, "bottom");
        let b = item("at://x/p/b", Some("at://x/p/a"), "one branch");
        let c = item("at://x/p/c", Some("at://x/p/a"), "other branch");
        let items = vec![&a, &b, &c];

        for entry in ["at://x/p/a", "at://x/p/b", "at://x/p/c"] {
            let err = chain_containing(&items, entry, &nothing_closed())
                .unwrap_err()
                .to_string();
            assert!(err.contains("at://x/p/b"), "from {entry}: {err}");
            assert!(err.contains("at://x/p/c"), "from {entry}: {err}");
            assert!(err.contains("refusing"), "from {entry}: {err}");
        }
    }

    /// A closed pull does not put a second branch on its parent.
    ///
    /// The shape a retired stack member leaves behind: the record still says
    /// it sits on the bottom, and the live member above it was relinked to
    /// the same place. Ordering that as a fork is how the sanctioned way out
    /// of a stack produced a stack no stack command would read.
    #[test]
    fn a_closed_member_is_not_a_second_branch() {
        let a = item("at://x/p/a", None, "bottom");
        let b = item("at://x/p/b", Some("at://x/p/a"), "retired middle");
        let c = item("at://x/p/c", Some("at://x/p/a"), "top, relinked past it");
        let items = vec![&a, &b, &c];
        let closed = HashSet::from(["at://x/p/b"]);

        for entry in ["at://x/p/a", "at://x/p/c"] {
            let chain = chain_containing(&items, entry, &closed)
                .expect("a retired member is not a fork")
                .expect("stacked");
            assert_eq!(
                uris(&chain.members),
                ["at://x/p/a", "at://x/p/c"],
                "from {entry}"
            );
        }
    }

    /// A closed pull with nothing live beside it is still on the chain.
    ///
    /// The other half of the tie rule, and the half that keeps a reconcile
    /// honest. A retired *top* has no live sibling to lose to, and it is
    /// still the only pull on record above the member below it — dropping it
    /// here would report a stack shorter than the one the records describe,
    /// to the command whose job is to reconcile the two.
    #[test]
    fn a_closed_member_with_no_live_sibling_stays_on_the_chain() {
        let a = item("at://x/p/a", None, "bottom");
        let b = item("at://x/p/b", Some("at://x/p/a"), "middle");
        let c = item("at://x/p/c", Some("at://x/p/b"), "retired top");
        let items = vec![&a, &b, &c];
        let closed = HashSet::from(["at://x/p/c"]);

        let chain = chain_containing(&items, "at://x/p/a", &closed)
            .expect("linear")
            .expect("stacked");
        assert_eq!(
            uris(&chain.members),
            ["at://x/p/a", "at://x/p/b", "at://x/p/c"]
        );
    }

    /// Entered *through* a closed pull, its own history still orders.
    ///
    /// Reading past a retired member is about not letting it stand between
    /// anyone and their stack. Asked about that pull directly — `pr view` on
    /// a closed member of a stack — the answer is still what it sat on.
    #[test]
    fn a_closed_member_entered_directly_keeps_its_own_history() {
        let a = item("at://x/p/a", None, "bottom");
        let b = item("at://x/p/b", Some("at://x/p/a"), "retired middle");
        let items = vec![&a, &b];
        let closed = HashSet::from(["at://x/p/b"]);

        let chain = chain_containing(&items, "at://x/p/b", &closed)
            .expect("linear")
            .expect("stacked");
        assert_eq!(uris(&chain.members), ["at://x/p/a", "at://x/p/b"]);
    }

    /// A fork between two *open* pulls is still a fork. Nothing about
    /// reading past closed records may soften the case the refusal is for.
    #[test]
    fn an_open_fork_is_still_refused_when_something_else_is_closed() {
        let a = item("at://x/p/a", None, "bottom");
        let b = item("at://x/p/b", Some("at://x/p/a"), "one branch");
        let c = item("at://x/p/c", Some("at://x/p/a"), "other branch");
        let d = item("at://x/p/d", Some("at://x/p/a"), "retired branch");
        let items = vec![&a, &b, &c, &d];
        let closed = HashSet::from(["at://x/p/d"]);

        let err = chain_containing(&items, "at://x/p/a", &closed)
            .unwrap_err()
            .to_string();
        assert!(err.contains("at://x/p/b"), "{err}");
        assert!(err.contains("at://x/p/c"), "{err}");
        assert!(!err.contains("at://x/p/d"), "named the retired one: {err}");
    }

    /// A loop can order nothing. Refused from every entry point, not just
    /// the one that happens to close the cycle.
    #[test]
    fn a_cycle_is_refused() {
        let a = item("at://x/p/a", Some("at://x/p/b"), "chicken");
        let b = item("at://x/p/b", Some("at://x/p/a"), "egg");
        let items = vec![&a, &b];

        for start in ["at://x/p/a", "at://x/p/b"] {
            let err = chain_containing(&items, start, &nothing_closed())
                .unwrap_err()
                .to_string();
            assert!(err.contains("loops"), "{err}");
        }
    }

    // -----------------------------------------------------------------------
    // What a writer is allowed to leave behind
    // -----------------------------------------------------------------------

    use super::{damage, project, refuse_new_damage};
    use crate::clients::atproto::record::Op;
    use jacquard::types::recordkey::RecordKey;

    const NSID: &str = crate::lexicon::tangled::PULL_NSID;

    fn key(rkey: &str) -> RecordKey<jacquard::types::recordkey::Rkey> {
        RecordKey::any_owned(rkey).expect("a record key")
    }

    /// A pull record's at-uri in the acting account's repository, which is
    /// where every op these tests build would land.
    fn mine(rkey: &str) -> String {
        format!("at://me/{NSID}/{rkey}")
    }

    fn relink(rkey: &str, dependent_on: Option<&str>) -> Op {
        let mut value = json!({"title": rkey});
        if let Some(dep) = dependent_on {
            value["dependentOn"] = json!(dep);
        }
        Op::Update {
            nsid: crate::lexicon::tangled::PULL_NSID,
            rkey: key(rkey),
            value,
        }
    }

    /// The projection is what the *reader* would see afterwards, so it is
    /// checked by reading it rather than by inspecting it.
    #[test]
    fn a_projection_is_the_listing_the_ops_would_leave() {
        let a = item(&mine("a"), None, "bottom");
        let items = [&a];
        let after = project(&items, "me", &[relink("b", Some(&mine("a")))]);
        assert_eq!(
            after.len(),
            2,
            "the created record is not in the projection"
        );
        let refs: Vec<&Value> = after.iter().collect();
        let chain = chain_containing(&refs, &mine("a"), &nothing_closed())
            .expect("a chain")
            .expect("two members");
        assert_eq!(uris(&chain.members), vec![mine("a"), mine("b")]);
    }

    /// The fork that cost this epic its most expensive bug: a second pull
    /// pointed at a parent that already had one. Invisible from the branch
    /// you did not enter through, which is why `damage` walks from every
    /// member rather than from one.
    #[test]
    fn a_write_that_would_fork_the_chain_is_refused() {
        let a = item(&mine("a"), None, "bottom");
        let b = item(&mine("b"), Some(&mine("a")), "top");
        let items = [&a, &b];

        let err = refuse_new_damage(
            &items,
            "me",
            &[relink("c", Some(&mine("a")))],
            &nothing_closed(),
        )
        .expect_err("a second dependent on `a` is a fork");
        let said = format!("{err:#}");
        assert!(said.contains("a stack is a chain, not a fork"), "{said}");
        assert!(said.contains("nothing was written"), "{said}");
    }

    /// Deleting a member out of the middle leaves the one above it pointing
    /// at a record that is gone. `stack unlink` exists because this is a real
    /// thing to want; doing it without relinking is the part that is refused.
    #[test]
    fn a_write_that_would_strand_the_member_above_it_is_refused() {
        let a = item(&mine("a"), None, "bottom");
        let b = item(&mine("b"), Some(&mine("a")), "middle");
        let c = item(&mine("c"), Some(&mine("b")), "top");
        let items = [&a, &b, &c];

        let err = refuse_new_damage(
            &items,
            "me",
            &[Op::Delete {
                nsid: crate::lexicon::tangled::PULL_NSID,
                rkey: key("b"),
            }],
            &nothing_closed(),
        )
        .expect_err("c is left depending on a record that no longer exists");
        assert!(format!("{err:#}").contains("continues below"), "{err:#}");

        // The same delete *with* the relink is the operation `unlink`
        // performs, and it has to stay allowed.
        refuse_new_damage(
            &items,
            "me",
            &[
                relink("c", Some(&mine("a"))),
                Op::Delete {
                    nsid: crate::lexicon::tangled::PULL_NSID,
                    rkey: key("b"),
                },
            ],
            &nothing_closed(),
        )
        .expect("relinking past the deleted member is a well-formed chain");
    }

    /// **A repair is not refused for the damage it repairs.** A stack can
    /// already be forked when a command starts — that is exactly when
    /// somebody reaches for `unlink` — so the comparison is against the
    /// damage on record now, and only what the write adds is refused.
    #[test]
    fn an_existing_fork_does_not_block_a_write_that_does_not_worsen_it() {
        let a = item(&mine("a"), None, "bottom");
        let b = item(&mine("b"), Some(&mine("a")), "one branch");
        let c = item(&mine("c"), Some(&mine("a")), "the other");
        let items = [&a, &b, &c];
        assert!(
            !damage(&items, &nothing_closed()).is_empty(),
            "the fixture is not forked"
        );

        // Touching a title while the fork stands is allowed: it is no worse.
        refuse_new_damage(
            &items,
            "me",
            &[relink("b", Some(&mine("a")))],
            &nothing_closed(),
        )
        .expect("a write that leaves the fork exactly as it was");

        // And the repair — dropping one of the two links — is allowed too.
        refuse_new_damage(&items, "me", &[relink("c", None)], &nothing_closed())
            .expect("unforking is the whole point of being able to write here");
    }

    /// A batch that mints a member depending on one it deletes in the same
    /// batch. `applyWrites` is atomic, so both land together and the new
    /// record points at nothing — a shape no single op looks wrong in, and
    /// exactly what a verb that reasoned one op at a time would produce.
    #[test]
    fn a_batch_that_deletes_what_it_depends_on_is_refused() {
        let a = item(&mine("a"), None, "bottom");
        let b = item(&mine("b"), Some(&mine("a")), "top");
        let items = [&a, &b];

        let err = refuse_new_damage(
            &items,
            "me",
            &[
                relink("c", Some(&mine("b"))),
                Op::Delete {
                    nsid: crate::lexicon::tangled::PULL_NSID,
                    rkey: key("b"),
                },
            ],
            &nothing_closed(),
        )
        .expect_err("c would depend on a record the same batch removed");
        assert!(format!("{err:#}").contains("continues below"), "{err:#}");
    }

    /// A relink that closes the chain into a loop. Nothing orders a cycle,
    /// and the reader has always refused one; this is the writer refusing to
    /// create one, which is the half that was missing.
    #[test]
    fn a_batch_that_closes_the_chain_into_a_loop_is_refused() {
        let a = item(&mine("a"), None, "bottom");
        let b = item(&mine("b"), Some(&mine("a")), "top");
        let items = [&a, &b];

        let err = refuse_new_damage(
            &items,
            "me",
            // The bottom now depends on the top.
            &[relink("a", Some(&mine("b")))],
            &nothing_closed(),
        )
        .expect_err("a loop orders nothing");
        assert!(format!("{err:#}").contains("loops back"), "{err:#}");
    }

    /// A member depending on itself: the shortest loop there is, and the one
    /// an off-by-one in a relink produces.
    #[test]
    fn a_member_that_depends_on_itself_is_refused() {
        let a = item(&mine("a"), None, "bottom");
        let items = [&a];
        let err = refuse_new_damage(
            &items,
            "me",
            &[relink("a", Some(&mine("a")))],
            &nothing_closed(),
        )
        .expect_err("a pull cannot sit on itself");
        assert!(format!("{err:#}").contains("loops back"), "{err:#}");
    }

    /// **Damage the write did not cause is not the write's to answer for**,
    /// and this is the case that decides it: a chain already dangling below
    /// what the listing holds — deep pagination, or a member in a
    /// contributor's PDS — must not turn every subsequent write into a
    /// refusal.
    #[test]
    fn a_chain_already_hanging_off_a_record_nobody_holds_is_not_the_writes_fault() {
        let a = item(&mine("a"), Some("at://somebody/else/p/x"), "bottom");
        let items = [&a];
        assert!(
            !damage(&items, &nothing_closed()).is_empty(),
            "the fixture is not dangling"
        );

        refuse_new_damage(
            &items,
            "me",
            &[relink("b", Some(&mine("a")))],
            &nothing_closed(),
        )
        .expect("adding a member above it introduces nothing");
    }

    /// Two ops on one record in a single batch, which `applyWrites` applies
    /// in order. The projection has to apply them in order too, or it
    /// answers about a record state that never exists.
    #[test]
    fn a_projection_applies_two_ops_to_one_record_in_order() {
        let a = item(&mine("a"), None, "bottom");
        let b = item(&mine("b"), Some(&mine("a")), "top");
        let items = [&a, &b];

        // Relinked to nothing, then relinked back: the net effect is none,
        // and a projection that took the *first* write would see the top
        // detached.
        let after = project(
            &items,
            "me",
            &[relink("b", None), relink("b", Some(&mine("a")))],
        );
        let refs: Vec<&Value> = after.iter().collect();
        let chain = chain_containing(&refs, &mine("a"), &nothing_closed())
            .expect("a chain")
            .expect("two members");
        assert_eq!(uris(&chain.members), vec![mine("a"), mine("b")]);
    }

    /// A record created and then deleted in the same batch is not there
    /// afterwards, so nothing may be left depending on it.
    #[test]
    fn a_record_created_and_deleted_in_one_batch_is_absent() {
        let a = item(&mine("a"), None, "bottom");
        let items = [&a];
        let after = project(
            &items,
            "me",
            &[
                relink("b", Some(&mine("a"))),
                Op::Delete {
                    nsid: crate::lexicon::tangled::PULL_NSID,
                    rkey: key("b"),
                },
            ],
        );
        assert_eq!(after.len(), 1, "the deleted record survived the projection");
    }

    /// Ops on other collections change no chain and must not be projected as
    /// pulls. `pr close` writes a status record beside these, and a
    /// projection that treated one as a pull would invent a member.
    #[test]
    fn a_projection_ignores_records_that_are_not_pulls() {
        let a = item(&mine("a"), None, "bottom");
        let items = [&a];
        let after = project(
            &items,
            "me",
            &[Op::Create {
                nsid: crate::lexicon::tangled::PULL_STATUS_NSID,
                rkey: key("s"),
                value: json!({"pull": mine("a"), "status": "closed"}),
            }],
        );
        assert_eq!(after.len(), 1, "a status record was projected as a pull");
    }

    /// **A stack whose members live in two accounts is ordinary**, not
    /// damage: a contributor's pull can sit in a chain with the repo
    /// owner's, and the listing carries both. The gate walks every pull in
    /// the repo, so a false positive here would refuse writes on any repo
    /// where two people stack.
    #[test]
    fn a_chain_across_two_accounts_is_not_damage() {
        let theirs = item("at://them/sh.tangled.repo.pull/x", None, "their bottom");
        let mine_on_top = item(
            &mine("a"),
            Some("at://them/sh.tangled.repo.pull/x"),
            "my member",
        );
        let items = [&theirs, &mine_on_top];
        assert!(
            damage(&items, &nothing_closed()).is_empty(),
            "a two-account chain read as damaged"
        );
        refuse_new_damage(
            &items,
            "me",
            &[relink("b", Some(&mine("a")))],
            &nothing_closed(),
        )
        .expect("stacking another member of mine on top is ordinary");
    }

    /// Adding a *third* branch to a parent that is already forked is allowed,
    /// and that is a decision rather than an oversight. The chain at that
    /// parent is already unorderable; a third dependent does not make it
    /// less readable, and refusing here would block the writes that repair
    /// the fork — `stack unlink` relinks a dependent before dropping a
    /// member, and the intermediate state it projects is exactly this.
    #[test]
    fn widening_a_fork_that_already_exists_is_allowed() {
        let a = item(&mine("a"), None, "bottom");
        let b = item(&mine("b"), Some(&mine("a")), "one branch");
        let c = item(&mine("c"), Some(&mine("a")), "another");
        let items = [&a, &b, &c];

        refuse_new_damage(
            &items,
            "me",
            &[relink("d", Some(&mine("a")))],
            &nothing_closed(),
        )
        .expect("the parent was already unorderable");
    }

    /// A *second* fork, at a parent that did not have one, is refused even
    /// while the first stands. The identity is the parent, so damage
    /// elsewhere is not a licence.
    #[test]
    fn a_second_fork_elsewhere_is_still_refused() {
        let a = item(&mine("a"), None, "bottom");
        let b = item(&mine("b"), Some(&mine("a")), "one branch");
        let c = item(&mine("c"), Some(&mine("a")), "another");
        let d = item(&mine("d"), Some(&mine("b")), "above b");
        let items = [&a, &b, &c, &d];

        let err = refuse_new_damage(
            &items,
            "me",
            &[relink("e", Some(&mine("b")))],
            &nothing_closed(),
        )
        .expect_err("b gains a fork of its own");
        assert!(format!("{err:#}").contains(&mine("b")), "{err:#}");
    }

    /// An empty batch changes nothing and must not refuse, however damaged
    /// the records already are.
    #[test]
    fn an_empty_batch_is_never_refused() {
        let a = item(&mine("a"), None, "bottom");
        let b = item(&mine("b"), Some(&mine("a")), "one branch");
        let c = item(&mine("c"), Some(&mine("a")), "another");
        let items = [&a, &b, &c];
        refuse_new_damage(&items, "me", &[], &nothing_closed()).expect("nothing was written");
    }

    /// A closed member keeps its link and is read past rather than counted as
    /// a fork, which is the rule `stack retire` broke. The writer has to
    /// share it, or every reconcile that retires a member would refuse
    /// itself.
    #[test]
    fn a_closed_sibling_is_not_a_fork_to_the_writer_either() {
        let a = item(&mine("a"), None, "bottom");
        let retired = item(&mine("b"), Some(&mine("a")), "retired");
        let items = [&a, &retired];
        let closed: HashSet<&str> = [retired["uri"].as_str().unwrap()].into_iter().collect();

        refuse_new_damage(&items, "me", &[relink("c", Some(&mine("a")))], &closed)
            .expect("a live member above a retired one shares its parent by design");
    }
}