//! A repository kept on a [`Store`] (Oxygen Strata, docs/stores.md): what
//! `oxy push dir://…` or `oxy clone s3://…` talk to.
//!
//! The store holds three kinds of key:
//!
//! - `oxygen.toml`: the layout's format and the repository's id, written
//!   once by the first push;
//! - `bundles/<seq>-<hash>.bundle`: each push's changes, signatures,
//!   seals, objects and policies, as the wire protocol encodes a bundle,
//!   written once and named by content;
//! - `refs`: every bookmark with its heads and attestation, every tag, the
//!   policy and its history, encoded as the protocol's `refs` response,
//!   moved by compare-and-swap.
//!
//! A store cannot run code, so the client does what a hub does on a push,
//! against a **mirror**: a bare repository of its own, kept beside the
//! remote's other state under `.oxy/remotes/<name>/strata/`, brought level
//! with the store before every use (the record read, the bundles it has
//! not seen imported, the view set from the record), and then used as a
//! [`LocalRemote`] would be. A push runs the hub's own checks
//! (`receive_push`) on the mirror, uploads the bundle, and moves `refs`
//! from the version it read; if the record moved meanwhile, the mirror is
//! put back and the push reports it, exactly as a hub reports a stale
//! push. Nothing in the store is trusted: every object is re-hashed as it
//! is imported, and the fetch that follows verifies each bookmark's
//! attestation under the policy, as it does for any remote.

use super::{fail, Repo, RepoError, Result};
use crate::hash::{ChangeHash, ObjectRef, ObjectType};
use crate::objects::StateObject;
use crate::protocol::{
    wire, Bundle, RefResult, RefUpdate, RefsResponse, Remote, RemoteOptions, SignedPolicy, TagRef,
    UNBOUND_ID,
};
use crate::strata::{Error as StoreError, Store, Version};
use std::cell::RefCell;
use std::collections::BTreeSet;
use std::fs;
use std::path::{Path, PathBuf};

const MARKER_KEY: &str = "oxygen.toml";
const REFS_KEY: &str = "refs";
const BUNDLES_PREFIX: &str = "bundles/";
const FORMAT: u32 = 1;

impl From<StoreError> for RepoError {
    fn from(e: StoreError) -> RepoError {
        RepoError::Msg(e.to_string())
    }
}

/// The `oxygen.toml` marker: two lines, written once.
fn marker(id: [u8; 16]) -> Vec<u8> {
    format!("format = {FORMAT}\nrepo = \"{}\"\n", hex::encode(id)).into_bytes()
}

fn parse_marker(bytes: &[u8], what: &str) -> Result<[u8; 16]> {
    let text = String::from_utf8_lossy(bytes);
    let mut format = None;
    let mut repo = None;
    for line in text.lines() {
        let Some((k, v)) = line.split_once('=') else {
            continue;
        };
        let v = v.trim().trim_matches('"');
        match k.trim() {
            "format" => format = v.parse::<u32>().ok(),
            "repo" => {
                repo = hex::decode(v)
                    .ok()
                    .and_then(|b| <[u8; 16]>::try_from(b).ok())
            }
            _ => {}
        }
    }
    match (format, repo) {
        (Some(f), _) if f > FORMAT => fail(format!(
            "{what} holds a repository in Strata format {f}, newer than this oxy understands (format {FORMAT}): upgrade oxy"
        )),
        (Some(_), Some(id)) => Ok(id),
        _ => fail(format!("{what}: its oxygen.toml is not a Strata marker")),
    }
}

/// A remote kept on a store.
pub struct BucketRemote {
    store: Box<dyn Store>,
    /// Where the mirror lives: the remote's directory under `.oxy/remotes`,
    /// or a temporary directory when there is no repository yet (a clone),
    /// removed on drop.
    dir: PathBuf,
    temporary: bool,
    claimed_id: Option<[u8; 16]>,
    mirror: RefCell<Option<Repo>>,
}

