crates/store/src/lib.rs

//! The record: entries as versioned objects behind Apache OpenDAL.
//!
//! Four things hold here, because the rest of the system depends on them:
//!
//! - every write is a new version; the previous text is kept, byte for byte;
//! - nothing is edited in place: the current version is replaced, never patched;
//! - demotion leaves a tombstone, which is the same entry with a status;
//! - the whole record can be read back, so an index can be rebuilt from it alone.
//!
//! The store is rooted. Every path it touches is relative to the operator's root, names
//! are validated before they become paths, and nothing can name a file outside the root.
//! On a laptop the root is a directory; in an institution it is a bucket. The code does
//! not know which.
//!
//! Layout, per domain:
//!
//! ```text
//! record/<domain>/<id>.md                             the current version
//! record/<domain>/.history/<id>/<stamp>-<digest>.md   every previous version
//! record/<domain>/.erasures/<id>.yaml                 the receipt of an erasure
//! record/<domain>/.transcripts/<session>.json         a transcript: a source, not an entry
//! record/<domain>/.excerpts/<id>/<name>.json          what an entry kept of a transcript
//! record/<domain>/.contexts/<session>/<n>.json        a model call, as the egress saw it
//! record/<domain>/.sources/<id>.json                  what a tool brought in
//! record/<domain>/.tapes/<session>/<n>.json           a memory tool call and its result
//! ```
//!
//! Transcripts and excerpts are objects beside the entries, never entries: nothing that
//! lists, reads back or rebuilds entries sees them, because they live in directories a
//! listing of entries skips. What they hold is the server's business; the store only
//! keeps the bytes under a validated name.
//!
//! Erasure is the one exception to the first two rules: the body leaves every version,
//! current and archived, and the stub is written over each of them in place, with the
//! receipt beside the record saying what was removed.

use dm_core::erasure::{body_digest, Receipt, Removed};
use dm_core::{digest, Entry, Status};
use opendal::Operator;
use std::path::Path;
use time::{format_description::FormatItem, macros::format_description, OffsetDateTime};

const RECORD: &str = "record";
const HISTORY: &str = ".history";
const ERASURES: &str = ".erasures";
const TRANSCRIPTS: &str = ".transcripts";
const EXCERPTS: &str = ".excerpts";
const CONTEXTS: &str = ".contexts";
const SOURCES: &str = ".sources";
const TAPES: &str = ".tapes";
const STAMP: &[FormatItem<'_>] =
    format_description!("[year][month][day]T[hour][minute][second].[subsecond digits:6]Z");

/// What can go wrong at the record.
#[derive(Debug, thiserror::Error)]
pub enum Error {
    /// The backend refused or failed.
    #[error("record backend: {0}")]
    Backend(#[from] opendal::Error),
    /// Text in the record is not a valid entry.
    #[error("entry {path}: {source}")]
    Entry {
        /// The path that failed to parse.
        path: String,
        /// Why.
        source: dm_core::Error,
    },
    /// A domain or id that cannot be a path segment.
    #[error("bad name {0:?}: letters, digits, '.', '_' and '-' only, not starting with '.'")]
    BadName(String),
    /// Bytes in the record that are not UTF-8.
    #[error("entry {0} is not UTF-8")]
    NotText(String),
    /// Asked for an entry that is not there.
    #[error("no entry {id} in {domain}")]
    NotFound {
        /// The domain asked for.
        domain: String,
        /// The id asked for.
        id: String,
    },
    /// Asked to write a version with text over an entry that was erased, when the put
    /// reads the stub: a writer that read the entry before the erasure does not write
    /// what it holds of it. A put whose read comes before the erasure's write is not
    /// refused: see [`Store::put`].
    #[error("entry {id} in {domain} is erased: no version with text is written over its stub")]
    Erased {
        /// The domain of the stub.
        domain: String,
        /// Its id.
        id: String,
    },
}

/// Whether `text`, the current version of an entry, is an erased stub. Text that is not
/// an entry is not one: the write is then refused or accepted elsewhere.
fn is_erased(text: &str) -> bool {
    Entry::parse(text).is_ok_and(|e| e.front.status == Some(Status::Erased))
}

/// The outcome of a write.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Written {
    /// The entry's id.
    pub id: String,
    /// Digest of the text now current.
    pub digest: String,
    /// Where the previous version went, if there was one and it differed.
    pub archived: Option<String>,
    /// False when the text was already current and nothing was written.
    pub changed: bool,
}

/// One previous version of an entry.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Version {
    /// `<stamp>-<digest>`, which sorts by time.
    pub name: String,
    /// The text as it was.
    pub text: String,
}

