//! An account's repository read whole and checked before anything in it is
//! believed: its commit signed by the key the account's DID document names,
//! and the records the ones that commit's MST holds.
//!
//! It comes from a mirror when one is given and has it, as STAR-lite
//! (microcosm's: every key and record in key order, the commit in its
//! header and no tree), else as a CAR from the account's PDS
//! (`com.atproto.sync.getRepo`). STAR-lite is read as it streams, the MST
//! root rebuilt from its keys one node a layer at a time; a CAR is held
//! whole to walk its tree.
//! <https://tangled.org/microcosm.blue/star/tree/main/v0-star/star-lite>

use anyhow::{Context, Result, ensure};
use ciborium::Value as V;
use knotless::crypto::PublicKey;
use knotless::drisl;
use knotless::env::Ctx;
use reqwest::StatusCode;
use sha2::{Digest, Sha256};
use std::collections::HashMap;

/// The most of a repository read, in bytes: a larger one is not read.
pub const MAX: usize = 256 << 20;

/// What a mirror is asked for: STAR-lite, or a CAR.
const ACCEPT: &str = "application/x.microcosm.star-lite, application/vnd.ipld.car;q=0.5";

/// STAR-lite's first bytes, of its version 0: `*l\0`.
const STAR_LITE: [u8; 3] = [0x2a, 0x6c, 0x00];

/// The first bytes of a DAG-CBOR SHA-256 CIDv1, as STAR-lite's root is.
const CBOR_CID: [u8; 4] = [0x01, 0x71, 0x12, 0x20];

/// The longest key, record and commit STAR-lite is read with.
const MAX_KEY: u64 = 1024;
const MAX_RECORD: u64 = 1 << 20;
const MAX_COMMIT: u64 = 4096;

/// A repository as its latest commit holds it: the commit's revision, and
/// the records kept.
#[derive(Debug)]
pub struct Archive {
    pub rev: String,
    pub records: Vec<Record>,
}

#[derive(Debug, PartialEq)]
pub struct Record {
    pub collection: String,
    pub rkey: String,
    pub value: serde_json::Value,
}

/// Reads `did`'s repository, from `mirror` when one is given and answers,
/// else from its PDS, `pds`: at most `max` bytes of it, keeping the records
/// of the collections `keep` names. A mirror that asks to be asked less
/// fails the read, for it to be tried again later.
pub async fn fetch(
    ctx: &Ctx,
    did: &str,
    pds: &str,
    mirror: Option<&str>,
    max: usize,
    keep: &(dyn Fn(&str) -> bool + Sync),
) -> Result<Archive> {
    let doc = knotless::identity::document(ctx, did).await?;
    let key = atproto::did::signing_key(&doc)?;
    let res = download(ctx, did, pds, mirror).await?;
    read(res, did, &key, max, keep).await
}

/// `did`'s repository, as `mirror` answers it, or else its PDS.
async fn download(
    ctx: &Ctx,
    did: &str,
    pds: &str,
    mirror: Option<&str>,
) -> Result<reqwest::Response> {
    let get_repo = |host: &str| {
        let url = format!(
            "{}/xrpc/com.atproto.sync.getRepo",
            host.trim_end_matches('/')
        );
        ctx.http.get(url).query(&[("did", did)])
    };
    if let Some(mirror) = mirror
        && let Ok(res) = get_repo(mirror).header("accept", ACCEPT).send().await
    {
        let slow = res.status() == StatusCode::TOO_MANY_REQUESTS;
        ensure!(!slow, "{mirror} asked to be asked less");
        if res.status().is_success() {
            return Ok(res);
        }
    }
    Ok(get_repo(pds).send().await?.error_for_status()?)
}

/// A body read as asked for, never more than `max` bytes of it.
struct Reader {
    body: reqwest::Response,
    buf: Vec<u8>,
    read: usize,
    max: usize,
}

impl Reader {
    /// Whether `n` bytes are in hand, reading on until they are or the
    /// body ends.
    async fn fill(&mut self, n: usize) -> Result<bool> {
        while self.buf.len() < n {
            let Some(chunk) = self.body.chunk().await? else {
                return Ok(false);
            };
            self.read += chunk.len();
            ensure!(
                self.read <= self.max,
                "the repository is over {} bytes",
                self.max
            );
            self.buf.extend_from_slice(&chunk);
        }
        Ok(true)
    }

