blockworx_store/

handle.rs

//! The store: a container and the [`Repo`] it holds, joined so that every
//! accepted commit reaches the files before the call returns.
//!
//! This is the only door. A caller can read the repo but never write it,
//! so "folded but not written" — the state a crash would silently keep —
//! is not something an outside caller can produce.

use std::path::Path;
use std::time::Duration;

use blockworx_doc::{
    commit::Commit,
    document::Document,
    id::EntityRef,
    repo::Repo,
    rev::Rev,
    trail::{Direction, Trail},
};

use super::Refusal;
use super::container::{Access, Container, ContainerError, Drained, ReadOnlyReason, WriteRefusal};
use super::manifest::{self, End, Fault, Row, RowKind, Tail};
use super::record::{Attribution, Digest, Identity, WallTime};
use super::revs;
use super::storage::{Name, Storage, ready_now};
use super::tags::{Tagging, Tags};

/// Where a row's wall time comes from. The system clock is the only
/// production value; the pinned arm is the seam a byte-stable golden and
/// a repeatable test need, and it advances so a fixture's rows are
/// ordered in time without depending on one.
#[derive(Clone, Copy, Debug)]
pub enum Clock {
    System,
    Pinned { at: WallTime, step: Duration },
}

impl Clock {
    pub(super) fn tick(&mut self) -> WallTime {
        match self {
            Clock::System => super::history::now(),
            Clock::Pinned { at, step } => {
                let now = *at;
                *at = WallTime::from_unix_millis(now.unix_millis() + step.as_millis() as u64);
                now
            }
        }
    }
}

/// Why a container could not be seeded from a session's commits — the
/// Save-As-container path, which lays out a container and then replays a
/// scratch session's commits into it through the ordinary write door.
#[derive(Debug, thiserror::Error)]
pub enum SeedFailure {
    #[error(transparent)]
    Create(#[from] ContainerError),
    #[error("seeding stopped at commit {at}: {why}")]
    Append { at: usize, why: Refusal },
}

/// What a history step is, before the repo has taken it.
enum Step {
    Edit(Commit),
    /// A commit that is already the document's past rather than a step this
    /// session took — the seeding path. Written like an edit, trailed like
    /// nothing: no one may take back a step they did not make.
    Seed(Commit),
    Undo(Rev),
    Redo(Rev),
}

pub struct Store<S: Storage> {
    container: Container<S>,
    repo: Repo,
    /// Where this session's history stands. Beside the repo rather than
    /// in it: a step adopts the rev copy this container holds, which is
    /// something only the store can read (`docs/log-vs-snapshot.md`).
    trail: Trail,
    clock: Clock,
    /// The digest the next row must name as its parent.
    head: Digest,
    /// Every row this container's manifest holds that spends a rev, in
    /// rev order: `rows()[i]` describes rev `i + 1`. What the history
    /// panel, the rev pick and `blockworx log` all read.
    rows: Vec<Row>,
    /// What the manifest's tag rows name its revs. Rebuilt at every open
    /// and never handed to the repo.
    tags: Tags,
    tail: Tail,
}

impl<S: Storage> Store<S> {
    /// Lay out a new container and open it at the empty document.
    ///
    /// # Errors
    /// As [`Container::create`].
    pub fn create(storage: S, clock: Clock) -> Result<Self, ContainerError> {
        ready_now(Self::creating(storage, clock))
    }

    /// [`Self::create`] for a storage whose futures are not ready when they
    /// are made.
    ///
    /// # Errors
    /// As [`Self::create`].
    pub async fn creating(storage: S, mut clock: Clock) -> Result<Self, ContainerError> {
        let container = Container::creating(storage, clock.tick()).await?;
        Ok(Self {
            container,
            repo: Repo::default(),
            trail: Trail::default(),
            clock,
            head: Digest::genesis(),
            rows: Vec::new(),
            tags: Tags::default(),
            tail: Tail::Whole,
        })
    }

