1use super::*;
2use crate::query_mapped_rows;
3#[cfg(any(test, feature = "test-utils"))]
4use crate::store::store_session::StoreRecords;
5
6impl StoreSession<'_> {
7 fn prepare_circle_restore_selection(&self) -> Result<CircleRestoreSelectionIndex, DbError> {
8 let tx = self.conn.unchecked_transaction().map_err(DbError::from)?;
9 let rows = query_mapped_rows(
10 &tx,
11 "SELECT circle_id, control_coord FROM circle_control_activations
12 ORDER BY circle_id, control_coord",
13 [],
14 |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
15 )?;
16 let mut circles: Vec<(
17 coven_protocol::circle::CircleId,
18 Vec<coven_protocol::circle::CircleControlCoord>,
19 )> = Vec::new();
20 for (circle_id, control_coord) in rows {
21 let circle_id: coven_protocol::circle::CircleId = circle_id
22 .parse()
23 .map_err(|error| DbError::context("parse retained Circle id", error))?;
24 let control: coven_protocol::circle::CircleControlCoord =
25 serde_json::from_str(&control_coord).map_err(|error| {
26 DbError::context("parse retained Circle control coordinate", error)
27 })?;
28 match circles.last_mut() {
29 Some((last_circle, controls)) if *last_circle == circle_id => {
30 controls.push(control)
31 }
32 _ => circles.push((circle_id, vec![control])),
33 }
34 }
35 let preserved_bootstraps = circle_bootstrap_coverage_refs_on(&tx)?;
36 tx.commit().map_err(DbError::from)?;
37 Ok(CircleRestoreSelectionIndex {
38 circles,
39 preserved_bootstraps,
40 })
41 }
42
43 fn retained_merge_materialization_by_ref(
44 &mut self,
45 root: &coven_protocol::store_commit::StoreRootRef,
46 reference: &StoreBatchCommitRef,
47 ) -> Result<OwnedVerifiedMergeMaterialization, DbError> {
48 let retained = self
49 .verified_store_authority
50 .retained_materialization_by_ref_on(
51 crate::store::store_session::StoreRecords::new(self.conn, self.store_dir),
52 reference,
53 )?;
54 if retained.root() != root {
55 return Err(DbError::Message(
56 "retained Merge materialization belongs to another Store root".to_string(),
57 ));
58 }
59 Ok(retained)
60 }
61
62 fn circle_replay_epoch_index(
63 &mut self,
64 root: &coven_protocol::store_commit::StoreRootRef,
65 ) -> Result<CircleReplayEpochIndex, DbError> {
66 self.verified_store_authority.retained_replay_inputs_on(
67 crate::store::store_session::StoreRecords::new(self.conn, self.store_dir),
68 root,
69 )?;
70 self.verified_store_authority.circle_replay_epoch_index_on(
71 crate::store::store_session::StoreRecords::new(self.conn, self.store_dir),
72 )
73 }
74}
75
76pub(crate) fn circle_bootstrap_coverage_ref_on(
77 conn: &Connection,
78 circle_id: coven_protocol::circle::CircleId,
79) -> Result<Option<coven_protocol::circle::CircleBootstrapCoverageRef>, DbError> {
80 let row: Option<(String, String, String, String, Vec<u8>)> = conn
81 .query_row(
82 "SELECT control_coord, activation_commit, exact_cut, image_hash, bootstrap_ref
83 FROM circle_bootstrap_coverage WHERE circle_id = ?1",
84 [circle_id.to_string()],
85 |row| {
86 Ok((
87 row.get(0)?,
88 row.get(1)?,
89 row.get(2)?,
90 row.get(3)?,
91 row.get(4)?,
92 ))
93 },
94 )
95 .optional()
96 .map_err(DbError::from)?;
97 let Some((control, activation_commit, exact_cut, image_hash, bootstrap_ref)) = row else {
98 return Ok(None);
99 };
100 decode_circle_bootstrap_coverage_ref(
101 circle_id,
102 control,
103 activation_commit,
104 exact_cut,
105 image_hash,
106 bootstrap_ref,
107 )
108 .map(Some)
109}
110
111pub(crate) fn circle_bootstrap_coverage_refs_on(
112 conn: &Connection,
113) -> Result<Vec<coven_protocol::circle::CircleBootstrapCoverageRef>, DbError> {
114 let rows = query_mapped_rows(
115 conn,
116 "SELECT circle_id, control_coord, activation_commit, exact_cut,
117 image_hash, bootstrap_ref
118 FROM circle_bootstrap_coverage ORDER BY circle_id",
119 [],
120 |row| {
121 Ok((
122 row.get::<_, String>(0)?,
123 row.get::<_, String>(1)?,
124 row.get::<_, String>(2)?,
125 row.get::<_, String>(3)?,
126 row.get::<_, String>(4)?,
127 row.get::<_, Vec<u8>>(5)?,
128 ))
129 },
130 )?;
131 let mut bootstraps = Vec::with_capacity(rows.len());
132 for (circle_id, control, activation_commit, exact_cut, image_hash, encoded_reference) in rows {
133 let circle_id: coven_protocol::circle::CircleId = circle_id
134 .parse()
135 .map_err(|error| DbError::context("parse retained Circle bootstrap id", error))?;
136 bootstraps.push(decode_circle_bootstrap_coverage_ref(
137 circle_id,
138 control,
139 activation_commit,
140 exact_cut,
141 image_hash,
142 encoded_reference,
143 )?);
144 }
145 Ok(bootstraps)
146}
147
148fn decode_circle_bootstrap_coverage_ref(
149 circle_id: coven_protocol::circle::CircleId,
150 control: String,
151 activation_commit: String,
152 exact_cut: String,
153 image_hash: String,
154 encoded_reference: Vec<u8>,
155) -> Result<coven_protocol::circle::CircleBootstrapCoverageRef, DbError> {
156 let control = serde_json::from_str(&control)
157 .map_err(|error| DbError::context("parse retained Circle bootstrap control", error))?;
158 let activation_commit = serde_json::from_str(&activation_commit)
159 .map_err(|error| DbError::context("parse retained Circle bootstrap activation", error))?;
160 let exact_cut: CommitFrontier = serde_json::from_str(&exact_cut)
161 .map_err(|error| DbError::context("parse retained Circle bootstrap coverage", error))?;
162 let bootstrap: coven_protocol::circle::CircleBootstrapRef =
163 serde_json::from_slice(&encoded_reference).map_err(|error| {
164 DbError::context("parse retained Circle bootstrap reference", error)
165 })?;
166 if serde_json::to_vec(&bootstrap)
167 .map_err(|error| DbError::context("serialize retained Circle bootstrap reference", error))?
168 != encoded_reference
169 || bootstrap.coverage != exact_cut
170 || bootstrap.image.image_hash.to_string() != image_hash
171 {
172 return Err(DbError::Message(
173 "retained Circle bootstrap row differs from its exact reference".to_string(),
174 ));
175 }
176 Ok(coven_protocol::circle::CircleBootstrapCoverageRef {
177 circle_id,
178 control,
179 activation_commit,
180 bootstrap,
181 })
182}
183
184impl StoreDatabase {
185 pub(crate) fn snapshot_circle_packages_after(
186 snapshot: &coven_protocol::store_commit::RetainedReplaySnapshotAuthority,
187 epochs: &CircleReplayEpochIndex,
188 cuts: &BTreeMap<coven_protocol::circle::CircleId, CommitFrontier>,
189 ) -> Result<Vec<coven_protocol::store_commit::RetainedPackageActivation>, DbError> {
190 let mut packages = Vec::new();
191 for retained in snapshot.metadata.history_summary.reclaim.packages.values() {
192 let coven_protocol::reclaim::AudienceBlobBindingPackage::Circle(package) =
193 &retained.package
194 else {
195 continue;
196 };
197 let Some(cut) = cuts.get(&package.circle_id) else {
198 continue;
199 };
200 if !cut.covers_commit(&retained.activation)
201 && epochs.permits(&retained.activation, package.circle_id, &package.control)?
202 {
203 packages.push(retained.clone());
204 }
205 }
206 Ok(packages)
207 }
208
209 pub async fn circle_snapshot_package_inputs(
210 &self,
211 cuts: BTreeMap<coven_protocol::circle::CircleId, CommitFrontier>,
212 ) -> Result<Vec<coven_protocol::store_commit::RetainedPackageActivation>, DbError> {
213 self.call_store(move |session| {
214 let root = session.required_root_authority()?;
215 let baseline = session
216 .verified_store_authority
217 .retained_replay_baseline_on(crate::store::store_session::StoreRecords::new(
218 session.conn,
219 session.store_dir,
220 ))?
221 .clone();
222 let RetainedReplayAuthority::InstalledSnapshot(snapshot) = &baseline.authority else {
223 return Err(DbError::Message(
224 "Circle package restoration requires an installed snapshot".into(),
225 ));
226 };
227 let epochs = session.circle_replay_epoch_index(&root)?;
228 Self::snapshot_circle_packages_after(snapshot, &epochs, &cuts)
229 })
230 .await
231 }
232
233 pub async fn prepare_circle_restore_selection(
234 &self,
235 ) -> Result<CircleRestoreSelectionIndex, DbError> {
236 self.call_store(|session| session.prepare_circle_restore_selection())
237 .await
238 }
239
240 pub async fn retained_merge_materialization_by_ref(
241 &self,
242 root: coven_protocol::store_commit::StoreRootRef,
243 reference: StoreBatchCommitRef,
244 ) -> Result<OwnedVerifiedMergeMaterialization, DbError> {
245 self.call_store(move |session| {
246 session.retained_merge_materialization_by_ref(&root, &reference)
247 })
248 .await
249 }
250
251 pub async fn circle_replay_epoch_index(
252 &self,
253 root: coven_protocol::store_commit::StoreRootRef,
254 ) -> Result<CircleReplayEpochIndex, DbError> {
255 self.call_store(move |session| session.circle_replay_epoch_index(&root))
256 .await
257 }
258
259 #[cfg(any(test, feature = "test-utils"))]
260 pub(crate) fn circle_bootstrap_replay_inputs_on(
261 records: StoreRecords<'_>,
262 ) -> Result<
263 Vec<(
264 StoreBatchCommitRef,
265 coven_protocol::circle_activation::VerifiedCircleImage,
266 )>,
267 DbError,
268 > {
269 records
270 .claimed_circle_bootstrap_coverage_refs()?
271 .into_iter()
272 .map(|coverage| {
273 let image_bytes = records.payload(coverage.bootstrap.image.image_hash)?;
274 let image =
275 coven_protocol::circle_activation::VerifiedCircleImage::from_stored_image(
276 coverage.circle_id,
277 coverage.control,
278 coverage.bootstrap,
279 image_bytes,
280 )
281 .map_err(DbError::from)?;
282 Ok((coverage.activation_commit, image))
283 })
284 .collect()
285 }
286}
287
288#[cfg(any(test, feature = "test-utils"))]
289impl crate::store::store_session::StoreTransaction<'_, '_> {
290 pub(crate) fn circle_bootstrap_replay_inputs(
291 self,
292 ) -> Result<
293 Vec<(
294 StoreBatchCommitRef,
295 coven_protocol::circle_activation::VerifiedCircleImage,
296 )>,
297 DbError,
298 > {
299 StoreDatabase::circle_bootstrap_replay_inputs_on(
300 crate::store::store_session::StoreRecords::new(self.transaction, self.store_dir),
301 )
302 }
303}