1use std::collections::BTreeSet;
2
3use crate::*;
4use coven_protocol::circle::CircleId;
5use coven_protocol::store_commit::{
6 circle_snapshot_image_semantic_prefix, circle_snapshot_slot_prefix, CircleSnapshotMeta,
7 CircleSnapshotRef,
8};
9use rusqlite::OptionalExtension;
10
11use super::*;
12
13impl StoreSession<'_> {
14 fn outbound_circle_snapshot_publication(
15 &mut self,
16 circle_id: CircleId,
17 ) -> Result<Option<DurableCircleSnapshotPublication>, DbError> {
18 let authority = self.local_store_authority()?;
19 load_outbound_circle_snapshot_on(self.conn, self.store_dir, &authority, circle_id)
20 }
21
22 fn latest_local_circle_snapshot(
23 &mut self,
24 circle_id: CircleId,
25 ) -> Result<Option<PublishedCircleSnapshot>, DbError> {
26 let authority = self.local_store_authority()?;
27 load_published_circle_snapshot_on(self.conn, &authority, circle_id)
28 }
29
30 fn stage_circle_snapshot_publication(
31 &mut self,
32 meta: CircleSnapshotMeta,
33 meta_prepared: PreparedExactObject,
34 image: SnapshotDatabaseImage,
35 image_prepared: PreparedExactObject,
36 blobs: Vec<PreparedSnapshotBlob>,
37 ) -> Result<CircleSnapshotRef, DbError> {
38 let authority = self.local_store_authority()?;
39 let tx = self.conn.unchecked_transaction().map_err(DbError::from)?;
40 let image_facts =
41 crate::payload_store::write_payload_file_blocking(&tx, self.store_dir, image.path())
42 .map_err(|source| SnapshotImageError::ProjectionPayloadStore {
43 operation: "spool Circle snapshot image".to_string(),
44 source,
45 });
46 let (image_hash, _) = image.finish(image_facts).map_err(snapshot_image_db_error)?;
47 let image_prepared_hash = crate::payload_store::write_payload_blocking(
48 &tx,
49 self.store_dir,
50 image_prepared.stored_bytes(),
51 )
52 .map_err(|error| DbError::context("spool prepared Circle snapshot image", error))?;
53 let image_prepared_size = image_prepared.stored_bytes().len() as u64;
54 let registration_ref = authority.reference();
55 let registration = authority.value();
56 validate_snapshot_author(&meta.author_registration, registration_ref, "Circle")?;
57 let device_id = registration.device_id.to_string();
58 validate_snapshot_image(
59 &meta.bootstrap.image,
60 &image_prepared,
61 image_hash,
62 image_prepared_hash,
63 image_prepared_size,
64 format!(
65 "{}.db",
66 circle_snapshot_image_semantic_prefix(
67 meta.circle_id,
68 &device_id,
69 meta.bootstrap.image.image_hash,
70 )
71 ),
72 "Circle",
73 )?;
74 let reference = CircleSnapshotRef {
75 generation: meta.generation,
76 snapshot_hash: meta.snapshot_hash(),
77 object: meta_prepared.reference().clone(),
78 };
79 let verified = CircleSnapshotMeta::parse_at(
80 &meta.to_bytes(),
81 registration.store_root.store_root_hash,
82 &reference,
83 registration,
84 )
85 .map_err(|error| DbError::context("verify staged Circle snapshot metadata", error))?;
86 if verified != meta {
87 return Err(DbError::Message(
88 "staged Circle snapshot changed during exact verification".to_string(),
89 ));
90 }
91 let previous = load_published_circle_snapshot_on(&tx, &authority, meta.circle_id)?;
92 let (expected_generation, expected_slot) = match &previous {
93 Some(previous) => (
94 previous
95 .reference
96 .generation
97 .checked_add(1)
98 .ok_or_else(|| {
99 DbError::Message("Circle snapshot generation overflow".to_string())
100 })?,
101 previous.successor_slot.clone(),
102 ),
103 None => (
104 0,
105 coven_protocol::objects::ObjectSlot::logical(format!(
106 "{}.json",
107 circle_snapshot_slot_prefix(meta.circle_id, &device_id, 0)
108 ))
109 .map_err(DbError::from)?,
110 ),
111 };
112 if meta.generation != expected_generation
113 || meta_prepared.reference().slot() != &expected_slot
114 || meta.successor.predecessor != previous.as_ref().map(|value| value.reference.clone())
115 {
116 return Err(DbError::Message(
117 "Circle snapshot does not extend the exact local stream".to_string(),
118 ));
119 }
120 let next_generation = meta
121 .generation
122 .checked_add(1)
123 .ok_or_else(|| DbError::Message("Circle snapshot generation overflow".to_string()))?;
124 let activation = coven_protocol::store_commit::circle_snapshot_stream_activation(
125 registration.store_root.store_root_hash,
126 registration_ref,
127 meta.circle_id,
128 &device_id,
129 )
130 .map_err(DbError::from)?;
131 if meta.successor.activation != activation
132 || meta.successor.next_slot.logical_key()
133 != format!(
134 "{}.json",
135 circle_snapshot_slot_prefix(meta.circle_id, &device_id, next_generation)
136 )
137 {
138 return Err(DbError::Message(
139 "Circle snapshot successor is outside its activated exact stream".to_string(),
140 ));
141 }
142 let snapshot_owner = coven_protocol::remote_object::SnapshotObjectOwner {
143 activation: meta.successor.activation,
144 generation: meta.generation,
145 };
146 validate_snapshot_blob_plans_on(
147 self.conn,
148 self.gates,
149 self.synced_tables,
150 &snapshot_owner,
151 &blobs,
152 )?;
153 tx.execute(
154 "INSERT INTO outbound_circle_snapshot \
155 (circle_id, snapshot_ref, meta_prepared, image_ref, meta_bytes, blobs) \
156 VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
157 rusqlite::params![
158 meta.circle_id.to_string(),
159 serde_json::to_string(&reference).map_err(|error| {
160 DbError::context("serialize exact Circle snapshot ref", error)
161 })?,
162 serde_json::to_string(&meta_prepared).map_err(|error| {
163 DbError::context("serialize prepared Circle snapshot metadata", error)
164 })?,
165 serde_json::to_string(&meta.bootstrap.image).map_err(|error| {
166 DbError::context("serialize exact Circle snapshot image ref", error)
167 })?,
168 meta.to_bytes(),
169 serde_json::to_string(&blobs).map_err(|error| {
170 DbError::context("serialize prepared Circle snapshot blobs", error)
171 })?,
172 ],
173 )
174 .map_err(DbError::from)?;
175 crate::payload_store::set_payload_owner_claims_on(
176 &tx,
177 &crate::payload_store::outbound_circle_snapshot_owner_key(meta.circle_id),
178 &BTreeSet::from([image_hash, image_prepared_hash]),
179 )?;
180 tx.commit().map_err(DbError::from)?;
181 Ok(reference)
182 }
183
184 fn complete_circle_snapshot_publication(
185 &mut self,
186 accepted: CircleSnapshotRef,
187 ) -> Result<(), DbError> {
188 let authority = self.local_store_authority()?;
189 let tx = self.conn.unchecked_transaction().map_err(DbError::from)?;
190 let circle_id = {
191 let bytes: Vec<u8> = tx
192 .query_row(
193 "SELECT meta_bytes FROM outbound_circle_snapshot \
194 WHERE snapshot_ref = ?1",
195 [serde_json::to_string(&accepted).map_err(|error| {
196 DbError::context("serialize accepted Circle snapshot ref", error)
197 })?],
198 |row| row.get::<_, Vec<u8>>(0),
199 )
200 .optional()
201 .map_err(DbError::from)?
202 .ok_or_else(|| {
203 DbError::Message("outbound Circle snapshot is absent".to_string())
204 })?;
205 let meta: CircleSnapshotMeta = serde_json::from_slice(&bytes)
206 .map_err(|error| DbError::context("accepted Circle snapshot metadata", error))?;
207 meta.circle_id
208 };
209 let outbound =
210 load_outbound_circle_snapshot_on(&tx, self.store_dir, &authority, circle_id)?
211 .ok_or_else(|| {
212 DbError::Message("outbound Circle snapshot is absent".to_string())
213 })?;
214 if outbound.reference != accepted {
215 return Err(DbError::Message(
216 "accepted Circle snapshot differs from the prepared exact object".to_string(),
217 ));
218 }
219 install_snapshot_blob_plans_on(&tx, &outbound.blobs)?;
220 let snapshot_owner = coven_protocol::remote_object::SnapshotObjectOwner {
221 activation: outbound.meta.value.successor.activation,
222 generation: outbound.meta.value.generation,
223 };
224 persist_snapshot_image_on(
225 &tx,
226 self.store_dir,
227 &outbound.meta.value.bootstrap.image,
228 snapshot_owner,
229 "Circle snapshot image",
230 )?;
231 let deleted = tx
232 .execute(
233 "DELETE FROM outbound_circle_snapshot WHERE circle_id = ?1",
234 [circle_id.to_string()],
235 )
236 .map_err(DbError::from)?;
237 if deleted != 1 {
238 return Err(DbError::Message(
239 "outbound Circle snapshot ownership row is absent or changed".to_string(),
240 ));
241 }
242 crate::payload_store::release_payload_owner_on(
243 &tx,
244 &crate::payload_store::outbound_circle_snapshot_owner_key(circle_id),
245 )?;
246 let accepted_generation =
247 snapshot_generation_as_i64(accepted.generation, "Circle snapshot")?;
248 tx.execute(
249 "INSERT INTO published_circle_snapshot \
250 (circle_id, generation, snapshot_ref, successor_slot, cut, meta_bytes) \
251 VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
252 rusqlite::params![
253 circle_id.to_string(),
254 accepted_generation,
255 serde_json::to_string(&accepted).map_err(|error| {
256 DbError::context("serialize published Circle snapshot ref", error)
257 })?,
258 serde_json::to_string(&outbound.meta.value.successor.next_slot).map_err(
259 |error| DbError::context("serialize Circle snapshot successor slot", error)
260 )?,
261 serde_json::to_string(&outbound.meta.value.bootstrap.coverage)
262 .map_err(|error| DbError::context("serialize Circle snapshot cut", error))?,
263 outbound.meta.bytes,
264 ],
265 )
266 .map_err(DbError::from)?;
267 tx.commit().map_err(DbError::from)
268 }
269}
270
271impl StoreDatabase {
272 pub async fn outbound_circle_snapshot_publication(
273 &self,
274 circle_id: CircleId,
275 ) -> Result<Option<DurableCircleSnapshotPublication>, DbError> {
276 let pending = self
277 .call_store(move |session| session.outbound_circle_snapshot_publication(circle_id))
278 .await?;
279 if let Some(pending) = &pending {
280 verify_snapshot_blob_spools(&pending.blobs, "prepared Circle").await?;
281 }
282 Ok(pending)
283 }
284
285 pub async fn latest_local_circle_snapshot(
286 &self,
287 circle_id: CircleId,
288 ) -> Result<Option<PublishedCircleSnapshot>, DbError> {
289 self.call_store(move |session| session.latest_local_circle_snapshot(circle_id))
290 .await
291 }
292
293 pub async fn stage_circle_snapshot_publication(
294 &self,
295 meta: CircleSnapshotMeta,
296 meta_prepared: PreparedExactObject,
297 image: SnapshotDatabaseImage,
298 image_prepared: PreparedExactObject,
299 blobs: Vec<PreparedSnapshotBlob>,
300 ) -> Result<CircleSnapshotRef, DbError> {
301 self.call_store(move |session| {
302 session.stage_circle_snapshot_publication(
303 meta,
304 meta_prepared,
305 image,
306 image_prepared,
307 blobs,
308 )
309 })
310 .await
311 }
312
313 pub async fn complete_circle_snapshot_publication(
314 &self,
315 accepted: CircleSnapshotRef,
316 ) -> Result<(), DbError> {
317 self.call_store(move |session| session.complete_circle_snapshot_publication(accepted))
318 .await
319 }
320}