    /// Lay out a new container and put `commits` into it, each as one row
    /// and one rev of its own.
    ///
    /// This is how a *scratch* session becomes durable: each commit lands
    /// as its own rev, so the container's history *is* the session's
    /// history. Nothing is trailed — the commits are the new container's
    /// past.
    ///
    /// A session that already has a container does **not** come this way:
    /// commits are all this can carry, so the wall times, the authors, the
    /// edit/undo/redo kinds and the tags would be flattened into a run of
    /// fresh edits by whoever pressed Save-as. That path copies the
    /// manifest's own rows instead ([`super::prefix::save_through`]).
    ///
    /// # Errors
    /// [`SeedFailure::Create`] as [`Self::create`], and
    /// [`SeedFailure::Append`] naming the commit that did not land — which
    /// leaves a container holding the prefix that did.
    pub fn seeded(
        storage: S,
        clock: Clock,
        commits: &[Commit],
        author: &Identity,
    ) -> Result<Self, SeedFailure> {
        Self::create(storage, clock)?.seeded_with(commits, author)
    }

    /// [`Self::seeded`] for a storage whose futures are not ready when they
    /// are made. The commits themselves are folded and written
    /// synchronously either way; what a resident container owes its storage
    /// afterwards is [`Self::drain`]'s.
    ///
    /// # Errors
    /// As [`Self::seeded`].
    pub async fn seeding(
        storage: S,
        clock: Clock,
        commits: &[Commit],
        author: &Identity,
    ) -> Result<Self, SeedFailure> {
        Self::creating(storage, clock)
            .await?
            .seeded_with(commits, author)
    }

    fn seeded_with(mut self, commits: &[Commit], author: &Identity) -> Result<Self, SeedFailure> {
        for (at, commit) in commits.iter().enumerate() {
            self.write(Step::Seed(commit.clone()), author.into())
                .map_err(|why| SeedFailure::Append { at, why })?;
        }
        Ok(self)
    }

    /// Open a container: read the manifest, verify its chain, and read the
    /// head rev.
    ///
    /// A manifest that does not verify is not a failure to open: the
    /// container opens read-only at the last good prefix, with the break as
    /// its [`ReadOnlyReason`], because the reader is owed a look at the
    /// past that *is* intact. A manifest with a partial trailing row opens
    /// writable, the row dropped and the file cut back to the last whole
    /// line.
    ///
    /// # Errors
    /// As [`Container::open`] — there is no container, or it cannot be
    /// read.
    pub fn open(storage: S, clock: Clock) -> Result<Self, ContainerError> {
        ready_now(Self::opening(storage, clock))
    }

    /// [`Self::open`] for a storage whose futures are not ready when they
    /// are made.
    ///
    /// The open is where a resident container is read in — the manifest,
    /// every rev and every payload — so everything after it, this
    /// function's own second half included, reads out of memory.
    ///
    /// # Errors
    /// As [`Self::open`].
    pub async fn opening(storage: S, mut clock: Clock) -> Result<Self, ContainerError> {
        Self::over(Container::opening(storage, clock.tick()).await?, clock)
    }

    /// Open a container to be read and never written.
    ///
    /// The same read as [`Self::open`] over a handle that never claimed
    /// the lock ([`Container::reading`]), so reading someone's document —
    /// or a fixture out of `fixtures/` — cannot cost its owner the right
    /// to write it. Every write door refuses with [`Refusal::ReadOnly`],
    /// which is why this takes no [`Clock`]: nothing here mints a row.
    ///
    /// # Errors
    /// As [`Self::open`].
    pub fn reading(storage: S) -> Result<Self, ContainerError> {
        ready_now(Self::opening_to_read(storage))
    }

    /// [`Self::reading`] for a storage whose futures are not ready when
    /// they are made.
    ///
    /// # Errors
    /// As [`Self::reading`].
    pub async fn opening_to_read(storage: S) -> Result<Self, ContainerError> {
        Self::over(Container::opening_to_read(storage).await?, Clock::System)
    }

