coven_database/

blob_declarations.rs

//! The resolved blob-declaration model: coven's own derivation of which rows
//! carry blobs and where each blob's columns live.
//!
//! The gate-sibling of [`Gates`](crate::Gates). A host declares per
//! table, via [`SyncedTable::carries_blob`], the columns that locate a blob (its
//! id, optional readable cloud path, and encryption-scope column) plus the
//! namespace and retention class. [`BlobDecls::from_tables`] resolves those column
//! *names* to indices against the live schema for the database handle — the same
//! `PRAGMA table_info` name→index resolution the gate runs — so coven reads a
//! row's blob straight off a changeset row or a live `SELECT` with no per-row host
//! callback.
//!
//! From that one model coven derives every blob set it needs:
//! [`BlobDecls::ref_from_change`] over a changeset row (push upload / pull
//! download / apply-side local-copy drop),
//! [`BlobDecls::publication_blobs_in_db`] over the whole database, and
//! [`BlobDecls::row_for_blob_in_namespace`] to map a blob back to its row by namespace
//! (the read-path locality dispatch and the make-Remote completion check).
//!
//! A declaration's three blob properties — [`Provenance`] (the Local story),
//! [`CacheFill`] (the Remote story), and [`BlobReplacement`] (whether the row may be
//! repointed at a different blob) — are described by the blob concept tree in
//! the replication layer. Write-once updates are refused here, where the declaration
//! and changed blob-id column are available. Immutable cloud-object identity comes
//! from the blob locator and retained exact object reference.

use std::collections::HashMap;

use rusqlite::{Connection, OptionalExtension};

use crate::{quote_ident, table_columns as session_table_columns};
use coven_foundation::changeset::{ChangeOp, RowChange};
use coven_protocol::blob::{BlobRef, BlobReplacement, BlobScope, CacheFill, Provenance};
use coven_protocol::synced_schema::SyncedTable;

/// Why building the blob-declaration model failed.
#[derive(Debug)]
pub enum BlobDeclError {
    /// A declared blob column is absent from the table's live schema.
    MissingColumn { table: String, column: String },
    /// A schema read (`PRAGMA table_info`) failed.
    Sqlite(rusqlite::Error),
    /// A captured host changeset could not be read.
    Changeset(crate::ChangesetError),
    /// A row's declared size column is negative.
    InvalidSize { table: String, value: i64 },
    /// A row names a blob but has no content hash.
    MissingHash { table: String, row_id: String },
    /// New and old changeset walks produced different row counts.
    ChangesetWalkMismatch { old_count: usize, new_count: usize },
    /// A blob-bearing INSERT or UPDATE has no primary key.
    MissingPublicationPrimaryKey { table: String },
    /// The transaction row named by a blob-bearing change is absent.
    MissingPublicationRow { table: String, primary_key: String },
    /// The transaction row named by a blob-bearing change no longer carries a blob.
    MissingPublicationBlob { table: String, primary_key: String },
    /// The transaction row carries a different blob than its change introduced.
    PublicationBlobMismatch {
        table: String,
        primary_key: String,
        changed_blob_id: String,
        row_blob_id: String,
    },
    /// A [`WriteOnce`](BlobReplacement::WriteOnce) changeset update changed the
    /// row's blob-id column, contrary to its declaration.
    WriteOnceBlobRepointed { table: String, blob_id: String },
}

