coven_protocol/

blob.rs

//! The blob engine: coven's single owner of a blob's whole durability lifecycle.
//!
//! coven syncs opaque encrypted blobs referenced by DB rows. By default it owns
//! the cloud layout (the content-addressed `{namespace}/{ab}/{cd}/{id}`) and
//! encryption; the host decides which rows carry blobs, where their plaintext
//! lives locally, and how each is scoped for encryption. A home configured for
//! the unobfuscated blob-path scheme instead stores each blob at the consumer's
//! readable [`BlobRef::cloud_path`] so the bucket is browsable.
//!
//! # The coven concept tree
//!
//! A blob has two **declared** properties — [`Provenance`] (its Local story) and
//! [`CacheFill`] (its Remote story) — and one **state**, locality, flipped by the
//! transitions. The cache is a *mechanism* that serves Remote blobs; it is not a
//! kind of blob.
//!
//! ```text
//! A blob the host declares with:
//!
//! provenance — its LOCAL story: where the bytes live when Local, and the
//!              Remote→Local path requirement
//!    ├─ user-provided   the user's file at a path; coven references it.
//!    │                  Remote→Local writes the bytes back to a user file → NEEDS A PATH.
//!    └─ host-provided   bae hands coven the data; coven keeps it in its local store.
//!                       Remote→Local restores it to the local store → no path.
//!
//! cache fill — its REMOTE story: how a device gets the bytes when the release is
//!              Remote. A cache-mechanism setting; applies to ANY blob, regardless of
//!              provenance, once it is Remote.
//!    ├─ CacheEager   fetched into the cache on pull, with the SQL row   (covers)
//!    └─ CacheLazy    fetched into the cache on first read               (audio — big, fetch what you play)
//!
//! and a current state:
//!
//! locality
//!    ├─ Local    bytes on-device — the user's path (user-provided) or coven's local store (host-provided)
//!    └─ Remote   bytes in the cloud; each device's local copy is a CACHE copy, filled
//!                per `cache fill`, kept-or-evicted per `pin`
//!
//! namespace (bucket)   the blob's category — release_files · covers · artist_images
//!
//! transitions
//!    ├─ Local → Remote   upload the bytes; now cache-distributed to every device per cache fill
//!    └─ Remote → Local   bring the bytes back to a local file — path required iff user-provided
//!
//! cache budget   per-NAMESPACE size limit; each namespace evicts independently, so
//!                evicting release_files (big) never touches covers (small reserved slice)
//! pin            keep one specific Remote blob's cache copy from eviction (e.g. a
//!                release the user pinned for offline)
//! ```
//!
//! ## The cache vs local files
//!
//! The cache holds local copies of **Remote** blobs (filled per `cache fill`,
//! evicted per budget unless pinned). It is **segmented by namespace**: each
//! namespace has its own configurable cache budget and evicts independently, so
//! evicting `release_files` (big) never touches `covers` (a small reserved slice). A
//! `CacheEager` cover that falls out of its namespace budget shows a placeholder
//! until the next read re-fetches it — covers are not pinned. A **Local** blob is not
//! in the cache: a user-provided Local blob is the user's file at its path (an
//! external ref); a host-provided Local blob is in coven's local store, whose
//! paths and file operations are owned by [`coven_foundation::store_dir::StoreDir`]. The cache is the mechanism for *remoteness* — so
//! `CacheEager`/`CacheLazy`/pin/budget describe a blob only while it is Remote, never
//! while it is Local.
//!
//! # The engine's halves
//!
//! This module is the engine; its halves move a blob through its lifecycle:
//!
//! - `blob::cache` — the device-local cache for **Remote** blobs: bytes on disk keyed
//!   by exact locator hash, with the folder a file lives in as the only retention truth
//!   (`storage/pinned/` protected, `storage/cache/` evictable). Reads — one-shot
//!   whole, which checks the plaintext against the row's hash because it reads
//!   every byte anyway, or an opened stream whose ranges each cost their own
//!   bytes: a positioned read of a local file, or the sealed chunks covering the
//!   range fetched from the cloud object and opened — plus pin/unpin, clear, and
//!   budget eviction.
//! - [`coven_foundation::store_dir::StoreDir`] — coven's own copy of a **host-provided Local**
//!   blob, in `storage/local/<namespace>/<id>`. Never evicted; the budget sweep
//!   never walks it.
//! - `blob::upload` — the cloud-write half: drain the durable upload queue, sealing
//!   each blob under its scope and writing it to the cloud with coalesced progress,
//!   so a local-only blob becomes uploaded. The sync cycle calls the drain
//!   each round before it pushes.
//! - `blob::delete` — the cloud-delete half: turn a queued deletion into a signed
//!   cloud tombstone, hold the blob for a convergence grace so a lagging peer
//!   isn't stranded, then GC the blob once the grace has passed. The sync cycle
//!   drains tombstones and runs the GC each round after it pulls.
//!
//! The types below ([`BlobRef`], [`BlobScope`], [`Provenance`],
//! [`CacheFill`], [`BlobTransitionObserver`]) are the vocabulary both halves and
//! the host speak. Which rows carry blobs is not a runtime callback but a per-table
//! declaration ([`crate::synced_schema::BlobDecl`]) coven resolves into a
//! the database's `BlobDecls` each cycle to derive the blob set itself.
//!
//! coven also owns the two locality transitions (`blob::transition`): `make_remote`
//! (Local → Remote: upload the bytes, then flip the gate) and `make_local`
//! (Remote → Local: bring each blob back to a local file, then retract). The
//! The upload drain advances the durable make-Remote intent after every exact
//! object lands. The Store publication activates the resulting gate change;
//! hosts observe both handoffs through the durable cloud-outbox query.