    /// Read an already-attached container — [`Self::open`]'s second half,
    /// shared with the prefix Save-as, which lays a container out with a
    /// manifest already in it.
    ///
    /// # Errors
    /// The manifest that could not be read.
    pub(super) fn over(mut container: Container<S>, clock: Clock) -> Result<Self, ContainerError> {
        let text = container.read_manifest()?;
        let scanned = manifest::scan(&text);
        // The head is the document this session opens on, so its bytes are
        // checked here — one blake3 over one file. Every *other* rev is
        // `blockworx verify`'s business.
        let (rows, broken) = whole(&container, scanned.rows);
        let history = manifest::history(&rows);
        let head = match container.read_rev(history.rev()) {
            Ok(document) => document,
            Err(fault) => {
                // The prefix `whole` handed back ends at a rev whose bytes
                // hash to what its row says, so this is artwork the
                // container cannot show rather than a rev that is not
                // itself. Either way the reader is owed the empty document
                // over a crash.
                tracing::error!("{fault}");
                container.demote(ReadOnlyReason::WriteFailed(std::io::Error::other(
                    fault.to_string(),
                )));
                Document::default()
            }
        };
        let mut tail = Tail::Whole;
        match (broken, scanned.end) {
            (Some(report), _) | (None, End::Broken(report)) => {
                container.demote(ReadOnlyReason::HistoryBroken(report));
            }
            (None, End::Truncated(dropped)) => {
                tail = Tail::Dropped(dropped);
                if let Err(WriteRefusal::Io(error)) =
                    container.truncate_manifest(dropped.good_bytes)
                {
                    container.demote(ReadOnlyReason::WriteFailed(error));
                }
            }
            (None, End::Whole) => {}
        }
        Ok(Self {
            container,
            repo: Repo::at(head),
            trail: history.trail,
            clock,
            head: history.head,
            rows: history.rows,
            tags: history.tags,
            tail,
        })
    }

    /// What this document is called.
    pub fn name(&self) -> Name {
        self.container.name()
    }

    /// Where this container sits on a filesystem, for a shell that can
    /// name one. `None` for a container that is not on one.
    pub fn path(&self) -> Option<&Path> {
        self.container.storage().disk_path()
    }

    /// Rename this store's container, keeping the lock and the manifest
    /// with it. The document is untouched: only what it is called changes,
    /// so nothing is appended and no rev is spent.
    ///
    /// # Errors
    /// [`Refusal::ReadOnly`] on a container this session may not write —
    /// renaming someone else's open document is not a read — and
    /// [`Refusal::Rename`] when the container does not move, which leaves
    /// the store where it was.
    pub fn rename(&mut self, to: &Name) -> Result<(), Refusal> {
        if self.read_only_reason().is_some() {
            return Err(Refusal::ReadOnly);
        }
        self.container.rename(to).map_err(Refusal::Rename)
    }

    /// [`Self::rename`] for a storage whose futures are not ready when they
    /// are made.
    ///
    /// # Errors
    /// As [`Self::rename`].
    pub async fn renaming_to(&mut self, to: &Name) -> Result<(), Refusal> {
        if self.read_only_reason().is_some() {
            return Err(Refusal::ReadOnly);
        }
        self.container
            .renaming_to(to)
            .await
            .map_err(Refusal::Rename)
    }

    /// How many writes this session has made to its container's memory and
    /// not yet to its storage — what a status line reads "writing…" off.
    /// Always zero where the writes are made as they are asked for.
    pub fn pending(&self) -> usize {
        self.container.pending()
    }

    /// Make the writes this session owes its storage, in the order it made
    /// them. See [`Container::drain`]: a write that does not land costs the
    /// container its lock, as a failed append does.
    ///
    /// # Errors
    /// The write that did not land.
    pub async fn drain(&mut self) -> std::io::Result<Drained> {
        self.container.drain().await
    }

    /// Readable, never writable: [`Self::submit_edit`] and its two
    /// neighbours are the only way a commit gets into this repo, which is
    /// what keeps the repo and the files in step.
    pub fn repo(&self) -> &Repo {
        &self.repo
    }

