1use super::{author_exclusion_activation_evidence, 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 store_device_state_snapshot_refs(&self) -> Result<Vec<String>, DbError> {
184 self.query(
185 "SELECT commit_ref FROM store_device_state_snapshots ORDER BY commit_ref",
186 [],
187 |row| row.get(0),
188 )
189 .map_err(DbError::from)
190 }
191
192 pub fn materialization_graph_counts(&self) -> Result<(i64, i64, i64), DbError> {
193 Ok((
194 table_row_count(
195 &self.connection,
196 DatabaseTestTable::named("materialized_commits"),
197 )?,
198 table_row_count(
199 &self.connection,
200 DatabaseTestTable::named("retained_merge_materializations"),
201 )?,
202 table_row_count(
203 &self.connection,
204 DatabaseTestTable::named("retained_replay_objects"),
205 )?,
206 ))
207 }
208
209 pub fn author_exclusion_activation_evidence(
210 &self,
211 ) -> Result<(String, String, String, String), DbError> {
212 author_exclusion_activation_evidence(&self.connection)
213 }
214
215 pub fn snapshot_blob_graph(
216 &self,
217 ) -> Result<
218 (
219 String,
220 String,
221 String,
222 String,
223 String,
224 coven_protocol::remote_object::RemoteObjectRecord,
225 ),
226 DbError,
227 > {
228 let (table, row_id, column, row_stamp, locator_hash, remote_state) = self
229 .connection
230 .query_row(
231 "SELECT binding.table_name, binding.row_id, binding.column_name,
232 binding.row_stamp, locator.locator_hash, remote.state
233 FROM row_blob_locators AS binding
234 JOIN blob_locators AS locator
235 ON locator.remote_object_id = binding.remote_object_id
236 JOIN remote_objects AS remote
237 ON remote.object_id = locator.remote_object_id",
238 [],
239 |row| {
240 Ok((
241 row.get(0)?,
242 row.get(1)?,
243 row.get(2)?,
244 row.get(3)?,
245 row.get(4)?,
246 row.get::<_, String>(5)?,
247 ))
248 },
249 )
250 .map_err(DbError::from)?;
251 let remote = serde_json::from_str(&remote_state)
252 .map_err(|error| DbError::context("parse snapshot remote blob", error))?;
253 Ok((table, row_id, column, row_stamp, locator_hash, remote))
254 }
255
256 pub fn install_snapshot_blob_binding(
257 &self,
258 binding: &coven_protocol::audience_package::RowBlobLocatorBinding,
259 remote: &coven_protocol::remote_object::RemoteObjectRecord,
260 ) -> Result<(), DbError> {
261 let object_id = remote.object_id().to_string();
262 self.connection
263 .execute(
264 "INSERT INTO remote_objects (object_id, state) VALUES (?1, ?2)",
265 rusqlite::params![
266 object_id,
267 serde_json::to_string(remote).map_err(DbError::from)?
268 ],
269 )
270 .map_err(DbError::from)?;
271 self.connection
272 .execute(
273 "INSERT INTO blob_locators (remote_object_id, locator_hash) VALUES (?1, ?2)",
274 rusqlite::params![
275 object_id,
276 binding.blob().locator().locator_hash().to_string()
277 ],
278 )
279 .map_err(DbError::from)?;
280 self.connection
281 .execute(
282 "INSERT INTO row_blob_locators
283 (table_name, row_id, column_name, row_stamp, audience_authority, remote_object_id)
284 VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
285 rusqlite::params![
286 binding.table(),
287 binding.row_id(),
288 binding.column(),
289 binding.row_stamp(),
290 serde_json::to_string(
291 &coven_protocol::audience_package::PackageAudience::Store
292 )
293 .map_err(DbError::from)?,
294 object_id,
295 ],
296 )
297 .map(|_| ())
298 .map_err(DbError::from)
299 }
300
301 pub fn create_interrupted_coven_schema(&self) -> Result<(), DbError> {
302 self.connection
303 .execute_batch(
304 "CREATE TABLE protocol_state (
305 key TEXT PRIMARY KEY,
306 value TEXT NOT NULL
307 ) STRICT;",
308 )
309 .map_err(DbError::from)
310 }
311
312 pub fn into_bytes(self) -> Result<Vec<u8>, DbError> {
313 self.connection
314 .serialize(rusqlite::MAIN_DB)
315 .map(|bytes| bytes.to_vec())
316 .map_err(DbError::from)
317 }
318}