pub mod locator;

#[cfg(test)]
mod row_ref_tests;

use sha2::{Digest, Sha256};

/// The content hash a blob-bearing row carries: the lowercase-hex SHA-256 of the
/// blob's plaintext bytes, computed at import and stored in the row's blob columns
/// alongside the declared size. The row is carried in a signed changeset (and in a
/// signed snapshot), so this hash is signed by the row's author — that is what
/// makes it authoritative: on download coven hashes the decrypted plaintext and
/// requires equality with the row's hash, so the bytes are pinned by the author,
/// not by the cloud key they happened to arrive under. A host computes this over a
/// blob's plaintext at import and writes it into the row's declared hash column,
/// the same way it writes the plaintext length into the size column.
pub fn content_hash(plaintext: &[u8]) -> String {
    hex::encode(Sha256::digest(plaintext))
}

/// An incremental SHA-256 over a blob's plaintext, so the streaming download path
/// verifies a blob's content hash without holding the whole plaintext in memory:
/// feed each decrypted chunk to [`update`](Self::update), call
/// [`finish`](Self::finish), and compare the returned digest with the row's hash
/// before committing the bytes to the cache. The hex-encoded digest matches
/// [`content_hash`] over the same bytes.
pub struct ContentHasher(Sha256);

impl ContentHasher {
    pub fn new() -> Self {
        ContentHasher(Sha256::new())
    }

    /// Fold the next plaintext chunk into the running digest.
    pub fn update(&mut self, chunk: &[u8]) {
        self.0.update(chunk);
    }

    /// The lowercase-hex digest of everything fed so far.
    pub fn finish(self) -> String {
        hex::encode(self.0.finalize())
    }
}

impl Default for ContentHasher {
    fn default() -> Self {
        Self::new()
    }
}

/// How many blob transfers coven runs at once in each of its two transfer loops:
/// the upload drain and the pin/download loop. An
/// open-time blob-engine tunable the host sets on the builder,
/// carried on the `Database` alongside the other open-time
/// blob config and read back by each loop, which holds `&Database`.
///
/// Each bound is a [`NonZeroUsize`], so a zero — which would leave a loop admitting
/// nothing and never completing — is unrepresentable rather than clamped or rejected
/// at open. `one_at_a_time()` (both `1`) is the default: transfers run one at a
/// time in queue order.
///
/// [`NonZeroUsize`]: std::num::NonZeroUsize
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TransferLimits {
    /// Maximum concurrent blob uploads in one upload-drain pass.
    pub uploads: std::num::NonZeroUsize,
    /// Maximum concurrent blob downloads (fetches) in one pin call.
    pub downloads: std::num::NonZeroUsize,
}

impl TransferLimits {
    /// One at a time in each loop.
    pub fn one_at_a_time() -> Self {
        Self {
            uploads: std::num::NonZeroUsize::MIN,
            downloads: std::num::NonZeroUsize::MIN,
        }
    }
}

// The cache's own tests: real `Database` + `TestStore` over a temp store
// dir, asserting hits/misses, the pinned/cache folder split, and pin/unpin/clear.
// These drive a real temp directory on the filesystem. See `blob::cache`.
// The upload drain's tests: real `Database` (the `cloud_outbox` queue) driven
// against `InMemoryCloudHome`/`FailingCloudHome`, asserting record-and-continue,
// per-entry backoff, scope-resolved sealing, and the observer callbacks. See
// `blob::upload`.
// The coven-owned make-Remote / make-Local transition tests: multi-device
// make_remote + make_local through the real cycle, cancel both directions, the
// drain's completion flip, durable cancellation, crash-idempotency at each commit
// boundary, and a round-trip. Uses a `watch` cancel signal and retained test devices,
// See `blob::transition`.
// The local-files store's tests: store/read round-trip, a host-provided Local blob
// surviving a budget sweep (the sweep never walks `local/`), and drop. These
// drive a real temp directory through `StoreDir`.
// The delete half's tests: tombstone signing, the drain that writes tombstones,
// the graced GC that reclaims exact immutable objects, and the delete-outbox row
// shape. Driven against `InMemoryCloudHome` and
// `TestStore`. See `blob::delete`.