    async fn take(&mut self, n: usize) -> Result<Vec<u8>> {
        ensure!(self.fill(n).await?, "the archive ends early");
        Ok(self.buf.drain(..n).collect())
    }

    /// An unsigned LEB128 integer, at most `most`.
    async fn varint(&mut self, most: u64) -> Result<u64> {
        let mut value = 0u64;
        for i in 0..10 {
            let byte = self.take(1).await?[0];
            value |= u64::from(byte & 0x7f) << (7 * i);
            if byte & 0x80 == 0 {
                ensure!(value <= most, "a length in the archive is over {most}");
                return Ok(value);
            }
        }
        anyhow::bail!("a length in the archive runs on")
    }

    /// The rest of the body.
    async fn rest(mut self) -> Result<Vec<u8>> {
        while self.fill(self.buf.len() + 1).await? {}
        Ok(self.buf)
    }
}

/// The records of `body`, STAR-lite or a CAR as its first bytes say, at
/// most `max` bytes of it.
async fn read(
    body: reqwest::Response,
    did: &str,
    key: &PublicKey,
    max: usize,
    keep: &(dyn Fn(&str) -> bool + Sync),
) -> Result<Archive> {
    let mut r = Reader {
        body,
        buf: Vec::new(),
        read: 0,
        max,
    };
    let star = r.fill(STAR_LITE.len()).await? && r.buf.starts_with(&STAR_LITE);
    match star {
        true => star_lite(r, did, key, keep).await,
        false => open(&r.rest().await?, did, key, keep),
    }
}

/// The records of a STAR-lite archive, once its commit is found to be
/// `did`'s and signed with `key`, and its keys and records to make the MST
/// root the commit names.
async fn star_lite(
    mut r: Reader,
    did: &str,
    key: &PublicKey,
    keep: &(dyn Fn(&str) -> bool + Sync),
) -> Result<Archive> {
    r.take(STAR_LITE.len()).await?;
    let root = r.take(36).await?;
    ensure!(
        root.starts_with(&CBOR_CID),
        "the archive's root is not an MST node's CID"
    );
    let len = r.varint(MAX_COMMIT).await?;
    let commit: V = ciborium::from_reader(&r.take(usize::try_from(len)?).await?[..])?;
    // The commit as signed names the root the archive's records make.
    let mut fields = commit.as_map().context("the commit is not a map")?.clone();
    fields.retain(|(k, _)| k.as_text() != Some("data"));
    fields.push((V::Text("data".into()), link(&root)));
    let rev = checked(&V::Map(fields), did, key)?;
    let mut tree = Tree::default();
    let mut records = Vec::new();
    while r.fill(1).await? {
        let len = r.varint(MAX_KEY).await?;
        let path = String::from_utf8(r.take(usize::try_from(len)?).await?)?;
        let len = r.varint(MAX_RECORD).await?;
        let record = r.take(usize::try_from(len)?).await?;
        tree.add(&path, &record)?;
        records.extend(kept(&path, &record, keep)?);
    }
    ensure!(
        tree.root() == root,
        "the archive's records do not make the MST its commit signed"
    );
    Ok(Archive { rev, records })
}

/// The record at `path`, if its collection is one `keep` names. One DRISL
/// holds that no record does, as a float, is left out.
fn kept(path: &str, block: &[u8], keep: &(dyn Fn(&str) -> bool + Sync)) -> Result<Option<Record>> {
    let (collection, rkey) = path.split_once('/').context("a path is not a record's")?;
    let value = keep(collection).then(|| drisl::record(block)).flatten();
    Ok(value.map(|value| Record {
        collection: collection.to_string(),
        rkey: rkey.to_string(),
        value,
    }))
}

/// The revision of `commit`, once it is found to be `did`'s and signed with
/// `key`.
fn checked(commit: &V, did: &str, key: &PublicKey) -> Result<String> {
    let field = |name: &str| map_get(commit, name);
    let author = field("did").and_then(V::as_text);
    ensure!(author == Some(did), "the commit is not {did}'s");
    let rev = field("rev")
        .and_then(V::as_text)
        .context("the commit has no rev")?;
    let sig = field("sig")
        .and_then(V::as_bytes)
        .context("the commit is unsigned")?;
    let signed = key.verify(&unsigned(commit)?, sig);
    ensure!(signed, "{did}'s commit is not signed by its key");
    Ok(rev.to_string())
}

