1use std::collections::HashSet;
2use std::path::Path;
3use std::time::Instant;
4
5use anyhow::{Context, Result, bail};
6use sqlx::{QueryBuilder, Sqlite, SqliteConnection};
7
8use super::apply::apply_remote_change;
9use super::wire::{ChangeRow, ChangeWire, PushAck, SyncRequest, SyncResponse};
10use crate::change_log::op_type;
11use crate::db::{Database, begin_immediate, get_meta, set_meta};
12
13#[derive(Debug, Clone, Default, PartialEq, Eq)]
14pub struct SyncPersistenceStatus {
15 pub pinned_server: Option<String>,
16 pub pending_changes: i64,
17 pub conflicts: i64,
18 pub sync_cursor: Option<String>,
19 pub local_sequence: Option<String>,
20 pub last_attempt: Option<String>,
21 pub last_success: Option<String>,
22 pub last_error: Option<String>,
23 pub last_pushed: Option<String>,
24 pub last_pulled: Option<String>,
25 pub last_cursor: Option<String>,
26}
27
28#[derive(Debug, Clone)]
29pub struct ClientSyncPage {
30 pub request: SyncRequest,
31 pub pending: usize,
32}
33
34#[derive(Debug)]
35pub struct ApplySyncPage {
36 pub request: SyncRequest,
37 pub response: SyncResponse,
38 pub attempted_at: String,
39 pub previous_pushed: i64,
40 pub previous_pulled: usize,
41}
42
43#[derive(Debug)]
44pub struct ServerSyncPage {
45 pub request: SyncRequest,
46}
47
48#[derive(Debug)]
49pub struct ServerSyncResult {
50 pub accepted_count: i64,
51 pub push_acks: Vec<PushAck>,
52 pub changes: Vec<ChangeWire>,
53 pub has_more: bool,
54 pub assign_ms: u128,
55 pub pull_query_ms: u128,
56}
57
58impl Database {
59 pub async fn sync_persistence_status(&self) -> Result<SyncPersistenceStatus> {
60 let mut conn = self.acquire().await?;
61 let pending_changes =
62 sqlx::query_scalar("SELECT count(*) FROM changes WHERE server_seq IS NULL")
63 .fetch_one(&mut *conn)
64 .await?;
65 let conflicts = sqlx::query_scalar("SELECT count(*) FROM conflicts WHERE resolved = 0")
66 .fetch_one(&mut *conn)
67 .await?;
68 Ok(SyncPersistenceStatus {
69 pinned_server: get_meta(&mut conn, "sync_server_url").await?,
70 pending_changes,
71 conflicts,
72 sync_cursor: get_meta(&mut conn, "sync_cursor").await?,
73 local_sequence: get_meta(&mut conn, "local_seq").await?,
74 last_attempt: get_meta(&mut conn, "sync_last_attempt_at").await?,
75 last_success: get_meta(&mut conn, "sync_last_success_at").await?,
76 last_error: get_meta(&mut conn, "sync_last_error").await?,
77 last_pushed: get_meta(&mut conn, "sync_last_pushed").await?,
78 last_pulled: get_meta(&mut conn, "sync_last_pulled").await?,
79 last_cursor: get_meta(&mut conn, "sync_last_cursor").await?,
80 })
81 }
82
83 pub async fn begin_sync_attempt(&self, attempted_at: String) -> Result<()> {
84 let mut conn = self.acquire().await?;
85 set_meta(&mut conn, "sync_last_attempt_at", &attempted_at).await
86 }
87
88 pub async fn record_sync_error(&self, error: String) -> Result<()> {
89 let mut conn = self.acquire().await?;
90 set_meta(&mut conn, "sync_last_error", &error).await
91 }
92
93 pub(super) async fn pending_sync_change_count(&self) -> Result<i64> {
94 let mut conn = self.acquire().await?;
95 Ok(
96 sqlx::query_scalar("SELECT COUNT(*) FROM changes WHERE server_seq IS NULL")
97 .fetch_one(&mut *conn)
98 .await?,
99 )
100 }
101
102 pub(super) async fn pending_blob_contracts(
103 &self,
104 ) -> Result<Vec<super::wire::BlobUploadContract>> {
105 let mut conn = self.acquire().await?;
106 let changes = load_unsynced_changes(&mut conn, i64::MAX as usize).await?;
107 super::blob::unique_blob_contracts(&changes)
108 }
109
110 pub async fn prepare_client_sync_page(
111 &self,
112 server: String,
113 push_limit: usize,
114 pull_limit: u32,
115 ) -> Result<ClientSyncPage> {
116 let mut conn = self.acquire().await?;
117 validate_sync_server(&mut conn, &server).await?;
118 let client_id = get_meta(&mut conn, "client_id")
119 .await?
120 .context("missing client id")?;
121 let after = sync_cursor(&mut conn).await?;
122 let changes = load_unsynced_changes(&mut conn, push_limit).await?;
123 let pending = changes.len();
124 Ok(ClientSyncPage {
125 request: SyncRequest {
126 protocol_version: Some(super::wire::SYNC_PROTOCOL_VERSION),
127 client_id,
128 after,
129 pull_limit: Some(pull_limit),
130 changes,
131 },
132 pending,
133 })
134 }
135
136 pub async fn apply_client_sync_page(&self, page: ApplySyncPage) -> Result<usize> {
137 let envelope = super::wire::validate_sync_request_envelope(&page.request)?;
138 let request_change_ids = page
139 .request
140 .changes
141 .iter()
142 .map(|change| change.change_id.clone())
143 .collect::<Vec<_>>();
144 super::wire::validate_sync_response_for_request(
145 envelope.after,
146 envelope.pull_limit,
147 &request_change_ids,
148 &page.response,
149 )?;
150 let mut conn = self.acquire().await?;
151 apply_sync_response(&mut conn, page).await
152 }
153
154 pub async fn persist_server_sync_page(&self, page: ServerSyncPage) -> Result<ServerSyncResult> {
155 self.persist_server_sync_page_inner(page, None).await
156 }
157
158 pub async fn persist_server_sync_page_with_blobs(
159 &self,
160 page: ServerSyncPage,
161 blob_dir: &Path,
162 ) -> Result<ServerSyncResult> {
163 self.persist_server_sync_page_inner(page, Some(blob_dir))
164 .await
165 }
166
167 async fn persist_server_sync_page_inner(
168 &self,
169 page: ServerSyncPage,
170 blob_dir: Option<&Path>,
171 ) -> Result<ServerSyncResult> {
172 let envelope = super::wire::validate_sync_request_envelope(&page.request)?;
173 for change in &page.request.changes {
174 super::wire::validate_pushed_change(change)?;
175 }
176 let mut conn = self.acquire().await?;
177 let assign_started = Instant::now();
178 let (accepted_count, push_acks) =
179 assign_server_sequences(&mut conn, page.request.changes, blob_dir).await?;
180 let assign_ms = assign_started.elapsed().as_millis();
181 let pull_query_started = Instant::now();
182 let (changes, has_more) =
183 load_server_changes_after(&mut conn, envelope.after, envelope.pull_limit).await?;
184 let pull_query_ms = pull_query_started.elapsed().as_millis();
185 Ok(ServerSyncResult {
186 accepted_count,
187 push_acks,
188 changes,
189 has_more,
190 assign_ms,
191 pull_query_ms,
192 })
193 }
194}
195
196async fn sync_cursor(conn: &mut SqliteConnection) -> Result<i64> {
197 Ok(get_meta(conn, "sync_cursor")
198 .await?
199 .unwrap_or_else(|| "0".to_string())
200 .parse::<i64>()?)
201}
202
203async fn validate_sync_server(conn: &mut SqliteConnection, server: &str) -> Result<()> {
204 let normalized = server.trim_end_matches('/');
205 if let Some(existing) = get_meta(conn, "sync_server_url").await? {
206 if existing != normalized {
207 bail!(
208 "error sync-server-changed existing={} requested={} hint=\"use a fresh database for a different sync server\"",
209 existing,
210 normalized
211 );
212 }
213 } else {
214 set_meta(conn, "sync_server_url", normalized).await?;
215 }
216 Ok(())
217}
218
219async fn load_unsynced_changes(
220 conn: &mut SqliteConnection,
221 limit: usize,
222) -> Result<Vec<ChangeWire>> {
223 let limit = limit as i64;
224 let rows = sqlx::query_as!(
225 ChangeRow,
226 r#"SELECT change_id AS "change_id!: String", client_id AS "client_id!: String",
227 local_seq AS "local_seq!: i64", entity_type AS "entity_type!: String",
228 entity_id AS "entity_id!: String", field, op_type AS "op_type!: String",
229 payload AS "payload!: String", base_version, created_at AS "created_at!: String",
230 server_seq
231 FROM changes WHERE server_seq IS NULL ORDER BY local_seq, created_at LIMIT ?"#,
232 limit,
233 )
234 .fetch_all(&mut *conn)
235 .await?;
236 Ok(rows.into_iter().map(ChangeRow::into_wire).collect())
237}
238
239async fn apply_sync_response(conn: &mut SqliteConnection, page: ApplySyncPage) -> Result<usize> {
240 let mut applied = 0;
241 let mut tx = begin_immediate(conn).await?;
242 let current_cursor = sync_cursor(&mut tx).await?;
243 if current_cursor != page.request.after {
244 bail!(
245 "error stale-sync-page expected_cursor={} request_cursor={}",
246 current_cursor,
247 page.request.after
248 );
249 }
250 update_change_server_seqs_if_missing(&mut tx, &page.response.push_acks).await?;
251 let existing_change_ids = load_existing_change_ids(&mut tx, &page.response.changes).await?;
252 for change in &page.response.changes {
253 if existing_change_ids.contains(change.change_id.as_str()) {
254 update_change_server_seq(&mut tx, &change.change_id, change.server_seq).await?;
255 continue;
256 }
257 apply_remote_change(&mut tx, change).await?;
258 insert_wire_change(&mut tx, change).await?;
259 applied += 1;
260 }
261 let pushed = page.previous_pushed + page.response.push_acks.len() as i64;
262 let pulled = page.previous_pulled + applied;
263 set_meta(&mut tx, "sync_cursor", &page.response.cursor.to_string()).await?;
264 set_meta(&mut tx, "sync_last_success_at", &page.attempted_at).await?;
265 set_meta(&mut tx, "sync_last_error", "").await?;
266 set_meta(&mut tx, "sync_last_pushed", &pushed.to_string()).await?;
267 set_meta(&mut tx, "sync_last_pulled", &pulled.to_string()).await?;
268 set_meta(
269 &mut tx,
270 "sync_last_cursor",
271 &page.response.cursor.to_string(),
272 )
273 .await?;
274 tx.commit().await?;
275 Ok(applied)
276}
277
278async fn load_existing_change_ids(
279 conn: &mut SqliteConnection,
280 changes: &[ChangeWire],
281) -> Result<HashSet<String>> {
282 if changes.is_empty() {
283 return Ok(HashSet::new());
284 }
285 let mut query_builder =
286 QueryBuilder::<Sqlite>::new("SELECT change_id FROM changes WHERE change_id IN (");
287 let mut separated = query_builder.separated(", ");
288 for change in changes {
289 separated.push_bind(&change.change_id);
290 }
291 separated.push_unseparated(")");
292 Ok(query_builder
293 .build_query_scalar::<String>()
294 .fetch_all(&mut *conn)
295 .await?
296 .into_iter()
297 .collect())
298}
299
300async fn update_change_server_seq(
301 conn: &mut SqliteConnection,
302 change_id: &str,
303 server_seq: Option<i64>,
304) -> Result<()> {
305 if let Some(server_seq) = server_seq {
306 sqlx::query!(
307 "UPDATE changes SET server_seq = ? WHERE change_id = ? AND server_seq IS NULL",
308 server_seq,
309 change_id,
310 )
311 .execute(&mut *conn)
312 .await?;
313 }
314 Ok(())
315}
316
317async fn update_change_server_seqs_if_missing(
318 conn: &mut SqliteConnection,
319 push_acks: &[PushAck],
320) -> Result<()> {
321 if push_acks.is_empty() {
322 return Ok(());
323 }
324 let mut query_builder = QueryBuilder::<Sqlite>::new("WITH updates(change_id, server_seq) AS (");
325 query_builder.push_values(push_acks, |mut row, ack| {
326 row.push_bind(&ack.change_id).push_bind(ack.server_seq);
327 });
328 query_builder.push(
329 ") UPDATE changes
330 SET server_seq = (
331 SELECT updates.server_seq
332 FROM updates
333 WHERE updates.change_id = changes.change_id
334 )
335 WHERE server_seq IS NULL
336 AND change_id IN (SELECT change_id FROM updates)",
337 );
338 query_builder.build().execute(&mut *conn).await?;
339 Ok(())
340}
341
342async fn insert_wire_change(conn: &mut SqliteConnection, change: &ChangeWire) -> Result<()> {
343 let payload = change.payload.to_string();
344 sqlx::query!(
345 "INSERT OR IGNORE INTO changes(change_id, client_id, local_seq, entity_type, entity_id, field,
346 op_type, payload, base_version, created_at, server_seq)
347 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
348 change.change_id,
349 change.client_id,
350 change.local_seq,
351 change.entity_type,
352 change.entity_id,
353 change.field,
354 change.op_type,
355 payload,
356 change.base_version,
357 change.created_at,
358 change.server_seq,
359 )
360 .execute(&mut *conn)
361 .await?;
362 Ok(())
363}
364
365async fn assign_server_sequences(
366 conn: &mut SqliteConnection,
367 changes: Vec<ChangeWire>,
368 blob_dir: Option<&Path>,
369) -> Result<(i64, Vec<PushAck>)> {
370 if changes.is_empty() {
371 return Ok((0, Vec::new()));
372 }
373 let mut tx = begin_immediate(conn).await?;
374 let existing_change_ids = load_existing_change_ids(&mut tx, &changes).await?;
375 let unassigned_changes = changes
376 .iter()
377 .filter(|change| !existing_change_ids.contains(&change.change_id))
378 .cloned()
379 .collect::<Vec<_>>();
380 if unassigned_changes
381 .iter()
382 .any(|change| change.op_type == op_type::ATTACHMENT_ADD)
383 && blob_dir.is_none()
384 {
385 bail!("error attachment-blob-storage-required");
386 }
387 if let Some(blob_dir) = blob_dir {
388 ensure_attachment_blobs_present(&mut tx, blob_dir, &unassigned_changes).await?;
389 }
390 let mut next_server_seq = next_available_server_seq(&mut tx).await?;
391 let mut accepted_count = 0_i64;
392 let mut push_acks = Vec::with_capacity(changes.len());
393 for change in changes {
394 let existing_server_seq = sqlx::query_scalar::<_, Option<i64>>(
395 "SELECT server_seq FROM changes WHERE change_id = ?",
396 )
397 .bind(&change.change_id)
398 .fetch_optional(&mut *tx)
399 .await?
400 .flatten();
401 let server_seq = if let Some(server_seq) = existing_server_seq {
402 server_seq
403 } else {
404 let server_seq = next_server_seq;
405 next_server_seq += 1;
406 let payload = change.payload.to_string();
407 sqlx::query(
408 "INSERT INTO changes(change_id, client_id, local_seq, entity_type, entity_id, field,
409 op_type, payload, base_version, created_at, server_seq)
410 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
411 )
412 .bind(&change.change_id)
413 .bind(&change.client_id)
414 .bind(change.local_seq)
415 .bind(&change.entity_type)
416 .bind(&change.entity_id)
417 .bind(&change.field)
418 .bind(&change.op_type)
419 .bind(payload)
420 .bind(&change.base_version)
421 .bind(&change.created_at)
422 .bind(server_seq)
423 .execute(&mut *tx)
424 .await?;
425 apply_server_blob_reference(&mut tx, &change).await?;
426 accepted_count += 1;
427 server_seq
428 };
429 push_acks.push(PushAck {
430 change_id: change.change_id,
431 server_seq,
432 });
433 }
434 tx.commit().await?;
435 Ok((accepted_count, push_acks))
436}
437
438async fn next_available_server_seq(conn: &mut SqliteConnection) -> Result<i64> {
439 Ok(sqlx::query_scalar!(
440 r#"SELECT COALESCE(MAX(server_seq), 0) + 1 AS "seq!: i64" FROM changes"#
441 )
442 .fetch_one(&mut *conn)
443 .await?)
444}
445
446async fn apply_server_blob_reference(
447 conn: &mut SqliteConnection,
448 change: &ChangeWire,
449) -> Result<()> {
450 match change.op_type.as_str() {
451 op_type::ATTACHMENT_ADD => {
452 let workspace_id = change.payload["workspace_id"]
453 .as_str()
454 .context("payload missing workspace_id")?;
455 let attachment_id = change.payload["attachment_id"]
456 .as_str()
457 .context("payload missing attachment_id")?;
458 let sha256 = change.payload["sha256"]
459 .as_str()
460 .context("payload missing sha256")?;
461 let byte_size = change.payload["byte_size"]
462 .as_i64()
463 .context("payload missing byte_size")?;
464 sqlx::query(
465 "INSERT INTO server_blob_references(
466 workspace_id, attachment_id, task_id, sha256, byte_size, deleted
467 ) VALUES (?, ?, ?, ?, ?, 0)
468 ON CONFLICT(workspace_id, attachment_id) DO UPDATE SET
469 task_id = excluded.task_id, sha256 = excluded.sha256,
470 byte_size = excluded.byte_size, deleted = 0",
471 )
472 .bind(workspace_id)
473 .bind(attachment_id)
474 .bind(&change.entity_id)
475 .bind(sha256)
476 .bind(byte_size)
477 .execute(&mut *conn)
478 .await?;
479 sqlx::query(
480 "DELETE FROM blob_upload_reservations WHERE workspace_id = ? AND sha256 = ?",
481 )
482 .bind(workspace_id)
483 .bind(sha256)
484 .execute(&mut *conn)
485 .await?;
486 }
487 op_type::ATTACHMENT_DELETE => {
488 sqlx::query(
489 "UPDATE server_blob_references SET deleted = 1
490 WHERE workspace_id = ? AND attachment_id = ?",
491 )
492 .bind(
493 change.payload["workspace_id"]
494 .as_str()
495 .context("payload missing workspace_id")?,
496 )
497 .bind(
498 change.payload["attachment_id"]
499 .as_str()
500 .context("payload missing attachment_id")?,
501 )
502 .execute(&mut *conn)
503 .await?;
504 }
505 op_type::SET_FIELD if change.field.as_deref() == Some("deleted") => {
506 let deleted = change.payload["value"]
507 .as_str()
508 .is_some_and(|value| value == "1");
509 sqlx::query(
510 "INSERT INTO server_task_tombstones(workspace_id, task_id, deleted)
511 VALUES (?, ?, ?)
512 ON CONFLICT(workspace_id, task_id) DO UPDATE SET deleted = excluded.deleted",
513 )
514 .bind(
515 change.payload["workspace_id"]
516 .as_str()
517 .context("payload missing workspace_id")?,
518 )
519 .bind(&change.entity_id)
520 .bind(i64::from(deleted))
521 .execute(&mut *conn)
522 .await?;
523 }
524 _ => {}
525 }
526 Ok(())
527}
528
529async fn ensure_attachment_blobs_present(
530 conn: &mut SqliteConnection,
531 blob_dir: &Path,
532 changes: &[ChangeWire],
533) -> Result<()> {
534 for change in changes {
535 if change.op_type != op_type::ATTACHMENT_ADD {
536 continue;
537 }
538 let contract = super::blob::attachment_blob_contract(change)?
539 .context("error attachment-blob-missing")?;
540 let Some(row) =
541 crate::attachments::storage::blob_inventory_row(conn, &contract.sha256).await?
542 else {
543 bail!("error attachment-blob-missing");
544 };
545 if !row.available || !crate::attachments::object_path(blob_dir, &contract.sha256)?.exists()
546 {
547 bail!("error attachment-blob-missing");
548 }
549 if row.byte_size != contract.byte_size || row.media_type != contract.media_type {
550 bail!("error blob-inventory-metadata-mismatch");
551 }
552 let bytes =
553 tokio::fs::read(crate::attachments::object_path(blob_dir, &contract.sha256)?).await?;
554 if i64::try_from(bytes.len()).ok() != Some(contract.byte_size)
555 || crate::attachments::storage::sha256_hex(&bytes) != contract.sha256
556 {
557 bail!("error attachment-blob-content-mismatch");
558 }
559 let validated =
560 crate::attachments::decode::validate_image(bytes, Some(contract.media_type.clone()))
561 .await?;
562 if (validated.facts.width, validated.facts.height) != (contract.width, contract.height) {
563 bail!("error blob-inventory-metadata-mismatch");
564 }
565 let admitted: bool = sqlx::query_scalar(
566 "SELECT EXISTS(
567 SELECT 1 FROM server_blob_references sbr
568 LEFT JOIN server_task_tombstones st
569 ON st.workspace_id = sbr.workspace_id AND st.task_id = sbr.task_id
570 WHERE sbr.workspace_id = ? AND sbr.sha256 = ? AND sbr.deleted = 0
571 AND COALESCE(st.deleted, 0) = 0
572 ) OR EXISTS(
573 SELECT 1 FROM blob_upload_reservations
574 WHERE workspace_id = ? AND sha256 = ? AND byte_size = ? AND expires_at > ?
575 )",
576 )
577 .bind(&contract.workspace_id)
578 .bind(&contract.sha256)
579 .bind(&contract.workspace_id)
580 .bind(&contract.sha256)
581 .bind(contract.byte_size)
582 .bind(crate::ids::now())
583 .fetch_one(&mut *conn)
584 .await?;
585 if !admitted {
586 bail!("error attachment-blob-unreserved");
587 }
588 }
589 Ok(())
590}
591
592async fn load_server_changes_after(
593 conn: &mut SqliteConnection,
594 after: i64,
595 pull_limit: u32,
596) -> Result<(Vec<ChangeWire>, bool)> {
597 let fetch_limit = i64::from(pull_limit) + 1;
598 let rows = sqlx::query_as!(
599 ChangeRow,
600 r#"SELECT change_id AS "change_id!: String", client_id AS "client_id!: String",
601 local_seq AS "local_seq!: i64", entity_type AS "entity_type!: String",
602 entity_id AS "entity_id!: String", field, op_type AS "op_type!: String",
603 payload AS "payload!: String", base_version, created_at AS "created_at!: String",
604 server_seq
605 FROM changes WHERE server_seq > ? ORDER BY server_seq LIMIT ?"#,
606 after,
607 fetch_limit,
608 )
609 .fetch_all(&mut *conn)
610 .await?;
611 let has_more = rows.len() > pull_limit as usize;
612 let changes = rows
613 .into_iter()
614 .take(pull_limit as usize)
615 .map(ChangeRow::into_wire)
616 .collect();
617 Ok((changes, has_more))
618}