1use super::*;
2
3pub struct CloudOutboxRecords<'connection> {
4 connection: &'connection Connection,
5}
6
7impl<'connection> CloudOutboxRecords<'connection> {
8 pub(crate) fn new(connection: &'connection Connection) -> Self {
9 Self { connection }
10 }
11
12 fn upload_entry_for_identity(
13 &self,
14 table: &str,
15 row_id: &str,
16 column: &str,
17 row_stamp: &str,
18 ) -> Result<Option<OutboxEntry>, DbError> {
19 self.connection
20 .query_row(
21 "SELECT id, operation, row_ref, stored_ref, source_path, retain_pinned,
22 upload_state, attempt_count, last_attempt_at, root_table, root_id
23 FROM cloud_outbox
24 WHERE operation = 'upload' AND table_name = ?1 AND row_id = ?2
25 AND column_name = ?3 AND row_stamp = ?4",
26 rusqlite::params![table, row_id, column, row_stamp],
27 row_to_outbox_entry,
28 )
29 .optional()
30 .map_err(DbError::from)
31 }
32
33 pub fn consume_created_upload_handoff(
34 &self,
35 package: &AudiencePackage,
36 binding: &RowBlobLocatorBinding,
37 ) -> Result<bool, DbError> {
38 let Some(entry) = self.upload_entry_for_identity(
39 binding.table(),
40 binding.row_id(),
41 binding.column(),
42 binding.row_stamp(),
43 )?
44 else {
45 return Ok(false);
46 };
47 let OutboxOperation::Upload {
48 row,
49 state: OutboxUploadState::Created {
50 authority, stored, ..
51 },
52 ..
53 } = &entry.operation
54 else {
55 return Err(DbError::Message(format!(
56 "activated blob binding {}/{}/{} at {} has an upload that is not Created",
57 binding.table(),
58 binding.row_id(),
59 binding.column(),
60 binding.row_stamp()
61 )));
62 };
63 if row.table() != binding.table()
64 || row.row_id() != binding.row_id()
65 || row.column() != binding.column()
66 || row.row_stamp() != binding.row_stamp()
67 || authority != package.audience()
68 || stored != binding.blob()
69 {
70 return Err(DbError::Message(format!(
71 "activated blob binding {}/{}/{} at {} differs from its Created upload handoff",
72 binding.table(),
73 binding.row_id(),
74 binding.column(),
75 binding.row_stamp()
76 )));
77 }
78 self.remove_entry(&entry)?;
79 Ok(true)
80 }
81
82 pub fn created_upload_handoff(
83 &self,
84 table: &str,
85 row_id: &str,
86 column: &str,
87 row_stamp: &str,
88 ) -> Result<Option<StoreWriteRemoteBlob>, DbError> {
89 let Some(entry) = self.upload_entry_for_identity(table, row_id, column, row_stamp)? else {
90 return Ok(None);
91 };
92 let OutboxOperation::Upload { row, state, .. } = entry.operation else {
93 return Err(DbError::Message(
94 "upload identity query returned a non-upload operation".to_string(),
95 ));
96 };
97 if row.table() != table
98 || row.row_id() != row_id
99 || row.column() != column
100 || row.row_stamp() != row_stamp
101 {
102 return Err(DbError::Message(format!(
103 "upload outbox row facts differ from identity {table}/{row_id}/{column} at {row_stamp}"
104 )));
105 }
106 match state {
107 OutboxUploadState::Created {
108 authority, stored, ..
109 } => Ok(Some(StoreWriteRemoteBlob { authority, stored })),
110 OutboxUploadState::Pending | OutboxUploadState::Prepared { .. } => Ok(None),
111 }
112 }
113
114 pub fn upload_entries_for_rows(
115 &self,
116 rows: &[RowBlobRef],
117 ) -> Result<Vec<OutboxEntry>, DbError> {
118 rows.iter()
119 .filter_map(|row| {
120 match self.upload_entry_for_identity(
121 row.table(),
122 row.row_id(),
123 row.column(),
124 row.row_stamp(),
125 ) {
126 Ok(Some(entry)) => Some(Ok(entry)),
127 Ok(None) => None,
128 Err(error) => Some(Err(error)),
129 }
130 })
131 .collect()
132 }
133
134 pub fn upload_entries_for_root(
135 &self,
136 gates: &Gates,
137 tables: &[SyncedTable],
138 root_table: &str,
139 root_id: &str,
140 ) -> Result<Vec<OutboxEntry>, DbError> {
141 let rows = Database::row_blob_refs_for_root_on(
142 self.connection,
143 gates,
144 tables,
145 root_table,
146 root_id,
147 )?;
148 self.upload_entries_for_rows(&rows)
149 }
150
151 #[allow(clippy::too_many_arguments)]
152 pub fn enqueue_upload(
153 &self,
154 root_table: &str,
155 root_id: &str,
156 root_label: &str,
157 row: &RowBlobRef,
158 source_path: &Path,
159 retain_pinned: bool,
160 created_at: &str,
161 ) -> Result<(), DbError> {
162 if row.authority() != &RowBlobAuthority::Local || row.stored().is_some() {
163 return Err(DbError::Message(
164 "cloud upload requires an exact Local row blob reference".to_string(),
165 ));
166 }
167 let source_path = source_path.to_str().ok_or_else(|| {
168 DbError::Message(format!(
169 "blob source path for {}/{}/{} is not UTF-8: {source_path:?}",
170 row.table(),
171 row.row_id(),
172 row.column()
173 ))
174 })?;
175 let encoded = serde_json::to_string(row)
176 .map_err(|error| DbError::context("serialize row blob ref", error))?;
177 let pending = serde_json::to_string(&OutboxUploadState::Pending)
178 .map_err(|error| DbError::context("serialize pending blob upload state", error))?;
179 self.connection
180 .execute(
181 "INSERT INTO cloud_outbox
182 (operation, table_name, row_id, column_name, row_stamp, root_table, root_id,
183 root_label, row_ref, upload_state, source_path, retain_pinned, created_at)
184 VALUES ('upload', ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)
185 ON CONFLICT(operation, table_name, row_id, column_name, row_stamp) DO UPDATE SET
186 root_table = excluded.root_table,
187 root_id = excluded.root_id,
188 root_label = excluded.root_label,
189 source_path = excluded.source_path,
190 retain_pinned = excluded.retain_pinned,
191 attempt_count = 0,
192 last_error = NULL,
193 last_attempt_at = NULL
194 WHERE cloud_outbox.row_ref = excluded.row_ref
195 AND cloud_outbox.root_table = excluded.root_table
196 AND cloud_outbox.root_id = excluded.root_id",
197 rusqlite::params![
198 row.table(),
199 row.row_id(),
200 row.column(),
201 row.row_stamp(),
202 root_table,
203 root_id,
204 root_label,
205 encoded,
206 pending,
207 source_path,
208 retain_pinned,
209 created_at,
210 ],
211 )
212 .map_err(DbError::from)
213 .and_then(|changed| {
214 if changed == 1 {
215 Ok(())
216 } else {
217 Err(DbError::Message(format!(
218 "upload outbox identity {}/{}/{}/{} carries different row facts",
219 row.table(),
220 row.row_id(),
221 row.column(),
222 row.row_stamp()
223 )))
224 }
225 })
226 }
227
228 pub fn enqueue_delete(&self, stored: &StoredBlobRef, created_at: &str) -> Result<(), DbError> {
229 let encoded = serde_json::to_string(stored)
230 .map_err(|error| DbError::context("serialize stored blob ref", error))?;
231 crate::with_coven_sql_authority(|| {
232 self.connection
233 .execute(
234 "INSERT INTO cloud_outbox (operation, stored_ref, created_at)
235 VALUES ('delete', ?1, ?2)
236 ON CONFLICT(stored_ref) DO UPDATE SET
237 created_at = excluded.created_at,
238 attempt_count = 0,
239 last_error = NULL,
240 last_attempt_at = NULL",
241 (encoded, created_at),
242 )
243 .map(|_| ())
244 .map_err(DbError::from)
245 })
246 }
247
248 pub fn remove_entry(&self, entry: &OutboxEntry) -> Result<(), DbError> {
249 let identity = outbox_identity(&entry.operation)?;
250 let removed = match identity {
251 OutboxIdentity::Upload {
252 table,
253 row_id,
254 column,
255 row_stamp,
256 } => self.connection.execute(
257 "DELETE FROM cloud_outbox WHERE id = ?1 AND operation = 'upload'
258 AND table_name = ?2 AND row_id = ?3 AND column_name = ?4 AND row_stamp = ?5",
259 rusqlite::params![entry.id, table, row_id, column, row_stamp],
260 ),
261 OutboxIdentity::Stored { operation, stored } => self.connection.execute(
262 "DELETE FROM cloud_outbox WHERE id = ?1 AND operation = ?2 AND stored_ref = ?3",
263 rusqlite::params![entry.id, operation, stored],
264 ),
265 }
266 .map_err(DbError::from)?;
267 if removed != 1 {
268 return Err(DbError::Message(
269 "cloud outbox entry changed before exact dequeue".to_string(),
270 ));
271 }
272 Ok(())
273 }
274
275 pub fn finish_cancelled_upload(&self, entry: &OutboxEntry) -> Result<bool, DbError> {
276 let OutboxOperation::Upload {
277 root_table,
278 root_id,
279 ..
280 } = &entry.operation
281 else {
282 return Err(DbError::Message(
283 "make_remote cleanup requires an upload entry".to_string(),
284 ));
285 };
286 if !matches!(
287 Database::make_remote_intent_state(self.connection, root_table, root_id)?,
288 Some(MakeRemoteIntentState::Cancelling)
289 ) {
290 return Err(DbError::Message(format!(
291 "make_remote cleanup for {root_table:?}/{root_id:?} lost cancellation ownership"
292 )));
293 }
294 self.remove_entry(entry)?;
295 let remaining: i64 = self
296 .connection
297 .query_row(
298 "SELECT COUNT(*) FROM cloud_outbox
299 WHERE operation = 'upload' AND root_table = ?1 AND root_id = ?2",
300 (root_table, root_id),
301 |row| row.get(0),
302 )
303 .map_err(DbError::from)?;
304 if remaining != 0 {
305 return Ok(false);
306 }
307 let removed = self
308 .connection
309 .execute(
310 "DELETE FROM blob_make_remote_intents
311 WHERE root_table = ?1 AND root_id = ?2 AND state = 'cancelling'",
312 (root_table, root_id),
313 )
314 .map_err(DbError::from)?;
315 if removed != 1 {
316 return Err(DbError::Message(format!(
317 "make_remote cancellation {root_table:?}/{root_id:?} changed before completion"
318 )));
319 }
320 Ok(true)
321 }
322}
323
324pub enum OutboxIdentity {
325 Upload {
326 table: String,
327 row_id: String,
328 column: String,
329 row_stamp: String,
330 },
331 Stored {
332 operation: &'static str,
333 stored: String,
334 },
335}
336
337pub fn outbox_identity(operation: &OutboxOperation) -> Result<OutboxIdentity, DbError> {
338 match operation {
339 OutboxOperation::Upload { row, .. } => Ok(OutboxIdentity::Upload {
340 table: row.table().to_string(),
341 row_id: row.row_id().to_string(),
342 column: row.column().to_string(),
343 row_stamp: row.row_stamp().to_string(),
344 }),
345 OutboxOperation::Delete { stored } => Ok(OutboxIdentity::Stored {
346 operation: "delete",
347 stored: serde_json::to_string(stored).map_err(|error| {
348 DbError::context("serialize stored blob outbox identity", error)
349 })?,
350 }),
351 }
352}
353
354pub fn row_to_outbox_entry(row: &rusqlite::Row<'_>) -> rusqlite::Result<OutboxEntry> {
355 fn invalid(
356 index: usize,
357 source: impl std::error::Error + Send + Sync + 'static,
358 ) -> rusqlite::Error {
359 rusqlite::Error::FromSqlConversionFailure(
360 index,
361 rusqlite::types::Type::Text,
362 Box::new(source),
363 )
364 }
365
366 let tag: String = row.get(1)?;
367 let operation = match tag.as_str() {
368 "upload" => {
369 let encoded: String = row.get(2)?;
370 let reference: RowBlobRef =
371 serde_json::from_str(&encoded).map_err(|error| invalid(2, error))?;
372 let source_path: String = row.get(4)?;
373 let state_json: String = row.get(6)?;
374 let state: OutboxUploadState =
375 serde_json::from_str(&state_json).map_err(|error| invalid(6, error))?;
376 if let OutboxUploadState::Prepared {
377 authority, stored, ..
378 }
379 | OutboxUploadState::Created {
380 authority, stored, ..
381 } = &state
382 {
383 let locator = stored.locator();
384 if !coven_protocol::blob::locator_describes_row(
385 locator,
386 reference.blob(),
387 reference.plaintext_size(),
388 reference.plaintext_hash(),
389 ) {
390 return Err(invalid(
391 6,
392 std::io::Error::other("prepared upload differs from its exact row version"),
393 ));
394 }
395 if locator.audience() != authority.remote_audience() {
396 return Err(invalid(
397 6,
398 std::io::Error::other(
399 "upload package authority differs from its stored locator",
400 ),
401 ));
402 }
403 }
404 OutboxOperation::Upload {
405 root_table: row.get(9)?,
406 root_id: row.get(10)?,
407 row: reference,
408 source_path: PathBuf::from(source_path),
409 retain_pinned: row.get(5)?,
410 state,
411 }
412 }
413 "delete" => {
414 let encoded: String = row.get(3)?;
415 let stored: StoredBlobRef =
416 serde_json::from_str(&encoded).map_err(|error| invalid(3, error))?;
417 OutboxOperation::Delete { stored }
418 }
419 _ => {
420 return Err(invalid(
421 1,
422 std::io::Error::other(format!("invalid cloud outbox operation {tag:?}")),
423 ))
424 }
425 };
426 Ok(OutboxEntry {
427 id: row.get(0)?,
428 attempt_count: row.get(7)?,
429 last_attempt_at: row.get(8)?,
430 operation,
431 })
432}