/// Where beside a domain's entries an object lives that is not an entry.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Area {
    /// Session transcripts: sources, reached only by citation.
    Transcripts,
    /// Excerpts of transcripts, one directory per entry that keeps them.
    Excerpts,
    /// Decision contexts: what each model call of a session was sent and returned, as
    /// the egress saw it; one directory per session.
    Contexts,
    /// What a tool, fetch or search, brought in from outside: kept on a clock and
    /// reached by citation, as transcripts are.
    Sources,
    /// Tapes: each memory tool call of a session and its full result, for a re-run to
    /// be answered from; one directory per session.
    Tapes,
}

impl Area {
    fn dir(self) -> &'static str {
        match self {
            Area::Transcripts => TRANSCRIPTS,
            Area::Excerpts => EXCERPTS,
            Area::Contexts => CONTEXTS,
            Area::Sources => SOURCES,
            Area::Tapes => TAPES,
        }
    }
}

fn object_path(domain: &str, area: Area, parts: &[&str]) -> Result<String, Error> {
    let mut path = format!("{RECORD}/{}/{}", valid_name(domain)?, area.dir());
    for p in parts {
        path.push('/');
        path.push_str(valid_name(p)?);
    }
    Ok(path)
}

/// The record, behind one operator.
#[derive(Clone)]
pub struct Store {
    op: Operator,
}

fn valid_name(s: &str) -> Result<&str, Error> {
    let ok = !s.is_empty()
        && !s.starts_with('.')
        && s.len() <= 128
        && s.chars()
            .all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-'));
    if ok {
        Ok(s)
    } else {
        Err(Error::BadName(s.to_string()))
    }
}

fn current_path(domain: &str, id: &str) -> Result<String, Error> {
    Ok(format!(
        "{RECORD}/{}/{}.md",
        valid_name(domain)?,
        valid_name(id)?
    ))
}

fn history_dir(domain: &str, id: &str) -> Result<String, Error> {
    Ok(format!(
        "{RECORD}/{}/{HISTORY}/{}/",
        valid_name(domain)?,
        valid_name(id)?
    ))
}

fn receipt_path(domain: &str, id: &str) -> Result<String, Error> {
    Ok(format!(
        "{RECORD}/{}/{ERASURES}/{}.yaml",
        valid_name(domain)?,
        valid_name(id)?
    ))
}

fn stamp() -> String {
    OffsetDateTime::now_utc()
        .format(STAMP)
        .unwrap_or_else(|_| "00000000T000000.000000Z".to_string())
}

impl Store {
    /// A store over any operator. The operator's root is the record's boundary.
    #[must_use]
    pub fn from_operator(op: Operator) -> Store {
        Store { op }
    }

    /// A store in a directory on disk, created if missing.
    ///
    /// # Errors
    ///
    /// When the backend cannot be built, for instance when the path is not a directory.
    pub fn open_dir(root: &Path) -> Result<Store, Error> {
        let root = root.to_string_lossy();
        let op = Operator::new(opendal::services::Fs::default().root(&root))?;
        Ok(Store::from_operator(op))
    }

    /// A store that lives only in memory, for tests and dry runs.
    ///
    /// # Errors
    ///
    /// Never in practice; the memory backend has nothing to fail on.
    pub fn in_memory() -> Result<Store, Error> {
        let op = Operator::new(opendal::services::Memory::default())?;
        Ok(Store::from_operator(op))
    }

    /// Write an entry as the current version of its id in its domain.
    ///
    /// If a different version is current, it is archived first. If the same text is
    /// already current, nothing is written. If the entry supersedes another that is still
    /// active, that one is marked superseded, which is a write of its own.
    ///
    /// A version with text is refused over an erased stub that this put reads. The read
    /// and the writes after it are not one step: an [`Store::erase`] between them is not
    /// seen. The version read before the erasure then goes into the history, and this
    /// version, with its text, is written over the stub, so the entry is live again.
    ///
    /// # Errors
    ///
    /// [`Error::BadName`] for a domain or id that cannot be a path, [`Error::Erased`] for
    /// a version with text over an erased entry's stub, or a backend error.
    pub async fn put(&self, entry: &Entry) -> Result<Written, Error> {
        let domain = entry.front.domain.as_str();
        let id = entry.front.id.as_str();
        let path = current_path(domain, id)?;
        let text = entry.render();
        let new_digest = digest(&text);

        let previous = self.read_text(&path).await?;
        if previous.as_deref() == Some(text.as_str()) {
            return Ok(Written {
                id: id.to_string(),
                digest: new_digest,
                archived: None,
                changed: false,
            });
        }
        // a writer that read the entry before the erasure, and now writes a version of
        // what it read, would bring the text back over the stub. The check holds only when
        // this read sees the stub: an erasure that writes its stub between this read and
        // the writes below is not seen. The text read here then goes into the history,
        // and this version is written over the stub, so the entry is live again with its
        // text. Closing that needs a conditional put or a lock for each entry
        if entry.front.status != Some(Status::Erased) && previous.as_deref().is_some_and(is_erased)
        {
            return Err(Error::Erased {
                domain: domain.to_string(),
                id: id.to_string(),
            });
        }

        let mut archived = None;
        if let Some(old) = previous {
            let name = format!("{}-{}", stamp(), &digest(&old)[..8]);
            let hist = format!("{}{name}.md", history_dir(domain, id)?);
            self.op.write(&hist, old.into_bytes()).await?;
            archived = Some(name);
        }
        self.op.write(&path, text.into_bytes()).await?;

        if let Some(older) = &entry.front.supersedes {
            if let Some(mut old) = self.get(domain, older).await? {
                if old.is_active() {
                    old.front.status = Some(Status::Superseded);
                    Box::pin(self.put(&old)).await?;
                }
            }
        }

        Ok(Written {
            id: id.to_string(),
            digest: new_digest,
            archived,
            changed: true,
        })
    }