impl std::fmt::Display for BlobDeclError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            BlobDeclError::MissingColumn { table, column } => {
                write!(
                    f,
                    "blob declaration names column {column:?} absent from {table:?}"
                )
            }
            BlobDeclError::Sqlite(e) => write!(f, "blob declaration schema read failed: {e}"),
            BlobDeclError::Changeset(error) => {
                write!(f, "blob declaration changeset read failed: {error}")
            }
            BlobDeclError::InvalidSize { table, value } => {
                write!(f, "blob declaration found invalid size in {table}: {value}")
            }
            BlobDeclError::MissingHash { table, row_id } => write!(
                f,
                "blob-bearing row {table:?}/{row_id:?} has no content hash"
            ),
            BlobDeclError::ChangesetWalkMismatch {
                old_count,
                new_count,
            } => write!(
                f,
                "blob declaration changeset walk mismatch: old={old_count}, new={new_count}"
            ),
            BlobDeclError::MissingPublicationPrimaryKey { table } => {
                write!(
                    f,
                    "blob-bearing Store write row in {table:?} has no primary key"
                )
            }
            BlobDeclError::MissingPublicationRow { table, primary_key } => write!(
                f,
                "blob-bearing Store write row {table:?}/{primary_key:?} is absent before commit"
            ),
            BlobDeclError::MissingPublicationBlob { table, primary_key } => write!(
                f,
                "blob-bearing Store write row {table:?}/{primary_key:?} no longer carries a blob"
            ),
            BlobDeclError::PublicationBlobMismatch {
                table,
                primary_key,
                changed_blob_id,
                row_blob_id,
            } => write!(
                f,
                "blob-bearing Store write row {table:?}/{primary_key:?} changed from introduced blob \
                 {changed_blob_id:?} to {row_blob_id:?} before commit"
            ),
            BlobDeclError::WriteOnceBlobRepointed { table, blob_id } => write!(
                f,
                "write-once row in {table} was repointed at blob {blob_id}: its declaration \
                 forbids changing the blob-id column. Declare the table replaceable if its \
                 rows are meant to be repointed"
            ),
        }
    }
}

impl std::error::Error for BlobDeclError {}

impl From<rusqlite::Error> for BlobDeclError {
    fn from(e: rusqlite::Error) -> Self {
        BlobDeclError::Sqlite(e)
    }
}

/// Exact row facts captured with a durable Store write for one blob-bearing row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PublicationBlob {
    pub table: String,
    pub row_id: String,
    pub row_stamp: String,
    pub column: String,
    pub blob: BlobRef,
    pub plaintext_size: u64,
    pub plaintext_hash: String,
}

/// A blob-bearing table's columns resolved to indices in the live schema (the
/// same order a changeset reports its columns, so an index reads either source).
struct TableBlob {
    namespace: String,
    provenance: Provenance,
    fill: CacheFill,
    columns: BlobColumns,
    /// Name of the blob-id column. The index reads a row top-to-bottom; the name
    /// keys a lookup the other way ([`BlobDecls::row_for_blob_in_namespace`]: which row
    /// carries a given blob id), so both directions resolve off the same declaration.
    id_col_name: String,
    /// The encryption scope, fixed per table by the declaration.
    scope: BlobScope,
    /// This table's row-repointing policy, enforced by [`TableBlob::ref_from_change`].
    replacement: BlobReplacement,
}

/// The declared content fields of one blob, resolved against its table schema.
/// These fields form one merge value even when only some change in an UPDATE.
pub(crate) struct BlobColumns {
    id: usize,
    size: usize,
    hash: usize,
    cloud_path: Option<usize>,
}

impl BlobColumns {
    pub(crate) fn resolve(
        table: &str,
        declaration: &coven_protocol::synced_schema::BlobDecl,
        columns: &[String],
    ) -> Result<Self, BlobDeclError> {
        let index = |name: &str| {
            columns
                .iter()
                .position(|column| column == name)
                .ok_or_else(|| BlobDeclError::MissingColumn {
                    table: table.to_string(),
                    column: name.to_string(),
                })
        };
        Ok(Self {
            id: index(&declaration.id_column)?,
            size: index(&declaration.size_column)?,
            hash: index(&declaration.hash_column)?,
            cloud_path: declaration
                .cloud_path_column
                .as_deref()
                .map(index)
                .transpose()?,
        })
    }

    pub(crate) fn iter(&self) -> impl Iterator<Item = usize> + '_ {
        [
            Some(self.id),
            Some(self.size),
            Some(self.hash),
            self.cloud_path,
        ]
        .into_iter()
        .flatten()
    }
}

impl TableBlob {
    /// Build the row's blob from its declared identity and readable path.
    /// Shared by changeset and live-row readers; locator construction owns
    /// path validation and exact cloud-object identity.
    fn blob_ref(&self, id: String, cloud_path: Option<String>) -> BlobRef {
        BlobRef {
            namespace: self.namespace.clone(),
            id,
            scope: self.scope.clone(),
            cloud_path,
            provenance: self.provenance,
            fill: self.fill,
        }
    }

