1use super::table_row_count;
2use crate::{Connection, DatabaseTestTable, DbError};
3
4pub struct DatabaseImageTest {
5 connection: Connection,
6}
7
8impl DatabaseImageTest {
9 pub fn open(path: &std::path::Path) -> Result<Self, DbError> {
10 Ok(Self {
11 connection: Connection::open(path).map_err(DbError::from)?,
12 })
13 }
14
15 pub fn from_bytes(bytes: &[u8]) -> Result<Self, DbError> {
16 let mut connection = Connection::open_in_memory().map_err(DbError::from)?;
17 crate::connection_io::deserialize_database_image_into(&mut connection, bytes)?;
18 Ok(Self { connection })
19 }
20
21 pub fn execute<P>(&self, sql: &str, params: P) -> rusqlite::Result<usize>
22 where
23 P: rusqlite::Params,
24 {
25 self.connection.execute(sql, params)
26 }
27
28 pub fn execute_batch(&self, sql: &str) -> rusqlite::Result<()> {
29 self.connection.execute_batch(sql)
30 }
31
32 pub fn query_row<T, P, F>(&self, sql: &str, params: P, map: F) -> rusqlite::Result<T>
33 where
34 P: rusqlite::Params,
35 F: FnOnce(&rusqlite::Row<'_>) -> rusqlite::Result<T>,
36 {
37 self.connection.query_row(sql, params, map)
38 }
39
40 pub fn query<T, P, F>(&self, sql: &str, params: P, map: F) -> rusqlite::Result<Vec<T>>
41 where
42 P: rusqlite::Params,
43 F: FnMut(&rusqlite::Row<'_>) -> rusqlite::Result<T>,
44 {
45 let mut statement = self.connection.prepare(sql)?;
46 let values = statement.query_map(params, map)?.collect();
47 values
48 }
49
50 pub fn apply_coven_schema(&self) -> Result<(), DbError> {
51 crate::apply_coven_schema(&self.connection).map_err(DbError::from)
52 }
53
54 pub fn downgrade_coven_schema_to_v0(&self, include_routing: bool) -> Result<(), DbError> {
55 crate::coven_schema::downgrade_coven_schema_to_v0_for_test(
56 &self.connection,
57 include_routing,
58 )
59 }
60
61 pub fn validate_uninitialized_coven_schema_v0(
62 &self,
63 include_routing: bool,
64 ) -> Result<(), crate::CovenMigrationError> {
65 crate::coven_migration::validate_uninitialized_coven_schema_v0_for_test(
66 &self.connection,
67 include_routing,
68 )
69 }
70
71 pub fn validate_current_initialized_coven_schema(
72 &self,
73 include_routing: bool,
74 ) -> Result<(), crate::OpenError> {
75 crate::database_open::load_coven_metadata(&self.connection)?;
76 crate::validate_coven_schema_for_reader(&self.connection, include_routing)?;
77 Ok(())
78 }
79
80 pub fn payload(
81 &self,
82 store_dir: &coven_foundation::store_dir::StoreDir,
83 encoded_hash: String,
84 ) -> Result<Vec<u8>, DbError> {
85 let hash = encoded_hash
86 .parse()
87 .map_err(|error| DbError::context("parse image payload hash", error))?;
88 crate::payload_store::read_payload_blocking(&self.connection, store_dir, hash)
89 .map_err(DbError::from)
90 }
91
92 pub fn scoped_routing_id(&self, table: &str, row_id: &str) -> String {
93 crate::DatabaseTestSql::new(&self.connection)
94 .row_routing_id([7; 32], table, row_id)
95 .expect("derive test row-routing id")
96 .to_string()
97 }
98
99 pub fn seed_active_circle(&self, label: &str) -> (String, String) {
100 let database = crate::DatabaseTestSql::new(&self.connection);
101 database
102 .install_test_store_root_authority("scoped-routing-root")
103 .expect("install scoped-routing Store root authority");
104 let (circle_id, control) = database.install_test_active_circle(label);
105 (
106 circle_id.to_string(),
107 serde_json::to_string(&control).expect("serialize active Circle control"),
108 )
109 }
110
111 pub fn seed_inactive_circle(&self, label: &str) -> String {
112 let database = crate::DatabaseTestSql::new(&self.connection);
113 database
114 .install_test_store_root_authority("scoped-routing-root")
115 .expect("install scoped-routing Store root authority");
116 database.install_test_inactive_circle(label).0.to_string()
117 }
118
119 pub fn coven_table_row_count(&self, table: DatabaseTestTable) -> Result<i64, DbError> {
120 table_row_count(&self.connection, table)
121 }
122
123 pub fn install_row_route(
124 &self,
125 routing_id: &str,
126 table: &str,
127 row_id: &str,
128 row_stamp: &str,
129 ) -> Result<(), DbError> {
130 self.connection
131 .execute(
132 "INSERT INTO _coven_row_routes
133 (routing_id, table_name, row_id, _updated_at) VALUES (?1, ?2, ?3, ?4)",
134 rusqlite::params![routing_id, table, row_id, row_stamp],
135 )
136 .map(|_| ())
137 .map_err(DbError::from)
138 }
139
140 pub fn install_audience_mirror(
141 &self,
142 routing_id: &str,
143 circle_id: Option<&str>,
144 row_stamp: &str,
145 ) -> Result<(), DbError> {
146 self.connection
147 .execute(
148 "INSERT INTO _coven_audience (routing_id, circle_id, _updated_at)
149 VALUES (?1, ?2, ?3)",
150 rusqlite::params![routing_id, circle_id, row_stamp],
151 )
152 .map(|_| ())
153 .map_err(DbError::from)
154 }
155
156 pub fn corrupt_document_route_id(&self) -> Result<(), DbError> {
157 self.connection
158 .execute(
159 "UPDATE _coven_row_routes
160 SET routing_id =
161 '0000000000000000000000000000000000000000000000000000000000000000'
162 WHERE table_name = 'documents'",
163 [],
164 )
165 .map(|_| ())
166 .map_err(DbError::from)
167 }
168
169 pub fn replace_first_circle_audience(&self, circle_id: Option<&str>) -> Result<(), DbError> {
170 self.connection
171 .execute(
172 "UPDATE _coven_audience SET circle_id = ?1
173 WHERE routing_id = (
174 SELECT routing_id FROM _coven_audience
175 WHERE circle_id IS NOT NULL ORDER BY routing_id LIMIT 1
176 )",
177 [circle_id],
178 )
179 .map(|_| ())
180 .map_err(DbError::from)
181 }
182
183 pub fn replace_store_device_snapshot(
186 &self,
187 reference: &coven_protocol::store_commit::StoreBatchCommitRef,
188 state: &coven_protocol::store_commit::ResolvedStoreDeviceState,
189 ) -> Result<(), DbError> {
190 let transaction = self.connection.unchecked_transaction()?;
191 transaction.execute(
192 "DELETE FROM store_device_state_snapshots WHERE commit_ref = ?1",
193 [serde_json::to_string(reference).map_err(DbError::from)?],
194 )?;
195 crate::store::record_store_device_snapshot_on(&transaction, reference, state)?;
196 transaction.commit().map_err(DbError::from)
197 }
198
199 pub fn store_device_state_snapshot_refs(&self) -> Result<Vec<String>, DbError> {
200 self.query(
201 "SELECT commit_ref FROM store_device_state_snapshots ORDER BY commit_ref",
202 [],
203 |row| row.get(0),
204 )
205 .map_err(DbError::from)
206 }
207
208 pub fn materialization_graph_counts(&self) -> Result<(i64, i64, i64), DbError> {
209 Ok((
210 table_row_count(
211 &self.connection,
212 DatabaseTestTable::named("materialized_commits"),
213 )?,
214 table_row_count(
215 &self.connection,
216 DatabaseTestTable::named("retained_merge_materializations"),
217 )?,
218 table_row_count(
219 &self.connection,
220 DatabaseTestTable::named("retained_replay_objects"),
221 )?,
222 ))
223 }
224
225 pub fn retained_materialization_bytes(&self) -> Result<Vec<Vec<u8>>, DbError> {
226 self.query(
227 "SELECT canonical_input FROM retained_merge_materializations",
228 [],
229 |row| row.get(0),
230 )
231 .map_err(DbError::from)
232 }
233
234 pub fn circle_states_containing(&self, text: &str) -> Result<i64, DbError> {
235 self.connection
236 .query_row(
237 "SELECT COUNT(*) FROM circle_current_state
238 WHERE instr(CAST(state AS TEXT), ?1) > 0",
239 [text],
240 |row| row.get(0),
241 )
242 .map_err(DbError::from)
243 }
244
245 pub fn remote_object(
246 &self,
247 object: &coven_protocol::objects::ExactObjectRef,
248 ) -> Result<coven_protocol::remote_object::RemoteObjectRecord, DbError> {
249 crate::load_remote_object_on(
250 &self.connection,
251 coven_protocol::remote_object::remote_object_id(object),
252 )
253 }
254
255 pub fn row_blob_remote_object(
256 &self,
257 table: &str,
258 row_id: &str,
259 column: &str,
260 row_stamp: &str,
261 ) -> Result<coven_protocol::remote_object::RemoteObjectRecord, DbError> {
262 let (object_id, encoded): (String, String) = self.connection.query_row(
263 "SELECT remote.object_id, remote.state FROM row_blob_locators AS binding
264 JOIN remote_objects AS remote ON remote.object_id = binding.remote_object_id
265 WHERE binding.table_name = ?1 AND binding.row_id = ?2
266 AND binding.column_name = ?3 AND binding.row_stamp = ?4",
267 [table, row_id, column, row_stamp],
268 |row| Ok((row.get(0)?, row.get(1)?)),
269 )?;
270 let remote: coven_protocol::remote_object::RemoteObjectRecord =
271 serde_json::from_str(&encoded)
272 .map_err(|error| DbError::context("decode image row blob ownership", error))?;
273 if remote.object_id().to_string() != object_id {
274 return Err(DbError::Message(
275 "image row blob binding differs from its remote object identity".into(),
276 ));
277 }
278 Ok(remote)
279 }
280
281 pub fn snapshot_blob_graph(
282 &self,
283 ) -> Result<
284 (
285 String,
286 String,
287 String,
288 String,
289 String,
290 coven_protocol::remote_object::RemoteObjectRecord,
291 ),
292 DbError,
293 > {
294 let (table, row_id, column, row_stamp, locator_hash, remote_state) = self
295 .connection
296 .query_row(
297 "SELECT binding.table_name, binding.row_id, binding.column_name,
298 binding.row_stamp, locator.locator_hash, remote.state
299 FROM row_blob_locators AS binding
300 JOIN blob_locators AS locator
301 ON locator.remote_object_id = binding.remote_object_id
302 JOIN remote_objects AS remote
303 ON remote.object_id = locator.remote_object_id",
304 [],
305 |row| {
306 Ok((
307 row.get(0)?,
308 row.get(1)?,
309 row.get(2)?,
310 row.get(3)?,
311 row.get(4)?,
312 row.get::<_, String>(5)?,
313 ))
314 },
315 )
316 .map_err(DbError::from)?;
317 let remote = serde_json::from_str(&remote_state)
318 .map_err(|error| DbError::context("parse snapshot remote blob", error))?;
319 Ok((table, row_id, column, row_stamp, locator_hash, remote))
320 }
321
322 pub fn install_snapshot_blob_binding(
323 &self,
324 binding: &coven_protocol::audience_package::RowBlobLocatorBinding,
325 remote: &coven_protocol::remote_object::RemoteObjectRecord,
326 ) -> Result<(), DbError> {
327 let object_id = remote.object_id().to_string();
328 self.connection
329 .execute(
330 "INSERT INTO remote_objects (object_id, state) VALUES (?1, ?2)",
331 rusqlite::params![
332 object_id,
333 serde_json::to_string(remote).map_err(DbError::from)?
334 ],
335 )
336 .map_err(DbError::from)?;
337 self.connection
338 .execute(
339 "INSERT INTO blob_locators (remote_object_id, locator_hash) VALUES (?1, ?2)",
340 rusqlite::params![
341 object_id,
342 binding.blob().locator().locator_hash().to_string()
343 ],
344 )
345 .map_err(DbError::from)?;
346 self.connection
347 .execute(
348 "INSERT INTO row_blob_locators
349 (table_name, row_id, column_name, row_stamp, audience_authority, remote_object_id)
350 VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
351 rusqlite::params![
352 binding.table(),
353 binding.row_id(),
354 binding.column(),
355 binding.row_stamp(),
356 serde_json::to_string(
357 &coven_protocol::audience_package::PackageAudience::Store
358 )
359 .map_err(DbError::from)?,
360 object_id,
361 ],
362 )
363 .map(|_| ())
364 .map_err(DbError::from)
365 }
366
367 pub fn create_interrupted_coven_schema(&self) -> Result<(), DbError> {
368 self.connection
369 .execute_batch(
370 "CREATE TABLE protocol_state (
371 key TEXT PRIMARY KEY,
372 value TEXT NOT NULL
373 ) STRICT;",
374 )
375 .map_err(DbError::from)
376 }
377
378 pub fn into_bytes(self) -> Result<Vec<u8>, DbError> {
379 self.connection
380 .serialize(rusqlite::MAIN_DB)
381 .map(|bytes| bytes.to_vec())
382 .map_err(DbError::from)
383 }
384}