impl Drop for BucketRemote {
    fn drop(&mut self) {
        if self.temporary {
            let _ = fs::remove_dir_all(&self.dir);
        }
    }
}

/// What a sync found: the record's version (to move it from) and the
/// repository id the store holds, if any.
struct Synced {
    version: Option<Version>,
    id: Option<[u8; 16]>,
}

impl BucketRemote {
    pub fn open(store: Box<dyn Store>, options: &RemoteOptions) -> Result<BucketRemote> {
        let (dir, temporary) = match &options.cache_dir {
            Some(d) => (d.clone(), false),
            None => {
                let mut nonce = [0u8; 6];
                rand::RngCore::fill_bytes(&mut rand::rngs::OsRng, &mut nonce);
                (
                    std::env::temp_dir().join(format!(
                        "oxy-strata-{}-{}",
                        std::process::id(),
                        hex::encode(nonce)
                    )),
                    true,
                )
            }
        };
        Ok(BucketRemote {
            store,
            dir,
            temporary,
            claimed_id: options.repo_id,
            mirror: RefCell::new(None),
        })
    }

    fn what(&self) -> String {
        self.store.describe()
    }

    /// The repository id the store holds: `None` until a first push.
    fn stored_id(&self) -> Result<Option<[u8; 16]>> {
        match self.store.get(MARKER_KEY) {
            Ok(b) => Ok(Some(parse_marker(&b, &self.what())?)),
            Err(StoreError::NotFound(_)) => Ok(None),
            Err(e) => Err(e.into()),
        }
    }

    fn mirror_dir(&self) -> PathBuf {
        self.dir.join("mirror")
    }

    fn seen_path(&self) -> PathBuf {
        self.dir.join("seen")
    }

    /// The bundles already imported into the mirror.
    fn seen(&self) -> BTreeSet<String> {
        fs::read_to_string(self.seen_path())
            .map(|s| s.lines().map(str::to_string).collect())
            .unwrap_or_default()
    }

    fn mark_seen(&self, key: &str) -> Result<()> {
        let mut seen = self.seen();
        if seen.insert(key.to_string()) {
            let text: String = seen.iter().map(|k| format!("{k}\n")).collect();
            crate::durable::write_atomic(&self.seen_path(), text.as_bytes())?;
        }
        Ok(())
    }

    /// Open the mirror for repository `id`, creating it on first use.
    fn open_mirror(&self, id: [u8; 16]) -> Result<()> {
        if self.mirror.borrow().is_some() {
            return Ok(());
        }
        let dir = self.mirror_dir();
        let repo = if dir.join(super::OXY_DIR).is_dir() {
            let repo = Repo::open(&dir)?;
            if repo.id()? != id {
                // The store was emptied and reused for another repository:
                // start the mirror over.
                drop(repo);
                fs::remove_dir_all(&self.dir)?;
                fs::create_dir_all(&self.dir)?;
                Repo::init_bare_with_id(&dir, id)?
            } else {
                repo
            }
        } else {
            fs::create_dir_all(&self.dir)?;
            Repo::init_bare_with_id(&dir, id)?
        };
        *self.mirror.borrow_mut() = Some(repo);
        Ok(())
    }

    fn with_mirror<T>(&self, f: impl FnOnce(&Repo) -> Result<T>) -> Result<T> {
        let m = self.mirror.borrow();
        match m.as_ref() {
            Some(repo) => f(repo),
            None => fail(format!(
                "{} holds no repository yet: push one to it first",
                self.what()
            )),
        }
    }