    /// The blob a changeset row references.
    ///
    /// The gate a [`WriteOnce`](BlobReplacement::WriteOnce) row passes through. A
    /// changeset UPDATE marks only the columns whose values changed. The decoded row
    /// also carries unchanged primary-key values, so write-once enforcement must
    /// inspect that marker before treating its blob id as a repointing. A real repoint
    /// is refused here, where the change is read and the declaration is available.
    fn ref_from_change(
        &self,
        table: &str,
        change: &RowChange,
    ) -> Result<Option<BlobRef>, BlobDeclError> {
        let Some(id) = change.col(self.columns.id).map(str::to_string) else {
            return Ok(None);
        };
        if self.replacement == BlobReplacement::WriteOnce
            && change.op == ChangeOp::Update
            && change.column_changed(self.columns.id)
        {
            return Err(BlobDeclError::WriteOnceBlobRepointed {
                table: table.to_string(),
                blob_id: id,
            });
        }
        let cloud_path = self
            .columns
            .cloud_path
            .and_then(|i| change.col(i))
            .map(str::to_string);
        Ok(Some(self.blob_ref(id, cloud_path)))
    }

    /// Build the [`BlobRef`] for a live `SELECT *` row of this table, or `None` when
    /// the row's blob id is NULL. The resolved
    /// indices address a `SELECT *` row in schema order, exactly as they address a
    /// changeset row.
    fn ref_from_row(&self, row: &rusqlite::Row<'_>) -> Result<Option<BlobRef>, BlobDeclError> {
        let Some(id) = row.get::<_, Option<String>>(self.columns.id)? else {
            return Ok(None);
        };
        let cloud_path = match self.columns.cloud_path {
            Some(i) => row.get::<_, Option<String>>(i)?,
            None => None,
        };
        Ok(Some(self.blob_ref(id, cloud_path)))
    }

    fn size_from_row(&self, table: &str, row: &rusqlite::Row<'_>) -> Result<u64, BlobDeclError> {
        let value = row.get::<_, i64>(self.columns.size)?;
        u64::try_from(value).map_err(|_| BlobDeclError::InvalidSize {
            table: table.to_string(),
            value,
        })
    }

    fn hash_from_row(
        &self,
        table: &str,
        row_id: &str,
        row: &rusqlite::Row<'_>,
    ) -> Result<String, BlobDeclError> {
        row.get::<_, Option<String>>(self.columns.hash)?
            .ok_or_else(|| BlobDeclError::MissingHash {
                table: table.to_string(),
                row_id: row_id.to_string(),
            })
    }
}

/// The blob declarations for a database handle, resolved from the declared set +
/// the live schema at open. A synced table absent from this map carries no blob.
pub struct BlobDecls {
    tables: HashMap<String, TableBlob>,
}

#[cfg(any(test, feature = "test-utils"))]
thread_local! {
    static FROM_TABLES_CALLS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}

#[cfg(any(test, feature = "test-utils"))]
pub fn reset_from_tables_call_count() {
    FROM_TABLES_CALLS.with(|calls| calls.set(0));
}

#[cfg(any(test, feature = "test-utils"))]
pub fn from_tables_call_count() -> usize {
    FROM_TABLES_CALLS.with(std::cell::Cell::get)
}