    /// The current version of an entry, parsed. `None` when there is none.
    ///
    /// # Errors
    ///
    /// A bad name, a backend error, or text in the record that is not an entry.
    pub async fn get(&self, domain: &str, id: &str) -> Result<Option<Entry>, Error> {
        Ok(self
            .get_with_text(domain, id)
            .await?
            .map(|(entry, _)| entry))
    }

    /// The current version of an entry, parsed, with the text it was parsed from: one
    /// read, so the two are the same version even while a write lands. `None` when there
    /// is none.
    ///
    /// # Errors
    ///
    /// A bad name, a backend error, or text in the record that is not an entry.
    pub async fn get_with_text(
        &self,
        domain: &str,
        id: &str,
    ) -> Result<Option<(Entry, String)>, Error> {
        let path = current_path(domain, id)?;
        match self.read_text(&path).await? {
            None => Ok(None),
            Some(text) => match Entry::parse(&text) {
                Ok(entry) => Ok(Some((entry, text))),
                Err(source) => Err(Error::Entry { path, source }),
            },
        }
    }

    /// The current version of an entry as text. `None` when there is none.
    ///
    /// # Errors
    ///
    /// A bad name or a backend error.
    pub async fn get_text(&self, domain: &str, id: &str) -> Result<Option<String>, Error> {
        self.read_text(&current_path(domain, id)?).await
    }

    /// Demote an entry: a new version with `status: demoted`, the body kept.
    ///
    /// # Errors
    ///
    /// [`Error::NotFound`] when there is no such entry, or what [`Store::put`] returns.
    pub async fn demote(&self, domain: &str, id: &str) -> Result<Written, Error> {
        let mut entry = self.get(domain, id).await?.ok_or_else(|| Error::NotFound {
            domain: domain.to_string(),
            id: id.to_string(),
        })?;
        entry.front.status = Some(Status::Demoted);
        self.put(&entry).await
    }

    /// Every version of an entry with the digest of its text, the current one first, as
    /// an erasure will list them: read before anything is removed, so the receipt can be
    /// written first.
    ///
    /// # Errors
    ///
    /// [`Error::NotFound`] when there is no such entry, a bad name, or a backend error.
    pub async fn digests(&self, domain: &str, id: &str) -> Result<Vec<Removed>, Error> {
        let current = self
            .read_text(&current_path(domain, id)?)
            .await?
            .ok_or_else(|| Error::NotFound {
                domain: domain.to_string(),
                id: id.to_string(),
            })?;
        let of_body = |text: &str| Entry::parse(text).ok().map(|e| body_digest(&e.body));
        let mut out = vec![Removed {
            version: "current".into(),
            digest: digest(&current),
            body_digest: of_body(&current),
        }];
        for version in self.history(domain, id).await? {
            out.push(Removed {
                version: version.name,
                digest: digest(&version.text),
                body_digest: of_body(&version.text),
            });
        }
        Ok(out)
    }

    /// Erase the entry at `domain` and `id`: `stub` is written over its current version
    /// and over every archived one, so no version keeps the body, and nothing is
    /// archived on the way. Returns what was removed, as [`Store::digests`] lists it.
    ///
    /// # Errors
    ///
    /// [`Error::NotFound`] when there is no such entry, a bad name, or a backend error.
    pub async fn erase(&self, domain: &str, id: &str, stub: &Entry) -> Result<Vec<Removed>, Error> {
        let removed = self.digests(domain, id).await?;
        let text = stub.render();
        let dir = history_dir(domain, id)?;
        for version in removed.iter().skip(1) {
            self.op
                .write(
                    &format!("{dir}{}.md", version.version),
                    text.clone().into_bytes(),
                )
                .await?;
        }
        self.op
            .write(&current_path(domain, id)?, text.into_bytes())
            .await?;
        Ok(removed)
    }