/// A DAG-CBOR link to `cid`.
fn link(cid: &[u8]) -> V {
    V::Tag(42, Box::new(V::Bytes([&[0], cid].concat())))
}

/// The CID of a DAG-CBOR block.
fn cid_of(block: &[u8]) -> Vec<u8> {
    [&CBOR_CID[..], &Sha256::digest(block)[..]].concat()
}

/// A DAG-CBOR block's bytes, its maps' keys in DRISL's order: shorter
/// first, then by their bytes.
fn encode(value: &V) -> Result<Vec<u8>> {
    let mut out = Vec::new();
    ciborium::into_writer(&ordered(value), &mut out)?;
    Ok(out)
}

fn ordered(value: &V) -> V {
    match value {
        V::Map(fields) => {
            let mut fields: Vec<(V, V)> = fields
                .iter()
                .map(|(k, v)| (k.clone(), ordered(v)))
                .collect();
            let order = |k: &V| k.as_text().map(|t| (t.len(), t.to_string()));
            fields.sort_by_key(|(k, _)| order(k));
            V::Map(fields)
        }
        V::Array(items) => V::Array(items.iter().map(ordered).collect()),
        other => other.clone(),
    }
}

/// The MST root, rebuilt from its keys in order: every node left of the
/// stack is finished, so only one node a layer is held at a time.
#[derive(Default)]
struct Tree {
    stack: Vec<Node>,
    prev: Option<usize>,
    last: String,
}

#[derive(Default)]
struct Node {
    left: Option<Vec<u8>>,
    entries: Vec<(String, Vec<u8>, Option<Vec<u8>>)>,
}

impl Node {
    fn is_empty(&self) -> bool {
        self.left.is_none() && self.entries.is_empty()
    }

    /// Its CID, as atproto writes a node: each key shortened by what it
    /// shares with the one before.
    fn cid(&self) -> Vec<u8> {
        let mut last = "";
        let entries = self.entries.iter().map(|(key, value, right)| {
            let shared = last
                .bytes()
                .zip(key.bytes())
                .take_while(|(a, b)| a == b)
                .count();
            last = key;
            V::Map(vec![
                (
                    V::Text("k".into()),
                    V::Bytes(key.as_bytes()[shared..].to_vec()),
                ),
                (V::Text("p".into()), V::Integer(shared.into())),
                (V::Text("t".into()), right.as_deref().map_or(V::Null, link)),
                (V::Text("v".into()), link(value)),
            ])
        });
        let node = V::Map(vec![
            (V::Text("e".into()), V::Array(entries.collect())),
            (
                V::Text("l".into()),
                self.left.as_deref().map_or(V::Null, link),
            ),
        ]);
        let mut out = Vec::new();
        ciborium::into_writer(&node, &mut out).expect("a node encodes");
        cid_of(&out)
    }
}

impl Tree {
    fn add(&mut self, key: &str, record: &[u8]) -> Result<()> {
        ensure!(
            key > self.last.as_str(),
            "the archive's keys are out of order at {key}"
        );
        self.last = key.to_string();
        let layer = layer_of(key);
        while self.stack.len() <= layer {
            self.stack.push(Node::default());
        }
        if self.prev.is_none_or(|prev| layer > prev) {
            self.fold(layer);
        }
        self.stack[layer]
            .entries
            .push((key.to_string(), cid_of(record), None));
        self.prev = Some(layer);
        Ok(())
    }

    /// Finishes every node below `layer`, each hung under the one above:
    /// right of its last entry, or as its left.
    fn fold(&mut self, layer: usize) {
        for i in 0..layer {
            if self.stack[i].is_empty() {
                continue;
            }
            let cid = std::mem::take(&mut self.stack[i]).cid();
            let parent = &mut self.stack[i + 1];
            match parent.entries.last_mut() {
                Some(last) => last.2 = Some(cid),
                None => parent.left = Some(cid),
            }
        }
    }

    fn root(mut self) -> Vec<u8> {
        let top = self.stack.len().saturating_sub(1);
        self.fold(top);
        self.stack.pop().unwrap_or_default().cid()
    }
}