impl BlobDecls {
    /// Resolve every [`SyncedTable::carries_blob`] declaration's column names to
    /// indices against the live schema, mirroring
    /// [`Gates::from_tables`](crate::Gates::from_tables). A declared
    /// column absent from the table is a host error surfaced here, never a silent
    /// drop.
    pub(crate) fn from_tables(
        conn: &Connection,
        tables: &[SyncedTable],
    ) -> Result<Self, BlobDeclError> {
        #[cfg(any(test, feature = "test-utils"))]
        FROM_TABLES_CALLS.with(|calls| calls.set(calls.get() + 1));

        let mut map = HashMap::new();
        for t in tables {
            let Some(decl) = t.blob() else {
                continue;
            };
            // Column names in declared (schema) order — the index of a name here
            // is the index a changeset reports for that column.
            let cols = session_table_columns(conn, t.name()).map_err(BlobDeclError::from)?;
            let columns = BlobColumns::resolve(t.name(), decl, &cols)?;

            map.insert(
                t.name().to_string(),
                TableBlob {
                    namespace: decl.namespace.clone(),
                    provenance: decl.provenance,
                    fill: decl.fill,
                    columns,
                    id_col_name: decl.id_column.clone(),
                    scope: decl.scope.clone(),
                    replacement: decl.replacement,
                },
            );
        }
        Ok(BlobDecls { tables: map })
    }

    /// Retain complete old/new content values for an actual blob edit. Metadata
    /// updates stay sparse, so they cannot overwrite independently edited content.
    pub(crate) fn complete_blob_changeset(
        &self,
        conn: &Connection,
        changeset: &[u8],
    ) -> Result<Vec<u8>, crate::DbError> {
        if self.tables.is_empty() || changeset.is_empty() {
            return Ok(changeset.to_vec());
        }
        let group = crate::gate::Changegroup::new().map_err(crate::DbError::from)?;
        unsafe {
            group
                .set_schema(conn.handle())
                .map_err(crate::DbError::from)?;
            crate::gate::for_each_change(changeset, |iter, change| {
                let Some(blob) = self
                    .tables
                    .get(&change.table)
                    .filter(|_| change.op == rusqlite::ffi::SQLITE_UPDATE)
                else {
                    return group.add_change(iter);
                };
                let (mut old, mut new, indirect) = crate::gate::update_values(iter)?;
                let edited = blob.columns.iter().filter(|index| *index != 0).any(|index| {
                    matches!((&old[index], &new[index]), (Some(old), Some(new)) if old != new)
                });
                if !edited {
                    return group.add_change(iter);
                }
                let pk = change.pk().ok_or_else(|| {
                    crate::GateError::MissingChangesetPrimaryKey(change.table.clone())
                })?;
                let sql = format!("SELECT * FROM {} WHERE id = ?1", quote_ident(&change.table));
                let values = conn
                    .query_row(&sql, [pk], |row| {
                        blob.columns
                            .iter()
                            .filter(|index| *index != 0)
                            .map(|index| {
                                row.get::<_, rusqlite::types::Value>(index)
                                    .map(|value| (index, value))
                            })
                            .collect::<rusqlite::Result<Vec<_>>>()
                    })
                    .map_err(|error| {
                        crate::GateError::Sql(
                            format!("capture blob content in {}", change.table),
                            error,
                        )
                    })?;
                for (index, value) in values {
                    if old[index].is_none() && new[index].is_none() {
                        old[index] = Some(value.clone());
                        new[index] = Some(value);
                    }
                }
                group.add_update(&change.table, &old, &new, indirect)
            })
            .map_err(crate::DbError::from)?;
        }
        group.output().map_err(crate::DbError::from)
    }

    /// Install connection-local guards that keep a blob cleanup intent exclusive
    /// until its filesystem deletion finishes. A cleanup intent is committed
    /// before the database releases the row; while either cleanup queue holds it,
    /// no INSERT or UPDATE may make the same `(namespace, blob id)` live again.
    /// TEMP triggers keep this runtime guard out of snapshots and use each
    /// declaration's resolved blob-id column rather than assuming the row primary
    /// key carries the blob id.
    pub(crate) fn install_cleanup_guards(&self, conn: &Connection) -> Result<(), BlobDeclError> {
        for (table, blob) in &self.tables {
            let table_ident = quote_ident(table);
            let id_ident = quote_ident(&blob.id_col_name);
            let namespace_literal: String =
                conn.query_row("SELECT quote(?1)", [&blob.namespace], |row| row.get(0))?;
            for (trigger_kind, event_clause) in [
                ("insert", "BEFORE INSERT".to_string()),
                ("update", format!("BEFORE UPDATE OF {id_ident}")),
            ] {
                let trigger = quote_ident(&format!(
                    "{}{trigger_kind}_{table}",
                    super::COVEN_CLEANUP_GUARD_PREFIX
                ));
                conn.execute_batch(&format!(
                    "CREATE TEMP TRIGGER {trigger} \
                     {event_clause} ON main.{table_ident} \
                     WHEN NEW.{id_ident} IS NOT NULL AND (\
                         EXISTS (\
                             SELECT 1 FROM local_cleanup_intents \
                             WHERE namespace = {namespace_literal} \
                               AND blob_id = NEW.{id_ident}\
                         ) OR EXISTS (\
                             SELECT 1 FROM published_blob_drop_intents \
                             WHERE namespace = {namespace_literal} \
                               AND blob_id = NEW.{id_ident}\
                         )\
                     ) \
                     BEGIN \
                         SELECT RAISE(ABORT, 'blob local cleanup in progress'); \
                     END;"
                ))?;
            }
        }
        Ok(())
    }

