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 private_routing_state_for_test(&self) -> Result<String, DbError> {
290 self.test_sql(|database| {
291 database.query_row(
292 "SELECT json_array(
293 (SELECT json_group_array(json_array(routing_id, table_name, row_id, _updated_at))
294 FROM (SELECT * FROM _coven_row_routes ORDER BY routing_id)),
295 (SELECT json_group_array(json_array(routing_id, circle_id, _updated_at))
296 FROM (SELECT * FROM _coven_audience ORDER BY routing_id)))",
297 [],
298 |row| row.get(0),
299 ).map_err(DbError::from)
300 }).await
301 }
302
303 pub async fn scoped_routing_state_for_test(
304 &self,
305 row_id: &str,
306 ) -> crate::ScopedRoutingStateForTest {
307 let row_id = row_id.to_string();
308 self.test_sql(move |database| {
309 let (row, route, mirror) = database.scoped_note_routing_state(&row_id, [42; 32])?;
310 Ok(crate::ScopedRoutingStateForTest { row, route, mirror })
311 })
312 .await
313 .expect("read scoped routing state")
314 }
315
316 pub async fn circle_control_activation_count_for_test(
317 &self,
318 circle_id: coven_protocol::circle::CircleId,
319 ) -> i64 {
320 self.test_sql(move |database| database.circle_control_activation_count(circle_id))
321 .await
322 .expect("count Circle control activations")
323 }
324
325 pub async fn row_blob_binding_count_for_test(&self, row_id: &str) -> i64 {
326 let row_id = row_id.to_string();
327 self.test_sql(move |database| database.row_blob_binding_count(&row_id))
328 .await
329 .expect("count row blob bindings")
330 }
331
332 pub async fn table_has_rows_for_test(
333 &self,
334 table: crate::DatabaseTestTable,
335 ) -> Result<bool, DbError> {
336 self.test_sql(move |database| database.table_has_rows(table))
337 .await
338 }
339
340 pub async fn store_partition_changesets_for_test(&self) -> Result<Vec<Vec<u8>>, DbError> {
341 self.test_sql(|database| database.store_partition_changesets())
342 .await
343 }
344
345 pub async fn has_store_partition_for_test(&self) -> Result<bool, DbError> {
346 self.test_sql(|database| database.has_store_partition())
347 .await
348 }
349
350 pub async fn delete_make_remote_intent_for_test(
351 &self,
352 root_table: &str,
353 root_id: &str,
354 ) -> Result<(), DbError> {
355 let root_table = root_table.to_string();
356 let root_id = root_id.to_string();
357 self.test_sql(move |database| database.delete_make_remote_intent(&root_table, &root_id))
358 .await
359 }
360
361 pub async fn make_remote_intent_exists_for_test(
362 &self,
363 root_table: &str,
364 root_id: &str,
365 ) -> Result<bool, DbError> {
366 let root_table = root_table.to_string();
367 let root_id = root_id.to_string();
368 self.test_sql(move |database| database.make_remote_intent_exists(&root_table, &root_id))
369 .await
370 }
371
372 pub async fn published_blob_drop_intent_exists_for_test(
373 &self,
374 blob_id: &str,
375 ) -> Result<bool, DbError> {
376 let blob_id = blob_id.to_string();
377 self.test_sql(move |database| database.published_blob_drop_intent_exists(&blob_id))
378 .await
379 }
380
381 pub async fn insert_published_blob_drop_intent_for_test(
382 &self,
383 sequence: u64,
384 namespace: &str,
385 blob_id: &str,
386 bytes: &[u8],
387 locator_hash: coven_protocol::store_commit::ObjectHash,
388 disposition: coven_protocol::blob::DeferredLocalBlobDisposition,
389 ) -> Result<(), DbError> {
390 let drop = coven_protocol::blob::DeferredLocalBlobDrop {
391 namespace: namespace.to_string(),
392 id: blob_id.to_string(),
393 size: bytes.len() as u64,
394 plaintext_hash: coven_protocol::store_commit::ObjectHash::digest(bytes),
395 locator_hash,
396 disposition,
397 };
398 self.test_sql(move |database| database.insert_published_blob_drop_intent(sequence, &drop))
399 .await
400 }
401
402 pub async fn remote_object_for_test(
403 &self,
404 object: coven_protocol::objects::ExactObjectRef,
405 ) -> Result<coven_protocol::remote_object::RemoteObjectRecord, DbError> {
406 self.test_sql(move |database| database.remote_object(&object))
407 .await
408 }
409
410 pub async fn retained_store_package_pin_for_test(
411 &self,
412 commit: &coven_protocol::store_commit::StoreBatchCommitRef,
413 ) -> Result<
414 (
415 coven_protocol::remote_object::RetainedReplayOwner,
416 coven_protocol::store_commit::StorePackageRef,
417 coven_protocol::remote_object::RemoteObjectRecord,
418 ),
419 DbError,
420 > {
421 let stream_id = commit.coord.stream_id.to_string();
422 let sequence = commit.coord.sequence();
423 let (input_hash, canonical_input) = self
424 .test_sql(move |database| database.retained_merge_input(&stream_id, sequence))
425 .await?;
426 let retained: serde_json::Value = serde_json::from_slice(&canonical_input)
427 .map_err(|error| DbError::context("parse retained package input", error))?;
428 let reference: coven_protocol::store_commit::StorePackageRef = serde_json::from_value(
429 retained["packages"][0]["store"]["reference"].clone(),
430 )
431 .map_err(|error| DbError::context("parse retained Store package reference", error))?;
432 let remote = self
433 .remote_object_for_test(reference.object.clone())
434 .await?;
435 Ok((
436 coven_protocol::remote_object::RetainedReplayOwner::Commit {
437 commit: commit.clone(),
438 input_hash,
439 },
440 reference,
441 remote,
442 ))
443 }
444
445 pub async fn remote_objects_for_test(
446 &self,
447 ) -> Result<Vec<coven_protocol::remote_object::RemoteObjectRecord>, DbError> {
448 self.test_sql(|database| database.remote_objects()).await
449 }
450
451 pub async fn remote_object_exists_for_test(
452 &self,
453 object: coven_protocol::objects::ExactObjectRef,
454 ) -> Result<bool, DbError> {
455 self.test_sql(move |database| database.remote_object_exists(&object))
456 .await
457 }
458
459 pub async fn remote_object_id_exists_for_test(
460 &self,
461 object_id: coven_protocol::store_commit::ObjectHash,
462 ) -> Result<bool, DbError> {
463 self.test_sql(move |database| database.remote_object_id_exists(object_id))
464 .await
465 }
466
467 pub async fn replace_remote_object_for_test(
468 &self,
469 object: coven_protocol::objects::ExactObjectRef,
470 remote: coven_protocol::remote_object::RemoteObjectRecord,
471 ) -> Result<(), DbError> {
472 self.test_sql(move |database| database.replace_remote_object(&object, &remote))
473 .await
474 }
475
476 pub async fn delete_remote_object_for_test(
477 &self,
478 object: coven_protocol::objects::ExactObjectRef,
479 ) -> Result<(), DbError> {
480 self.test_sql(move |database| database.delete_remote_object(&object))
481 .await
482 }
483
484 pub async fn enqueue_blob_delete_for_test(
485 &self,
486 stored: &coven_protocol::blob::locator::StoredBlobRef,
487 created_at: &str,
488 ) -> Result<(), DbError> {
489 let stored = stored.clone();
490 let created_at = created_at.to_string();
491 self.test_sql(move |database| database.enqueue_blob_delete(&stored, &created_at))
492 .await
493 }
494
495 pub async fn delete_outbox_attempt_for_test(
496 &self,
497 id: i64,
498 ) -> Result<Option<crate::OutboxAttempt>, DbError> {
499 self.test_sql(move |database| database.delete_outbox_attempt(id))
500 .await
501 }
502
503 pub async fn insert_local_blob_row_for_test(
504 &self,
505 root_id: &str,
506 row_id: &str,
507 blob_id: &str,
508 cloud_path: Option<&str>,
509 bytes: &[u8],
510 ) -> Result<(), crate::HostWriteError<DbError>> {
511 let root_id = root_id.to_string();
512 let row_id = row_id.to_string();
513 let blob_id = blob_id.to_string();
514 let cloud_path = cloud_path.map(str::to_string);
515 let size = i64::try_from(bytes.len()).expect("test blob size fits SQLite");
516 let hash = coven_protocol::store_commit::ObjectHash::digest(bytes).to_string();
517 crate::StoreDatabase::new(self)
518 .run_host_store_write_for_test(None, None, move |transaction| {
519 transaction
520 .execute(
521 "INSERT INTO notes
522 (id, title, body, shared, _updated_at, created_at)
523 VALUES (?1, 'blob root', NULL, 0, '0000000001000-0000-dev1', '2026-01-01')",
524 [root_id.as_str()],
525 )
526 .map_err(DbError::from)?;
527 transaction
528 .execute(
529 "INSERT INTO note_photos
530 (id, note_id, kind, size, hash, cloud_path, blob_id, _updated_at, created_at)
531 VALUES (?1, ?2, 'cover', ?3, ?4, ?5, ?6,
532 '0000000001000-0000-dev1', '2026-01-01')",
533 rusqlite::params![row_id, root_id, size, hash, cloud_path, blob_id],
534 )
535 .map_err(DbError::from)?;
536 Ok(())
537 })
538 .await
539 .map(|_| ())
540 }
541
542 pub async fn capture_circle_document_for_test(
543 &self,
544 row_id: &str,
545 circle_id: coven_protocol::circle::CircleId,
546 stamp: &str,
547 ) -> Result<coven_protocol::write::WriteId, crate::HostWriteError<DbError>> {
548 let routing = coven_keys::encryption::EncryptionService::from_key([42; 32]);
549 let audience_value = circle_id.to_string();
550 let row_id = row_id.to_string();
551 let stamp = stamp.to_string();
552 let receipt = crate::StoreDatabase::new(self)
553 .run_host_store_write_for_test(Some(routing), None, move |transaction| {
554 transaction
555 .execute(
556 "INSERT INTO documents (id, audience, _updated_at)
557 VALUES (?1, ?2, ?3)",
558 rusqlite::params![row_id, audience_value, stamp],
559 )
560 .map(|_| ())
561 .map_err(DbError::from)
562 })
563 .await?;
564 Ok(receipt.write_id)
565 }
566
567 pub async fn circle_document_present_for_test(&self, row_id: &str) -> Result<bool, DbError> {
568 let row_id = row_id.to_string();
569 self.test_sql(move |database| {
570 database
571 .query_row(
572 "SELECT EXISTS(SELECT 1 FROM documents WHERE id = ?1)",
573 [row_id],
574 |row| row.get::<_, bool>(0),
575 )
576 .map_err(DbError::from)
577 })
578 .await
579 }
580
581 pub async fn local_store_device_id_for_test(
582 &self,
583 ) -> Result<coven_protocol::store_commit::StoreDeviceId, DbError> {
584 self.get_protocol_state(crate::LOCAL_DEVICE_ID_STATE_KEY)
585 .await?
586 .ok_or_else(|| DbError::Message("local device id is not installed".to_string()))?
587 .parse()
588 .map_err(|error| DbError::context("parse local device id", error))
589 }
590
591 pub async fn capture_document_for_test(
592 &self,
593 row_id: &str,
594 audience: Option<coven_protocol::circle::CircleId>,
595 stamp: &str,
596 ) -> Result<coven_protocol::write::WriteId, crate::HostWriteError<DbError>> {
597 let routing = coven_keys::encryption::EncryptionService::from_key([42; 32]);
598 let audience = audience.map(|circle_id| circle_id.to_string());
599 let row_id = row_id.to_string();
600 let stamp = stamp.to_string();
601 let receipt = crate::StoreDatabase::new(self)
602 .run_host_store_write_for_test(Some(routing), None, move |transaction| {
603 transaction
604 .execute(
605 "INSERT INTO documents (id, audience, _updated_at)
606 VALUES (?1, ?2, ?3)",
607 rusqlite::params![row_id, audience, stamp],
608 )
609 .map(|_| ())
610 .map_err(DbError::from)
611 })
612 .await?;
613 Ok(receipt.write_id)
614 }
615
616 pub async fn capture_document_with_file_for_test(
617 &self,
618 document_id: &str,
619 file_id: &str,
620 audience: Option<coven_protocol::circle::CircleId>,
621 bytes: &[u8],
622 stamp: &str,
623 ) -> Result<coven_protocol::write::WriteId, crate::HostWriteError<DbError>> {
624 let routing = coven_keys::encryption::EncryptionService::from_key([42; 32]);
625 let document_id = document_id.to_string();
626 let file_id = file_id.to_string();
627 let audience = audience.map(|circle_id| circle_id.to_string());
628 let size = i64::try_from(bytes.len()).expect("test blob size fits SQLite");
629 let hash = coven_protocol::blob::content_hash(bytes);
630 let stamp = stamp.to_string();
631 let receipt = crate::StoreDatabase::new(self)
632 .run_host_store_write_for_test(Some(routing), None, move |transaction| {
633 transaction
634 .execute(
635 "INSERT INTO documents (id, audience, _updated_at)
636 VALUES (?1, ?2, ?3)",
637 rusqlite::params![document_id, audience, stamp],
638 )
639 .map_err(DbError::from)?;
640 transaction
641 .execute(
642 "INSERT INTO document_files
643 (id, document_id, size, hash, _updated_at)
644 VALUES (?1, ?2, ?3, ?4, ?5)",
645 rusqlite::params![file_id, document_id, size, hash, stamp],
646 )
647 .map(|_| ())
648 .map_err(DbError::from)
649 })
650 .await?;
651 Ok(receipt.write_id)
652 }
653
654 pub async fn document_file_stamp_for_test(&self, file_id: &str) -> Result<String, DbError> {
655 let file_id = file_id.to_string();
656 self.test_sql(move |database| {
657 database
658 .query_row(
659 "SELECT _updated_at FROM document_files WHERE id = ?1",
660 [file_id],
661 |row| row.get::<_, String>(0),
662 )
663 .map_err(DbError::from)
664 })
665 .await
666 }
667
668 pub async fn release_retained_replay_ownership_for_test(&self) -> Result<(), DbError> {
669 self.test_sql(|database| {
670 database.transaction(|transaction| {
671 transaction.remove_retained_replay_ownership_from_snapshot()
672 })
673 })
674 .await
675 }
676
677 pub async fn insert_browsable_blob_row_for_test(
678 &self,
679 blob_id: &str,
680 cloud_path: &str,
681 bytes: &[u8],
682 ) -> Result<(), DbError> {
683 let note = format!("note-{blob_id}");
684 let blob_id = blob_id.to_string();
685 let cloud_path = cloud_path.to_string();
686 let size = i64::try_from(bytes.len()).expect("test blob size fits SQLite");
687 let hash = coven_protocol::blob::content_hash(bytes);
688 self.test_sql(move |database| {
689 database
690 .execute(
691 "INSERT INTO notes (id, title, shared, _updated_at, created_at)
692 VALUES (?1, 'browsable-test', 1, '0000000001000-0000-dev1', '2026-01-01')",
693 [note.as_str()],
694 )
695 .map_err(DbError::from)?;
696 database
697 .execute(
698 "INSERT INTO note_photos
699 (id, note_id, kind, size, hash, _updated_at, created_at)
700 VALUES (?1, ?2, ?3, ?4, ?5,
701 '0000000001000-0000-dev1', '2026-01-01')",
702 rusqlite::params![blob_id, note, cloud_path, size, hash],
703 )
704 .map_err(DbError::from)?;
705 Ok(())
706 })
707 .await
708 }
709
710 pub async fn bind_stored_blob_to_row_for_test(
711 &self,
712 stored: &coven_protocol::blob::locator::StoredBlobRef,
713 table: &str,
714 id: &str,
715 owner: coven_protocol::store_commit::StoreBatchCommitRef,
716 ) -> Result<(), DbError> {
717 let locator = stored.locator().clone();
718 let record =
719 coven_protocol::remote_object::RemoteObjectRecord::activated_blob(stored, owner)
720 .map_err(DbError::from)?
721 .into_record();
722 let object_id = record.object_id().to_string();
723 let state = serde_json::to_string(&record).map_err(DbError::from)?;
724 let locator_hash = locator.locator_hash().to_string();
725 let authority =
726 serde_json::to_string(&coven_protocol::audience_package::PackageAudience::Store)
727 .map_err(DbError::from)?;
728 let id_for_insert = id.to_string();
729 let table_for_insert = table.to_string();
730 let stamp_table = table.to_string();
731 let stamp_id = id.to_string();
732 self.test_sql(move |database| {
733 let row_stamp = database
734 .query_row(
735 &format!(
736 "SELECT _updated_at FROM {} WHERE id = ?1",
737 crate::quote_ident(&stamp_table)
738 ),
739 [stamp_id],
740 |row| row.get::<_, String>(0),
741 )
742 .map_err(DbError::from)?;
743 database.install_blob_binding(
744 &object_id,
745 &state,
746 &locator_hash,
747 &table_for_insert,
748 &id_for_insert,
749 "id",
750 &row_stamp,
751 &authority,
752 )
753 })
754 .await
755 }
756
757 pub async fn store_package_is_retained_for_replay_for_test(
758 &self,
759 package: coven_protocol::store_commit::StorePackageRef,
760 activation: coven_protocol::store_commit::StoreBatchCommitRef,
761 ) -> Result<bool, DbError> {
762 let database = crate::StoreDatabase::new(self);
763 let root = database
764 .local_store_root_ref()
765 .await?
766 .ok_or_else(|| DbError::Message("test Store root is not installed".to_string()))?;
767 database
768 .store_package_is_retained_for_replay(root, package, activation)
769 .await
770 }
771
772 pub async fn circle_state_counts_for_test(
773 &self,
774 circle_id: coven_protocol::circle::CircleId,
775 ) -> Result<(i64, i64, i64), DbError> {
776 self.test_sql(move |database| database.circle_state_counts(circle_id))
777 .await
778 }
779
780 pub async fn upload_outbox_attempt_for_test(
781 &self,
782 row_id: &str,
783 ) -> Result<Option<crate::OutboxAttempt>, DbError> {
784 let row_id = row_id.to_string();
785 self.test_sql(move |database| database.upload_outbox_attempt(&row_id))
786 .await
787 }
788
789 pub async fn corrupt_upload_outbox_attempt_time_for_test(
790 &self,
791 id: i64,
792 ) -> Result<(), DbError> {
793 self.test_sql(move |database| database.corrupt_upload_outbox_attempt_time(id))
794 .await
795 }
796
797 pub async fn corrupt_delete_outbox_attempt_time_for_test(
798 &self,
799 id: i64,
800 ) -> Result<(), DbError> {
801 self.test_sql(move |database| database.corrupt_delete_outbox_attempt_time(id))
802 .await
803 }
804
805 #[allow(clippy::too_many_arguments)]
806 pub async fn enqueue_blob_upload_with_retention_for_test(
807 &self,
808 root_table: &str,
809 root_id: &str,
810 row: coven_protocol::blob::RowBlobRef,
811 source_path: std::path::PathBuf,
812 retain_pinned: bool,
813 created_at: &str,
814 ) -> Result<(), DbError> {
815 let root_table = root_table.to_string();
816 let root_id = root_id.to_string();
817 let root_label = format!("{root_table}/{root_id}");
818 let created_at = created_at.to_string();
819 self.test_sql(move |database| {
820 database.enqueue_blob_upload(
821 &root_table,
822 &root_id,
823 &root_label,
824 &row,
825 &source_path,
826 retain_pinned,
827 &created_at,
828 )
829 })
830 .await
831 }
832
833 pub async fn roll_back_blob_upload_for_test(
834 &self,
835 root_table: &str,
836 root_id: &str,
837 row: coven_protocol::blob::RowBlobRef,
838 source_path: std::path::PathBuf,
839 created_at: &str,
840 ) -> Result<(), DbError> {
841 let root_table = root_table.to_string();
842 let root_id = root_id.to_string();
843 let root_label = format!("{root_table}/{root_id}");
844 let created_at = created_at.to_string();
845 self.test_sql(move |database| {
846 database.rolled_back_transaction(|transaction| {
847 transaction.enqueue_blob_upload(
848 &root_table,
849 &root_id,
850 &root_label,
851 &row,
852 &source_path,
853 false,
854 &created_at,
855 )
856 })
857 })
858 .await
859 }
860
861 pub async fn published_blob_drop_intent_count_for_test(
862 &self,
863 sequence: i64,
864 namespace: &str,
865 blob_id: &str,
866 ) -> Result<i64, DbError> {
867 let namespace = namespace.to_string();
868 let blob_id = blob_id.to_string();
869 self.test_sql(move |database| {
870 database.published_blob_drop_intent_count(sequence, &namespace, &blob_id)
871 })
872 .await
873 }
874
875 pub async fn first_published_blob_drop_intent_for_test(
876 &self,
877 namespace: &str,
878 blob_id: &str,
879 ) -> Result<(i64, coven_protocol::blob::DeferredLocalBlobDisposition), DbError> {
880 let namespace = namespace.to_string();
881 let blob_id = blob_id.to_string();
882 self.test_sql(move |database| {
883 let (sequence, disposition): (i64, String) = database
884 .query_row(
885 "SELECT seq, disposition FROM published_blob_drop_intents
886 WHERE namespace = ?1 AND blob_id = ?2
887 ORDER BY seq LIMIT 1",
888 (&namespace, &blob_id),
889 |row| Ok((row.get(0)?, row.get(1)?)),
890 )
891 .map_err(DbError::from)?;
892 let disposition =
893 coven_protocol::blob::DeferredLocalBlobDisposition::from_db(&disposition)
894 .map_err(|error| DbError::Message(error.to_string()))?;
895 Ok((sequence, disposition))
896 })
897 .await
898 }
899
900 pub async fn scoped_store_state_counts_for_test(&self) -> Result<[i64; 4], DbError> {
901 self.test_sql(|database| database.scoped_store_state_counts())
902 .await
903 }
904
905 pub async fn install_make_local_commit_failure_for_test(&self) -> Result<(), DbError> {
906 self.test_sql(|database| {
907 database
908 .execute_batch(
909 "CREATE TRIGGER reject_make_local_gate_update
910 BEFORE UPDATE OF shared ON notes
911 WHEN NEW.id = 'n1' AND NEW.shared = 0
912 BEGIN
913 SELECT RAISE(ABORT, 'forced make_local commit failure');
914 END;",
915 )
916 .map_err(DbError::from)
917 })
918 .await
919 }
920
921 pub async fn store_device_registration_activation_for_test(
922 &self,
923 device_id: &str,
924 ) -> Result<coven_protocol::store_commit::StoreDeviceRegistrationActivation, DbError> {
925 let device_id = device_id.to_string();
926 self.test_sql(move |database| database.store_device_registration_activation(&device_id))
927 .await
928 }
929
930 pub async fn latest_published_store_snapshot_for_test(
931 &self,
932 ) -> Result<(i64, Vec<u8>), DbError> {
933 self.test_sql(|database| database.latest_published_store_snapshot())
934 .await
935 }
936
937 pub async fn latest_published_store_snapshot_bytes_for_test(&self) -> Result<Vec<u8>, DbError> {
938 self.test_sql(|database| database.latest_published_store_snapshot_bytes())
939 .await
940 }
941
942 pub async fn materialized_commits_without_device_state_count_for_test(
943 &self,
944 ) -> Result<i64, DbError> {
945 self.test_sql(|database| database.materialized_commits_without_device_state_count())
946 .await
947 }
948
949 pub async fn store_device_state_snapshot_refs_for_test(
950 &self,
951 ) -> Result<Vec<coven_protocol::store_commit::StoreBatchCommitRef>, DbError> {
952 self.test_sql(|database| database.store_device_state_snapshot_refs())
953 .await
954 }
955
956 pub async fn restored_row_graph_counts_for_test(
957 &self,
958 ) -> Result<(i64, i64, i64, i64), DbError> {
959 self.test_sql(|database| {
960 Ok((
961 database.query_row("SELECT COUNT(*) FROM notes WHERE id = 'n1'", [], |row| {
962 row.get::<_, i64>(0)
963 })?,
964 database.query_row(
965 "SELECT COUNT(*) FROM note_photos WHERE id = 'photo1'",
966 [],
967 |row| row.get::<_, i64>(0),
968 )?,
969 database.query_row(
970 "SELECT COUNT(*) FROM note_photos AS photo
971 JOIN notes AS note ON note.id = photo.note_id
972 WHERE photo.id = 'photo1' AND note.id = 'n1'",
973 [],
974 |row| row.get::<_, i64>(0),
975 )?,
976 database.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |row| {
977 row.get::<_, i64>(0)
978 })?,
979 ))
980 })
981 .await
982 }
983
984 pub(super) async fn test_sql<F, R>(&self, operation: F) -> Result<R, DbError>
985 where
986 F: for<'connection> FnOnce(DatabaseTestSql<'connection>) -> Result<R, DbError>
987 + Send
988 + 'static,
989 R: Send + 'static,
990 {
991 self.call_database(move |session| session.run_test_sql(operation))
992 .await
993 }
994}