/// Which key encrypts a blob, as a host names it on a [`BlobRef`].
///
/// The host names *what* a blob is scoped to — the whole store or a derived
/// per-scope key — never the raw key bytes. Storage and encryption consume this
/// same type; there is no key material in it to leak.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum BlobScope {
    /// The store master key — every member reads it.
    Master,
    /// A per-scope key derived from the master key (e.g. one key per item).
    Derived(String),
}

/// A blob's **Local story**: where its bytes live while the blob is Local, and
/// whether bringing it back from Remote needs a destination path. Orthogonal to
/// [`CacheFill`] (the Remote story) — a blob declares both.
///
/// The cache never enters into this: a Local blob is not a cache copy. Provenance
/// decides which of the two Local homes holds it, and what `make_local` does to
/// restore it.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum Provenance {
    /// The user's own file at a path; coven references it but does not own it
    /// (tracked as an external ref — see `local_blob_refs`). `make_local` writes
    /// the bytes back to a user file, so it **needs a destination path**.
    UserProvided,
    /// The host hands coven the data; coven keeps its own copy in the local store
    /// (`storage/local/<namespace>/<id>`, owned by [`coven_foundation::store_dir::StoreDir`]). `make_local`
    /// restores it to the local store, so it needs **no path**.
    HostProvided,
}

/// A blob's **Remote story**: how a device gets the bytes once the blob is Remote.
/// A cache-mechanism setting — it describes a blob only while Remote — that applies
/// to ANY blob regardless of [`Provenance`]. Orthogonal to provenance; a blob
/// declares both.
///
/// Both classes are declared per blob and are global (every device reads the same
/// class from the blob's [`BlobRef`]); the difference is what a device does with
/// the blob on pull. The distinction has to be a declared property and not a
/// per-device choice: device B, deciding during its own pull whether to fetch a
/// blob, can only read the blob's declared class — it cannot see what device A
/// chose locally.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum CacheFill {
    /// Fetched into the cache on pull, right away, on every device — part of
    /// "having the store" (e.g. cover art, so the grid renders from local bytes
    /// without a fetch). The cache copy is evictable + re-fetchable, not pinned.
    CacheEager,
    /// Not fetched on pull: a pulling device skips it and fetches it into the cache
    /// on first read — e.g. audio, which is big and streams on demand.
    CacheLazy,
}

/// A blob's **replacement story**: whether the row carrying it may ever be repointed at
/// a different blob. Orthogonal to [`Provenance`] and [`CacheFill`]; a blob declares all
/// three.
///
/// It exists because a cloud object must never be rewritten with different bytes. The
/// pull verifies an object against its row's content hash and a position advances only over
/// a fully-realized changeset, so a key whose content can change leaves a device that
/// pulls an older changeset unable to satisfy it — wedged there for good, not merely
/// missing a blob. Two declarations reach that guarantee by different routes, and coven
/// enforces whichever one the blob declares:
///
/// - [`Replaceable`](Self::Replaceable) — the row may be repointed, so the *key* must
///   move with the blob: a readable `cloud_path` has to name its blob
///   (`cloud_path_names_blob`), and a replacement then writes a new
///   object beside the one it replaces instead of over it.
/// - [`WriteOnce`](Self::WriteOnce) — the row is never repointed, so the object at its
///   key is written once and there is nothing to protect it from. Its path is free to be
///   a stable, fully readable name. coven refuses the repointing.
///
/// `Replaceable` is the default: its guarantee is the airtight one (the key itself
/// carries the blob id, so no path can ever be reused), while `WriteOnce` is a weaker
/// contract a consumer opts into knowingly — see its docs.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BlobReplacement {
    /// The row may be repointed at a new blob id — replacing a cover, swapping an
    /// attachment. Requires a readable `cloud_path` that names its blob, so that the
    /// replacement's fresh blob id yields a fresh key.
    Replaceable,
    /// The row is never repointed: the blob it names when it is inserted is the blob it
    /// names for life. Repointing one is refused.
    ///
    /// This buys a stable, fully readable cloud path — `Live at Leeds/01 Sonata.flac`
    /// rather than `01 Sonata-0ef7a1c9.flac` — for content that is written once and never
    /// rewritten: an imported file, whose bytes are what they are.
    ///
    /// **What coven enforces, and what it does not.** coven refuses to repoint the row,
    /// which is the reuse it can see. It cannot see a consumer *deleting* a row and
    /// inserting a different blob at the same `cloud_path` — the deleted row is gone, and
    /// coven keeps no history of the paths it has used. Declaring `WriteOnce` is therefore
    /// also a promise that the path is never reused by a different blob. Derive it from
    /// data that never repeats and it holds by construction: a path carrying a freshly
    /// minted id for the thing being imported can never be handed out twice.
    WriteOnce,
}