    /// The blob a single changeset row references, or `None` when the row's table
    /// carries no blob or the blob id is absent/NULL. Reads the declared columns
    /// off the changeset row (which reports columns in schema order, the order the
    /// resolved indices address).
    pub fn ref_from_change(&self, change: &RowChange) -> Result<Option<BlobRef>, BlobDeclError> {
        let Some(tb) = self.tables.get(&change.table) else {
            return Ok(None);
        };
        tb.ref_from_change(&change.table, change)
    }

    /// The exact blob reference and declared size owned by an INSERT or UPDATE,
    /// completed from that row inside the transaction that produced the change.
    /// An UPDATE can omit its unchanged blob-id column. The live transaction row
    /// supplies that identity and its content facts before a later write can
    /// repoint or delete it; an identity present in the change must still match.
    pub(crate) fn publication_blob_from_change(
        &self,
        conn: &Connection,
        change: &RowChange,
    ) -> Result<Option<PublicationBlob>, BlobDeclError> {
        if !matches!(change.op, ChangeOp::Insert | ChangeOp::Update) {
            return Ok(None);
        }
        let Some(tb) = self.tables.get(&change.table) else {
            return Ok(None);
        };
        let changed_blob = tb.ref_from_change(&change.table, change)?;
        if changed_blob.is_none()
            && (change.op == ChangeOp::Insert || change.column_changed(tb.columns.id))
        {
            return Ok(None);
        }
        let pk = change
            .pk()
            .ok_or_else(|| BlobDeclError::MissingPublicationPrimaryKey {
                table: change.table.clone(),
            })?;
        let publication = match self.publication_blob_for_row(conn, &change.table, pk) {
            Ok(Some(publication)) => publication,
            Ok(None) => {
                return Err(BlobDeclError::MissingPublicationRow {
                    table: change.table.clone(),
                    primary_key: pk.to_string(),
                });
            }
            Err(BlobDeclError::MissingPublicationBlob { .. }) if changed_blob.is_none() => {
                return Ok(None);
            }
            Err(error) => return Err(error),
        };
        if let Some(changed_blob) = changed_blob {
            if publication.blob.id != changed_blob.id {
                return Err(BlobDeclError::PublicationBlobMismatch {
                    table: change.table.clone(),
                    primary_key: pk.to_string(),
                    changed_blob_id: changed_blob.id,
                    row_blob_id: publication.blob.id.clone(),
                });
            }
        }
        Ok(Some(publication))
    }

    pub(crate) fn publication_blob_for_row(
        &self,
        conn: &Connection,
        table: &str,
        row_id: &str,
    ) -> Result<Option<PublicationBlob>, BlobDeclError> {
        let Some(blob) = self.tables.get(table) else {
            return Ok(None);
        };
        let sql = format!("SELECT * FROM {} WHERE id = ?1", quote_ident(table));
        let mut statement = conn.prepare(&sql)?;
        let mut rows = statement.query([row_id])?;
        rows.next()?
            .map(|row| publication_blob_from_row(table, blob, row))
            .transpose()
    }