    /// Keep the receipt of an erasure beside the record.
    ///
    /// # Errors
    ///
    /// A bad name or a backend error.
    pub async fn put_receipt(&self, receipt: &Receipt) -> Result<(), Error> {
        let path = receipt_path(&receipt.domain, &receipt.id)?;
        self.op.write(&path, receipt.to_yaml().into_bytes()).await?;
        Ok(())
    }

    /// The receipt of an erasure, if the entry was erased.
    ///
    /// # Errors
    ///
    /// A bad name, a backend error, or a receipt that does not parse.
    pub async fn receipt(&self, domain: &str, id: &str) -> Result<Option<Receipt>, Error> {
        let path = receipt_path(domain, id)?;
        match self.read_text(&path).await? {
            None => Ok(None),
            Some(text) => Receipt::from_yaml(&text)
                .map(Some)
                .map_err(|source| Error::Entry { path, source }),
        }
    }

    /// Every receipt in a domain, by id.
    ///
    /// # Errors
    ///
    /// A bad name, a backend error, or a receipt that does not parse.
    pub async fn receipts(&self, domain: &str) -> Result<Vec<Receipt>, Error> {
        let dir = format!("{RECORD}/{}/{ERASURES}/", valid_name(domain)?);
        if !self.op.exists(&dir).await? {
            return Ok(Vec::new());
        }
        let mut ids: Vec<String> = self
            .op
            .list(&dir)
            .await?
            .into_iter()
            .filter(|e| e.metadata().is_file())
            .filter_map(|e| e.name().strip_suffix(".yaml").map(str::to_string))
            .collect();
        ids.sort();
        let mut out = Vec::with_capacity(ids.len());
        for id in ids {
            if let Some(r) = self.receipt(domain, &id).await? {
                out.push(r);
            }
        }
        Ok(out)
    }

    /// Ids with a current version in a domain, tombstones included, sorted.
    ///
    /// # Errors
    ///
    /// A bad name or a backend error.
    pub async fn list(&self, domain: &str) -> Result<Vec<String>, Error> {
        let dir = format!("{RECORD}/{}/", valid_name(domain)?);
        let mut ids: Vec<String> = self
            .op
            .list(&dir)
            .await?
            .into_iter()
            .filter(|e| e.metadata().is_file())
            .filter_map(|e| e.name().strip_suffix(".md").map(str::to_string))
            .collect();
        ids.sort();
        Ok(ids)
    }

    /// Whether `id` names a current entry in any domain of the record. An id names one
    /// entry across the record, since a reader asks for one by id alone, so a copy that
    /// would land under an id in use is given another; the record answers, not an index
    /// that can lag it.
    ///
    /// # Errors
    ///
    /// [`Error::BadName`] for an id that cannot be a path, or a backend error.
    pub async fn holds(&self, id: &str) -> Result<bool, Error> {
        for domain in self.domains().await? {
            if self.read_text(&current_path(&domain, id)?).await?.is_some() {
                return Ok(true);
            }
        }
        Ok(false)
    }

    /// Every domain with a directory in the record, sorted.
    ///
    /// # Errors
    ///
    /// A backend error.
    pub async fn domains(&self) -> Result<Vec<String>, Error> {
        let mut names: Vec<String> = self
            .op
            .list(&format!("{RECORD}/"))
            .await?
            .into_iter()
            // the listing names the directory itself as well as what is in it
            .filter(|e| e.metadata().is_dir() && e.path() != format!("{RECORD}/"))
            .map(|e| e.name().trim_end_matches('/').to_string())
            .filter(|n| !n.is_empty() && !n.starts_with('.'))
            .collect();
        names.sort();
        Ok(names)
    }

    /// Every previous version of an entry, oldest first.
    ///
    /// # Errors
    ///
    /// A bad name, a backend error, or bytes that are not text.
    pub async fn history(&self, domain: &str, id: &str) -> Result<Vec<Version>, Error> {
        let dir = history_dir(domain, id)?;
        if !self.op.exists(&dir).await? {
            return Ok(Vec::new());
        }
        let mut names: Vec<String> = self
            .op
            .list(&dir)
            .await?
            .into_iter()
            .filter(|e| e.metadata().is_file())
            .filter_map(|e| e.name().strip_suffix(".md").map(str::to_string))
            .collect();
        names.sort();
        let mut out = Vec::with_capacity(names.len());
        for name in names {
            let path = format!("{dir}{name}.md");
            let text = self
                .read_text(&path)
                .await?
                .ok_or_else(|| Error::NotText(path.clone()))?;
            out.push(Version { name, text });
        }
        Ok(out)
    }