/// Whether the readable `cloud_path` a consumer supplied names the blob `blob_id` — what
/// coven requires of a [`Replaceable`](BlobReplacement::Replaceable) blob's key on a
/// browsable home, and what a hashed key gets for free by carrying the id itself.
///
/// The path's file name (its last `/`-segment), with any extension stripped, must be the
/// blob id or end with `-{blob_id}`:
///
/// ```text
/// covers/Live at Leeds/cover-0ef7a1c9.jpg   ✓   stem `cover-0ef7a1c9` ends with -0ef7a1c9
/// covers/Live at Leeds/0ef7a1c9.jpg         ✓   stem is the blob id
/// covers/Live at Leeds/cover.jpg            ✗   names no blob
/// ```
///
/// A blob id names one immutable byte-string and is minted fresh for every stored blob, so
/// a path carrying it moves whenever the bytes do — which is what leaves a replaced blob's
/// object standing at its own key instead of overwritten.
///
/// The `-` delimiter is what makes this a near-injective mapping where a bare substring
/// test would not be: without it, blob `1` would satisfy blob `11`'s path and the two could
/// be keyed at one object. Two ids can still collide if one is a `-`-suffix of the other
/// AND the consumer builds paths that land on the same file name — which ids drawn from any
/// of the usual generators do not do.
pub fn cloud_path_names_blob(cloud_path: &str, blob_id: &str) -> bool {
    let file_name = cloud_path.rsplit('/').next().unwrap_or(cloud_path);
    let stem = file_name
        .rsplit_once('.')
        .map_or(file_name, |(stem, _extension)| stem);
    stem == blob_id
        || stem
            .strip_suffix(blob_id)
            .is_some_and(|prefix| prefix.ends_with('-'))
}

/// A blob a row references: its cloud identity, encryption scope, and the two
/// declared properties ([`provenance`](BlobRef::provenance) +
/// [`fill`](BlobRef::fill)). coven derives it from the row's declared columns
/// ([`crate::synced_schema::BlobDecl`]) via the database's `BlobDecls`. Where its bytes
/// live depends on its locality and provenance: a user-provided Local blob is the
/// user's file at its path; a host-provided Local blob is in coven's local store
/// (`storage/local/<namespace>/<id>`); a Remote blob's device-local copy is a cache
/// copy (`storage/pinned/<namespace>/<locator-hash>` /
/// `storage/cache/<namespace>/<locator-hash>`, built
/// from the validated namespace + exact locator hash — see `blob::cache`).
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct BlobRef {
    /// Cloud namespace, e.g. `"images"`. Becomes `{namespace}/{ab}/{cd}/{id}`
    /// under the hashed scheme, or `{namespace}/{cloud_path}` under the plain one.
    pub namespace: String,
    /// Blob id (typically the id of the blob-bearing row).
    pub id: String,
    /// Encryption scope for this blob.
    pub scope: BlobScope,
    /// The consumer's readable cloud-relative path for this blob, e.g.
    /// `"Artist - Album/cover.jpg"`. Used as the object key under `namespace` when
    /// the home's storage `BlobPathScheme` is `Plain`;
    /// ignored when `Hashed`. `None` is only valid for a `Hashed` home — a `Plain`
    /// home with no `cloud_path` is a surfaced error, never a silent fallback.
    pub cloud_path: Option<String>,
    /// The blob's **Local story**: where its bytes live while Local, and whether
    /// `make_local` needs a destination path. See [`Provenance`].
    pub provenance: Provenance,
    /// The blob's **Remote story**: whether a pulling device fetches it into the
    /// cache right away ([`CacheFill::CacheEager`]) or on first read
    /// ([`CacheFill::CacheLazy`]). See [`CacheFill`].
    pub fill: CacheFill,
}

/// One exact blob-bearing row version. A reference becomes stale when the live
/// row stamp or any declared blob value changes.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct RowBlobRef {
    table: String,
    row_id: String,
    row_stamp: String,
    column: String,
    blob: BlobRef,
    plaintext_size: u64,
    plaintext_hash: crate::store_commit::ObjectHash,
    authority: RowBlobAuthority,
    stored: Option<locator::StoredBlobRef>,
}

/// The authority state that determines where one row version's blob lives.
/// A remote-audience blob remains `PendingRemote` while its verified plaintext
/// is local and no cloud object has been created; `Remote` carries the exact
/// package authority needed to open its committed object.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case", deny_unknown_fields)]
pub enum RowBlobAuthority {
    Local,
    PendingRemote(locator::RemoteAudience),
    Remote(crate::audience_package::PackageAudience),
}

pub enum BlobOpeningAuthority<'a> {
    Store,
    Circle {
        circle_id: crate::circle::CircleId,
        control: &'a crate::circle::CircleControlCoord,
        key_fingerprint: coven_keys::encryption::KeyFingerprint,
    },
}