/// How many layers up the MST a key sits: the leading zeros of its
/// SHA-256, counted two bits at a time.
fn layer_of(key: &str) -> usize {
    let hash = Sha256::digest(key.as_bytes());
    let bytes = hash.iter().position(|b| *b != 0).unwrap_or(hash.len());
    let bits = hash.get(bytes).map_or(0, |b| b.leading_zeros() as usize);
    (bytes * 8 + bits) / 2
}

/// The records of `car`, a repository's CAR, once its commit is found to
/// be `did`'s, signed with `key`, and each block on the way to a record
/// kept is found to be the one its CID names.
pub fn open(
    car: &[u8],
    did: &str,
    key: &PublicKey,
    keep: &(dyn Fn(&str) -> bool + Sync),
) -> Result<Archive> {
    let (root, blocks) = blocks(car)?;
    let commit: V = decode(&blocks, &root)?;
    let rev = checked(&commit, did, key)?;
    let data = map_get(&commit, "data").and_then(drisl::link);
    let data = data.context("the commit names no data")?;
    let mut entries = Vec::new();
    walk(&blocks, data, &mut entries)?;
    let mut records = Vec::new();
    for (path, cid) in entries {
        records.extend(kept(&path, block(&blocks, &cid)?, keep)?);
    }
    Ok(Archive { rev, records })
}

/// The bytes the commit's signature is over: the commit without it.
fn unsigned(commit: &V) -> Result<Vec<u8>> {
    let fields = commit.as_map().context("the commit is not a map")?;
    let kept = fields.iter().filter(|(k, _)| k.as_text() != Some("sig"));
    encode(&V::Map(kept.cloned().collect()))
}

/// Each key under the MST node at `cid`, in order, with its value's CID.
fn walk(blocks: &Blocks, cid: &[u8], out: &mut Vec<(String, Vec<u8>)>) -> Result<()> {
    let node = decode(blocks, cid)?;
    if let Some(left) = map_get(&node, "l").and_then(drisl::link) {
        walk(blocks, left, out)?;
    }
    let entries = map_get(&node, "e").and_then(V::as_array);
    let mut last: Vec<u8> = Vec::new();
    for entry in entries.context("an MST node has no entries")? {
        let shared = map_get(entry, "p").and_then(V::as_integer);
        let shared = shared.and_then(|p| usize::try_from(p).ok());
        let suffix = map_get(entry, "k").and_then(V::as_bytes);
        let value = map_get(entry, "v").and_then(drisl::link);
        let parts = shared.zip(suffix).zip(value);
        let ((shared, suffix), value) = parts.context("an MST entry is malformed")?;
        let mut key = last
            .get(..shared)
            .context("an MST key shares more than it has")?
            .to_vec();
        key.extend_from_slice(suffix);
        out.push((String::from_utf8(key.clone())?, value.to_vec()));
        if let Some(right) = map_get(entry, "t").and_then(drisl::link) {
            walk(blocks, right, out)?;
        }
        last = key;
    }
    Ok(())
}

fn map_get<'a>(value: &'a V, name: &str) -> Option<&'a V> {
    let fields = value.as_map()?;
    fields
        .iter()
        .find(|(k, _)| k.as_text() == Some(name))
        .map(|(_, v)| v)
}

/// A CAR's blocks, by their CIDs' bytes.
type Blocks<'a> = HashMap<&'a [u8], &'a [u8]>;

/// The root a CAR names, and its blocks.
fn blocks(car: &[u8]) -> Result<(Vec<u8>, Blocks<'_>)> {
    let mut rest = car;
    let mut sections = Vec::new();
    while !rest.is_empty() {
        let len = varint(&mut rest).context("a CAR section has no length")?;
        let len = usize::try_from(len)?;
        ensure!(len <= rest.len(), "a CAR section runs past its end");
        sections.push(&rest[..len]);
        rest = &rest[len..];
    }
    let (header, sections) = sections.split_first().context("the CAR is empty")?;
    let header: V = ciborium::from_reader(*header)?;
    let roots = map_get(&header, "roots").and_then(V::as_array);
    let root = roots.and_then(|r| drisl::link(r.first()?));
    let root = root.context("the CAR names no root")?.to_vec();
    let mut blocks = HashMap::new();
    for section in sections {
        // A CIDv1 of a SHA-256: version, codec, the hash's code and length.
        ensure!(section.len() >= 36, "a CAR block is too short for its CID");
        blocks.insert(&section[..36], &section[36..]);
    }
    Ok((root, blocks))
}

