1use crate::{Database, DatabaseTestSql, DbError};
2use rusqlite::OptionalExtension;
3
4impl Database {
5 pub async fn remove_store_protocol_root_for_test(&self) {
6 self.test_sql(|database| database.remove_store_protocol_root())
7 .await
8 .expect("remove exact Store root authority");
9 }
10
11 pub async fn remove_retained_replay_baseline_for_test(&self) {
12 self.test_sql(|database| {
13 database
14 .execute("DELETE FROM retained_replay_baselines", [])
15 .map(|_| ())
16 .map_err(DbError::from)
17 })
18 .await
19 .expect("remove retained replay baseline");
20 }
21
22 pub async fn tamper_retained_recovery_registration_for_test(
23 &self,
24 reference: &coven_protocol::store_commit::StoreBatchCommitRef,
25 tamper: crate::RetainedRegistrationTamper,
26 ) {
27 let reference = reference.clone();
28 self.test_sql(move |database| {
29 database.tamper_retained_recovery_registration(&reference, tamper)
30 })
31 .await
32 .expect("install tampered retained recovery registration");
33 }
34
35 pub async fn execute_test_sql(&self, sql: &str) {
36 let sql = sql.to_string();
37 self.test_sql(move |database| database.execute_batch(&sql).map_err(DbError::from))
38 .await
39 .unwrap_or_else(|error| panic!("test SQL execution failed: {error}"));
40 }
41
42 pub async fn execute_test_host_write(&self, sql: &str) {
43 let sql = sql.to_string();
44 crate::StoreDatabase::new(self)
45 .run_host_store_write_for_test(None, None, move |transaction| {
46 transaction.execute_batch(&sql).map_err(DbError::from)
47 })
48 .await
49 .unwrap_or_else(|error| panic!("test host write failed: {error}"));
50 }
51
52 pub async fn add_local_photo_for_test(
53 &self,
54 note_id: &str,
55 photo_id: &str,
56 cloud_path: &str,
57 bytes: &[u8],
58 source: &std::path::Path,
59 ) {
60 let note_id = note_id.to_string();
61 let photo_id = photo_id.to_string();
62 let cloud_path = cloud_path.to_string();
63 let size = i64::try_from(bytes.len()).expect("test blob size fits SQLite");
64 let hash = coven_protocol::blob::content_hash(bytes);
65 self.execute_test_host_write(&format!(
66 "INSERT INTO note_photos
67 (id, note_id, kind, size, hash, _updated_at, created_at, cloud_path)
68 VALUES ('{photo_id}', '{note_id}', 'image', {size}, '{hash}',
69 '0000000001000-0000-A', '2026-01-01', '{cloud_path}')"
70 ))
71 .await;
72 crate::StoreDatabase::new(self)
73 .register_external_blob_for_test("note_photos", &photo_id, source)
74 .await;
75 }
76
77 pub async fn insert_local_upload_rows_for_test(
78 &self,
79 root_id: &str,
80 rows: &[(&str, &[u8])],
81 ) -> Result<(), DbError> {
82 let root_id = root_id.to_string();
83 let rows = rows
84 .iter()
85 .map(|(id, bytes)| {
86 (
87 id.to_string(),
88 i64::try_from(bytes.len()).expect("test blob size fits SQLite"),
89 coven_protocol::blob::content_hash(bytes),
90 )
91 })
92 .collect::<Vec<_>>();
93 self.test_sql(move |database| {
94 database
95 .execute(
96 "INSERT INTO notes (id, title, shared, _updated_at, created_at)
97 VALUES (?1, 'upload', 0, '0000000001000-0000-test', '2024-01-01')",
98 [&root_id],
99 )
100 .map_err(DbError::from)?;
101 for (id, size, hash) in rows {
102 database
103 .execute(
104 "INSERT INTO note_photos
105 (id, note_id, kind, size, hash, _updated_at, created_at)
106 VALUES (?1, ?2, 'attach', ?3, ?4,
107 '0000000001000-0000-test', '2024-01-01')",
108 rusqlite::params![id, root_id, size, hash],
109 )
110 .map_err(DbError::from)?;
111 }
112 Ok(())
113 })
114 .await
115 }
116
117 pub async fn seed_stuck_blob_upload_for_test(&self, created_at: &str) -> Result<(), DbError> {
118 let hash = coven_protocol::blob::content_hash(b"x");
119 self.execute_test_sql(&format!(
120 "INSERT INTO notes (id, title, body, shared, _updated_at, created_at) \
121 VALUES ('pending-root', 'Pending', NULL, 0, \
122 '0000000000001-0000-M', '2026-01-01'); \
123 INSERT INTO note_photos \
124 (id, note_id, kind, size, hash, _updated_at, created_at) \
125 VALUES ('pending-blob', 'pending-root', 'cover', 1, '{hash}', \
126 '0000000000001-0000-M', '2026-01-01')"
127 ))
128 .await;
129 let row = crate::StoreDatabase::new(self)
130 .row_blob_ref("note_photos", "pending-blob")
131 .await?;
132 let created_at = created_at.to_string();
133 self.test_sql(move |database| {
134 database.enqueue_blob_upload(
135 "notes",
136 "pending-root",
137 "Pending Root",
138 &row,
139 std::path::Path::new("/nonexistent/pending-blob"),
140 false,
141 &created_at,
142 )
143 })
144 .await
145 }
146
147 pub async fn query_test_text(&self, sql: &str) -> String {
148 let sql = sql.to_string();
149 self.test_sql(move |database| {
150 database
151 .query_row(&sql, [], |row| row.get::<_, String>(0))
152 .map_err(DbError::from)
153 })
154 .await
155 .unwrap_or_else(|error| panic!("test text query failed: {error}"))
156 }
157
158 pub async fn test_row_exists(&self, sql: &str) -> bool {
159 let sql = sql.to_string();
160 self.test_sql(move |database| {
161 database
162 .query_row(&sql, [], |_| Ok(()))
163 .optional()
164 .map(|row| row.is_some())
165 .map_err(DbError::from)
166 })
167 .await
168 .unwrap_or_else(|error| panic!("test row-existence query failed: {error}"))
169 }
170
171 pub async fn capture_test_changeset(&self, statements: &[&str]) -> Vec<u8> {
172 let statements = statements
173 .iter()
174 .map(|statement| statement.to_string())
175 .collect::<Vec<_>>();
176 self.call_store(move |session| session.capture_test_changeset(&statements))
177 .await
178 .unwrap_or_else(|error| panic!("test changeset capture failed: {error}"))
179 }
180
181 pub async fn capture_test_changeset_for_tables(&self, tables: &[&str], sql: &str) -> Vec<u8> {
182 let tables = tables
183 .iter()
184 .map(|table| table.to_string())
185 .collect::<Vec<_>>();
186 let sql = sql.to_string();
187 self.test_sql(move |database| database.capture_changeset(&tables, &[sql]))
188 .await
189 .unwrap_or_else(|error| panic!("raw test changeset capture failed: {error}"))
190 }
191
192 async fn apply_test_changeset_result(
193 &self,
194 bytes: &[u8],
195 ) -> Result<crate::ApplyResult, DbError> {
196 let bytes = bytes.to_vec();
197 self.call_store(move |session| session.apply_test_changeset(&bytes))
198 .await
199 }
200
201 pub async fn try_apply_test_changeset(&self, bytes: &[u8]) -> Result<(), DbError> {
202 self.apply_test_changeset_result(bytes).await.map(|_| ())
203 }
204
205 pub async fn apply_test_changeset(&self, bytes: &[u8]) {
206 self.try_apply_test_changeset(bytes)
207 .await
208 .expect("apply test changeset");
209 }
210
211 pub async fn apply_test_changeset_reporting_foreign_key_violations(
212 &self,
213 bytes: &[u8],
214 ) -> Result<bool, DbError> {
215 self.apply_test_changeset_result(bytes)
216 .await
217 .map(|result| result.had_fk_violations)
218 }
219
220 pub async fn plant_blob_row_for_test(&self, blob_id: &str, remote: bool, bytes: &[u8]) {
221 self.plant_blob_row_with_facts_for_test(
222 blob_id,
223 remote,
224 bytes.len() as u64,
225 Some(&coven_protocol::blob::content_hash(bytes)),
226 )
227 .await;
228 }
229
230 pub async fn plant_blob_row_with_facts_for_test(
231 &self,
232 blob_id: &str,
233 remote: bool,
234 size: u64,
235 hash: Option<&str>,
236 ) {
237 let note = format!("note-{blob_id}");
238 let blob_id = blob_id.to_string();
239 let hash = hash.map(str::to_string);
240 self.test_sql(move |database| {
241 database
242 .execute(
243 "INSERT INTO notes (id, title, shared, _updated_at, created_at) \
244 VALUES (?1, 'read-test', ?2, '0000000001000-0000-dev1', '2026-01-01')",
245 (note.as_str(), remote as i64),
246 )
247 .map_err(DbError::from)?;
248 database
249 .execute(
250 "INSERT INTO note_photos (id, note_id, kind, size, hash, _updated_at, created_at) \
251 VALUES (?1, ?2, 'attach', ?3, ?4, '0000000001000-0000-dev1', '2026-01-01')",
252 rusqlite::params![blob_id.as_str(), note.as_str(), size as i64, hash],
253 )
254 .map_err(DbError::from)?;
255 Ok(())
256 })
257 .await
258 .expect("plant test blob row");
259 }
260
261 pub async fn set_blob_remote_for_test(&self, blob_id: &str, remote: bool) {
262 let note = format!("note-{blob_id}");
263 self.test_sql(move |database| {
264 database
265 .execute(
266 "UPDATE notes SET shared = ?1 WHERE id = ?2",
267 (remote as i64, note.as_str()),
268 )
269 .map(|_| ())
270 .map_err(DbError::from)
271 })
272 .await
273 .expect("change test blob locality");
274 }
275
276 pub async fn run_scoped_host_write_for_test(&self, sql: String) {
277 crate::StoreDatabase::new(self)
278 .run_host_store_write_for_test(
279 Some(coven_keys::encryption::EncryptionService::from_key(
280 [42; 32],
281 )),
282 None,
283 move |transaction| transaction.execute_batch(&sql).map_err(DbError::from),
284 )
285 .await
286 .expect("commit scoped host write");
287 }
288
289 pub async fn scoped_routing_state_for_test(
290 &self,
291 row_id: &str,
292 ) -> crate::ScopedRoutingStateForTest {
293 let row_id = row_id.to_string();
294 self.test_sql(move |database| {
295 let (row, route, mirror) = database.scoped_note_routing_state(&row_id, [42; 32])?;
296 Ok(crate::ScopedRoutingStateForTest { row, route, mirror })
297 })
298 .await
299 .expect("read scoped routing state")
300 }
301
302 pub async fn circle_control_activation_count_for_test(
303 &self,
304 circle_id: coven_protocol::circle::CircleId,
305 ) -> i64 {
306 self.test_sql(move |database| database.circle_control_activation_count(circle_id))
307 .await
308 .expect("count Circle control activations")
309 }
310
311 pub async fn row_blob_binding_count_for_test(&self, row_id: &str) -> i64 {
312 let row_id = row_id.to_string();
313 self.test_sql(move |database| database.row_blob_binding_count(&row_id))
314 .await
315 .expect("count row blob bindings")
316 }
317
318 pub async fn bind_circle_row_blob_for_test(&self, row_id: &str) {
319 let row_id = row_id.to_string();
320 let object_id = "0".repeat(64);
321 self.test_sql(move |database| {
322 database.install_blob_binding(
323 &object_id,
324 "{}",
325 &"1".repeat(64),
326 "notes",
327 &row_id,
328 "attachment",
329 "0000000002000-0000-owner",
330 "{}",
331 )
332 })
333 .await
334 .expect("bind Circle row blob");
335 }
336
337 pub async fn table_has_rows_for_test(
338 &self,
339 table: crate::DatabaseTestTable,
340 ) -> Result<bool, DbError> {
341 self.test_sql(move |database| database.table_has_rows(table))
342 .await
343 }
344
345 pub async fn store_partition_changesets_for_test(&self) -> Result<Vec<Vec<u8>>, DbError> {
346 self.test_sql(|database| database.store_partition_changesets())
347 .await
348 }
349
350 pub async fn has_store_partition_for_test(&self) -> Result<bool, DbError> {
351 self.test_sql(|database| database.has_store_partition())
352 .await
353 }
354
355 pub async fn delete_make_remote_intent_for_test(
356 &self,
357 root_table: &str,
358 root_id: &str,
359 ) -> Result<(), DbError> {
360 let root_table = root_table.to_string();
361 let root_id = root_id.to_string();
362 self.test_sql(move |database| database.delete_make_remote_intent(&root_table, &root_id))
363 .await
364 }
365
366 pub async fn make_remote_intent_exists_for_test(
367 &self,
368 root_table: &str,
369 root_id: &str,
370 ) -> Result<bool, DbError> {
371 let root_table = root_table.to_string();
372 let root_id = root_id.to_string();
373 self.test_sql(move |database| database.make_remote_intent_exists(&root_table, &root_id))
374 .await
375 }
376
377 pub async fn published_blob_drop_intent_exists_for_test(
378 &self,
379 blob_id: &str,
380 ) -> Result<bool, DbError> {
381 let blob_id = blob_id.to_string();
382 self.test_sql(move |database| database.published_blob_drop_intent_exists(&blob_id))
383 .await
384 }
385
386 pub async fn insert_published_blob_drop_intent_for_test(
387 &self,
388 sequence: u64,
389 namespace: &str,
390 blob_id: &str,
391 bytes: &[u8],
392 locator_hash: coven_protocol::store_commit::ObjectHash,
393 disposition: coven_protocol::blob::DeferredLocalBlobDisposition,
394 ) -> Result<(), DbError> {
395 let drop = coven_protocol::blob::DeferredLocalBlobDrop {
396 namespace: namespace.to_string(),
397 id: blob_id.to_string(),
398 size: bytes.len() as u64,
399 plaintext_hash: coven_protocol::store_commit::ObjectHash::digest(bytes),
400 locator_hash,
401 disposition,
402 };
403 self.test_sql(move |database| database.insert_published_blob_drop_intent(sequence, &drop))
404 .await
405 }
406
407 pub async fn remote_object_for_test(
408 &self,
409 object: coven_protocol::objects::ExactObjectRef,
410 ) -> Result<coven_protocol::remote_object::RemoteObjectRecord, DbError> {
411 self.test_sql(move |database| database.remote_object(&object))
412 .await
413 }
414
415 pub async fn retained_store_package_pin_for_test(
416 &self,
417 commit: &coven_protocol::store_commit::StoreBatchCommitRef,
418 ) -> Result<
419 (
420 coven_protocol::remote_object::RetainedReplayOwner,
421 coven_protocol::store_commit::StorePackageRef,
422 coven_protocol::remote_object::RemoteObjectRecord,
423 ),
424 DbError,
425 > {
426 let stream_id = commit.coord.stream_id.to_string();
427 let sequence = commit.coord.sequence();
428 let (input_hash, canonical_input) = self
429 .test_sql(move |database| database.retained_merge_input(&stream_id, sequence))
430 .await?;
431 let retained: serde_json::Value = serde_json::from_slice(&canonical_input)
432 .map_err(|error| DbError::context("parse retained package input", error))?;
433 let reference: coven_protocol::store_commit::StorePackageRef = serde_json::from_value(
434 retained["packages"][0]["store"]["reference"].clone(),
435 )
436 .map_err(|error| DbError::context("parse retained Store package reference", error))?;
437 let remote = self
438 .remote_object_for_test(reference.object.clone())
439 .await?;
440 Ok((
441 coven_protocol::remote_object::RetainedReplayOwner::Commit {
442 commit: commit.clone(),
443 input_hash,
444 },
445 reference,
446 remote,
447 ))
448 }
449
450 pub async fn remote_objects_for_test(
451 &self,
452 ) -> Result<Vec<coven_protocol::remote_object::RemoteObjectRecord>, DbError> {
453 self.test_sql(|database| database.remote_objects()).await
454 }
455
456 pub async fn remote_object_exists_for_test(
457 &self,
458 object: coven_protocol::objects::ExactObjectRef,
459 ) -> Result<bool, DbError> {
460 self.test_sql(move |database| database.remote_object_exists(&object))
461 .await
462 }
463
464 pub async fn remote_object_id_exists_for_test(
465 &self,
466 object_id: coven_protocol::store_commit::ObjectHash,
467 ) -> Result<bool, DbError> {
468 self.test_sql(move |database| database.remote_object_id_exists(object_id))
469 .await
470 }
471
472 pub async fn replace_remote_object_for_test(
473 &self,
474 object: coven_protocol::objects::ExactObjectRef,
475 remote: coven_protocol::remote_object::RemoteObjectRecord,
476 ) -> Result<(), DbError> {
477 self.test_sql(move |database| database.replace_remote_object(&object, &remote))
478 .await
479 }
480
481 pub async fn delete_remote_object_for_test(
482 &self,
483 object: coven_protocol::objects::ExactObjectRef,
484 ) -> Result<(), DbError> {
485 self.test_sql(move |database| database.delete_remote_object(&object))
486 .await
487 }
488
489 pub async fn enqueue_blob_delete_for_test(
490 &self,
491 stored: &coven_protocol::blob::locator::StoredBlobRef,
492 created_at: &str,
493 ) -> Result<(), DbError> {
494 let stored = stored.clone();
495 let created_at = created_at.to_string();
496 self.test_sql(move |database| database.enqueue_blob_delete(&stored, &created_at))
497 .await
498 }
499
500 pub async fn delete_outbox_attempt_for_test(
501 &self,
502 id: i64,
503 ) -> Result<Option<crate::OutboxAttempt>, DbError> {
504 self.test_sql(move |database| database.delete_outbox_attempt(id))
505 .await
506 }
507
508 pub async fn insert_local_blob_row_for_test(
509 &self,
510 root_id: &str,
511 row_id: &str,
512 blob_id: &str,
513 cloud_path: Option<&str>,
514 bytes: &[u8],
515 ) -> Result<(), crate::HostWriteError<DbError>> {
516 let root_id = root_id.to_string();
517 let row_id = row_id.to_string();
518 let blob_id = blob_id.to_string();
519 let cloud_path = cloud_path.map(str::to_string);
520 let size = i64::try_from(bytes.len()).expect("test blob size fits SQLite");
521 let hash = coven_protocol::store_commit::ObjectHash::digest(bytes).to_string();
522 crate::StoreDatabase::new(self)
523 .run_host_store_write_for_test(None, None, move |transaction| {
524 transaction
525 .execute(
526 "INSERT INTO notes
527 (id, title, body, shared, _updated_at, created_at)
528 VALUES (?1, 'blob root', NULL, 0, '0000000001000-0000-dev1', '2026-01-01')",
529 [root_id.as_str()],
530 )
531 .map_err(DbError::from)?;
532 transaction
533 .execute(
534 "INSERT INTO note_photos
535 (id, note_id, kind, size, hash, cloud_path, blob_id, _updated_at, created_at)
536 VALUES (?1, ?2, 'cover', ?3, ?4, ?5, ?6,
537 '0000000001000-0000-dev1', '2026-01-01')",
538 rusqlite::params![row_id, root_id, size, hash, cloud_path, blob_id],
539 )
540 .map_err(DbError::from)?;
541 Ok(())
542 })
543 .await
544 .map(|_| ())
545 }
546
547 pub async fn capture_circle_document_for_test(
548 &self,
549 row_id: &str,
550 circle_id: coven_protocol::circle::CircleId,
551 stamp: &str,
552 ) -> Result<coven_protocol::write::WriteId, crate::HostWriteError<DbError>> {
553 let routing = coven_keys::encryption::EncryptionService::from_key([42; 32]);
554 let audience_value = circle_id.to_string();
555 let row_id = row_id.to_string();
556 let stamp = stamp.to_string();
557 let receipt = crate::StoreDatabase::new(self)
558 .run_host_store_write_for_test(Some(routing), None, move |transaction| {
559 transaction
560 .execute(
561 "INSERT INTO documents (id, audience, _updated_at)
562 VALUES (?1, ?2, ?3)",
563 rusqlite::params![row_id, audience_value, stamp],
564 )
565 .map(|_| ())
566 .map_err(DbError::from)
567 })
568 .await?;
569 Ok(receipt.write_id)
570 }
571
572 pub async fn circle_document_present_for_test(&self, row_id: &str) -> Result<bool, DbError> {
573 let row_id = row_id.to_string();
574 self.test_sql(move |database| {
575 database
576 .query_row(
577 "SELECT EXISTS(SELECT 1 FROM documents WHERE id = ?1)",
578 [row_id],
579 |row| row.get::<_, bool>(0),
580 )
581 .map_err(DbError::from)
582 })
583 .await
584 }
585
586 pub async fn local_store_device_id_for_test(
587 &self,
588 ) -> Result<coven_protocol::store_commit::StoreDeviceId, DbError> {
589 self.get_protocol_state(crate::LOCAL_DEVICE_ID_STATE_KEY)
590 .await?
591 .ok_or_else(|| DbError::Message("local device id is not installed".to_string()))?
592 .parse()
593 .map_err(|error| DbError::context("parse local device id", error))
594 }
595
596 pub async fn capture_document_for_test(
597 &self,
598 row_id: &str,
599 audience: Option<coven_protocol::circle::CircleId>,
600 stamp: &str,
601 ) -> Result<coven_protocol::write::WriteId, crate::HostWriteError<DbError>> {
602 let routing = coven_keys::encryption::EncryptionService::from_key([42; 32]);
603 let audience = audience.map(|circle_id| circle_id.to_string());
604 let row_id = row_id.to_string();
605 let stamp = stamp.to_string();
606 let receipt = crate::StoreDatabase::new(self)
607 .run_host_store_write_for_test(Some(routing), None, move |transaction| {
608 transaction
609 .execute(
610 "INSERT INTO documents (id, audience, _updated_at)
611 VALUES (?1, ?2, ?3)",
612 rusqlite::params![row_id, audience, stamp],
613 )
614 .map(|_| ())
615 .map_err(DbError::from)
616 })
617 .await?;
618 Ok(receipt.write_id)
619 }
620
621 pub async fn capture_document_with_file_for_test(
622 &self,
623 document_id: &str,
624 file_id: &str,
625 audience: Option<coven_protocol::circle::CircleId>,
626 bytes: &[u8],
627 stamp: &str,
628 ) -> Result<coven_protocol::write::WriteId, crate::HostWriteError<DbError>> {
629 let routing = coven_keys::encryption::EncryptionService::from_key([42; 32]);
630 let document_id = document_id.to_string();
631 let file_id = file_id.to_string();
632 let audience = audience.map(|circle_id| circle_id.to_string());
633 let size = i64::try_from(bytes.len()).expect("test blob size fits SQLite");
634 let hash = coven_protocol::blob::content_hash(bytes);
635 let stamp = stamp.to_string();
636 let receipt = crate::StoreDatabase::new(self)
637 .run_host_store_write_for_test(Some(routing), None, move |transaction| {
638 transaction
639 .execute(
640 "INSERT INTO documents (id, audience, _updated_at)
641 VALUES (?1, ?2, ?3)",
642 rusqlite::params![document_id, audience, stamp],
643 )
644 .map_err(DbError::from)?;
645 transaction
646 .execute(
647 "INSERT INTO document_files
648 (id, document_id, size, hash, _updated_at)
649 VALUES (?1, ?2, ?3, ?4, ?5)",
650 rusqlite::params![file_id, document_id, size, hash, stamp],
651 )
652 .map(|_| ())
653 .map_err(DbError::from)
654 })
655 .await?;
656 Ok(receipt.write_id)
657 }
658
659 pub async fn document_file_stamp_for_test(&self, file_id: &str) -> Result<String, DbError> {
660 let file_id = file_id.to_string();
661 self.test_sql(move |database| {
662 database
663 .query_row(
664 "SELECT _updated_at FROM document_files WHERE id = ?1",
665 [file_id],
666 |row| row.get::<_, String>(0),
667 )
668 .map_err(DbError::from)
669 })
670 .await
671 }
672
673 pub async fn release_retained_replay_ownership_for_test(&self) -> Result<(), DbError> {
674 self.test_sql(|database| {
675 database.transaction(|transaction| {
676 transaction.remove_retained_replay_ownership_from_snapshot()
677 })
678 })
679 .await
680 }
681
682 pub async fn insert_browsable_blob_row_for_test(
683 &self,
684 blob_id: &str,
685 cloud_path: &str,
686 bytes: &[u8],
687 ) -> Result<(), DbError> {
688 let note = format!("note-{blob_id}");
689 let blob_id = blob_id.to_string();
690 let cloud_path = cloud_path.to_string();
691 let size = i64::try_from(bytes.len()).expect("test blob size fits SQLite");
692 let hash = coven_protocol::blob::content_hash(bytes);
693 self.test_sql(move |database| {
694 database
695 .execute(
696 "INSERT INTO notes (id, title, shared, _updated_at, created_at)
697 VALUES (?1, 'browsable-test', 1, '0000000001000-0000-dev1', '2026-01-01')",
698 [note.as_str()],
699 )
700 .map_err(DbError::from)?;
701 database
702 .execute(
703 "INSERT INTO note_photos
704 (id, note_id, kind, size, hash, _updated_at, created_at)
705 VALUES (?1, ?2, ?3, ?4, ?5,
706 '0000000001000-0000-dev1', '2026-01-01')",
707 rusqlite::params![blob_id, note, cloud_path, size, hash],
708 )
709 .map_err(DbError::from)?;
710 Ok(())
711 })
712 .await
713 }
714
715 pub async fn bind_stored_blob_to_row_for_test(
716 &self,
717 stored: &coven_protocol::blob::locator::StoredBlobRef,
718 table: &str,
719 id: &str,
720 owner: coven_protocol::store_commit::StoreBatchCommitRef,
721 ) -> Result<(), DbError> {
722 let locator = stored.locator().clone();
723 let record =
724 coven_protocol::remote_object::RemoteObjectRecord::activated_blob(stored, owner)
725 .map_err(DbError::from)?
726 .into_record();
727 let object_id = record.object_id().to_string();
728 let state = serde_json::to_string(&record).map_err(DbError::from)?;
729 let locator_hash = locator.locator_hash().to_string();
730 let authority =
731 serde_json::to_string(&coven_protocol::audience_package::PackageAudience::Store)
732 .map_err(DbError::from)?;
733 let id_for_insert = id.to_string();
734 let table_for_insert = table.to_string();
735 let stamp_table = table.to_string();
736 let stamp_id = id.to_string();
737 self.test_sql(move |database| {
738 let row_stamp = database
739 .query_row(
740 &format!(
741 "SELECT _updated_at FROM {} WHERE id = ?1",
742 crate::quote_ident(&stamp_table)
743 ),
744 [stamp_id],
745 |row| row.get::<_, String>(0),
746 )
747 .map_err(DbError::from)?;
748 database.install_blob_binding(
749 &object_id,
750 &state,
751 &locator_hash,
752 &table_for_insert,
753 &id_for_insert,
754 "id",
755 &row_stamp,
756 &authority,
757 )
758 })
759 .await
760 }
761
762 pub async fn store_package_is_retained_for_replay_for_test(
763 &self,
764 package: coven_protocol::store_commit::StorePackageRef,
765 activation: coven_protocol::store_commit::StoreBatchCommitRef,
766 ) -> Result<bool, DbError> {
767 let database = crate::StoreDatabase::new(self);
768 let root = database
769 .local_store_root_ref()
770 .await?
771 .ok_or_else(|| DbError::Message("test Store root is not installed".to_string()))?;
772 database
773 .store_package_is_retained_for_replay(root, package, activation)
774 .await
775 }
776
777 pub async fn circle_state_counts_for_test(
778 &self,
779 circle_id: coven_protocol::circle::CircleId,
780 ) -> Result<(i64, i64, i64), DbError> {
781 self.test_sql(move |database| database.circle_state_counts(circle_id))
782 .await
783 }
784
785 pub async fn upload_outbox_attempt_for_test(
786 &self,
787 row_id: &str,
788 ) -> Result<Option<crate::OutboxAttempt>, DbError> {
789 let row_id = row_id.to_string();
790 self.test_sql(move |database| database.upload_outbox_attempt(&row_id))
791 .await
792 }
793
794 pub async fn corrupt_upload_outbox_attempt_time_for_test(
795 &self,
796 id: i64,
797 ) -> Result<(), DbError> {
798 self.test_sql(move |database| database.corrupt_upload_outbox_attempt_time(id))
799 .await
800 }
801
802 pub async fn corrupt_delete_outbox_attempt_time_for_test(
803 &self,
804 id: i64,
805 ) -> Result<(), DbError> {
806 self.test_sql(move |database| database.corrupt_delete_outbox_attempt_time(id))
807 .await
808 }
809
810 #[allow(clippy::too_many_arguments)]
811 pub async fn enqueue_blob_upload_with_retention_for_test(
812 &self,
813 root_table: &str,
814 root_id: &str,
815 row: coven_protocol::blob::RowBlobRef,
816 source_path: std::path::PathBuf,
817 retain_pinned: bool,
818 created_at: &str,
819 ) -> Result<(), DbError> {
820 let root_table = root_table.to_string();
821 let root_id = root_id.to_string();
822 let root_label = format!("{root_table}/{root_id}");
823 let created_at = created_at.to_string();
824 self.test_sql(move |database| {
825 database.enqueue_blob_upload(
826 &root_table,
827 &root_id,
828 &root_label,
829 &row,
830 &source_path,
831 retain_pinned,
832 &created_at,
833 )
834 })
835 .await
836 }
837
838 pub async fn roll_back_blob_upload_for_test(
839 &self,
840 root_table: &str,
841 root_id: &str,
842 row: coven_protocol::blob::RowBlobRef,
843 source_path: std::path::PathBuf,
844 created_at: &str,
845 ) -> Result<(), DbError> {
846 let root_table = root_table.to_string();
847 let root_id = root_id.to_string();
848 let root_label = format!("{root_table}/{root_id}");
849 let created_at = created_at.to_string();
850 self.test_sql(move |database| {
851 database.rolled_back_transaction(|transaction| {
852 transaction.enqueue_blob_upload(
853 &root_table,
854 &root_id,
855 &root_label,
856 &row,
857 &source_path,
858 false,
859 &created_at,
860 )
861 })
862 })
863 .await
864 }
865
866 pub async fn published_blob_drop_intent_count_for_test(
867 &self,
868 sequence: i64,
869 namespace: &str,
870 blob_id: &str,
871 ) -> Result<i64, DbError> {
872 let namespace = namespace.to_string();
873 let blob_id = blob_id.to_string();
874 self.test_sql(move |database| {
875 database.published_blob_drop_intent_count(sequence, &namespace, &blob_id)
876 })
877 .await
878 }
879
880 pub async fn first_published_blob_drop_intent_for_test(
881 &self,
882 namespace: &str,
883 blob_id: &str,
884 ) -> Result<(i64, coven_protocol::blob::DeferredLocalBlobDisposition), DbError> {
885 let namespace = namespace.to_string();
886 let blob_id = blob_id.to_string();
887 self.test_sql(move |database| {
888 let (sequence, disposition): (i64, String) = database
889 .query_row(
890 "SELECT seq, disposition FROM published_blob_drop_intents
891 WHERE namespace = ?1 AND blob_id = ?2
892 ORDER BY seq LIMIT 1",
893 (&namespace, &blob_id),
894 |row| Ok((row.get(0)?, row.get(1)?)),
895 )
896 .map_err(DbError::from)?;
897 let disposition =
898 coven_protocol::blob::DeferredLocalBlobDisposition::from_db(&disposition)
899 .map_err(|error| DbError::Message(error.to_string()))?;
900 Ok((sequence, disposition))
901 })
902 .await
903 }
904
905 pub async fn scoped_store_state_counts_for_test(&self) -> Result<[i64; 4], DbError> {
906 self.test_sql(|database| database.scoped_store_state_counts())
907 .await
908 }
909
910 pub async fn install_make_local_commit_failure_for_test(&self) -> Result<(), DbError> {
911 self.test_sql(|database| {
912 database
913 .execute_batch(
914 "CREATE TRIGGER reject_make_local_gate_update
915 BEFORE UPDATE OF shared ON notes
916 WHEN NEW.id = 'n1' AND NEW.shared = 0
917 BEGIN
918 SELECT RAISE(ABORT, 'forced make_local commit failure');
919 END;",
920 )
921 .map_err(DbError::from)
922 })
923 .await
924 }
925
926 pub async fn store_device_registration_activation_for_test(
927 &self,
928 device_id: &str,
929 ) -> Result<coven_protocol::store_commit::StoreDeviceRegistrationActivation, DbError> {
930 let device_id = device_id.to_string();
931 self.test_sql(move |database| database.store_device_registration_activation(&device_id))
932 .await
933 }
934
935 pub async fn latest_published_store_snapshot_for_test(
936 &self,
937 ) -> Result<(i64, Vec<u8>), DbError> {
938 self.test_sql(|database| database.latest_published_store_snapshot())
939 .await
940 }
941
942 pub async fn latest_published_store_snapshot_bytes_for_test(&self) -> Result<Vec<u8>, DbError> {
943 self.test_sql(|database| database.latest_published_store_snapshot_bytes())
944 .await
945 }
946
947 pub async fn materialized_commits_without_device_state_count_for_test(
948 &self,
949 ) -> Result<i64, DbError> {
950 self.test_sql(|database| database.materialized_commits_without_device_state_count())
951 .await
952 }
953
954 pub async fn store_device_state_snapshot_refs_for_test(
955 &self,
956 ) -> Result<Vec<coven_protocol::store_commit::StoreBatchCommitRef>, DbError> {
957 self.test_sql(|database| database.store_device_state_snapshot_refs())
958 .await
959 }
960
961 pub async fn restored_row_graph_counts_for_test(
962 &self,
963 ) -> Result<(i64, i64, i64, i64), DbError> {
964 self.test_sql(|database| {
965 Ok((
966 database.query_row("SELECT COUNT(*) FROM notes WHERE id = 'n1'", [], |row| {
967 row.get::<_, i64>(0)
968 })?,
969 database.query_row(
970 "SELECT COUNT(*) FROM note_photos WHERE id = 'photo1'",
971 [],
972 |row| row.get::<_, i64>(0),
973 )?,
974 database.query_row(
975 "SELECT COUNT(*) FROM note_photos AS photo
976 JOIN notes AS note ON note.id = photo.note_id
977 WHERE photo.id = 'photo1' AND note.id = 'n1'",
978 [],
979 |row| row.get::<_, i64>(0),
980 )?,
981 database.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |row| {
982 row.get::<_, i64>(0)
983 })?,
984 ))
985 })
986 .await
987 }
988
989 pub(super) async fn test_sql<F, R>(&self, operation: F) -> Result<R, DbError>
990 where
991 F: for<'connection> FnOnce(DatabaseTestSql<'connection>) -> Result<R, DbError>
992 + Send
993 + 'static,
994 R: Send + 'static,
995 {
996 self.call_database(move |session| session.run_test_sql(operation))
997 .await
998 }
999}