#[derive(Debug, thiserror::Error)]
pub enum BlobOpeningAuthorityError {
    #[error("blob {id} has no exact remote authority")]
    LocalityUnresolved { id: String },
    #[error(
        "Circle {circle_id} blob locator audience or key differs from its exact activated authority"
    )]
    CircleAuthorityMismatch { circle_id: crate::circle::CircleId },
}

impl<'de> serde::Deserialize<'de> for RowBlobRef {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        #[derive(serde::Deserialize)]
        #[serde(deny_unknown_fields)]
        struct Fields {
            table: String,
            row_id: String,
            row_stamp: String,
            column: String,
            blob: BlobRef,
            plaintext_size: u64,
            plaintext_hash: crate::store_commit::ObjectHash,
            authority: RowBlobAuthority,
            stored: Option<locator::StoredBlobRef>,
        }

        let fields = Fields::deserialize(deserializer)?;
        Self::new(
            fields.table,
            fields.row_id,
            fields.row_stamp,
            fields.column,
            fields.blob,
            fields.plaintext_size,
            fields.plaintext_hash,
            fields.authority,
            fields.stored,
        )
        .map_err(serde::de::Error::custom)
    }
}

impl RowBlobAuthority {
    pub fn audience(&self) -> crate::circle::Audience {
        match self {
            Self::Local => crate::circle::Audience::Local,
            Self::PendingRemote(locator::RemoteAudience::Store) => crate::circle::Audience::Store,
            Self::PendingRemote(locator::RemoteAudience::Circle(circle_id)) => {
                crate::circle::Audience::Circle(*circle_id)
            }
            Self::Remote(crate::audience_package::PackageAudience::Store) => {
                crate::circle::Audience::Store
            }
            Self::Remote(crate::audience_package::PackageAudience::Circle {
                circle_id, ..
            }) => crate::circle::Audience::Circle(*circle_id),
        }
    }

    pub fn opening_authority<'a>(
        &'a self,
        stored: &locator::StoredBlobRef,
    ) -> Result<BlobOpeningAuthority<'a>, BlobOpeningAuthorityError> {
        match self {
            Self::Local | Self::PendingRemote(_) => {
                Err(BlobOpeningAuthorityError::LocalityUnresolved {
                    id: stored.locator().blob_id().to_string(),
                })
            }
            Self::Remote(crate::audience_package::PackageAudience::Store) => {
                Ok(BlobOpeningAuthority::Store)
            }
            Self::Remote(crate::audience_package::PackageAudience::Circle {
                circle_id,
                control,
                key_fingerprint,
            }) => {
                if stored.locator().audience() != locator::RemoteAudience::Circle(*circle_id)
                    || stored.locator().key_fingerprint() != Some(*key_fingerprint)
                {
                    return Err(BlobOpeningAuthorityError::CircleAuthorityMismatch {
                        circle_id: *circle_id,
                    });
                }
                Ok(BlobOpeningAuthority::Circle {
                    circle_id: *circle_id,
                    control,
                    key_fingerprint: *key_fingerprint,
                })
            }
        }
    }
}

/// Whether `locator` describes exactly the row version it was minted for.
///
/// A stored blob's locator carries the namespace, id, plaintext size and hash,
/// and encryption scope of the row it was sealed from. Anything that hands a
/// `StoredBlobRef` and a row to each other checks this before trusting the pair.
///
/// [`RowBlobRef::new`] enforces the same facts and more, one field at a time so
/// it can name which one diverged; this is the yes-or-no form for callers that
/// answer a mismatch their own way.
/// Whether `locator` is the blob this row already has at the provider, for the
/// audience it is being published to.
///
/// This is deliberately **not** equality with a freshly minted locator. A
/// locator carries the fingerprint of the key that sealed its bytes, and that
/// key is whichever generation the keyring sealed under at upload time. Minting
/// one today and demanding the stored one match it says "any key rotation
/// re-identifies every blob in the Store" — which would ask the publisher to
/// re-upload bytes it has, under a name nothing has ever written, and for a
/// user-provided blob there is no local file left to re-upload from.
///
/// What actually identifies an already-uploaded blob is its content and where
/// it is readable from: namespace, id, plaintext size and hash, key scope, and
/// audience. An audience move is a genuine re-seal and is still refused here.
pub fn locator_is_this_rows_upload(
    locator: &locator::BlobLocator,
    blob: &BlobRef,
    plaintext_size: u64,
    plaintext_hash: crate::store_commit::ObjectHash,
    audience: &locator::RemoteAudience,
) -> bool {
    locator_describes_row(locator, blob, plaintext_size, plaintext_hash)
        && &locator.audience() == audience
}

