coven_database/
blob_records.rs

1use crate::cloud_outbox_records::CloudOutboxRecords;
2use crate::remote_object_records::load_remote_object_on;
3
4use super::*;
5
6/// The locator a stored blob's row carries.
7///
8/// The caller has already established the record is an activated stored blob,
9/// and that domain's payloads are the row-carried locator by the record's own
10/// validation. An absent locator is therefore a record contradicting itself, not
11/// a case to substitute empty bytes for.
12pub(crate) fn carried_blob_locator(
13    remote: &coven_protocol::remote_object::RemoteObjectRecord,
14    context: &str,
15) -> Result<BlobLocator, DbError> {
16    let locator_bytes = remote.payloads().carried_locator_bytes().ok_or_else(|| {
17        DbError::Message(format!(
18            "{context}: stored blob {} carries no locator in its row",
19            remote.object_id()
20        ))
21    })?;
22    BlobLocator::parse(locator_bytes).map_err(|error| DbError::context(context.to_string(), error))
23}
24
25pub(crate) struct LiveBlobRow {
26    pub stamp: String,
27    pub blob_id: String,
28    pub plaintext_size: u64,
29    pub plaintext_hash: ObjectHash,
30    pub cloud_path: Option<String>,
31}
32
33pub(crate) fn live_blob_row(
34    conn: &Connection,
35    table: &str,
36    row_id: &str,
37    declaration: &coven_protocol::synced_schema::BlobDecl,
38) -> Result<Option<LiveBlobRow>, DbError> {
39    let cloud_path = declaration
40        .cloud_path_column
41        .as_deref()
42        .map(quote_ident)
43        .unwrap_or_else(|| "NULL".to_string());
44    let sql = format!(
45        "SELECT {}, {}, {}, {}, {} FROM {} WHERE {} = ?1",
46        quote_ident(&declaration.id_column),
47        quote_ident(&declaration.size_column),
48        quote_ident(&declaration.hash_column),
49        cloud_path,
50        quote_ident("_updated_at"),
51        quote_ident(table),
52        quote_ident("id"),
53    );
54    let raw = conn
55        .query_row(&sql, [row_id], |row| {
56            Ok((
57                row.get::<_, String>(0)?,
58                row.get::<_, i64>(1)?,
59                row.get::<_, String>(2)?,
60                row.get::<_, Option<String>>(3)?,
61                row.get::<_, String>(4)?,
62            ))
63        })
64        .optional()
65        .map_err(DbError::from)?;
66    let Some((blob_id, plaintext_size, plaintext_hash, cloud_path, stamp)) = raw else {
67        return Ok(None);
68    };
69    let plaintext_size = u64::try_from(plaintext_size).map_err(|_| {
70        DbError::Message(format!(
71            "winning blob row {:?}/{:?} has negative plaintext size {plaintext_size}",
72            table, row_id
73        ))
74    })?;
75    let plaintext_hash = plaintext_hash.parse().map_err(|error| {
76        DbError::context(
77            format!(
78                "winning blob row {:?}/{:?} has invalid plaintext hash",
79                table, row_id
80            ),
81            error,
82        )
83    })?;
84    Ok(Some(LiveBlobRow {
85        stamp,
86        blob_id,
87        plaintext_size,
88        plaintext_hash,
89        cloud_path,
90    }))
91}
92
93pub(crate) fn blob_row_mismatch(
94    table: &str,
95    row_id: &str,
96    column: &str,
97    row_stamp: &str,
98) -> DbError {
99    DbError::Message(format!(
100        "blob locator does not match winning row values for {:?}/{:?}/{:?} at {:?}",
101        table, row_id, column, row_stamp
102    ))
103}
104
105#[allow(clippy::too_many_arguments)]
106pub(crate) fn validate_live_blob_locator(
107    table: &str,
108    row_id: &str,
109    column: &str,
110    row_stamp: &str,
111    stored: &StoredBlobRef,
112    declaration: &coven_protocol::synced_schema::BlobDecl,
113    row: &LiveBlobRow,
114    live_audience: &RemoteAudience,
115) -> Result<(), DbError> {
116    if !live_blob_locator_matches(stored, declaration, row, live_audience) {
117        return Err(blob_row_mismatch(table, row_id, column, row_stamp));
118    }
119    Ok(())
120}
121
122pub(crate) fn live_blob_locator_matches(
123    stored: &StoredBlobRef,
124    declaration: &coven_protocol::synced_schema::BlobDecl,
125    row: &LiveBlobRow,
126    live_audience: &RemoteAudience,
127) -> bool {
128    let locator = stored.locator();
129    locator.namespace() == declaration.namespace
130        && locator.blob_id() == row.blob_id
131        && locator.plaintext_size() == row.plaintext_size
132        && locator.plaintext_hash() == row.plaintext_hash
133        && &locator.audience() == live_audience
134        && locator
135            .scope()
136            .is_none_or(|scope| scope == &declaration.scope)
137        && locator
138            .cloud_path()
139            .is_none_or(|path| row.cloud_path.as_deref() == Some(path))
140}
141
142/// Index accepted blob provenance independently of whether its row version wins.
143/// The index remains until exact object retirement removes it with the remote record.
144pub(crate) fn record_stored_locator_on(
145    conn: &Connection,
146    stored: &StoredBlobRef,
147) -> Result<(), DbError> {
148    conn.execute(
149        "INSERT INTO blob_locators (remote_object_id, locator_hash) VALUES (?1, ?2)
150         ON CONFLICT(remote_object_id) DO NOTHING",
151        rusqlite::params![
152            remote_object_id(stored.object()).to_string(),
153            stored.locator().locator_hash().to_string(),
154        ],
155    )
156    .map_err(DbError::from)?;
157    validate_stored_locator_on(conn, stored)
158}
159
160pub(crate) fn validate_stored_locator_on(
161    conn: &Connection,
162    expected: &StoredBlobRef,
163) -> Result<(), DbError> {
164    let locator_hash = expected.locator().locator_hash().to_string();
165    let expected_remote_object_id = remote_object_id(expected.object());
166    let stored_locator_hash: String = conn
167        .query_row(
168            "SELECT locator_hash FROM blob_locators WHERE remote_object_id = ?1",
169            [expected_remote_object_id.to_string()],
170            |row| row.get(0),
171        )
172        .map_err(DbError::from)?;
173    if stored_locator_hash != locator_hash {
174        return Err(DbError::Message(format!(
175            "stored blob object {expected_remote_object_id} is indexed under locator {stored_locator_hash}, expected {locator_hash}"
176        )));
177    }
178    let remote = load_remote_object_on(conn, expected_remote_object_id)?;
179    if !remote.is_activated_stored_blob() {
180        return Err(DbError::Message(format!(
181            "stored blob locator {locator_hash} does not reference activated ownership"
182        )));
183    }
184    let locator = carried_blob_locator(
185        &remote,
186        &format!("stored blob locator {locator_hash} is invalid"),
187    )?;
188    let actual = StoredBlobRef::new(locator, remote.object().clone()).map_err(|error| {
189        DbError::context(
190            format!("stored blob reference {locator_hash} is invalid"),
191            error,
192        )
193    })?;
194    if &actual != expected {
195        return Err(DbError::Message(format!(
196            "blob object {expected_remote_object_id} differs from its exact stored reference"
197        )));
198    }
199    Ok(())
200}
201
202pub(crate) fn validate_stored_row_binding_on(
203    conn: &Connection,
204    binding: &RowBlobLocatorBinding,
205    expected_authority: &coven_protocol::audience_package::PackageAudience,
206    expected_remote_object_id: ObjectHash,
207) -> Result<(), DbError> {
208    let (audience_authority, remote_object_id): (String, String) = conn
209        .query_row(
210            "SELECT audience_authority, remote_object_id FROM row_blob_locators
211             WHERE table_name = ?1 AND row_id = ?2 AND column_name = ?3 AND row_stamp = ?4",
212            rusqlite::params![
213                binding.table(),
214                binding.row_id(),
215                binding.column(),
216                binding.row_stamp(),
217            ],
218            |row| Ok((row.get(0)?, row.get(1)?)),
219        )
220        .map_err(DbError::from)?;
221    let actual_authority: coven_protocol::audience_package::PackageAudience =
222        serde_json::from_str(&audience_authority)
223            .map_err(|error| DbError::context("parse stored row blob audience authority", error))?;
224    if &actual_authority != expected_authority
225        || remote_object_id != expected_remote_object_id.to_string()
226    {
227        return Err(DbError::Message(format!(
228            "row blob binding {:?}/{:?}/{:?} at {:?} is already bound to different exact content",
229            binding.table(),
230            binding.row_id(),
231            binding.column(),
232            binding.row_stamp()
233        )));
234    }
235    Ok(())
236}
237
238pub(crate) fn load_prepared_audience_objects_on(
239    conn: &Connection,
240    store_dir: &coven_foundation::store_dir::StoreDir,
241    write_id: &WriteId,
242) -> Result<PreparedAudienceObjects, DbError> {
243    let mut package_statement = conn
244        .prepare(
245            "SELECT remote_object_id FROM store_write_packages
246             WHERE write_id = ?1 ORDER BY audience",
247        )
248        .map_err(DbError::from)?;
249    let package_ids = package_statement
250        .query_map([write_id.as_str()], |row| row.get::<_, String>(0))
251        .map_err(DbError::from)?
252        .collect::<Result<Vec<_>, _>>()
253        .map_err(DbError::from)?;
254    let mut blob_statement = conn
255        .prepare(
256            "SELECT remote_object_id, audience, locator_hash, spool_path FROM store_write_blobs
257             WHERE write_id = ?1 ORDER BY audience, remote_object_id",
258        )
259        .map_err(DbError::from)?;
260    let blob_rows = blob_statement
261        .query_map([write_id.as_str()], |row| {
262            Ok((
263                row.get::<_, String>(0)?,
264                row.get::<_, String>(1)?,
265                row.get::<_, String>(2)?,
266                row.get::<_, Option<String>>(3)?,
267            ))
268        })
269        .map_err(DbError::from)?
270        .collect::<Result<Vec<_>, _>>()
271        .map_err(DbError::from)?;
272    let packages = package_ids
273        .into_iter()
274        .map(|encoded| {
275            let object_id = encoded
276                .parse()
277                .map_err(|error| DbError::context("stored remote object id", error))?;
278            PreparedAudiencePackage::from_remote(
279                conn,
280                store_dir,
281                load_remote_object_on(conn, object_id)?,
282            )
283        })
284        .collect::<Result<Vec<_>, DbError>>()?;
285    let blobs = blob_rows
286        .into_iter()
287        .map(|(encoded, audience, locator_hash, spool_path)| {
288            let object_id = encoded
289                .parse()
290                .map_err(|error| DbError::context("stored remote object id", error))?;
291            PreparedAudienceBlob::from_remote(
292                parse_remote_audience_db(&audience)?,
293                &locator_hash,
294                load_remote_object_on(conn, object_id)?,
295                spool_path.map(PathBuf::from),
296            )
297        })
298        .collect::<Result<Vec<_>, DbError>>()?;
299    Ok(PreparedAudienceObjects { packages, blobs })
300}
301
302pub(crate) fn load_activated_registration_on(
303    conn: &Connection,
304    root: &coven_protocol::store_commit::StoreRootRef,
305    reference: &StoreDeviceRegistrationRef,
306) -> Result<StoreDeviceRegistration, DbError> {
307    let (bytes, encoded): (Vec<u8>, String) = conn
308        .query_row(
309            "SELECT registration_bytes, registration_object \
310             FROM store_device_registration_activations \
311             WHERE device_id = ?1 AND registration_hash = ?2",
312            (
313                reference.device_id.to_string(),
314                reference.registration_hash.to_string(),
315            ),
316            |row| Ok((row.get(0)?, row.get(1)?)),
317        )
318        .map_err(DbError::from)?;
319    let stored: StoreDeviceRegistrationRef = serde_json::from_str(&encoded)
320        .map_err(|error| DbError::context("activated Store registration ref", error))?;
321    if stored != *reference {
322        return Err(DbError::Message(
323            "activated Store registration differs from its exact reference".to_string(),
324        ));
325    }
326    let registration = StoreDeviceRegistration::parse_at(&bytes, root, reference.device_id)
327        .map_err(|error| DbError::context("activated Store registration", error))?;
328    reference
329        .verify_registration(&registration)
330        .map_err(DbError::from)?;
331    Ok(registration)
332}
333
334#[allow(clippy::too_many_arguments)]
335pub(crate) fn previous_row_blob_for_write_on(
336    conn: &Connection,
337    table: &str,
338    row_id: &str,
339    row_stamp: &str,
340    column: &str,
341    blob: &BlobRef,
342    plaintext_size: u64,
343    plaintext_hash: ObjectHash,
344) -> Result<Option<StoreWriteRemoteBlob>, DbError> {
345    if let Some(handoff) =
346        CloudOutboxRecords::new(conn).created_upload_handoff(table, row_id, column, row_stamp)?
347    {
348        let locator = handoff.stored.locator();
349        if !coven_protocol::blob::locator_describes_row(
350            locator,
351            blob,
352            plaintext_size,
353            plaintext_hash,
354        ) {
355            return Err(DbError::Message(format!(
356                "created upload {table}/{row_id}/{column} at {row_stamp} differs from its captured row"
357            )));
358        }
359        return Ok(Some(handoff));
360    }
361    let Some(previous) = bound_row_blob_on(conn, table, row_id, column)? else {
362        return Ok(None);
363    };
364    if !coven_protocol::blob::locator_describes_row(
365        previous.stored.locator(),
366        blob,
367        plaintext_size,
368        plaintext_hash,
369    ) {
370        return Ok(None);
371    }
372    Ok(Some(previous))
373}
374
375/// The exact accepted object already bound to this row, before or after a row
376/// merge changes its timestamp. Consumers compare its content with their row.
377pub(crate) fn bound_row_blob_on(
378    conn: &Connection,
379    table: &str,
380    row_id: &str,
381    column: &str,
382) -> Result<Option<StoreWriteRemoteBlob>, DbError> {
383    let raw = conn
384        .query_row(
385            "SELECT row_blob_locators.audience_authority, blob_locators.remote_object_id
386             FROM row_blob_locators
387             JOIN blob_locators USING (remote_object_id)
388             WHERE table_name = ?1 AND row_id = ?2 AND column_name = ?3
389             ORDER BY row_stamp DESC LIMIT 1",
390            rusqlite::params![table, row_id, column],
391            |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
392        )
393        .optional()
394        .map_err(DbError::from)?;
395    let Some((authority, object_id)) = raw else {
396        return Ok(None);
397    };
398    let authority: coven_protocol::audience_package::PackageAudience =
399        serde_json::from_str(&authority)
400            .map_err(|error| DbError::context("prior row blob authority", error))?;
401    let object_id = object_id
402        .parse()
403        .map_err(|error| DbError::context("prior row blob object id", error))?;
404    let remote = load_remote_object_on(conn, object_id)?;
405    if !remote.is_activated_stored_blob() {
406        return Err(DbError::Message(format!(
407            "prior row blob {table}/{row_id}/{column} is not activated"
408        )));
409    }
410    let locator = carried_blob_locator(&remote, "prior row blob locator")?;
411    if locator.audience() != authority.remote_audience() {
412        return Err(DbError::Message(format!(
413            "prior row blob {table}/{row_id}/{column} authority differs from its locator"
414        )));
415    }
416    let stored = StoredBlobRef::new(locator, remote.object().clone())
417        .map_err(|error| DbError::context("prior row blob reference", error))?;
418    Ok(Some(StoreWriteRemoteBlob { authority, stored }))
419}
420
421pub fn remote_audience_to_db(audience: &RemoteAudience) -> String {
422    match audience {
423        RemoteAudience::Store => "store".to_string(),
424        RemoteAudience::Circle(circle_id) => circle_id.to_string(),
425    }
426}
427
428pub(crate) fn parse_remote_audience_db(value: &str) -> Result<RemoteAudience, DbError> {
429    if value == "store" {
430        return Ok(RemoteAudience::Store);
431    }
432    value
433        .parse()
434        .map(RemoteAudience::Circle)
435        .map_err(|error| DbError::context(format!("invalid stored blob audience {value:?}"), error))
436}
coven_database/blob_records.rs

coven_database/
blob_records.rs