    /// Require every inserted or updated blob-bearing row in `changeset` to
    /// carry complete final content facts before its transaction can commit.
    pub(crate) fn validate_changed_rows(
        &self,
        conn: &Connection,
        changeset: &[u8],
    ) -> Result<(), BlobDeclError> {
        let changes = crate::walk_changeset(changeset).map_err(BlobDeclError::Changeset)?;
        for change in changes {
            if !matches!(change.op, ChangeOp::Insert | ChangeOp::Update)
                || !self.tables.contains_key(&change.table)
            {
                continue;
            }
            let row_id =
                change
                    .pk()
                    .ok_or_else(|| BlobDeclError::MissingPublicationPrimaryKey {
                        table: change.table.clone(),
                    })?;
            match self.publication_blob_for_row(conn, &change.table, row_id) {
                Ok(_) | Err(BlobDeclError::MissingPublicationBlob { .. }) => {}
                Err(error) => return Err(error),
            }
        }
        Ok(())
    }

    /// Every exact blob-bearing row version currently present in `conn`.
    pub(crate) fn publication_blobs_in_db(
        &self,
        conn: &Connection,
    ) -> Result<Vec<PublicationBlob>, BlobDeclError> {
        let mut out = Vec::new();
        for (table, blob) in &self.tables {
            let sql = format!("SELECT * FROM {}", quote_ident(table));
            let mut statement = conn.prepare(&sql)?;
            let mut rows = statement.query([])?;
            while let Some(row) = rows.next()? {
                let Some(reference) = blob.ref_from_row(row)? else {
                    continue;
                };
                let row_id = row.get::<_, String>("id")?;
                out.push(PublicationBlob {
                    table: table.clone(),
                    row_id: row_id.clone(),
                    row_stamp: row.get("_updated_at")?,
                    column: blob.id_col_name.clone(),
                    blob: reference,
                    plaintext_size: blob.size_from_row(table, row)?,
                    plaintext_hash: blob.hash_from_row(table, &row_id, row)?,
                });
            }
        }
        out.sort_by(|left, right| {
            (&left.table, &left.row_id, &left.column, &left.row_stamp).cmp(&(
                &right.table,
                &right.row_id,
                &right.column,
                &right.row_stamp,
            ))
        });
        Ok(out)
    }

    /// The `(table, primary key)` of the row carrying `blob_id` in the table declared
    /// for `namespace` — the carrying table resolved from the blob's own namespace
    /// (part of its address), not by scanning every blob-bearing table. `None` when no
    /// declared table owns `namespace`, or that table has no row with the id. Both the
    /// read path (locality dispatch) and the make-Remote completion check use this, so
    /// a blob id that collides across namespaces always reads the right table's gate,
    /// never the first id match.
    pub(crate) fn row_for_blob_in_namespace(
        &self,
        conn: &Connection,
        namespace: &str,
        blob_id: &str,
    ) -> Result<Option<(String, String)>, BlobDeclError> {
        let Some((table, tb)) = self.table_for_namespace(namespace) else {
            return Ok(None);
        };
        let sql = format!(
            "SELECT id FROM {} WHERE {} = ?1",
            quote_ident(table),
            quote_ident(&tb.id_col_name),
        );
        conn.query_row(&sql, [blob_id], |row| row.get::<_, String>(0))
            .optional()
            .map(|primary_key| primary_key.map(|primary_key| (table.clone(), primary_key)))
            .map_err(BlobDeclError::from)
    }

    /// The one `(table, declaration)` whose blob namespace is `namespace`, or
    /// `None` when no declared table owns it. The namespace is part of a blob's
    /// address, so this resolves the carrying table without scanning every
    /// blob-bearing table's rows.
    fn table_for_namespace(&self, namespace: &str) -> Option<(&String, &TableBlob)> {
        self.tables.iter().find(|(_, tb)| tb.namespace == namespace)
    }