pub fn locator_describes_row(
    locator: &locator::BlobLocator,
    blob: &BlobRef,
    plaintext_size: u64,
    plaintext_hash: crate::store_commit::ObjectHash,
) -> bool {
    locator.namespace() == blob.namespace
        && locator.blob_id() == blob.id
        && locator.plaintext_size() == plaintext_size
        && locator.plaintext_hash() == plaintext_hash
        && locator.scope().is_none_or(|scope| scope == &blob.scope)
}

impl RowBlobRef {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        table: String,
        row_id: String,
        row_stamp: String,
        column: String,
        blob: BlobRef,
        plaintext_size: u64,
        plaintext_hash: crate::store_commit::ObjectHash,
        authority: RowBlobAuthority,
        stored: Option<locator::StoredBlobRef>,
    ) -> Result<Self, RowBlobRefError> {
        let remote = match &authority {
            RowBlobAuthority::Local => None,
            RowBlobAuthority::PendingRemote(audience) => Some(audience.clone()),
            RowBlobAuthority::Remote(package) => Some(package.remote_audience()),
        };
        match (&authority, remote.as_ref(), stored.as_ref()) {
            (RowBlobAuthority::Local, None, None)
            | (RowBlobAuthority::PendingRemote(_), Some(_), None) => {}
            (RowBlobAuthority::Remote(_), Some(expected), Some(stored))
                if &stored.locator().audience() == expected => {}
            (RowBlobAuthority::Local, None, Some(_)) => {
                return Err(RowBlobRefError::LocalHasLocator);
            }
            (RowBlobAuthority::PendingRemote(_), Some(_), Some(_)) => {
                return Err(RowBlobRefError::PendingHasLocator);
            }
            (RowBlobAuthority::Remote(_), Some(_), None) => {
                return Err(RowBlobRefError::RemoteMissingLocator);
            }
            (RowBlobAuthority::Remote(_), Some(expected), Some(stored)) => {
                return Err(RowBlobRefError::AudienceMismatch {
                    row: expected.clone(),
                    locator: stored.locator().audience(),
                });
            }
            _ => unreachable!("authority determines whether a remote audience exists"),
        }
        if let Some(stored) = &stored {
            let locator = stored.locator();
            if locator.namespace() != blob.namespace {
                return Err(RowBlobRefError::NamespaceMismatch {
                    row: blob.namespace.clone(),
                    locator: locator.namespace().to_string(),
                });
            }
            if locator.blob_id() != blob.id {
                return Err(RowBlobRefError::IdMismatch {
                    row: blob.id.clone(),
                    locator: locator.blob_id().to_string(),
                });
            }
            if locator.plaintext_size() != plaintext_size
                || locator.plaintext_hash() != plaintext_hash
            {
                return Err(RowBlobRefError::PlaintextMismatch);
            }
            match locator {
                locator::BlobLocator::Opaque {
                    scope,
                    key_fingerprint,
                    ..
                } => {
                    if scope != &blob.scope {
                        return Err(RowBlobRefError::ScopeMismatch);
                    }
                    if let RowBlobAuthority::Remote(
                        crate::audience_package::PackageAudience::Circle {
                            key_fingerprint: expected,
                            ..
                        },
                    ) = &authority
                    {
                        if key_fingerprint != expected {
                            return Err(RowBlobRefError::CircleKeyMismatch);
                        }
                    }
                }
                locator::BlobLocator::Browsable { cloud_path, .. } => {
                    if blob.cloud_path.as_deref() != Some(cloud_path) {
                        return Err(RowBlobRefError::CloudPathMismatch);
                    }
                }
            }
        }
        Ok(Self {
            table,
            row_id,
            row_stamp,
            column,
            blob,
            plaintext_size,
            plaintext_hash,
            authority,
            stored,
        })
    }

    pub fn table(&self) -> &str {
        &self.table
    }

    pub fn row_id(&self) -> &str {
        &self.row_id
    }

    pub fn row_stamp(&self) -> &str {
        &self.row_stamp
    }

    pub fn column(&self) -> &str {
        &self.column
    }

    pub fn blob(&self) -> &BlobRef {
        &self.blob
    }

    pub fn plaintext_size(&self) -> u64 {
        self.plaintext_size
    }

    pub fn plaintext_hash(&self) -> crate::store_commit::ObjectHash {
        self.plaintext_hash
    }

    pub fn authority(&self) -> &RowBlobAuthority {
        &self.authority
    }

    pub fn audience(&self) -> crate::circle::Audience {
        self.authority.audience()
    }

    pub fn stored(&self) -> Option<&locator::StoredBlobRef> {
        self.stored.as_ref()
    }
}