    /// Bring the mirror level with the store.
    fn sync(&self) -> Result<Synced> {
        let Some(id) = self.stored_id()? else {
            return Ok(Synced {
                version: None,
                id: None,
            });
        };
        self.open_mirror(id)?;
        let record = self.store.read_ref(REFS_KEY)?;
        let (resp, version) = match record {
            Some((bytes, version)) => {
                let (rid, resp) = wire::decode_refs(&bytes)?;
                if rid != id {
                    return fail(format!(
                        "{}: its refs record names a different repository than its oxygen.toml; the store is damaged",
                        self.what()
                    ));
                }
                (resp, Some(version))
            }
            None => (RefsResponse::default(), None),
        };
        self.with_mirror(|mirror| {
            // The policy first: bundles are then imported under it.
            if let Some(sp) = &resp.policy {
                let have = mirror.current_policy()?.map(|(_, p)| p.policy.serial);
                if have.is_none_or(|s| s < sp.policy.serial) {
                    mirror
                        .adopt_policy_chain(&resp.history, sp, "strata sync")
                        .map_err(|e| RepoError::Msg(format!("policy from {}: {e}", self.what())))?;
                }
            }
            self.import_new_bundles(mirror)?;
            self.set_view(mirror, &resp)?;
            Ok(())
        })?;
        Ok(Synced {
            version,
            id: Some(id),
        })
    }

    /// Import every bundle in the store the mirror has not seen. Bundles
    /// are named in the order they were pushed, but two pushes from the
    /// same base may interleave, so a bundle whose dependencies are not
    /// there yet is tried again after the others.
    fn import_new_bundles(&self, mirror: &Repo) -> Result<()> {
        let seen = self.seen();
        let mut pending: Vec<String> = self
            .store
            .list(BUNDLES_PREFIX)?
            .into_iter()
            .map(|e| e.key)
            .filter(|k| !seen.contains(k))
            .collect();
        let strict = mirror.current_policy()?.is_some();
        while !pending.is_empty() {
            let mut deferred = Vec::new();
            let mut last_error = None;
            for key in pending.drain(..) {
                let bytes = self.store.get(&key)?;
                let bundle = wire::decode_bundle(&bytes)
                    .map_err(|e| RepoError::Msg(format!("{}: {key}: {e}", self.what())))?;
                match mirror.import_bundle(&bundle, strict) {
                    Ok(_) => self.mark_seen(&key)?,
                    Err(RepoError::Msg(m)) if m.contains("neither stored nor in the bundle") => {
                        last_error = Some(m);
                        deferred.push(key);
                    }
                    Err(e) => return Err(RepoError::Msg(format!("{}: {key}: {e}", self.what()))),
                }
            }
            if deferred.len() == pending.len() + deferred.len() && !deferred.is_empty() {
                // No progress: something those bundles need was never
                // pushed, or a push is still in progress.
                return fail(format!(
                    "{}: {} cannot be imported ({}); a push may be in progress, so try again, else the store is missing changes",
                    self.what(),
                    deferred.join(", "),
                    last_error.unwrap_or_default()
                ));
            }
            pending = deferred;
        }
        Ok(())
    }

    /// Make the mirror's view say what the record says.
    fn set_view(&self, mirror: &Repo, resp: &RefsResponse) -> Result<()> {
        let (op_ref, op) = mirror.head_op()?;
        let mut view = op.view.clone();
        view.bookmarks.clear();
        for r in &resp.refs {
            for h in &r.heads {
                if !mirror.store.has(h) {
                    return fail(format!(
                        "{}: bookmark {:?} names change {}, which no bundle there holds; a push may be in progress, so try again",
                        self.what(),
                        r.name,
                        h.short(12)
                    ));
                }
            }
            let state = mirror.store.put_state(&StateObject::new(r.heads.clone()))?;
            if state != r.state {
                return fail(format!(
                    "{}: bookmark {:?} names a state that does not match its heads; the store is damaged",
                    self.what(),
                    r.name
                ));
            }
            if let Some(sa) = &r.attestation {
                let current = mirror.latest_ref_attest(&r.name)?.map(|(_, a)| a);
                if current.as_ref() != Some(sa) {
                    let a = mirror.store.put_attest(&sa.attest)?;
                    for sig in &sa.sigs {
                        mirror.store.put_signature(&a, sig)?;
                    }
                    mirror.set_ref_attest(&r.name, &a)?;
                }
            }
            view.bookmarks.insert(r.name.clone(), state);
        }
        let policy = mirror.current_policy()?.map(|(_, sp)| sp.policy);
        for t in &resp.tags {
            mirror.store.put_state(&StateObject::new(t.heads.clone()))?;
            mirror
                .adopt_tag(&mut view, t, policy.as_ref())
                .map_err(|e| RepoError::Msg(format!("{}: {e}", self.what())))?;
        }
        if view != op.view {
            mirror.record_op(&[op_ref], "strata sync", view)?;
        }
        Ok(())
    }