/// The block `cid` names, once it is found to hash to it.
fn block<'a>(blocks: &Blocks<'a>, cid: &[u8]) -> Result<&'a [u8]> {
    let data = blocks.get(cid).context("the CAR is missing a block")?;
    let sha256 = cid.len() == 36 && cid[0] == 1 && cid[2..4] == [0x12, 0x20];
    ensure!(sha256, "a CID is not a SHA-256 CIDv1");
    ensure!(
        cid[4..] == Sha256::digest(data)[..],
        "a block is not the one its CID names"
    );
    Ok(data)
}

fn decode(blocks: &Blocks, cid: &[u8]) -> Result<V> {
    Ok(ciborium::from_reader(block(blocks, cid)?)?)
}

/// An unsigned LEB128 integer off the front of `bytes`.
fn varint(bytes: &mut &[u8]) -> Option<u64> {
    let mut value = 0u64;
    for (i, &byte) in bytes.iter().enumerate().take(10) {
        value |= u64::from(byte & 0x7f) << (7 * i);
        if byte & 0x80 == 0 {
            *bytes = &bytes[i + 1..];
            return Some(value);
        }
    }
    None
}

#[cfg(test)]
mod tests {
    use super::*;
    use knotless::crypto::P256Key;

    fn cbor(value: &V) -> Vec<u8> {
        let mut out = Vec::new();
        ciborium::into_writer(value, &mut out).unwrap();
        out
    }

    fn cid(data: &[u8]) -> Vec<u8> {
        let mut cid = vec![1, 0x71, 0x12, 0x20];
        cid.extend_from_slice(&Sha256::digest(data));
        cid
    }

    fn link(cid: &[u8]) -> V {
        let mut bytes = vec![0];
        bytes.extend_from_slice(cid);
        V::Tag(42, Box::new(V::Bytes(bytes)))
    }

    fn map(pairs: Vec<(&str, V)>) -> V {
        V::Map(
            pairs
                .into_iter()
                .map(|(k, v)| (V::Text(k.into()), v))
                .collect(),
        )
    }

    fn put(car: &mut Vec<u8>, section: &[u8]) {
        let mut len = section.len();
        while len >= 0x80 {
            car.push((len as u8 & 0x7f) | 0x80);
            len >>= 7;
        }
        car.push(len as u8);
        car.extend_from_slice(section);
    }

    /// A repository of `did`'s holding two `records`, by path, the second
    /// in a subtree right of the first, its commit signed by `key` and then
    /// changed as `signed` changes it.
    fn repo(did: &str, key: &P256Key, records: &[(&str, V)], signed: impl Fn(&mut V)) -> Vec<u8> {
        let mut blocks: Vec<Vec<u8>> = Vec::new();
        let mut leaf = |value: &V| {
            let bytes = cbor(value);
            let cid = cid(&bytes);
            blocks.push(bytes);
            cid
        };
        let values: Vec<Vec<u8>> = records.iter().map(|(_, v)| leaf(v)).collect();
        // The second record in a subtree to the right of the first.
        let (first, second) = (records[0].0.as_bytes(), records[1].0.as_bytes());
        let right = map(vec![
            ("l", V::Null),
            (
                "e",
                V::Array(vec![map(vec![
                    ("p", V::Integer(0.into())),
                    ("k", V::Bytes(second.to_vec())),
                    ("v", link(&values[1])),
                    ("t", V::Null),
                ])]),
            ),
        ]);
        let right = leaf(&right);
        let root = map(vec![
            ("l", V::Null),
            (
                "e",
                V::Array(vec![map(vec![
                    ("p", V::Integer(0.into())),
                    ("k", V::Bytes(first.to_vec())),
                    ("v", link(&values[0])),
                    ("t", link(&right)),
                ])]),
            ),
        ]);
        let data = leaf(&root);
        let mut commit = map(vec![
            ("did", V::Text(did.into())),
            ("version", V::Integer(3.into())),
            ("data", link(&data)),
            ("rev", V::Text("3l".into())),
            ("prev", V::Null),
        ]);
        let sig = key.sign(&encode(&commit).unwrap());
        let fields = commit.as_map_mut().unwrap();
        fields.push((V::Text("sig".into()), V::Bytes(sig)));
        signed(&mut commit);
        let commit = leaf(&commit);
        let mut car = Vec::new();
        put(
            &mut car,
            &cbor(&map(vec![
                ("version", V::Integer(1.into())),
                ("roots", V::Array(vec![link(&commit)])),
            ])),
        );
        for block in &blocks {
            let mut section = cid(block);
            section.extend_from_slice(block);
            put(&mut car, &section);
        }
        car
    }