#[derive(Debug, thiserror::Error)]
pub enum RowBlobRefError {
    #[error("Local row blob carries a remote locator")]
    LocalHasLocator,
    #[error("pending remote row blob carries a cloud locator")]
    PendingHasLocator,
    #[error("remote row blob has no exact locator")]
    RemoteMissingLocator,
    #[error("row audience {row:?} differs from locator audience {locator:?}")]
    AudienceMismatch {
        row: locator::RemoteAudience,
        locator: locator::RemoteAudience,
    },
    #[error("row blob namespace {row:?} differs from locator namespace {locator:?}")]
    NamespaceMismatch { row: String, locator: String },
    #[error("row blob id {row:?} differs from locator id {locator:?}")]
    IdMismatch { row: String, locator: String },
    #[error("row blob plaintext size or hash differs from its exact locator")]
    PlaintextMismatch,
    #[error("row blob encryption scope differs from its exact locator")]
    ScopeMismatch,
    #[error("row blob Circle key differs from its exact locator")]
    CircleKeyMismatch,
    #[error("row blob cloud path differs from its exact locator")]
    CloudPathMismatch,
}

/// Notified about coven's blob transitions, for host-specific bookkeeping and UI:
/// per-blob upload progress while a make_remote uploads, per-blob materialize
/// progress while a make_local copies files back, and the synchronous
/// make-local completion the host turns into its own UI event.
///
/// The host no longer drives the transition — coven owns flipping the gate and
/// deciding when a cycle publishes — so this observer only *reports*. The upload
/// callbacks fire as the drain works: preparation starts while the plaintext is
/// verified and sealed into its durable spool, `on_blob_upload_started` fires
/// only when that prepared spool is handed to the provider,
/// `on_blob_upload_progress` fires zero or more times as encrypted bytes reach
/// the cloud (backends that can't report sub-file progress call it once at the end
/// with `bytes_done == bytes_total`), `on_blob_uploaded` on success (notification
/// only — the durable queue records Created and the Store publication later
/// activates the root),
/// and `on_blob_upload_failed` when an attempt fails and its entry stays queued.
///
/// A make-remote's root state is durable and belongs in
/// `CloudOutboxLiveQuery`, not an observer callback that can be lost across a
/// restart. `on_root_made_local` reports the synchronous opposite direction;
/// `on_blob_materialize_progress` moves its per-file progress bar.
///
/// The upload-pause methods let the host suspend the upload pipeline without
/// touching the queue or discarding an open provider upload. The drain checks
/// the absolute state before admitting work, stops polling active preparation,
/// and stops active provider request bodies from yielding bytes while paused;
/// resume continues those same operations.
///
#[async_trait::async_trait]
pub trait BlobTransitionObserver: Send + Sync {
    /// The plaintext source is being verified and sealed into its durable
    /// upload spool. Fires only for a Pending journal; a restart-resumed
    /// Prepared journal proceeds directly to upload.
    async fn on_blob_preparation_started(&self, upload: &RowBlobRef) {
        let _ = upload;
    }

    /// `bytes_done` of `bytes_total` plaintext source bytes have been consumed
    /// by preparation. Values are cumulative and monotonic.
    async fn on_blob_preparation_progress(
        &self,
        upload: &RowBlobRef,
        bytes_done: u64,
        bytes_total: u64,
    ) {
        let _ = (upload, bytes_done, bytes_total);
    }

    /// The durable spool is prepared and its provider upload is starting now.
    async fn on_blob_upload_started(&self, upload: &RowBlobRef);

    /// `bytes_done` of `bytes_total` encrypted bytes have reached the cloud for
    /// this in-flight blob. `bytes_done` is cumulative and monotonic within one
    /// upload attempt. The default is a no-op so observers that don't surface
    /// sub-file progress don't need a stub.
    async fn on_blob_upload_progress(
        &self,
        upload: &RowBlobRef,
        bytes_done: u64,
        bytes_total: u64,
    ) {
        let _ = (upload, bytes_done, bytes_total);
    }

    /// The blob was uploaded to the cloud successfully — notification only.
    /// coven owns the durable Created handoff and the Store publication that
    /// completes the make_remote.
    async fn on_blob_uploaded(&self, upload: &RowBlobRef);

    /// An upload attempt failed; the entry remains queued for retry.
    async fn on_blob_upload_failed(&self, upload: &RowBlobRef, error: &str);

    /// Whether upload work is currently paused. The drain checks this before
    /// admission and while provider work is active. The default is `false` so
    /// existing implementations don't need a stub.
    fn should_skip_uploads(&self) -> bool {
        false
    }

    /// Complete when the absolute upload-pause state becomes paused. The
    /// default never completes because the default state never pauses.
    async fn wait_until_uploads_paused(&self) {
        std::future::pending::<()>().await;
    }

    /// Complete when the absolute upload-pause state becomes running. An
    /// observer that can return `true` from [`Self::should_skip_uploads`] must
    /// override this so a suspended transfer can resume.
    async fn wait_until_uploads_resumed(&self) {
        std::future::pending::<()>().await;
    }