    /// Where this session's history stands: what one press would take
    /// back, what one would put back, and which rev's document the head
    /// holds.
    pub fn trail(&self) -> &Trail {
        &self.trail
    }

    pub fn document(&self) -> &Document {
        self.repo.document()
    }

    pub fn access(&self) -> &Access {
        self.container.access()
    }

    pub fn read_only_reason(&self) -> Option<&ReadOnlyReason> {
        match self.container.access() {
            Access::ReadOnly(reason) => Some(reason),
            Access::Writable => None,
        }
    }

    /// Whether load dropped an unfinished row, and which.
    pub fn tail(&self) -> Tail {
        self.tail
    }

    /// The manifest's rows, one per rev, in rev order.
    pub fn rows(&self) -> &[Row] {
        &self.rows
    }

    /// The row rev `at` was written under, or `None` for a rev this
    /// history does not hold.
    pub fn row(&self, at: Rev) -> Option<&Row> {
        let ndx = usize::try_from(at.get().checked_sub(1)?).ok()?;
        self.rows.get(ndx)
    }

    /// The row a reader frames rev `at` from: its own, or — for a step —
    /// the row of the act it moves.
    ///
    /// A step frames what it *moved*, so the scope it opens and the view
    /// it replays are the ones the act was made in, not where the hand
    /// that pressed undo happened to be standing. The row itself still
    /// records the undoer's own circumstances; the derivation happens at
    /// read time.
    pub fn framing(&self, at: Rev) -> Option<&Row> {
        let mut row = self.row(at)?;
        // Bounded by the rows behind it: every `of` names an earlier rev,
        // which the manifest checked when it was read.
        for _ in 0..self.rows.len() {
            let Some(of) = row.kind.steps() else {
                return Some(row);
            };
            row = self.row(of)?;
        }
        Some(row)
    }

    /// What this manifest's tag rows call its revs.
    pub fn tags(&self) -> &Tags {
        &self.tags
    }

    /// The document this container holds at `at`, payloads and all.
    ///
    /// # Errors
    /// [`Refusal::Unreachable`] for a rev whose copy the container cannot
    /// show.
    pub fn document_at(&self, at: Rev) -> Result<Document, Refusal> {
        self.container
            .read_rev(at)
            .map_err(|why| Refusal::Unreachable {
                at,
                why: why.to_string(),
            })
    }

    /// What the head row stamps. Read off the row rather than recomputed:
    /// the row was written after the bytes landed, and the open checked
    /// them against each other.
    fn head_hash(&self) -> Digest {
        self.rows
            .last()
            .map_or_else(|| Digest::of(&[]), |row| row.hash)
    }

    /// Fold `commit`, write its rev, and append its row. See [`Store`]'s
    /// note on the write path: a write that fails costs the container its
    /// lock.
    ///
    /// # Errors
    /// [`Refusal::ReadOnly`] when this handle may not write,
    /// [`Refusal::Fold`] when the fold refuses the commit, and
    /// [`Refusal::Append`] when the row does not reach the file.
    pub fn submit_edit<'a>(
        &mut self,
        commit: Commit,
        by: impl Into<Attribution<'a>>,
    ) -> Result<Rev, Refusal> {
        self.write(Step::Edit(commit), by.into())
    }