    fn records() -> Vec<(&'static str, V)> {
        vec![
            (
                "dev.knotless.repo/a",
                map(vec![("name", V::Text("a".into()))]),
            ),
            (
                "dev.knotless.star/b",
                map(vec![("subject", V::Text("s".into()))]),
            ),
        ]
    }

    #[test]
    fn a_signed_repository_opens_to_its_records() {
        let key = P256Key::generate();
        let car = repo("did:plc:a", &key, &records(), |_| {});
        let all = open(&car, "did:plc:a", &key.public(), &|_| true).unwrap();
        assert_eq!(all.rev, "3l");
        let paths: Vec<String> = all
            .records
            .iter()
            .map(|r| format!("{}/{}", r.collection, r.rkey))
            .collect();
        assert_eq!(paths, ["dev.knotless.repo/a", "dev.knotless.star/b"]);
        assert_eq!(all.records[0].value, serde_json::json!({"name": "a"}));
        // Only the collections kept.
        let repos = open(&car, "did:plc:a", &key.public(), &|c| {
            c == "dev.knotless.repo"
        })
        .unwrap();
        assert_eq!(repos.records.len(), 1);
    }

    #[test]
    fn what_is_not_the_accounts_is_refused() {
        let key = P256Key::generate();
        let car = repo("did:plc:a", &key, &records(), |_| {});
        // Another account's, or signed by another key.
        assert!(open(&car, "did:plc:b", &key.public(), &|_| true).is_err());
        let other = P256Key::generate().public();
        let err = open(&car, "did:plc:a", &other, &|_| true).unwrap_err();
        assert!(
            format!("{err:#}").contains("not signed by its key"),
            "{err:#}"
        );
        // A commit changed after it was signed.
        let changed = repo("did:plc:a", &key, &records(), |commit| {
            commit.as_map_mut().unwrap()[3].1 = V::Text("4z".into());
        });
        assert!(open(&changed, "did:plc:a", &key.public(), &|_| true).is_err());
        // A block changed in the CAR: it no longer hashes to its CID.
        let mut flipped = car.clone();
        let at = flipped.len() - 1;
        flipped[at] ^= 1;
        let err = open(&flipped, "did:plc:a", &key.public(), &|_| true).unwrap_err();
        assert!(format!("{err:#}").contains("CID"), "{err:#}");
        // Cut short, or no CAR at all.
        assert!(open(&car[..car.len() - 3], "did:plc:a", &key.public(), &|_| true).is_err());
        assert!(open(&[], "did:plc:a", &key.public(), &|_| true).is_err());
        // A length that never ends.
        assert!(open(&[0x80], "did:plc:a", &key.public(), &|_| true).is_err());
    }

    /// The repository harness/star-fixture.mjs made with @atproto/repo:
    /// its DID, the key that signed it, its revision, as a CAR and as
    /// STAR-lite.
    struct Fixture {
        did: String,
        key: PublicKey,
        rev: String,
        car: Vec<u8>,
        star: Vec<u8>,
    }

    fn fixture() -> Fixture {
        let about: serde_json::Value =
            serde_json::from_str(include_str!("../fixtures/repo.json")).unwrap();
        Fixture {
            did: about["did"].as_str().unwrap().into(),
            key: PublicKey::from_did_key(about["key"].as_str().unwrap()).unwrap(),
            rev: about["rev"].as_str().unwrap().into(),
            car: include_bytes!("../fixtures/repo.car").to_vec(),
            star: include_bytes!("../fixtures/repo.star").to_vec(),
        }
    }

    /// `bytes` read as it would stream, a few bytes at a time.
    fn streamed(
        bytes: &[u8],
        did: &str,
        key: &PublicKey,
        max: usize,
        keep: &(dyn Fn(&str) -> bool + Sync),
    ) -> Result<Archive> {
        let res = reqwest::Response::from(axum::http::Response::new(bytes.to_vec()));
        futures::executor::block_on(read(res, did, key, max, keep))
    }