    /// The record for the mirror's current references.
    fn record(&self, id: [u8; 16]) -> Result<Vec<u8>> {
        self.with_mirror(|mirror| {
            let mut resp = mirror.remote_refs()?;
            // A mirror has no checkout; the store's clone picks a bookmark
            // as any other remote's does.
            resp.default_bookmark = None;
            Ok(wire::encode_refs(id, &resp))
        })
    }

    /// Store a bundle under the next sequence number; returns its key.
    fn upload(&self, bundle: &Bundle) -> Result<String> {
        let bytes = wire::encode_bundle(bundle);
        let hash = blake3::hash(&bytes).to_hex();
        let seq = self.store.list(BUNDLES_PREFIX)?.len() + 1;
        let key = format!("{BUNDLES_PREFIX}{seq:08}-{}.bundle", &hash[..16]);
        match self.store.put_new(&key, &bytes) {
            Ok(()) | Err(StoreError::Exists(_)) => {}
            Err(e) => return Err(e.into()),
        }
        self.mark_seen(&key)?;
        Ok(key)
    }
}

impl Remote for BucketRemote {
    fn repo_id(&self) -> Result<[u8; 16]> {
        Ok(self.stored_id()?.unwrap_or(UNBOUND_ID))
    }

    fn refs(&self) -> Result<RefsResponse> {
        let synced = self.sync()?;
        if synced.id.is_none() {
            return Ok(RefsResponse::default());
        }
        self.with_mirror(|mirror| {
            let mut resp = mirror.remote_refs()?;
            resp.default_bookmark = None;
            Ok(resp)
        })
    }

    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
        self.with_mirror(|m| m.negotiate(have, want))
    }

    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {
        self.with_mirror(|m| m.bundle_raw_for(changes, max_blob))
    }

    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> {
        self.with_mirror(|m| m.objects_for(refs))
    }

    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle> {
        self.with_mirror(|m| m.checkpoint_bundle_for(heads))
    }

    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>> {
        self.with_mirror(|m| m.shallow_changes(heads, depth))
    }

    fn push(
        &self,
        bundle: Bundle,
        updates: Vec<RefUpdate>,
        policy: Option<SignedPolicy>,
        tags: Vec<TagRef>,
    ) -> Result<Vec<RefResult>> {
        let mut synced = self.sync()?;
        let id = match synced.id {
            Some(id) => id,
            None => {
                // The first push binds the store to this repository. Two
                // first pushes race on the marker; the loser reads the
                // winner's and goes on only if it is the same repository.
                let Some(id) = self.claimed_id else {
                    return fail(format!(
                        "{} is empty, and this client did not say which repository it is",
                        self.what()
                    ));
                };
                match self.store.put_new(MARKER_KEY, &marker(id)) {
                    Ok(()) => {}
                    Err(StoreError::Exists(_)) => {}
                    Err(e) => return Err(e.into()),
                }
                synced = self.sync()?;
                match synced.id {
                    Some(stored) if stored == id => id,
                    _ => {
                        return fail(format!(
                            "{} was just bound to another repository by a concurrent push",
                            self.what()
                        ))
                    }
                }
            }
        };
        if let Some(claimed) = self.claimed_id {
            if claimed != id {
                return fail(format!(
                    "{} holds a different repository (id {}) than this one ({}): a store holds one repository; use another prefix or bucket",
                    self.what(),
                    hex::encode(id),
                    hex::encode(claimed)
                ));
            }
        }
        let (before_ref, before) = self.with_mirror(|m| m.head_op())?;
        let results =
            self.with_mirror(|m| m.receive_push(bundle.clone(), updates, policy, tags))?;
        if !results.iter().any(|r| r.ok) {
            return Ok(results);
        }
        if !bundle.is_empty() {
            self.upload(&bundle)?;
        }
        let record = self.record(id)?;
        match self
            .store
            .write_ref(REFS_KEY, &record, synced.version.as_ref())
        {
            Ok(_) => Ok(results),
            Err(StoreError::Conflict(_)) => {
                // Someone else moved the record meanwhile. The bundle is in
                // the store and harmless; the mirror goes back to what the
                // store says, and the push is reported as a hub would.
                self.with_mirror(|m| {
                    let (now, _) = m.head_op()?;
                    if now != before_ref {
                        m.record_op(
                            &[now],
                            "strata push undone: the remote moved",
                            before.view.clone(),
                        )?;
                    }
                    Ok(())
                })?;
                let _ = self.sync();
                Ok(results
                    .into_iter()
                    .map(|r| RefResult {
                        name: r.name,
                        ok: false,
                        reason: "the remote moved while pushing (another push landed first); pull, then push again".into(),
                    })
                    .collect())
            }
            Err(e) => Err(e.into()),
        }
    }
}

