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