    #[test]
    fn star_lite_and_a_car_of_one_repository_read_the_same() {
        let f = fixture();
        let repos = |c: &str| c == "dev.knotless.repo";
        let star = streamed(&f.star, &f.did, &f.key, MAX, &repos).unwrap();
        let car = streamed(&f.car, &f.did, &f.key, MAX, &repos).unwrap();
        assert_eq!(
            (star.rev.as_str(), car.rev.as_str()),
            (f.rev.as_str(), f.rev.as_str())
        );
        assert_eq!(star.records.len(), 30);
        assert_eq!(star.records, car.records);
        assert_eq!(star.records[0].value["$type"], "dev.knotless.repo");
        let all = streamed(&f.star, &f.did, &f.key, MAX, &|_| true).unwrap();
        assert_eq!(all.records.len(), 60);
        // No records at all make the empty tree.
        assert_eq!(Tree::default().root(), Node::default().cid());
    }

    #[test]
    fn keys_sit_on_the_layers_atproto_puts_them() {
        // The examples of atproto's repository spec.
        assert_eq!(layer_of("2653ae71"), 0);
        assert_eq!(layer_of("blue"), 1);
        assert_eq!(layer_of("app.bsky.feed.post/454397e440ec"), 4);
        assert_eq!(layer_of("app.bsky.feed.post/9adeb165882c"), 8);
    }

    #[test]
    fn maps_in_arrays_are_ordered_too() {
        let n = |i: i32| V::Integer(i.into());
        let mixed = V::Array(vec![map(vec![("bb", n(1)), ("a", n(2))])]);
        let sorted = V::Array(vec![map(vec![("a", n(2)), ("bb", n(1))])]);
        let mut want = Vec::new();
        ciborium::into_writer(&sorted, &mut want).unwrap();
        assert_eq!(encode(&mixed).unwrap(), want);
    }

    #[test]
    fn star_lite_that_is_not_the_accounts_is_refused() {
        let f = fixture();
        let refused = |star: &[u8], did: &str, key: &PublicKey, max: usize| {
            let err = streamed(star, did, key, max, &|_| true).unwrap_err();
            format!("{err:#}")
        };
        let other = P256Key::generate().public();
        assert!(refused(&f.star, &f.did, &other, MAX).contains("not signed"));
        assert!(refused(&f.star, "did:plc:other", &f.key, MAX).contains("is not did:plc:other's"));
        assert!(refused(&f.star, &f.did, &f.key, 100).contains("over 100 bytes"));
        // A record changed: the keys and records no longer make the root.
        let mut changed = f.star.clone();
        let last = changed.len() - 2;
        changed[last] ^= 1;
        assert!(refused(&changed, &f.did, &f.key, MAX).contains("do not make the MST"));
        // Cut short, a root that is no node's, and lengths past what is read.
        let cut = &f.star[..f.star.len() - 3];
        assert!(refused(cut, &f.did, &f.key, MAX).contains("ends early"));
        let mut rootless = f.star.clone();
        rootless[4] = 0x55;
        assert!(refused(&rootless, &f.did, &f.key, MAX).contains("not an MST node's CID"));
        let mut rest = &f.star[39..];
        let commit = usize::try_from(varint(&mut rest).unwrap()).unwrap();
        let header = f.star.len() - rest.len() + commit;
        let long = [&f.star[..header], &[0x80, 0x10][..]].concat();
        assert!(refused(&long, &f.did, &f.key, MAX).contains("over 1024"));
        let endless = [&f.star[..header], &[0x80; 10][..]].concat();
        assert!(refused(&endless, &f.did, &f.key, MAX).contains("runs on"));
        // Keys out of order: the first record again after the rest.
        let mut first = &f.star[header..];
        let key_len = usize::try_from(varint(&mut first).unwrap()).unwrap();
        let mut after = &first[key_len..];
        let record_len = usize::try_from(varint(&mut after).unwrap()).unwrap();
        let one = f.star[header..].len() - after.len() + record_len;
        let again = [&f.star[..], &f.star[header..header + one]].concat();
        assert!(refused(&again, &f.did, &f.key, MAX).contains("out of order"));
    }
}