    /// Adopt the document the trail's top entry restores, recorded as an
    /// undo of `edit`.
    ///
    /// # Errors
    /// As [`Self::submit_edit`], with [`Refusal::Step`] in place of
    /// [`Refusal::Fold`] when the trail will not take the step, and
    /// [`Refusal::Unreachable`] for a rev this container cannot read back.
    pub fn undo<'a>(&mut self, edit: Rev, by: impl Into<Attribution<'a>>) -> Result<Rev, Refusal> {
        self.write(Step::Undo(edit), by.into())
    }

    /// # Errors
    /// As [`Self::undo`].
    pub fn redo<'a>(&mut self, edit: Rev, by: impl Into<Attribution<'a>>) -> Result<Rev, Refusal> {
        self.write(Step::Redo(edit), by.into())
    }

    /// Put a name on `at`, or take one off.
    ///
    /// One appended row, chained like any other, that no rev is spent on
    /// and the trail never hears about: tagging is not an edit, and undo
    /// after it still takes back the last one.
    ///
    /// # Errors
    /// [`Refusal::ReadOnly`] when this handle may not write,
    /// [`Refusal::NoSuchRev`] for a rev this history does not hold, and
    /// [`Refusal::Append`] when the row does not reach the file.
    pub fn tag<'a>(
        &mut self,
        at: Rev,
        name: &str,
        how: Tagging,
        by: impl Into<Attribution<'a>>,
    ) -> Result<(), Refusal> {
        if self.read_only_reason().is_some() {
            return Err(Refusal::ReadOnly);
        }
        if at == Rev::ZERO || at > self.repo.rev() {
            return Err(Refusal::NoSuchRev(at));
        }
        let by = by.into();
        let wall_time = self.clock.tick();
        // A tag has no rev file of its own, so it stamps the head it was
        // appended under — which is what makes a tag written against
        // another history detectable.
        let row = self.row_for(Written {
            rev: at,
            kind: match how {
                Tagging::Added => RowKind::Tag,
                Tagging::Removed => RowKind::Untag,
            },
            label: name.to_owned(),
            touched: Vec::new(),
            hash: self.head_hash(),
            wall_time,
            by,
        });
        match self.container.append_row(&row) {
            Ok(()) => {
                self.head = row.digest();
                self.tags.apply(at, name, how);
                Ok(())
            }
            Err(WriteRefusal::ReadOnly) => Err(Refusal::ReadOnly),
            Err(WriteRefusal::Io(error)) => Err(self.parted_from(error)),
        }
    }

    /// Take the step, write the rev, then append the row. A write that
    /// fails leaves this session's document ahead of the files, so the
    /// container gives up its lock rather than carrying on writing into a
    /// history with a hole in it.
    ///
    /// # Errors
    /// [`Refusal::ReadOnly`] when this handle may not write,
    /// [`Refusal::Fold`] or [`Refusal::Step`] when the repo refuses the
    /// step, and [`Refusal::Append`] when the row does not reach the file.
    fn write(&mut self, step: Step, by: Attribution<'_>) -> Result<Rev, Refusal> {
        if self.read_only_reason().is_some() {
            return Err(Refusal::ReadOnly);
        }
        let (rev, kind, label, touched) = match step {
            Step::Edit(commit) => {
                let (label, touched) = (commit.label().to_owned(), named(&commit));
                let _s = tracing::info_span!("fold", ops = commit.ops().len()).entered();
                let rev = self.repo.submit(commit, &mut self.trail)?;
                (rev, RowKind::Edit, label, touched)
            }
            Step::Seed(commit) => {
                let (label, touched) = (commit.label().to_owned(), named(&commit));
                let rev = self.repo.fold_one(commit)?.rev();
                self.trail.seeded(rev);
                (rev, RowKind::Edit, label, touched)
            }
            Step::Undo(edit) => self.step(edit, Direction::Undo)?,
            Step::Redo(edit) => self.step(edit, Direction::Redo)?,
        };
        let wall_time = self.clock.tick();
        // The rev's own copy of the document — and the payloads it names —
        // land before the row that names them, so the manifest never names
        // bytes a crash could still take away.
        let _write = tracing::info_span!("write_rev", rev = rev.get()).entered();
        let hash = match self.container.write_rev(rev, self.repo.document()) {
            Ok(hash) => hash,
            Err(WriteRefusal::ReadOnly) => return Err(Refusal::ReadOnly),
            Err(WriteRefusal::Io(error)) => return Err(self.parted_from(error)),
        };
        let row = self.row_for(Written {
            rev,
            kind,
            label,
            touched,
            hash,
            wall_time,
            by,
        });
        match self.container.append_row(&row) {
            Ok(()) => {
                self.head = row.digest();
                self.rows.push(row);
                Ok(rev)
            }
            Err(WriteRefusal::ReadOnly) => Err(Refusal::ReadOnly),
            Err(WriteRefusal::Io(error)) => Err(self.parted_from(error)),
        }
    }

    /// Take one history step, adopting the rev copy this container holds,
    /// and carry the stepped row's own `touched` names onto the row this
    /// step writes — so a step frames what it moved.
    fn step(
        &mut self,
        edit: Rev,
        direction: Direction,
    ) -> Result<(Rev, RowKind, String, Vec<EntityRef>), Refusal> {
        // The trail is asked first, so a spent step is reported as one
        // rather than as a rev this history does not hold.
        let entry = self.trail.stepping(edit, direction)?;
        let stepped = self.row(entry.rev).ok_or(Refusal::NoSuchRev(edit))?;
        let (label, touched) = (stepped.label.clone(), stepped.touched.clone());
        let rev = crate::doc::stepped(
            &mut self.repo,
            &mut self.trail,
            crate::doc::Stepping {
                edit,
                direction,
                label: &label,
            },
            |at| {
                self.container
                    .read_rev(at)
                    .map_err(|why| Refusal::Unreachable {
                        at,
                        why: why.to_string(),
                    })
            },
        )?;
        let kind = match direction {
            Direction::Undo => RowKind::Undo { of: edit },
            Direction::Redo => RowKind::Redo { of: edit },
        };
        Ok((rev, kind, format!("{} {label}", direction.verb()), touched))
    }

    fn row_for(&mut self, written: Written<'_>) -> Row {
        let (touched, truncated) = Row::naming(written.touched);
        Row {
            rev: written.rev,
            kind: written.kind,
            wall_time: written.wall_time,
            author: written.by.author.clone(),
            label: written.label,
            scope: written.by.standing.path().clone(),
            scope_names: written.by.standing.names().to_vec(),
            camera: written.by.camera,
            touched,
            truncated,
            hash: written.hash,
            parent: self.head,
        }
    }

    /// This session's document is now ahead of the files, so the container
    /// gives up its lock rather than carrying on writing into a history
    /// with a hole in it.
    fn parted_from(&mut self, error: std::io::Error) -> Refusal {
        let echo = std::io::Error::other(error.to_string());
        self.container.demote(ReadOnlyReason::WriteFailed(error));
        Refusal::Append(echo)
    }
}

