1use crate::{Database, DatabaseTestTable, DbError};
2
3impl Database {
4 pub async fn install_malformed_store_root_authority_for_test(&self) -> Result<(), DbError> {
5 self.test_sql(|database| {
6 database
7 .execute(
8 "INSERT INTO store_protocol_root_authority
9 (singleton, store_root_hash, store_protocol_root_bytes, store_root_object)
10 VALUES (1, ?1, X'00', '{}')",
11 ["00".repeat(32)],
12 )
13 .map(|_| ())
14 .map_err(DbError::from)
15 })
16 .await
17 }
18
19 pub async fn install_exact_store_root_authority_for_test(
20 &self,
21 reference: coven_protocol::store_commit::StoreRootRef,
22 bytes: Vec<u8>,
23 ) -> Result<(), DbError> {
24 self.test_sql(move |database| {
25 database.install_exact_store_root_authority(&reference, &bytes)
26 })
27 .await
28 }
29
30 pub async fn seed_existing_store_write_for_test(
31 &self,
32 changeset_hash: String,
33 ) -> Result<(), DbError> {
34 self.test_sql(move |database| {
35 database
36 .execute(
37 "INSERT INTO store_writes
38 (write_id, status, affected_rows, changeset_hash, base, blob_facts)
39 VALUES (
40 'existing-write', '\"pending\"', '[]', ?1,
41 '{\"dependencies\":{}}',
42 '{\"blobs\":[]}'
43 )",
44 [changeset_hash],
45 )
46 .map(|_| ())
47 .map_err(DbError::from)
48 })
49 .await
50 }
51
52 pub async fn host_identity_rollback_state_for_test(
53 &self,
54 ) -> Result<(i64, Vec<String>), DbError> {
55 self.test_sql(|database| {
56 let row_count = database
57 .query_row("SELECT COUNT(*) FROM things", [], |row| row.get(0))
58 .map_err(DbError::from)?;
59 let write_hashes = database
60 .query(
61 "SELECT changeset_hash FROM store_writes
62 WHERE changeset_hash IS NOT NULL ORDER BY ordinal",
63 [],
64 |row| row.get::<_, String>(0),
65 )
66 .map_err(DbError::from)?;
67 Ok((row_count, write_hashes))
68 })
69 .await
70 }
71
72 pub async fn seed_thing_for_test(&self, row_id: String) -> Result<(), DbError> {
73 self.test_sql(move |database| {
74 database
75 .execute(
76 "INSERT INTO things VALUES (?1, 'base', '0000000001000-0000-writer')",
77 [row_id],
78 )
79 .map(|_| ())
80 .map_err(DbError::from)
81 })
82 .await
83 }
84
85 pub async fn store_write_hashes_for_test(&self) -> Result<Vec<String>, DbError> {
86 self.test_sql(|database| {
87 database
88 .query(
89 "SELECT changeset_hash FROM store_writes
90 WHERE changeset_hash IS NOT NULL ORDER BY ordinal",
91 [],
92 |row| row.get::<_, String>(0),
93 )
94 .map_err(DbError::from)
95 })
96 .await
97 }
98
99 pub async fn thing_and_store_write_state_for_test(
100 &self,
101 ) -> Result<((String, String), Vec<String>), DbError> {
102 self.test_sql(|database| {
103 let row = database
104 .query_row("SELECT id, body FROM things", [], |row| {
105 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
106 })
107 .map_err(DbError::from)?;
108 let write_hashes = database
109 .query(
110 "SELECT changeset_hash FROM store_writes
111 WHERE changeset_hash IS NOT NULL ORDER BY ordinal",
112 [],
113 |row| row.get::<_, String>(0),
114 )
115 .map_err(DbError::from)?;
116 Ok((row, write_hashes))
117 })
118 .await
119 }
120
121 pub async fn make_remote_retain_pinned_column_for_test(
122 &self,
123 ) -> Result<Option<(i64, Option<String>)>, DbError> {
124 self.test_sql(|database| {
125 let rows = database
126 .query("PRAGMA table_info(blob_make_remote_intents)", [], |row| {
127 Ok((
128 row.get::<_, String>(1)?,
129 row.get::<_, i64>(3)?,
130 row.get::<_, Option<String>>(4)?,
131 ))
132 })
133 .map_err(DbError::from)?;
134 Ok(rows
135 .into_iter()
136 .find_map(|(name, not_null, default_value)| {
137 (name == "retain_pinned").then_some((not_null, default_value))
138 }))
139 })
140 .await
141 }
142
143 pub async fn vacuum_into_for_test(&self, destination: String) -> Result<(), DbError> {
144 self.test_sql(move |database| {
145 database
146 .execute("VACUUM INTO ?1", [destination])
147 .map(|_| ())
148 .map_err(DbError::from)
149 })
150 .await
151 }
152
153 pub async fn remove_store_publication_boundary_for_test(&self) -> Result<(), DbError> {
154 self.test_sql(|database| {
155 database.clear_table(DatabaseTestTable::named("store_publication_current"))
156 })
157 .await
158 }
159
160 pub async fn table_row_count_for_test(&self, table: DatabaseTestTable) -> Result<i64, DbError> {
161 self.test_sql(move |database| database.table_row_count(table))
162 .await
163 }
164
165 #[allow(clippy::too_many_arguments)]
166 pub async fn install_blob_binding_for_test(
167 &self,
168 object_id: String,
169 state: String,
170 locator_hash: String,
171 table: &str,
172 row_id: &str,
173 column: &str,
174 row_stamp: &str,
175 audience: String,
176 ) -> Result<(), DbError> {
177 let table = table.to_string();
178 let row_id = row_id.to_string();
179 let column = column.to_string();
180 let row_stamp = row_stamp.to_string();
181 self.test_sql(move |database| {
182 database.install_blob_binding(
183 &object_id,
184 &state,
185 &locator_hash,
186 &table,
187 &row_id,
188 &column,
189 &row_stamp,
190 &audience,
191 )
192 })
193 .await
194 }
195
196 pub async fn protocol_state_prefix_count_for_test(&self, prefix: &str) -> Result<i64, DbError> {
197 let prefix = prefix.to_string();
198 self.test_sql(move |database| database.protocol_state_prefix_count(&prefix))
199 .await
200 }
201
202 pub async fn exact_row_blob_locator_count_for_test(
203 &self,
204 table: &str,
205 row_id: &str,
206 column: &str,
207 stamp: &str,
208 ) -> Result<i64, DbError> {
209 let table = table.to_string();
210 let row_id = row_id.to_string();
211 let column = column.to_string();
212 let stamp = stamp.to_string();
213 self.test_sql(move |database| {
214 database.exact_row_blob_locator_count(&table, &row_id, &column, &stamp)
215 })
216 .await
217 }
218
219 pub async fn exact_upload_outbox_count_for_test(
220 &self,
221 table: &str,
222 row_id: &str,
223 column: &str,
224 stamp: &str,
225 ) -> Result<i64, DbError> {
226 let table = table.to_string();
227 let row_id = row_id.to_string();
228 let column = column.to_string();
229 let stamp = stamp.to_string();
230 self.test_sql(move |database| {
231 database.exact_upload_outbox_count(&table, &row_id, &column, &stamp)
232 })
233 .await
234 }
235
236 pub async fn install_outbound_preparation_failure_for_test(&self) -> Result<(), DbError> {
237 self.test_sql(|database| database.install_outbound_preparation_failure_trigger())
238 .await
239 }
240
241 pub async fn remove_outbound_preparation_failure_for_test(&self) -> Result<(), DbError> {
242 self.test_sql(|database| {
243 database
244 .execute_batch("DROP TRIGGER fail_outbound_preparation")
245 .map_err(DbError::from)
246 })
247 .await
248 }
249
250 pub async fn staged_circle_acknowledgement_object_for_test(
251 &self,
252 ) -> Result<coven_protocol::objects::PreparedExactObject, DbError> {
253 self.test_sql(|database| database.staged_circle_acknowledgement_object())
254 .await
255 }
256
257 pub async fn install_owner_anchor_failure_for_test(&self) -> Result<(), DbError> {
258 self.test_sql(|database| {
259 database
260 .execute_batch(
261 "CREATE TEMP TRIGGER fail_owner_anchor_baseline
262 BEFORE INSERT ON retained_replay_baselines
263 BEGIN
264 SELECT RAISE(ABORT, 'injected owner anchor failure');
265 END",
266 )
267 .map_err(DbError::from)
268 })
269 .await
270 }
271
272 pub async fn install_replay_image_corruption_for_test(&self) -> Result<(), DbError> {
273 self.test_sql(|database| {
274 database
275 .execute_batch(
276 "CREATE TEMP TRIGGER corrupt_owner_anchor_replay_image
277 AFTER INSERT ON retained_replay_baselines
278 BEGIN
279 UPDATE payload_storage
280 SET compressed_bytes = X'00', compressed_size = 1
281 WHERE payload_hash = NEW.image_payload_hash
282 AND storage = 'inline';
283 END",
284 )
285 .map_err(DbError::from)
286 })
287 .await
288 }
289
290 pub async fn remove_owner_anchor_failure_for_test(&self) -> Result<(), DbError> {
291 self.test_sql(|database| {
292 database
293 .execute_batch("DROP TRIGGER fail_owner_anchor_baseline")
294 .map_err(DbError::from)
295 })
296 .await
297 }
298
299 pub async fn corrupt_store_device_registration_bytes_for_test(
300 &self,
301 registration: coven_protocol::store_commit::StoreDeviceRegistrationRef,
302 ) -> Result<(), DbError> {
303 self.test_sql(move |database| {
304 database.corrupt_store_device_registration_bytes(®istration)
305 })
306 .await
307 }
308
309 pub async fn validate_retained_merge_replay_for_test(
310 &self,
311 root: coven_protocol::store_commit::StoreRootRef,
312 ) -> Result<(), DbError> {
313 self.test_sql(move |database| database.load_retained_merge_replay_inputs(&root).map(drop))
314 .await
315 }
316
317 pub async fn replace_retained_merge_input_for_test(
318 &self,
319 stream_id: String,
320 canonical_input: Vec<u8>,
321 ) -> Result<(), DbError> {
322 self.test_sql(move |database| {
323 database.replace_retained_merge_input(&stream_id, &canonical_input)
324 })
325 .await
326 }
327
328 pub async fn insert_invalid_materialized_commit_for_test(&self) -> Result<(), DbError> {
329 self.test_sql(|database| database.insert_invalid_materialized_commit())
330 .await
331 }
332
333 pub async fn retained_materialization_input_for_test(
334 &self,
335 stream_id: String,
336 sequence: u64,
337 ) -> Result<(Vec<u8>, String, String), DbError> {
338 self.test_sql(move |database| database.retained_materialization_input(&stream_id, sequence))
339 .await
340 }
341
342 pub async fn retained_canonical_input_for_test(
343 &self,
344 stream_id: String,
345 sequence: u64,
346 ) -> Result<Vec<u8>, DbError> {
347 self.test_sql(move |database| database.retained_canonical_input(&stream_id, sequence))
348 .await
349 }
350
351 pub async fn corrupt_retained_materialization_input_for_test(
352 &self,
353 stream_id: String,
354 sequence: u64,
355 ) -> Result<(), DbError> {
356 self.test_sql(move |database| {
357 database.corrupt_retained_materialization_input(&stream_id, sequence)
358 })
359 .await
360 }
361
362 pub async fn insert_retained_replay_object_for_test(
363 &self,
364 owner: coven_protocol::remote_object::RetainedReplayOwner,
365 object: coven_protocol::objects::ExactObjectRef,
366 ) -> Result<(), DbError> {
367 self.test_sql(move |database| database.insert_retained_replay_object(&owner, &object))
368 .await
369 }
370
371 pub async fn retained_merge_input_hash_for_test(
372 &self,
373 stream_id: String,
374 sequence: u64,
375 ) -> Result<coven_protocol::store_commit::ObjectHash, DbError> {
376 self.test_sql(move |database| {
377 database
378 .retained_merge_input(&stream_id, sequence)
379 .map(|(input_hash, _)| input_hash)
380 })
381 .await
382 }
383
384 pub async fn materialized_commit_exists_for_test(
385 &self,
386 stream_id: String,
387 sequence: u64,
388 ) -> Result<bool, DbError> {
389 self.test_sql(move |database| database.materialized_commit_exists(&stream_id, sequence))
390 .await
391 }
392
393 pub async fn remove_materialized_note_for_test(
394 &self,
395 stream_id: String,
396 sequence: u64,
397 row_id: String,
398 ) -> Result<(), DbError> {
399 self.test_sql(move |database| {
400 database.transaction(|transaction| {
401 transaction.delete_materialized_commit(&stream_id, sequence)?;
402 transaction
403 .execute("DELETE FROM notes WHERE id = ?1", [row_id])
404 .map(|_| ())
405 .map_err(DbError::from)
406 })
407 })
408 .await
409 }
410
411 pub async fn write_retains_prepared_for_test(
412 &self,
413 write_id: coven_protocol::write::WriteId,
414 ) -> Result<bool, DbError> {
415 self.test_sql(move |database| database.write_retains_prepared(&write_id))
416 .await
417 }
418
419 pub async fn install_outbound_completion_failure_for_test(&self) -> Result<(), DbError> {
420 self.test_sql(|database| database.install_outbound_completion_failure_trigger())
421 .await
422 }
423
424 pub async fn remove_outbound_completion_failure_for_test(&self) -> Result<(), DbError> {
425 self.test_sql(|database| {
426 database
427 .execute_batch("DROP TRIGGER fail_outbound_completion")
428 .map_err(DbError::from)
429 })
430 .await
431 }
432
433 pub async fn replace_store_root_hash_for_test(
434 &self,
435 value: Option<String>,
436 ) -> Result<(), DbError> {
437 self.test_sql(move |database| database.replace_store_root_hash(value.as_deref()))
438 .await
439 }
440
441 pub async fn delete_device_state_snapshot_for_test(
442 &self,
443 commit_ref: String,
444 ) -> Result<(), DbError> {
445 self.test_sql(move |database| database.delete_device_state_snapshot(&commit_ref))
446 .await
447 }
448
449 pub async fn delete_retained_materialization_without_foreign_keys_for_test(
450 &self,
451 reference: coven_protocol::store_commit::StoreBatchCommitRef,
452 ) -> Result<(), DbError> {
453 self.test_sql(move |database| {
454 database.delete_retained_materialization_without_foreign_keys(&reference)
455 })
456 .await
457 }
458
459 pub async fn replace_device_state_snapshot_for_test(
460 &self,
461 commit_ref: String,
462 state: coven_protocol::store_commit::ResolvedStoreDeviceState,
463 ) -> Result<(), DbError> {
464 self.test_sql(move |database| database.replace_device_state_snapshot(&commit_ref, &state))
465 .await
466 }
467
468 pub async fn forge_device_in_state_snapshots_for_test(
469 &self,
470 forged_device_id: coven_protocol::store_commit::StoreDeviceId,
471 ) -> Result<(), DbError> {
472 self.test_sql(move |database| database.forge_device_in_state_snapshots(forged_device_id))
473 .await
474 }
475
476 pub async fn delete_exact_materialized_commit_for_test(
477 &self,
478 reference: coven_protocol::store_commit::StoreBatchCommitRef,
479 ) -> Result<(), DbError> {
480 self.test_sql(move |database| database.delete_exact_materialized_commit(&reference))
481 .await
482 }
483
484 pub async fn install_protocol_state_key_insert_failure_for_test(
485 &self,
486 rejected_key: String,
487 ) -> Result<(), DbError> {
488 self.test_sql(move |database| {
489 database
490 .execute_batch(&format!(
491 "CREATE TRIGGER reject_protocol_state_key
492 BEFORE INSERT ON protocol_state
493 WHEN NEW.key = '{rejected_key}'
494 BEGIN SELECT RAISE(ABORT, 'forced cursor failure'); END;"
495 ))
496 .map_err(DbError::from)
497 })
498 .await
499 }
500
501 pub async fn install_protocol_state_insert_failure_for_test(&self) -> Result<(), DbError> {
502 self.test_sql(|database| database.install_protocol_state_insert_failure_trigger())
503 .await
504 }
505
506 pub async fn apply_changeset_for_test(
507 &self,
508 bytes: Vec<u8>,
509 tables: Vec<coven_protocol::synced_schema::SyncedTable>,
510 receiver_wall_ms: u64,
511 ) -> Result<crate::ApplyResult, DbError> {
512 self.test_sql(move |database| database.apply_changeset(&bytes, &tables, receiver_wall_ms))
513 .await
514 }
515
516 pub async fn apply_changesets_atomically_for_test(
517 &self,
518 changesets: Vec<Vec<u8>>,
519 tables: Vec<coven_protocol::synced_schema::SyncedTable>,
520 receiver_wall_ms: u64,
521 ) -> Result<(Vec<crate::ApplyResult>, bool), DbError> {
522 self.test_sql(move |database| {
523 database.apply_changesets_atomically(changesets, &tables, receiver_wall_ms)
524 })
525 .await
526 }
527
528 pub async fn store_write_partitions_in_audience_order_for_test(
529 &self,
530 ) -> Result<Vec<(String, Option<String>, Vec<u8>)>, DbError> {
531 self.test_sql(|database| database.store_write_partitions_in_audience_order())
532 .await
533 }
534
535 pub async fn first_store_write_partition_hash_for_test(
536 &self,
537 write_id: coven_protocol::write::WriteId,
538 ) -> Result<coven_protocol::store_commit::ObjectHash, DbError> {
539 self.test_sql(move |database| database.first_store_write_partition_hash(write_id.as_str()))
540 .await
541 }
542
543 pub async fn plant_control_on_local_partition_for_test(&self) -> Result<(), DbError> {
544 self.test_sql(|database| database.plant_control_on_the_local_partition())
545 .await
546 }
547
548 pub async fn store_write_row_for_test(
549 &self,
550 write_id: coven_protocol::write::WriteId,
551 ) -> Result<(String, Vec<u8>), DbError> {
552 self.test_sql(move |database| database.store_write_row(write_id.as_str()))
553 .await
554 }
555
556 pub async fn row_and_private_routing_presence_for_test(
557 &self,
558 table: &str,
559 row_id: &str,
560 ) -> Result<(bool, bool, bool), DbError> {
561 let table = table.to_string();
562 let row_id = row_id.to_string();
563 self.test_sql(move |database| database.row_and_private_routing_presence(&table, &row_id))
564 .await
565 }
566
567 pub async fn store_write_row_and_only_partition_for_test(
568 &self,
569 write_id: coven_protocol::write::WriteId,
570 ) -> Result<((String, Vec<u8>), (String, Option<String>, Vec<u8>)), DbError> {
571 self.test_sql(move |database| {
572 Ok((
573 database.store_write_row(write_id.as_str())?,
574 database.only_store_write_partition(write_id.as_str())?,
575 ))
576 })
577 .await
578 }
579
580 pub async fn store_write_partition_changesets_for_test(
581 &self,
582 write_id: coven_protocol::write::WriteId,
583 ) -> Result<Vec<(String, Vec<u8>)>, DbError> {
584 self.test_sql(move |database| database.store_write_partition_changesets(write_id.as_str()))
585 .await
586 }
587
588 pub async fn apply_coven_routing_schema_for_test(&self) -> Result<(), DbError> {
589 self.test_sql(|database| database.apply_coven_routing_schema())
590 .await
591 }
592
593 pub async fn circle_current_state_for_test(
594 &self,
595 circle_id: coven_protocol::circle::CircleId,
596 ) -> Result<Option<coven_protocol::circle_activation::CircleCurrentState>, DbError> {
597 self.test_sql(move |database| database.circle_current_state(circle_id))
598 .await
599 }
600
601 pub async fn record_verified_circle_activations_for_test(
602 &self,
603 commit: coven_protocol::store_commit::VerifiedStoreBatchCommit,
604 activations: Vec<coven_protocol::circle_activation::VerifiedCircleReference>,
605 ) -> Result<(), DbError> {
606 self.test_sql(move |database| {
607 database.record_verified_circle_activations(&commit, &activations)
608 })
609 .await
610 }
611
612 pub async fn circle_access_owner_for_test(
613 &self,
614 circle_id: coven_protocol::circle::CircleId,
615 ) -> Result<String, DbError> {
616 self.test_sql(move |database| database.circle_access_owner(circle_id))
617 .await
618 }
619
620 pub async fn clear_circle_access_cache_for_test(&self) -> Result<(), DbError> {
621 self.test_sql(|database| {
622 database.clear_table(DatabaseTestTable::named("circle_access_cache"))
623 })
624 .await
625 }
626
627 pub async fn replace_circle_operation_prepared_for_test(
628 &self,
629 operation_id: coven_protocol::circle::CircleOperationId,
630 substitute: coven_protocol::circle_journal::CircleOperationJournal,
631 ) -> Result<(), DbError> {
632 self.test_sql(move |database| {
633 database.replace_circle_operation_prepared(&operation_id, &substitute)
634 })
635 .await
636 }
637
638 pub async fn circle_state_table_counts_for_test(&self) -> Result<(i64, i64), DbError> {
639 self.test_sql(|database| database.circle_state_table_counts())
640 .await
641 }
642
643 pub async fn document_circle_route_for_test(
644 &self,
645 row_id: &str,
646 ) -> Result<(String, String, String), DbError> {
647 let row_id = row_id.to_string();
648 self.test_sql(move |database| database.document_circle_route(&row_id))
649 .await
650 }
651
652 pub async fn corrupt_live_document_route_id_for_test(
653 &self,
654 row_id: &str,
655 ) -> Result<(), DbError> {
656 let row_id = row_id.to_string();
657 self.test_sql(move |database| database.corrupt_live_document_route_id(&row_id))
658 .await
659 }
660
661 pub async fn materialization_graph_counts_for_test(&self) -> Result<(i64, i64, i64), DbError> {
662 self.test_sql(|database| {
663 Ok((
664 database.table_row_count(DatabaseTestTable::named("materialized_commits"))?,
665 database
666 .table_row_count(DatabaseTestTable::named("retained_merge_materializations"))?,
667 database.table_row_count(DatabaseTestTable::named("retained_replay_objects"))?,
668 ))
669 })
670 .await
671 }
672
673 pub async fn persist_exact_remote_object_for_test(
674 &self,
675 remote: coven_protocol::remote_object::ClosedRemoteObject,
676 context: String,
677 ) -> Result<(), DbError> {
678 self.test_sql(move |database| database.persist_exact_remote_object(&remote, &context))
679 .await
680 }
681
682 pub async fn remote_object_by_id_for_test(
683 &self,
684 object_id: coven_protocol::store_commit::ObjectHash,
685 ) -> Result<coven_protocol::remote_object::RemoteObjectRecord, DbError> {
686 self.test_sql(move |database| database.load_remote_object(object_id))
687 .await
688 }
689
690 pub async fn install_reclaimed_store_package_for_test(
691 &self,
692 operation: crate::DurableStoreReclaimOperation,
693 package: crate::ReclaimedStorePackage,
694 ) -> Result<(), DbError> {
695 self.test_sql(move |database| {
696 database.transaction(|transaction| {
697 transaction.insert_store_reclaim_operation(&operation)?;
698 transaction.record_reclaimed_store_package(&package)
699 })
700 })
701 .await
702 }
703
704 pub async fn reclaimed_store_package_for_test(
705 &self,
706 object_id: coven_protocol::store_commit::ObjectHash,
707 ) -> Result<Option<crate::ReclaimedStorePackage>, DbError> {
708 self.test_sql(move |database| database.load_reclaimed_store_package(object_id))
709 .await
710 }
711
712 pub async fn record_reclaimed_store_package_for_test(
713 &self,
714 package: crate::ReclaimedStorePackage,
715 ) -> Result<(), DbError> {
716 self.test_sql(move |database| database.record_reclaimed_store_package(&package))
717 .await
718 }
719
720 pub async fn scoped_routing_counts_for_test(
721 &self,
722 circle_id: coven_protocol::circle::CircleId,
723 ) -> Result<(i64, i64), DbError> {
724 self.test_sql(move |database| database.scoped_routing_counts(circle_id))
725 .await
726 }
727
728 pub async fn cleanup_intent_copy_identities_for_test(&self) -> Result<Vec<String>, DbError> {
729 self.test_sql(|database| database.cleanup_intent_copy_identities())
730 .await
731 }
732
733 pub async fn insert_cleanup_intent_for_test(
734 &self,
735 namespace: String,
736 blob_id: String,
737 copy_identity: String,
738 ) -> Result<(), DbError> {
739 self.test_sql(move |database| {
740 database.insert_cleanup_intent(&namespace, &blob_id, ©_identity)
741 })
742 .await
743 }
744
745 pub async fn seed_distinct_cleanup_bindings_for_test(
746 &self,
747 removed_locator: coven_protocol::store_commit::ObjectHash,
748 live_locator: coven_protocol::store_commit::ObjectHash,
749 removed_object: coven_protocol::store_commit::ObjectHash,
750 live_object: coven_protocol::store_commit::ObjectHash,
751 ) -> Result<(), DbError> {
752 self.test_sql(move |database| {
753 database
754 .execute_batch(&format!(
755 "INSERT INTO notes (id, title, shared, _updated_at, created_at)
756 VALUES ('parent', 'parent', 1, '0000000001000-0000-test', '2026-01-01');
757 INSERT INTO note_photos
758 (id, note_id, kind, size, hash, blob_id, _updated_at, created_at)
759 VALUES
760 ('removed-row', 'parent', 'cover', 5, '{hash}', 'shared-id',
761 '0000000001000-0000-test', '2026-01-01'),
762 ('live-row', 'parent', 'cover', 5, '{hash}', 'shared-id',
763 '0000000001001-0000-test', '2026-01-01');",
764 hash = coven_protocol::blob::content_hash(b"bytes"),
765 ))
766 .map_err(DbError::from)?;
767 for (object, locator) in [
768 (removed_object, removed_locator),
769 (live_object, live_locator),
770 ] {
771 database
772 .execute(
773 "INSERT INTO remote_objects (object_id, state) VALUES (?1, '{}')",
774 [object.to_string()],
775 )
776 .map_err(DbError::from)?;
777 database
778 .execute(
779 "INSERT INTO blob_locators (remote_object_id, locator_hash)
780 VALUES (?1, ?2)",
781 (object.to_string(), locator.to_string()),
782 )
783 .map_err(DbError::from)?;
784 }
785 for (row_id, row_stamp, object) in [
786 ("removed-row", "0000000001000-0000-test", removed_object),
787 ("live-row", "0000000001001-0000-test", live_object),
788 ] {
789 database
790 .execute(
791 "INSERT INTO row_blob_locators
792 (table_name, row_id, column_name, row_stamp,
793 audience_authority, remote_object_id)
794 VALUES ('note_photos', ?1, 'blob_id', ?2, '\"store\"', ?3)",
795 (row_id, row_stamp, object.to_string()),
796 )
797 .map_err(DbError::from)?;
798 }
799 database
800 .execute("DELETE FROM note_photos WHERE id = 'removed-row'", [])
801 .map(|_| ())
802 .map_err(DbError::from)
803 })
804 .await
805 }
806}