    /// Whether a live row still needs the logical-id-keyed local source for this
    /// blob. A row needs that source exactly when its current stamp has no installed
    /// remote locator binding.
    pub(crate) fn local_copy_is_referenced(
        &self,
        conn: &Connection,
        namespace: &str,
        blob_id: &str,
    ) -> Result<bool, BlobDeclError> {
        let Some((table, blob)) = self.table_for_namespace(namespace) else {
            return Ok(false);
        };
        let sql = format!(
            "SELECT EXISTS(
                 SELECT 1 FROM {table} AS live
                 WHERE CAST(live.{blob_column} AS TEXT) = ?1
                   AND NOT EXISTS (
                       SELECT 1 FROM row_blob_locators AS binding
                       WHERE binding.table_name = ?2
                         AND binding.row_id = CAST(live.id AS TEXT)
                         AND binding.column_name = ?3
                         AND binding.row_stamp = CAST(live._updated_at AS TEXT)
                   )
             )",
            table = quote_ident(table),
            blob_column = quote_ident(&blob.id_col_name),
        );
        conn.query_row(
            &sql,
            rusqlite::params![blob_id, table, blob.id_col_name],
            |row| row.get(0),
        )
        .map_err(BlobDeclError::from)
    }

    /// Whether any live row still carries this logical blob ID, independent of
    /// whether that row currently resolves to a local source or an exact remote
    /// locator.
    pub(crate) fn blob_id_is_referenced(
        &self,
        conn: &Connection,
        namespace: &str,
        blob_id: &str,
    ) -> Result<bool, BlobDeclError> {
        let Some((table, blob)) = self.table_for_namespace(namespace) else {
            return Ok(false);
        };
        let sql = format!(
            "SELECT EXISTS(
                 SELECT 1 FROM {table}
                 WHERE CAST({blob_column} AS TEXT) = ?1
             )",
            table = quote_ident(table),
            blob_column = quote_ident(&blob.id_col_name),
        );
        conn.query_row(&sql, [blob_id], |row| row.get(0))
            .map_err(BlobDeclError::from)
    }

    /// Whether a live row's current stamp still names one exact locator. A row
    /// that merely reuses the same logical blob id under another locator does not
    /// retain this locator's cache or pinned file.
    pub(crate) fn exact_copy_is_referenced(
        &self,
        conn: &Connection,
        namespace: &str,
        blob_id: &str,
        locator_hash: coven_protocol::store_commit::ObjectHash,
    ) -> Result<bool, BlobDeclError> {
        let Some((table, blob)) = self.table_for_namespace(namespace) else {
            return Ok(false);
        };
        let sql = format!(
            "SELECT EXISTS(
                 SELECT 1
                 FROM {table} AS live
                 JOIN row_blob_locators AS binding
                   ON binding.table_name = ?2
                  AND binding.row_id = CAST(live.id AS TEXT)
                  AND binding.column_name = ?3
                  AND binding.row_stamp = CAST(live._updated_at AS TEXT)
                 JOIN blob_locators AS locator
                   ON locator.remote_object_id = binding.remote_object_id
                 WHERE CAST(live.{blob_column} AS TEXT) = ?1
                   AND locator.locator_hash = ?4
             )",
            table = quote_ident(table),
            blob_column = quote_ident(&blob.id_col_name),
        );
        conn.query_row(
            &sql,
            rusqlite::params![blob_id, table, blob.id_col_name, locator_hash.to_string()],
            |row| row.get(0),
        )
        .map_err(BlobDeclError::from)
    }
}

fn publication_blob_from_row(
    table: &str,
    blob: &TableBlob,
    row: &rusqlite::Row<'_>,
) -> Result<PublicationBlob, BlobDeclError> {
    let row_id = row.get::<_, String>("id")?;
    let reference =
        blob.ref_from_row(row)?
            .ok_or_else(|| BlobDeclError::MissingPublicationBlob {
                table: table.to_string(),
                primary_key: row_id.clone(),
            })?;
    let plaintext_hash = blob.hash_from_row(table, &row_id, row)?;
    Ok(PublicationBlob {
        table: table.to_string(),
        row_id,
        row_stamp: row.get("_updated_at")?,
        column: blob.id_col_name.clone(),
        blob: reference,
        plaintext_size: blob.size_from_row(table, row)?,
        plaintext_hash,
    })
}

#[cfg(test)]
#[path = "blob_declarations_tests.rs"]
mod tests;