    /// Every current entry in every domain. What a rebuild reads.
    ///
    /// # Errors
    ///
    /// A backend error, or text in the record that is not an entry.
    pub async fn all(&self) -> Result<Vec<Entry>, Error> {
        let mut out = Vec::new();
        for domain in self.domains().await? {
            for id in self.list(&domain).await? {
                if let Some(e) = self.get(&domain, &id).await? {
                    out.push(e);
                }
            }
        }
        Ok(out)
    }

    /// Keep an object that is not an entry under a domain: `parts` name it inside the
    /// area, each a valid name. Replaces what was there.
    ///
    /// # Errors
    ///
    /// A bad name or a backend error.
    pub async fn put_object(
        &self,
        domain: &str,
        area: Area,
        parts: &[&str],
        bytes: Vec<u8>,
    ) -> Result<(), Error> {
        self.op
            .write(&object_path(domain, area, parts)?, bytes)
            .await?;
        Ok(())
    }

    /// An object's text, if it is there.
    ///
    /// # Errors
    ///
    /// A bad name, a backend error, or bytes that are not text.
    pub async fn get_object(
        &self,
        domain: &str,
        area: Area,
        parts: &[&str],
    ) -> Result<Option<String>, Error> {
        self.read_text(&object_path(domain, area, parts)?).await
    }

    /// Remove an object; one that is not there is not an error.
    ///
    /// # Errors
    ///
    /// A bad name or a backend error.
    pub async fn remove_object(
        &self,
        domain: &str,
        area: Area,
        parts: &[&str],
    ) -> Result<(), Error> {
        self.op.delete(&object_path(domain, area, parts)?).await?;
        Ok(())
    }

    /// The names under a directory of an area, sorted: its files with `dirs` false, its
    /// subdirectories with `dirs` true. Nothing when the directory is not there.
    ///
    /// # Errors
    ///
    /// A bad name or a backend error.
    pub async fn objects(
        &self,
        domain: &str,
        area: Area,
        dir: &[&str],
        dirs: bool,
    ) -> Result<Vec<String>, Error> {
        let path = format!("{}/", object_path(domain, area, dir)?);
        if !self.op.exists(&path).await? {
            return Ok(Vec::new());
        }
        let mut names: Vec<String> = self
            .op
            .list(&path)
            .await?
            .into_iter()
            .filter(|e| e.path() != path && e.metadata().is_dir() == dirs)
            .map(|e| e.name().trim_end_matches('/').to_string())
            .filter(|n| !n.is_empty())
            .collect();
        names.sort();
        Ok(names)
    }

