coven_database/store/
host_sql.rs

1use crate::Gates;
2use crate::{CloudOutboxRecords, Database, DbError, ExternalBlobRecords, PreparedExternalBlob};
3use coven_protocol::blob::Provenance;
4
5use coven_protocol::hlc::UpdatedAtStamper;
6use coven_protocol::synced_schema::SyncedTable;
7
8const EXTERNAL_BLOB_HASH_PARAMETER: &str = ":coven_external_blob_hash";
9
10/// Host SQL against Coven's retained database connection.
11///
12/// The connection remains private. This context exposes query operations while
13/// preventing callers from changing connection configuration or starting an
14/// independent transaction.
15///
16/// ```compile_fail
17/// fn cannot_write(sql: coven::SqlReadContext<'_>) {
18///     sql.execute("DELETE FROM notes", []);
19/// }
20/// ```
21pub struct SqlReadContext<'connection> {
22    connection: &'connection rusqlite::Connection,
23    dependencies: Option<crate::live_query::ReadDependencyCapture>,
24}
25
26impl<'connection> SqlReadContext<'connection> {
27    pub(crate) fn new(connection: &'connection rusqlite::Connection) -> Self {
28        Self {
29            connection,
30            dependencies: None,
31        }
32    }
33
34    pub(crate) fn tracking(
35        connection: &'connection rusqlite::Connection,
36        dependencies: crate::live_query::ReadDependencyCapture,
37    ) -> Self {
38        Self {
39            connection,
40            dependencies: Some(dependencies),
41        }
42    }
43
44    pub fn query_row<T, P, F>(&self, sql: &str, params: P, map: F) -> rusqlite::Result<T>
45    where
46        P: rusqlite::Params,
47        F: FnOnce(&rusqlite::Row<'_>) -> rusqlite::Result<T>,
48    {
49        let mut statement = self.prepare_tracked(sql)?;
50        if let Err(error) = params.__bind_in(&mut statement) {
51            self.finish_statement(None);
52            return Err(error);
53        }
54        self.finish_statement(statement.expanded_sql());
55        let mut rows = statement.raw_query();
56        match rows.next()? {
57            Some(row) => map(row),
58            None => Err(rusqlite::Error::QueryReturnedNoRows),
59        }
60    }
61
62    pub fn query<T, P, F>(&self, sql: &str, params: P, map: F) -> rusqlite::Result<Vec<T>>
63    where
64        P: rusqlite::Params,
65        F: FnMut(&rusqlite::Row<'_>) -> rusqlite::Result<T>,
66    {
67        let mut statement = self.prepare_tracked(sql)?;
68        if let Err(error) = params.__bind_in(&mut statement) {
69            self.finish_statement(None);
70            return Err(error);
71        }
72        self.finish_statement(statement.expanded_sql());
73        let mut rows = statement.raw_query();
74        let mut map = map;
75        let mut values = Vec::new();
76        while let Some(row) = rows.next()? {
77            values.push(map(row)?);
78        }
79        Ok(values)
80    }
81
82    fn prepare_tracked(&self, sql: &str) -> rusqlite::Result<rusqlite::Statement<'connection>> {
83        if let Some(dependencies) = &self.dependencies {
84            dependencies.begin_statement();
85        }
86        match self.connection.prepare(sql) {
87            Ok(statement) => Ok(statement),
88            Err(error) => {
89                self.finish_statement(None);
90                Err(error)
91            }
92        }
93    }
94
95    fn finish_statement(&self, expanded_sql: Option<String>) {
96        if let Some(dependencies) = &self.dependencies {
97            dependencies.finish_statement(expanded_sql);
98        }
99    }
100}
101
102/// Host SQL inside one journaled write transaction.
103///
104/// The underlying transaction remains private so host SQL cannot remove
105/// Coven's authorizer or address Coven-owned attached schemas.
106pub struct SqlContext<'context, 'connection> {
107    transaction: &'context rusqlite::Transaction<'connection>,
108    stamper: UpdatedAtStamper,
109    tables: &'context [SyncedTable],
110    gates: &'context Gates,
111}
112
113impl<'context, 'connection> SqlContext<'context, 'connection> {
114    pub(crate) fn new(
115        transaction: &'context rusqlite::Transaction<'connection>,
116        stamper: UpdatedAtStamper,
117        tables: &'context [SyncedTable],
118        gates: &'context Gates,
119    ) -> Self {
120        Self {
121            transaction,
122            stamper,
123            tables,
124            gates,
125        }
126    }
127
128    fn blob_table(&self, table: &str) -> Result<&SyncedTable, DbError> {
129        let declared = self
130            .tables
131            .iter()
132            .find(|candidate| candidate.name() == table)
133            .ok_or_else(|| DbError::Message(format!("undeclared synced table {table:?}")))?;
134        if declared.blob().is_none() {
135            return Err(DbError::Message(format!(
136                "synced table {table:?} has no blob declaration"
137            )));
138        }
139        Ok(declared)
140    }
141
142    fn user_provided_blob_table(&self, table: &str) -> Result<&SyncedTable, DbError> {
143        let declared = self.blob_table(table)?;
144        let blob = declared.blob().expect("blob_table requires a declaration");
145        if blob.provenance != Provenance::UserProvided {
146            return Err(DbError::Message(format!(
147                "table {table:?} declares host-provided blobs, which Coven copies; \
148                 an external file registration on it would never be read"
149            )));
150        }
151        Ok(declared)
152    }
153
154    fn register_prepared_external_blob(
155        &self,
156        declared: &SyncedTable,
157        table: &str,
158        row_id: &str,
159        prepared: PreparedExternalBlob,
160    ) -> Result<(), DbError> {
161        prepared.validate_current()?;
162        let blob = declared.blob().expect("blob table requires a declaration");
163        let table_ident = crate::quote_ident(table);
164        let size_ident = crate::quote_ident(&blob.size_column);
165        let hash_ident = crate::quote_ident(&blob.hash_column);
166        let select = format!("SELECT {size_ident}, {hash_ident} FROM {table_ident} WHERE id = ?1");
167        let (declared_size, declared_hash) =
168            self.transaction.query_row(&select, [row_id], |row| {
169                Ok((row.get::<_, i64>(0)?, row.get::<_, Option<String>>(1)?))
170            })?;
171        let declared_size = u64::try_from(declared_size).map_err(|_| {
172            DbError::Message(format!(
173                "external blob row {table:?}/{row_id:?} has negative size"
174            ))
175        })?;
176        if declared_size != prepared.size() {
177            return Err(DbError::Message(format!(
178                "external blob row {table:?}/{row_id:?} declares {declared_size} bytes, but Coven read {}",
179                prepared.size()
180            )));
181        }
182        match declared_hash {
183            Some(hash) if hash != prepared.hash() => {
184                return Err(DbError::Message(format!(
185                    "external blob row {table:?}/{row_id:?} already declares different content"
186                )));
187            }
188            Some(_) => {}
189            None => {
190                let update = format!(
191                    "UPDATE {table_ident} SET {hash_ident} = ?1 \
192                     WHERE id = ?2 AND {hash_ident} IS NULL"
193                );
194                let updated = self
195                    .transaction
196                    .execute(&update, rusqlite::params![prepared.hash(), row_id])?;
197                if updated != 1 {
198                    return Err(DbError::Message(format!(
199                        "external blob row {table:?}/{row_id:?} changed before registration"
200                    )));
201                }
202            }
203        }
204        let reference = Database::row_blob_ref_on(self.transaction, self.gates, declared, row_id)?;
205        ExternalBlobRecords::new(self.transaction).register(&reference, prepared.path())
206    }
207
208    pub fn execute<P>(&self, sql: &str, params: P) -> rusqlite::Result<usize>
209    where
210        P: rusqlite::Params,
211    {
212        self.transaction.execute(sql, params)
213    }
214
215    pub fn execute_batch(&self, sql: &str) -> rusqlite::Result<()> {
216        self.transaction.execute_batch(sql)
217    }
218
219    pub fn query_row<T, P, F>(&self, sql: &str, params: P, map: F) -> rusqlite::Result<T>
220    where
221        P: rusqlite::Params,
222        F: FnOnce(&rusqlite::Row<'_>) -> rusqlite::Result<T>,
223    {
224        self.transaction.query_row(sql, params, map)
225    }
226
227    pub fn query<T, P, F>(&self, sql: &str, params: P, map: F) -> rusqlite::Result<Vec<T>>
228    where
229        P: rusqlite::Params,
230        F: FnMut(&rusqlite::Row<'_>) -> rusqlite::Result<T>,
231    {
232        let mut statement = self.transaction.prepare(sql)?;
233        let values = statement.query_map(params, map)?.collect();
234        values
235    }
236
237    pub fn stamp(&self) -> String {
238        self.stamper.stamp()
239    }
240
241    /// Insert a user-provided blob row without exposing its content hash to the
242    /// host. `insert_sql` names `:coven_external_blob_hash` where the declared
243    /// hash column belongs; `params` supplies every other named parameter. Coven
244    /// binds its private digest, verifies the inserted row and prepared file,
245    /// and registers the exact source path in this transaction.
246    pub fn insert_external_blob(
247        &self,
248        table: &str,
249        row_id: &str,
250        prepared: PreparedExternalBlob,
251        insert_sql: &str,
252        params: &[(&str, &dyn rusqlite::ToSql)],
253    ) -> Result<(), DbError> {
254        crate::observe_host_sql_write();
255        crate::with_coven_sql_authority(|| {
256            let declared = self.user_provided_blob_table(table)?;
257            if params
258                .iter()
259                .any(|(name, _)| *name == EXTERNAL_BLOB_HASH_PARAMETER)
260            {
261                return Err(DbError::Message(format!(
262                    "{EXTERNAL_BLOB_HASH_PARAMETER} is reserved for Coven"
263                )));
264            }
265            let mut names = std::collections::HashSet::with_capacity(params.len());
266            if params.iter().any(|(name, _)| !names.insert(*name)) {
267                return Err(DbError::Message(
268                    "external blob insert parameters contain a duplicate name".to_string(),
269                ));
270            }
271
272            let mut statement = self.transaction.prepare(insert_sql)?;
273            let hash_index = statement
274                .parameter_index(EXTERNAL_BLOB_HASH_PARAMETER)?
275                .ok_or_else(|| {
276                    DbError::Message(format!(
277                        "external blob insert is missing {EXTERNAL_BLOB_HASH_PARAMETER}"
278                    ))
279                })?;
280            let expected_parameters = params.len() + 1;
281            if statement.parameter_count() != expected_parameters {
282                return Err(DbError::Message(format!(
283                    "external blob insert declares {} parameters, expected {expected_parameters}",
284                    statement.parameter_count()
285                )));
286            }
287            rusqlite::Params::__bind_in(params, &mut statement)?;
288            statement.raw_bind_parameter(hash_index, prepared.hash())?;
289            let inserted = statement.raw_execute()?;
290            if inserted != 1 {
291                return Err(DbError::Message(format!(
292                    "external blob insert changed {inserted} rows, expected 1"
293                )));
294            }
295            drop(statement);
296            self.register_prepared_external_blob(declared, table, row_id, prepared)
297        })
298    }
299
300    pub fn register_external_blob(
301        &self,
302        table: &str,
303        row_id: &str,
304        prepared: PreparedExternalBlob,
305    ) -> Result<(), DbError> {
306        crate::observe_host_sql_write();
307        crate::with_coven_sql_authority(|| {
308            let declared = self.user_provided_blob_table(table)?;
309            self.register_prepared_external_blob(declared, table, row_id, prepared)
310        })
311    }
312
313    /// Require an earlier blob reference to still describe this transaction's
314    /// current row and stored object. Call before changing or deleting the row;
315    /// a stale reference aborts the write when the error is propagated.
316    pub fn validate_row_blob_ref(
317        &self,
318        reference: &coven_protocol::blob::RowBlobRef,
319    ) -> Result<(), DbError> {
320        crate::with_coven_sql_authority(|| {
321            let table = self.blob_table(reference.table())?;
322            Database::validate_row_blob_ref_on(self.transaction, self.gates, table, reference)
323        })
324    }
325
326    pub fn enqueue_blob_delete(
327        &self,
328        blob: &coven_protocol::blob::RowBlobRef,
329    ) -> Result<(), DbError> {
330        crate::observe_host_sql_write();
331        let stored = blob.stored().ok_or_else(|| {
332            DbError::Message(format!(
333                "blob {:?} in {:?} has no cloud object to remove",
334                blob.blob().id,
335                blob.blob().namespace
336            ))
337        })?;
338        crate::with_coven_sql_authority(|| {
339            CloudOutboxRecords::new(self.transaction).enqueue_delete(stored, &self.stamp())
340        })
341    }
342
343    pub fn clear_external_blob(&self, table: &str, row_id: &str) -> Result<(), DbError> {
344        crate::observe_host_sql_write();
345        crate::with_coven_sql_authority(|| {
346            let declared = self.blob_table(table)?;
347            let reference =
348                Database::row_blob_ref_on(self.transaction, self.gates, declared, row_id)?;
349            ExternalBlobRecords::new(self.transaction).clear(&reference)
350        })
351    }
352
353    #[cfg(any(test, feature = "test-utils"))]
354    pub fn materialized_sequence(&self, stream_id: &str) -> Result<Option<u64>, DbError> {
355        use rusqlite::OptionalExtension;
356
357        crate::with_coven_sql_authority(|| {
358            self.transaction
359                .query_row(
360                    "SELECT seq FROM materialized_commits WHERE device_id = ?1",
361                    [stream_id],
362                    |row| row.get::<_, i64>(0),
363                )
364                .optional()
365                .map_err(DbError::from)?
366                .map(|sequence| {
367                    u64::try_from(sequence)
368                        .map_err(|error| DbError::context("invalid sequence", error))
369                })
370                .transpose()
371        })
372    }
373}
coven_database/store/host_sql.rs

coven_database/store/
host_sql.rs