    /// coven completed a make_local of `(root_table, root_id)`: every blob is back
    /// to a local file (a user file for user-provided, the local store for
    /// host-provided), the gate is flipped false (the subtree retracts from peers),
    /// and the cloud blobs are queued for tombstoning. The default is a no-op.
    async fn on_root_made_local(&self, root_table: &str, root_id: &str) {
        let _ = (root_table, root_id);
    }

    /// `done` of `total` of a make_local's blobs have been materialized back to a
    /// local file, so the host can move a per-file progress bar. The default is a
    /// no-op.
    async fn on_blob_materialize_progress(
        &self,
        root_table: &str,
        root_id: &str,
        blob_id: &str,
        done: u64,
        total: u64,
    ) {
        let _ = (root_table, root_id, blob_id, done, total);
    }
}

#[cfg(test)]
mod cloud_path_tests {
    use super::cloud_path_names_blob;

    /// A replaceable blob's readable path must name the blob standing at it, so that
    /// repointing the row moves its cloud key instead of overwriting the object it
    /// replaced. A path naming no blob — the natural-looking `cover.jpg` — is what makes a
    /// replacement rewrite the object its predecessor holds.
    #[test]
    fn a_cloud_path_names_the_blob_whose_id_ends_its_file_name() {
        assert!(cloud_path_names_blob(
            "Live at Leeds/cover-0ef7a1c9.jpg",
            "0ef7a1c9"
        ));
        assert!(cloud_path_names_blob(
            "Live at Leeds/0ef7a1c9.jpg",
            "0ef7a1c9"
        ));
        assert!(
            cloud_path_names_blob("0ef7a1c9", "0ef7a1c9"),
            "no directory and no extension: the whole path is the blob id",
        );

        assert!(
            !cloud_path_names_blob("Live at Leeds/cover.jpg", "0ef7a1c9"),
            "names no blob — the next cover would take this same name",
        );
        assert!(
            !cloud_path_names_blob("0ef7a1c9/cover.jpg", "0ef7a1c9"),
            "the id must name the OBJECT, not a directory above it — two blobs under one \
             directory would still collide on the file",
        );
        assert!(
            !cloud_path_names_blob("Live at Leeds/cover-0ef7a1c9-thumb.jpg", "0ef7a1c9"),
            "the id must END the file name's stem, not sit inside it",
        );
    }

    /// The `-` delimiter is what makes the path→blob mapping unambiguous. A bare substring
    /// test would let one blob satisfy another's path, and the two would be keyed at one
    /// cloud object — the exact collision the rule exists to prevent.
    #[test]
    fn one_blob_id_cannot_satisfy_another_s_path_by_being_a_tail_of_it() {
        assert!(cloud_path_names_blob("cover-10ef7a1c9.jpg", "10ef7a1c9"));
        assert!(
            !cloud_path_names_blob("cover-10ef7a1c9.jpg", "0ef7a1c9"),
            "blob 0ef7a1c9 must not claim blob 10ef7a1c9's object",
        );
    }
}

/// The default convergence window a host gets if it configures none: how long a
/// deleted blob is kept after its tombstone is written, before a GC pass
/// reclaims it. The host overrides it on the coven builder; the writer's
/// tombstone collection evaluates whatever grace it is handed against the
/// tombstone's `deleted_at`.
///
/// A device offline for less than the grace is never stranded by a deletion —
/// when it reconnects it pulls the removal of the row that referenced the blob,
/// and the blob is still present until then. The window is human-scale (days,
/// not the sub-second commit window the snapshot sweep's grace covers) because
/// the device it protects is a person's offline laptop or phone, not a
/// concurrent writer mid-publish.
pub const BLOB_TOMBSTONE_GRACE: chrono::Duration = chrono::Duration::days(7);

#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DeferredLocalBlobDisposition {
    Drop,
    Cache,
    Pin,
}

impl DeferredLocalBlobDisposition {
    pub fn as_db(self) -> &'static str {
        match self {
            Self::Drop => "drop",
            Self::Cache => "cache",
            Self::Pin => "pin",
        }
    }

    pub fn from_db(raw: &str) -> Result<Self, DeferredLocalBlobDispositionError> {
        match raw {
            "drop" => Ok(Self::Drop),
            "cache" => Ok(Self::Cache),
            "pin" => Ok(Self::Pin),
            other => Err(DeferredLocalBlobDispositionError {
                value: other.to_string(),
            }),
        }
    }
}

#[derive(Debug, thiserror::Error)]
#[error("unknown disposition in published blob drop intent: {value}")]
pub struct DeferredLocalBlobDispositionError {
    value: String,
}

#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DeferredLocalBlobDrop {
    pub namespace: String,
    pub id: String,
    pub size: u64,
    pub plaintext_hash: crate::store_commit::ObjectHash,
    pub locator_hash: crate::store_commit::ObjectHash,
    pub disposition: DeferredLocalBlobDisposition,
}