/// Everything a row is minted from beyond the chain it links into.
struct Written<'a> {
    rev: Rev,
    kind: RowKind,
    label: String,
    touched: Vec<EntityRef>,
    hash: Digest,
    wall_time: WallTime,
    by: Attribution<'a>,
}

/// The entities a commit named, spelled as the manifest records them.
fn named(commit: &Commit) -> Vec<EntityRef> {
    let mut touched: Vec<EntityRef> = Vec::new();
    for target in commit
        .ops()
        .iter()
        .map(blockworx_doc::opcode::OpCodes::target)
    {
        if !touched.contains(&target) {
            touched.push(target);
        }
    }
    touched
}

/// The longest prefix of `rows` whose last rev the container can show,
/// with the break that cut it short.
///
/// Only the *head* is checked at open: every other rev file is
/// `blockworx verify`'s business, and a hole in one of them is a finding
/// rather than a reason to refuse the document.
fn whole<S: Storage>(
    container: &Container<S>,
    mut rows: Vec<manifest::Verified>,
) -> (Vec<manifest::Verified>, Option<manifest::BreakReport>) {
    let backing = container.revs();
    let mut broken = None;
    // A tag spends no rev and stamps the head it was appended under, so
    // the row that names the *document* is the last one that journals.
    while let Some(head) = rows
        .iter()
        .rev()
        .find(|verified| verified.row.kind.takes_a_rev())
    {
        let (at, named) = (head.row.rev, head.row.hash);
        let Err(fault) = revs::witnessed(&backing, at, named) else {
            break;
        };
        broken.get_or_insert(manifest::BreakReport {
            at: head.at,
            fault: Fault::Unwitnessed {
                at,
                why: fault.to_string(),
            },
        });
        rows.retain(|verified| verified.row.rev < at);
    }
    (rows, broken)
}