    async fn read_text(&self, path: &str) -> Result<Option<String>, Error> {
        match self.op.read(path).await {
            Ok(buf) => String::from_utf8(buf.to_vec())
                .map(Some)
                .map_err(|_| Error::NotText(path.to_string())),
            Err(e) if e.kind() == opendal::ErrorKind::NotFound => Ok(None),
            Err(e) => Err(e.into()),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use dm_core::{EntryType, Frontmatter, Label, TrustTier};
    use time::macros::datetime;

    fn entry(domain: &str, id: &str, body: &str) -> Entry {
        Entry {
            front: Frontmatter {
                id: id.into(),
                entry_type: EntryType::Fact,
                domain: domain.into(),
                author: "human:test:me".into(),
                created: datetime!(2026-10-01 10:00 UTC),
                source: vec![],
                confidence: None,
                trust_tier: TrustTier::Raw,
                review_by: None,
                supersedes: None,
                policy_context: vec![],
                classification: Label::Public,
                status: None,
                origin: None,
                summary: None,
                title: None,
            },
            body: body.into(),
        }
    }

    #[tokio::test]
    async fn put_then_get_round_trips() {
        let s = Store::in_memory().unwrap();
        let e = entry("dom-a", "mem-a-1", "First.\n");
        let w = s.put(&e).await.unwrap();
        assert!(w.changed);
        assert!(w.archived.is_none());
        assert_eq!(w.digest, digest(&e.render()));
        assert_eq!(s.get("dom-a", "mem-a-1").await.unwrap(), Some(e));
        assert_eq!(s.get("dom-a", "mem-a-9").await.unwrap(), None);
    }

    /// A writer that read an entry before its erasure, and writes a version of what it
    /// read after, is refused: the stub stays, and so does the history as the erasure
    /// left it. A later erased version, the stub again under another reason, is written.
    #[tokio::test]
    async fn no_version_with_text_is_written_over_an_erased_stub() {
        let s = Store::in_memory().unwrap();
        let e = entry(
            "dom-a",
            "mem-a-1",
            "The vault opens with passphrase-Qz7xx.\n",
        );
        s.put(&e).await.unwrap();
        let before = e.clone();
        let mut stub = entry("dom-a", "mem-a-1", "");
        stub.front.status = Some(Status::Erased);
        s.put(&stub).await.unwrap();
        let mut promoted = before;
        promoted.front.trust_tier = TrustTier::Verified;
        let refused = s.put(&promoted).await;
        assert!(
            matches!(&refused, Err(Error::Erased { id, .. }) if id == "mem-a-1"),
            "{refused:?}"
        );
        assert_eq!(
            s.get("dom-a", "mem-a-1").await.unwrap(),
            Some(stub.clone()),
            "the stub stays"
        );
        assert_eq!(
            s.digests("dom-a", "mem-a-1").await.unwrap().len(),
            2,
            "and no version was added"
        );
        let mut again = stub.clone();
        again.front.title = Some("erased".into());
        assert!(
            s.put(&again).await.unwrap().changed,
            "a stub over a stub is written"
        );
    }

    #[tokio::test]
    async fn every_write_is_a_new_version_and_the_old_text_survives() {
        let s = Store::in_memory().unwrap();
        let v1 = entry("dom-a", "mem-a-1", "First.\n");
        let v2 = entry("dom-a", "mem-a-1", "Second.\n");
        s.put(&v1).await.unwrap();
        let w = s.put(&v2).await.unwrap();
        assert!(w.changed);
        let archived = w.archived.expect("previous version archived");
        let hist = s.history("dom-a", "mem-a-1").await.unwrap();
        assert_eq!(hist.len(), 1);
        assert_eq!(hist[0].name, archived);
        assert_eq!(hist[0].text, v1.render(), "archived byte for byte");
        assert_eq!(s.get("dom-a", "mem-a-1").await.unwrap(), Some(v2));
    }

    #[tokio::test]
    async fn identical_write_changes_nothing() {
        let s = Store::in_memory().unwrap();
        let e = entry("dom-a", "mem-a-1", "Same.\n");
        s.put(&e).await.unwrap();
        let w = s.put(&e).await.unwrap();
        assert!(!w.changed);
        assert!(w.archived.is_none());
        assert!(s.history("dom-a", "mem-a-1").await.unwrap().is_empty());
    }

    #[tokio::test]
    async fn history_is_oldest_first() {
        let s = Store::in_memory().unwrap();
        for i in 1..=3 {
            s.put(&entry("dom-a", "mem-a-1", &format!("v{i}\n")))
                .await
                .unwrap();
        }
        let hist = s.history("dom-a", "mem-a-1").await.unwrap();
        let bodies: Vec<_> = hist
            .iter()
            .map(|v| Entry::parse(&v.text).unwrap().body)
            .collect();
        assert_eq!(bodies, vec!["v1\n", "v2\n"]);
        assert!(hist[0].name < hist[1].name);
    }

    #[tokio::test]
    async fn demotion_leaves_a_tombstone_with_the_body() {
        let s = Store::in_memory().unwrap();
        s.put(&entry("dom-a", "mem-a-1", "Kept.\n")).await.unwrap();
        s.demote("dom-a", "mem-a-1").await.unwrap();
        let t = s.get("dom-a", "mem-a-1").await.unwrap().unwrap();
        assert_eq!(t.front.status, Some(Status::Demoted));
        assert!(!t.is_active());
        assert_eq!(t.body, "Kept.\n");
        assert_eq!(s.list("dom-a").await.unwrap(), vec!["mem-a-1"]);
        assert!(matches!(
            s.demote("dom-a", "mem-a-nope").await,
            Err(Error::NotFound { .. })
        ));
    }

    #[tokio::test]
    async fn superseding_marks_the_old_entry() {
        let s = Store::in_memory().unwrap();
        s.put(&entry("dom-a", "mem-a-1", "Old.\n")).await.unwrap();
        let mut newer = entry("dom-a", "mem-a-2", "New.\n");
        newer.front.supersedes = Some("mem-a-1".into());
        s.put(&newer).await.unwrap();
        let old = s.get("dom-a", "mem-a-1").await.unwrap().unwrap();
        assert_eq!(old.front.status, Some(Status::Superseded));
        assert_eq!(s.history("dom-a", "mem-a-1").await.unwrap().len(), 1);
        // superseding again is idempotent: the old one is no longer active
        s.put(&newer).await.unwrap();
        assert_eq!(s.history("dom-a", "mem-a-1").await.unwrap().len(), 1);
    }

    #[tokio::test]
    async fn list_domains_and_all_for_rebuild() {
        let s = Store::in_memory().unwrap();
        s.put(&entry("dom-b", "mem-b-1", "b1\n")).await.unwrap();
        s.put(&entry("dom-a", "mem-a-2", "a2\n")).await.unwrap();
        s.put(&entry("dom-a", "mem-a-1", "a1\n")).await.unwrap();
        s.put(&entry("dom-a", "mem-a-1", "a1 again\n"))
            .await
            .unwrap();
        assert_eq!(s.domains().await.unwrap(), vec!["dom-a", "dom-b"]);
        assert_eq!(s.list("dom-a").await.unwrap(), vec!["mem-a-1", "mem-a-2"]);
        let all = s.all().await.unwrap();
        let ids: Vec<_> = all.iter().map(|e| e.front.id.as_str()).collect();
        assert_eq!(ids, vec!["mem-a-1", "mem-a-2", "mem-b-1"]);
        assert_eq!(
            all[0].body, "a1 again\n",
            "all() reads current versions only"
        );
        assert!(s.list("dom-none").await.unwrap().is_empty());
    }

    #[tokio::test]
    async fn names_that_could_leave_the_root_are_refused() {
        let s = Store::in_memory().unwrap();
        for bad in ["../x", "a/b", ".hidden", "", "a b", "x\\y", "..", "."] {
            assert!(
                matches!(s.get(bad, "mem-a-1").await, Err(Error::BadName(_))),
                "domain {bad:?}"
            );
            assert!(
                matches!(s.get("dom-a", bad).await, Err(Error::BadName(_))),
                "id {bad:?}"
            );
        }
        let mut e = entry("dom-a", "../escape", "x\n");
        assert!(matches!(s.put(&e).await, Err(Error::BadName(_))));
        e.front.id = "ok".into();
        e.front.domain = "../escape".into();
        assert!(matches!(s.put(&e).await, Err(Error::BadName(_))));
    }

    #[tokio::test]
    async fn on_disk_layout_is_what_the_docs_say() {
        let dir = tempfile::tempdir().unwrap();
        let s = Store::open_dir(dir.path()).unwrap();
        s.put(&entry("dom-a", "mem-a-1", "v1\n")).await.unwrap();
        let w = s.put(&entry("dom-a", "mem-a-1", "v2\n")).await.unwrap();
        let current = dir.path().join("record/dom-a/mem-a-1.md");
        assert!(current.is_file());
        assert!(std::fs::read_to_string(&current).unwrap().ends_with("v2\n"));
        let hist = dir
            .path()
            .join("record/dom-a/.history/mem-a-1")
            .join(format!("{}.md", w.archived.unwrap()));
        assert!(hist.is_file(), "{}", hist.display());
        assert!(std::fs::read_to_string(&hist).unwrap().ends_with("v1\n"));
        // a second store over the same directory sees the same record
        let again = Store::open_dir(dir.path()).unwrap();
        assert_eq!(again.list("dom-a").await.unwrap(), vec!["mem-a-1"]);
        assert_eq!(again.history("dom-a", "mem-a-1").await.unwrap().len(), 1);
    }

    #[tokio::test]
    async fn text_that_is_not_an_entry_is_an_error_not_a_panic() {
        let s = Store::in_memory().unwrap();
        s.op.write("record/dom-a/mem-a-1.md", b"not an entry".to_vec())
            .await
            .unwrap();
        assert!(matches!(
            s.get("dom-a", "mem-a-1").await,
            Err(Error::Entry { .. })
        ));
        s.op.write("record/dom-a/mem-a-2.md", vec![0xff, 0xfe])
            .await
            .unwrap();
        assert!(matches!(
            s.get("dom-a", "mem-a-2").await,
            Err(Error::NotText(_))
        ));
    }

    #[tokio::test]
    async fn objects_beside_the_entries_are_not_entries() {
        let dir = tempfile::tempdir().unwrap();
        let s = Store::open_dir(dir.path()).unwrap();
        s.put(&entry("dom-a", "mem-a-1", "an entry\n"))
            .await
            .unwrap();
        s.put_object("dom-a", Area::Transcripts, &["sess-1.json"], b"{}".to_vec())
            .await
            .unwrap();
        s.put_object(
            "dom-a",
            Area::Excerpts,
            &["mem-a-1", "sess-1.3.json"],
            b"{}".to_vec(),
        )
        .await
        .unwrap();
        assert!(dir
            .path()
            .join("record/dom-a/.transcripts/sess-1.json")
            .is_file());
        assert_eq!(s.list("dom-a").await.unwrap(), vec!["mem-a-1"]);
        assert_eq!(s.all().await.unwrap().len(), 1);
        assert_eq!(s.domains().await.unwrap(), vec!["dom-a"]);
        assert_eq!(
            s.objects("dom-a", Area::Transcripts, &[], false)
                .await
                .unwrap(),
            vec!["sess-1.json"]
        );
        assert_eq!(
            s.objects("dom-a", Area::Excerpts, &[], true).await.unwrap(),
            vec!["mem-a-1"]
        );
        assert_eq!(
            s.objects("dom-a", Area::Excerpts, &["mem-a-1"], false)
                .await
                .unwrap(),
            vec!["sess-1.3.json"]
        );
        assert!(s
            .objects("dom-b", Area::Excerpts, &[], true)
            .await
            .unwrap()
            .is_empty());
        s.remove_object("dom-a", Area::Excerpts, &["mem-a-1", "sess-1.3.json"])
            .await
            .unwrap();
        s.remove_object("dom-a", Area::Excerpts, &["mem-a-1", "sess-1.3.json"])
            .await
            .unwrap();
        assert!(s
            .get_object("dom-a", Area::Excerpts, &["mem-a-1", "sess-1.3.json"])
            .await
            .unwrap()
            .is_none());
        assert!(matches!(
            s.put_object("dom-a", Area::Transcripts, &["../x"], vec![])
                .await,
            Err(Error::BadName(_))
        ));
    }

    /// How many files are under `dir`, and which of them mention `word`.
    fn files_under(dir: &Path, word: &str) -> (usize, Vec<std::path::PathBuf>) {
        let (mut files, mut holding) = (0, Vec::new());
        let mut stack = vec![dir.to_path_buf()];
        while let Some(d) = stack.pop() {
            for f in std::fs::read_dir(&d).unwrap().flatten() {
                if f.path().is_dir() {
                    stack.push(f.path());
                } else {
                    files += 1;
                    if std::fs::read_to_string(f.path()).unwrap().contains(word) {
                        holding.push(f.path());
                    }
                }
            }
        }
        (files, holding)
    }

    #[tokio::test]
    async fn erasure_writes_the_stub_over_every_version_and_keeps_a_receipt() {
        use dm_core::erasure::{stub, Reason, Receipt};
        let dir = tempfile::tempdir().unwrap();
        let s = Store::open_dir(dir.path()).unwrap();
        for i in 1..=3 {
            s.put(&entry(
                "dom-a",
                "mem-a-1",
                &format!("version {i} of the secret\n"),
            ))
            .await
            .unwrap();
        }
        s.put(&entry("dom-a", "mem-a-2", "another entry, kept\n"))
            .await
            .unwrap();
        let current = s.get("dom-a", "mem-a-1").await.unwrap().unwrap();
        let before = s.digests("dom-a", "mem-a-1").await.unwrap();
        assert_eq!(before.len(), 3, "read before anything is removed");
        let removed = s
            .erase("dom-a", "mem-a-1", &stub(&current, Reason::Mistaken))
            .await
            .unwrap();
        assert_eq!(removed, before, "the current version and two archived ones");
        assert_eq!(
            removed[0].body_digest.as_deref(),
            Some(dm_core::erasure::body_digest("version 3 of the secret\n").as_str())
        );
        assert_eq!(removed[0].version, "current");
        assert_eq!(removed[0].digest, digest(&current.render()));
        let after = s.get("dom-a", "mem-a-1").await.unwrap().unwrap();
        assert_eq!(after.front.status, Some(Status::Erased));
        assert_eq!(after.body, "");
        for v in s.history("dom-a", "mem-a-1").await.unwrap() {
            assert_eq!(Entry::parse(&v.text).unwrap().body, "", "{}", v.name);
        }
        let receipt = Receipt {
            id: "mem-a-1".into(),
            domain: "dom-a".into(),
            erased_at: OffsetDateTime::UNIX_EPOCH,
            erased_by: "human:test:me".into(),
            reason: Reason::Mistaken,
            removed,
            flagged: vec![],
            unresolved: vec![],
            promotion_whys_removed: 0,
            promotion_tasks_cleared: 0,
        };
        s.put_receipt(&receipt).await.unwrap();
        assert!(receipt.to_yaml().contains("body_digest:"));
        assert_eq!(
            s.receipt("dom-a", "mem-a-1").await.unwrap().as_ref(),
            Some(&receipt)
        );
        assert_eq!(s.receipt("dom-a", "mem-a-2").await.unwrap(), None);
        assert_eq!(s.receipts("dom-a").await.unwrap(), vec![receipt]);
        assert!(s.receipts("dom-b").await.unwrap().is_empty());
        // the word is in no file under the root, and the other entry is untouched
        let (files, holding) = files_under(dir.path(), "secret");
        assert!(holding.is_empty(), "{holding:?}");
        assert_eq!(
            files, 5,
            "current, two archived, the receipt, the other entry"
        );
        assert_eq!(s.list("dom-a").await.unwrap(), vec!["mem-a-1", "mem-a-2"]);
        let missing = stub(&entry("dom-a", "mem-a-9", "x\n"), Reason::Legal);
        assert!(matches!(
            s.erase("dom-a", "mem-a-9", &missing).await,
            Err(Error::NotFound { .. })
        ));
    }
}