/// Open a remote on `store` for the local repository described by `options`.
pub fn open_store_remote(
    store: Box<dyn Store>,
    options: &RemoteOptions,
) -> Result<Box<dyn Remote>> {
    Ok(Box::new(BucketRemote::open(store, options)?))
}

/// Whether `path` holds a Strata store rather than a repository.
pub fn is_store_dir(path: &Path) -> bool {
    path.join(MARKER_KEY).is_file() || path.join(REFS_KEY).is_file()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::repo::IdentityScope;
    use crate::strata::{Entry, LocalStore, MemoryStore};
    use std::ops::Range;
    use std::sync::Mutex;

    fn hermetic() {
        static ONCE: std::sync::Once = std::sync::Once::new();
        static DIR: std::sync::OnceLock<tempfile::TempDir> = std::sync::OnceLock::new();
        ONCE.call_once(|| {
            let d = tempfile::tempdir().unwrap();
            std::env::set_var("OXY_HOME", d.path());
            DIR.set(d).ok();
        });
    }

    fn write(root: &Path, rel: &str, content: &[u8]) {
        let p = root.join(rel);
        fs::create_dir_all(p.parent().unwrap()).unwrap();
        fs::write(p, content).unwrap();
    }

    fn add_remote(repo: &mut Repo, name: &str, url: &str) {
        repo.config.set(&format!("remote.{name}.url"), url).unwrap();
        repo.save_config().unwrap();
    }

    fn keys(store: &dyn Store, prefix: &str) -> Vec<String> {
        store
            .list(prefix)
            .unwrap()
            .into_iter()
            .map(|e| e.key)
            .collect()
    }

    #[test]
    fn push_clone_and_pull_through_a_directory_store() {
        hermetic();
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let url = format!("dir://{}", root.join("bucket").display());
        let store = LocalStore::open(root.join("bucket")).unwrap();

        // A signs, commits, tags and pushes: the store gets the marker, one
        // bundle, and a record.
        let mut a = Repo::init(&root.join("a"))
            .unwrap()
            .with_home(root.join("home-a"));
        a.create_identity("Ada", "ada@example.org", IdentityScope::Repo)
            .unwrap();
        add_remote(&mut a, "bucket", &url);
        write(&root.join("a"), "f.txt", b"a\n");
        let h1 = a.commit("first", &[]).unwrap();
        a.tag_create("v1", &[h1], "one").unwrap();
        let results = a.push("bucket", false).unwrap();
        assert!(results.iter().all(|r| r.ok), "{results:?}");
        assert_eq!(
            keys(&store, ""),
            [
                "bundles/00000001-".to_string() + &keys(&store, "bundles/")[0][17..],
                "oxygen.toml".into(),
                "refs".into()
            ]
        );
        assert_eq!(
            parse_marker(&store.get("oxygen.toml").unwrap(), "t").unwrap(),
            a.id().unwrap()
        );
        let again = a.push("bucket", false).unwrap();
        assert!(
            again[0].ok && again[0].reason.contains("up to date"),
            "{again:?}"
        );
        assert_eq!(
            keys(&store, "bundles/").len(),
            1,
            "a no-op push uploads nothing"
        );

        // B clones from the store: same id, same file, the tag, and the
        // seal on the change.
        let b = Repo::clone(&url, &root.join("b")).unwrap();
        assert_eq!(b.id().unwrap(), a.id().unwrap());
        assert_eq!(fs::read(root.join("b/f.txt")).unwrap(), b"a\n");
        assert_eq!(b.bookmark_heads().unwrap(), vec![h1]);
        assert_eq!(
            b.tags()
                .unwrap()
                .iter()
                .map(|t| t.name.clone())
                .collect::<Vec<_>>(),
            ["v1"]
        );
        assert!(matches!(
            b.authenticate_change(&h1).unwrap(),
            crate::repo::SignatureStatus::Valid { sealed: true, .. }
        ));
        assert!(
            root.join("b/.oxy/remotes/origin/strata/mirror/.oxy")
                .is_dir(),
            "the mirror lives with the remote's state"
        );
        assert!(
            !std::env::temp_dir()
                .read_dir()
                .unwrap()
                .flatten()
                .any(|e| e.file_name().to_string_lossy().starts_with("oxy-strata-")),
            "a clone's temporary mirror is removed"
        );

        // B commits and pushes; A pulls it. A then commits and pushes.
        write(&root.join("b"), "f.txt", b"a\nb\n");
        let h2 = b.commit("second", &[]).unwrap();
        assert!(b.push("origin", false).unwrap()[0].ok);
        assert_eq!(keys(&store, "bundles/").len(), 2);
        let pr = a.pull("bucket").unwrap();
        assert!(pr.changed && pr.checked_out);
        assert_eq!(fs::read(root.join("a/f.txt")).unwrap(), b"a\nb\n");
        write(&root.join("a"), "g.txt", b"g\n");
        let h3 = a.commit("third", &[]).unwrap();
        assert!(a.push("bucket", false).unwrap()[0].ok);
        let pr = b.pull("origin").unwrap();
        assert!(pr.changed);
        // g.txt is new, so its change commutes with B's: two heads.
        let mut want = vec![h2, h3];
        want.sort();
        assert_eq!(b.bookmark_heads().unwrap(), want);
        assert_eq!(fs::read(root.join("b/g.txt")).unwrap(), b"g\n");

        // A third clone gets everything from the three bundles.
        let c = Repo::clone(&url, &root.join("c")).unwrap();
        assert_eq!(c.bookmark_heads().unwrap(), want);
        assert_eq!(fs::read(root.join("c/f.txt")).unwrap(), b"a\nb\n");
    }

    #[test]
    fn a_push_behind_the_remote_is_refused_before_anything_is_uploaded() {
        hermetic();
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let url = format!("dir://{}", root.join("bucket").display());
        let store = LocalStore::open(root.join("bucket")).unwrap();
        let mut a = Repo::init(&root.join("a")).unwrap();
        add_remote(&mut a, "bucket", &url);
        write(&root.join("a"), "f.txt", b"a\n");
        a.commit("first", &[]).unwrap();
        assert!(a.push("bucket", false).unwrap()[0].ok);
        let b = Repo::clone(&url, &root.join("b")).unwrap();
        write(&root.join("b"), "f.txt", b"b\n");
        b.commit("from b", &[]).unwrap();
        assert!(b.push("origin", false).unwrap()[0].ok);
        write(&root.join("a"), "g.txt", b"g\n");
        a.commit("from a", &[]).unwrap();
        let e = a.push("bucket", false).unwrap_err().to_string();
        assert!(e.contains("pull first"), "{e}");
        assert_eq!(keys(&store, "bundles/").len(), 2);
        assert!(a.pull("bucket").unwrap().changed);
        assert!(a.push("bucket", false).unwrap()[0].ok);
        assert_eq!(keys(&store, "bundles/").len(), 3);
    }

    /// A store whose record moves under a writer the first time it is
    /// written to: what a concurrent push looks like between the read and
    /// the compare-and-swap.
    struct Racing {
        inner: MemoryStore,
        raced: Mutex<bool>,
    }

    impl Store for Racing {
        fn describe(&self) -> String {
            "racing".into()
        }
        fn has(&self, key: &str) -> crate::strata::Result<bool> {
            self.inner.has(key)
        }
        fn get(&self, key: &str) -> crate::strata::Result<Vec<u8>> {
            self.inner.get(key)
        }
        fn get_range(&self, key: &str, range: Range<u64>) -> crate::strata::Result<Vec<u8>> {
            self.inner.get_range(key, range)
        }
        fn put_new(&self, key: &str, bytes: &[u8]) -> crate::strata::Result<()> {
            self.inner.put_new(key, bytes)
        }
        fn list(&self, prefix: &str) -> crate::strata::Result<Vec<Entry>> {
            self.inner.list(prefix)
        }
        fn delete(&self, key: &str) -> crate::strata::Result<()> {
            self.inner.delete(key)
        }
        fn read_ref(&self, key: &str) -> crate::strata::Result<Option<(Vec<u8>, Version)>> {
            self.inner.read_ref(key)
        }
        fn write_ref(
            &self,
            key: &str,
            bytes: &[u8],
            expected: Option<&Version>,
        ) -> crate::strata::Result<Version> {
            let mut raced = self.raced.lock().unwrap();
            if !*raced && expected.is_some() {
                *raced = true;
                // Someone else lands a record: the same references with a
                // different detail, so the version moves.
                let (current, v) = self.inner.read_ref(key)?.unwrap();
                let (id, mut resp) =
                    wire::decode_refs(&current).map_err(|e| StoreError::Backend(e.to_string()))?;
                resp.default_bookmark = Some("raced".into());
                self.inner
                    .write_ref(key, &wire::encode_refs(id, &resp), Some(&v))?;
            }
            self.inner.write_ref(key, bytes, expected)
        }
    }

    #[test]
    fn a_record_moved_during_a_push_fails_the_push_and_keeps_the_mirror_honest() {
        hermetic();
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let a = Repo::init(&root.join("a")).unwrap();
        write(&root.join("a"), "f.txt", b"a\n");
        let h1 = a.commit("first", &[]).unwrap();
        let store = Box::new(Racing {
            inner: MemoryStore::new(),
            raced: Mutex::new(false),
        });
        let options = RemoteOptions {
            repo_id: Some(a.id().unwrap()),
            cache_dir: Some(root.join("cache")),
            ..Default::default()
        };
        let remote = BucketRemote::open(store, &options).unwrap();
        // The first push creates the record (expected None: no race yet).
        let state = a.store.put_state(&StateObject::new(vec![h1])).unwrap();
        let empty = a.store.put_state(&StateObject::new(vec![])).unwrap();
        let mut bundle = a.bundle_for(&[h1]).unwrap();
        bundle
            .objects
            .push((ObjectType::State, StateObject::new(vec![h1]).encode()));
        let update = |old, new| RefUpdate {
            name: "main".into(),
            old_state: old,
            new_state: new,
            force: false,
            attestation: None,
        };
        let r = remote
            .push(bundle, vec![update(empty, state)], None, vec![])
            .unwrap();
        assert!(r[0].ok, "{r:?}");
        // The second push meets the race: refused, nothing moved.
        write(&root.join("a"), "f.txt", b"ab\n");
        let h2 = a.commit("second", &[]).unwrap();
        let state2 = a.store.put_state(&StateObject::new(vec![h2])).unwrap();
        let mut bundle = a.bundle_for(&[h2]).unwrap();
        bundle
            .objects
            .push((ObjectType::State, StateObject::new(vec![h2]).encode()));
        let r = remote
            .push(bundle, vec![update(state, state2)], None, vec![])
            .unwrap();
        assert!(
            !r[0].ok && r[0].reason.contains("moved while pushing"),
            "{r:?}"
        );
        let refs = remote.refs().unwrap();
        assert_eq!(
            refs.refs[0].heads,
            vec![h1],
            "the mirror says what the store says"
        );
    }

    #[test]
    fn a_store_holds_one_repository() {
        hermetic();
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let url = format!("dir://{}", root.join("bucket").display());
        let mut a = Repo::init(&root.join("a")).unwrap();
        add_remote(&mut a, "bucket", &url);
        write(&root.join("a"), "f.txt", b"a\n");
        a.commit("first", &[]).unwrap();
        assert!(a.push("bucket", false).unwrap()[0].ok);
        let mut other = Repo::init(&root.join("other")).unwrap();
        add_remote(&mut other, "bucket", &url);
        write(&root.join("other"), "f.txt", b"o\n");
        other.commit("mine", &[]).unwrap();
        let e = other.push("bucket", false).unwrap_err().to_string();
        assert!(
            e.contains("different repository") || e.contains("another repository"),
            "{e}"
        );
        assert_eq!(
            keys(&LocalStore::open(root.join("bucket")).unwrap(), "bundles/").len(),
            1
        );
    }

    #[test]
    fn a_damaged_bundle_is_caught_and_an_empty_store_cannot_be_cloned() {
        hermetic();
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let url = format!("dir://{}", root.join("bucket").display());
        let e = Repo::clone(&url, &root.join("nothing"))
            .err()
            .expect("refused")
            .to_string();
        assert!(e.contains("empty and unbound"), "{e}");
        let mut a = Repo::init(&root.join("a")).unwrap();
        add_remote(&mut a, "bucket", &url);
        write(&root.join("a"), "f.txt", b"a\n");
        a.commit("first", &[]).unwrap();
        assert!(a.push("bucket", false).unwrap()[0].ok);
        let store = LocalStore::open(root.join("bucket")).unwrap();
        let key = keys(&store, "bundles/").remove(0);
        let mut bytes = store.get(&key).unwrap();
        let mid = bytes.len() / 2;
        bytes[mid] ^= 0xff;
        store.delete(&key).unwrap();
        store.put_new(&key, &bytes).unwrap();
        let e = Repo::clone(&url, &root.join("b"))
            .err()
            .expect("refused")
            .to_string();
        assert!(
            e.contains(&key) || e.contains("bundle") || e.contains("hash"),
            "{e}"
        );
        assert!(!root.join("b/f.txt").exists());
    }

    #[test]
    fn markers_parse_and_refuse_newer_formats() {
        let id = [7u8; 16];
        assert_eq!(parse_marker(&marker(id), "s").unwrap(), id);
        let e = parse_marker(
            b"format = 9\nrepo = \"07070707070707070707070707070707\"\n",
            "s",
        )
        .unwrap_err()
        .to_string();
        assert!(e.contains("format 9"), "{e}");
        assert!(parse_marker(b"hello", "s").is_err());
    }
}