1use std::collections::HashSet;
2
3use anyhow::{Context, Result, bail};
4use sqlx::SqliteConnection;
5
6use super::super::apply::apply_remote_change;
7use super::{ApplySyncPage, ClientSyncPage};
8use crate::change_log::op_type;
9use crate::db::{Database, begin_immediate, get_meta, set_meta};
10use crate::sync::wire::{
11 ChangeRow, ChangeWire, MAX_PUSH_BATCH, MAX_SYNC_REQUEST_BYTES, SyncRequest,
12};
13
14impl Database {
15 pub(in crate::sync) async fn pending_sync_changes_exist(&self) -> Result<bool> {
16 let mut conn = self.acquire_reader().await?;
17 Ok(
18 sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM changes WHERE server_seq IS NULL)")
19 .fetch_one(&mut *conn)
20 .await?,
21 )
22 }
23
24 pub(in crate::sync) async fn pending_blob_counts(
25 &self,
26 known_server_blobs: &HashSet<String>,
27 ) -> Result<crate::attachments::lifecycle::ByteCount> {
28 let mut conn = self.acquire_reader().await?;
29 let known = serde_json::to_string(known_server_blobs)?;
30 let (count, bytes): (i64, i64) = sqlx::query_as(
31 "WITH known(sha256) AS (
32 SELECT value FROM json_each(?)
33 ), pending AS (
34 SELECT json_extract(payload, '$.workspace_id') AS workspace_id,
35 json_extract(payload, '$.sha256') AS sha256,
36 MAX(CAST(json_extract(payload, '$.byte_size') AS INTEGER)) AS byte_size
37 FROM changes
38 WHERE server_seq IS NULL AND op_type = 'attachment_add'
39 GROUP BY workspace_id, sha256
40 )
41 SELECT COUNT(*), COALESCE(SUM(byte_size), 0)
42 FROM pending LEFT JOIN known USING (sha256)
43 WHERE known.sha256 IS NULL",
44 )
45 .bind(known)
46 .fetch_one(&mut *conn)
47 .await?;
48 Ok(crate::attachments::lifecycle::ByteCount {
49 count: u64::try_from(count)?,
50 bytes: u64::try_from(bytes)?,
51 })
52 }
53
54 pub(in crate::sync) async fn replica_sync_protocol(&self) -> Result<u32> {
55 let mut conn = self.acquire_reader().await?;
56 super::super::protocol::replica_protocol(&mut conn).await
57 }
58
59 pub(in crate::sync) async fn prepare_sync_discovery(&self, server: &str) -> Result<String> {
60 let mut conn = self.acquire_writer().await?;
61 validate_sync_server(&mut conn, server).await?;
62 super::super::protocol::replica_protocol(&mut conn).await?;
63 get_meta(&mut conn, "client_id")
64 .await?
65 .context("missing client id")
66 }
67
68 pub(in crate::sync) async fn block_sync_protocol(&self, protocol: u32) -> Result<()> {
69 let mut conn = self.acquire_writer().await?;
70 set_meta(&mut conn, "sync_blocked_protocol", &protocol.to_string()).await
71 }
72
73 pub async fn prepare_client_sync_page(
74 &self,
75 server: String,
76 push_limit: usize,
77 pull_limit: u32,
78 ) -> Result<ClientSyncPage> {
79 self.prepare_client_sync_page_at_protocol(server, push_limit, pull_limit, None)
80 .await
81 }
82
83 pub(in crate::sync) async fn prepare_client_sync_page_at_protocol(
84 &self,
85 server: String,
86 push_limit: usize,
87 pull_limit: u32,
88 protocol: Option<u32>,
89 ) -> Result<ClientSyncPage> {
90 let mut conn = self.acquire_writer().await?;
91 validate_sync_server(&mut conn, &server).await?;
92 let behavior_protocol = super::super::protocol::replica_protocol(&mut conn).await?;
93 let protocol = protocol.unwrap_or(behavior_protocol);
94 super::super::protocol::validate_behavior_protocol(protocol)?;
95 if protocol < behavior_protocol {
96 return Err(super::super::protocol::SyncCompatibilityError {
97 server_protocol: protocol,
98 client_protocol: behavior_protocol,
99 }
100 .into());
101 }
102 let client_id = get_meta(&mut conn, "client_id")
103 .await?
104 .context("missing client id")?;
105 let after = sync_cursor(&mut conn).await?;
106 let changes = load_unsynced_changes(&mut conn, push_limit.min(MAX_PUSH_BATCH)).await?;
107 let request = bound_push_request(
108 SyncRequest {
109 protocol_version: Some(protocol),
110 client_id,
111 after,
112 pull_limit: Some(pull_limit),
113 changes,
114 },
115 MAX_SYNC_REQUEST_BYTES,
116 )?;
117 for change in &request.changes {
118 super::super::protocol::validate_change(protocol, change)?;
119 }
120 Ok(ClientSyncPage {
121 behavior_protocol,
122 pending: request.changes.len(),
123 request,
124 })
125 }
126
127 pub async fn apply_client_sync_page(&self, page: ApplySyncPage) -> Result<usize> {
128 self.apply_client_sync_page_with_context(page, None).await
129 }
130
131 pub(in crate::sync) async fn apply_client_sync_page_with_context(
132 &self,
133 page: ApplySyncPage,
134 expected_behavior: Option<u32>,
135 ) -> Result<usize> {
136 let protocol = page
137 .request
138 .protocol_version
139 .context("missing selected protocol")?;
140 super::super::protocol::validate_behavior_protocol(protocol)?;
141 let envelope = crate::sync::wire::validate_request_at_protocol(&page.request, protocol)?;
142 let request_change_ids = page
143 .request
144 .changes
145 .iter()
146 .map(|change| change.change_id.clone())
147 .collect::<Vec<_>>();
148 crate::sync::wire::validate_response_at_protocol(
149 protocol,
150 envelope.after,
151 envelope.pull_limit,
152 &request_change_ids,
153 &page.response,
154 )?;
155 let mut conn = self.acquire_writer().await?;
156 apply_sync_response(&mut conn, page, expected_behavior).await
157 }
158}
159
160async fn sync_cursor(conn: &mut SqliteConnection) -> Result<i64> {
161 Ok(crate::db::get_meta(conn, "sync_cursor")
162 .await?
163 .unwrap_or_else(|| "0".to_string())
164 .parse::<i64>()?)
165}
166
167async fn validate_sync_server(conn: &mut SqliteConnection, server: &str) -> Result<()> {
168 let normalized = server.trim_end_matches('/');
169 if let Some(existing) = get_meta(conn, "sync_server_url").await? {
170 if existing != normalized {
171 bail!(
172 "error sync-server-changed existing={} requested={} hint=\"use a fresh database for a different sync server\"",
173 existing,
174 normalized
175 );
176 }
177 } else {
178 set_meta(conn, "sync_server_url", normalized).await?;
179 }
180 Ok(())
181}
182
183pub(super) fn bound_push_request(
184 mut request: SyncRequest,
185 byte_limit: usize,
186) -> Result<SyncRequest> {
187 let changes = std::mem::take(&mut request.changes);
188 let mut bytes = serde_json::to_vec(&request)?.len();
190 if bytes > byte_limit {
191 bail!("error sync-request-envelope-too-large limit={byte_limit}");
192 }
193 for change in changes.into_iter().take(MAX_PUSH_BATCH) {
194 let change_bytes = serde_json::to_vec(&change)?.len();
195 let separator = usize::from(!request.changes.is_empty());
196 if change_bytes + separator > byte_limit - bytes {
197 if request.changes.is_empty() {
198 bail!(
199 "error sync-change-exceeds-request-budget local_seq={} limit={byte_limit} hint=repair-pending-change",
200 change.local_seq
201 );
202 }
203 break;
204 }
205 bytes += change_bytes + separator;
206 request.changes.push(change);
207 }
208 Ok(request)
209}
210
211async fn load_unsynced_changes(
212 conn: &mut SqliteConnection,
213 limit: usize,
214) -> Result<Vec<ChangeWire>> {
215 let limit = limit as i64;
216 let rows = sqlx::query_as!(
217 ChangeRow,
218 r#"SELECT change_id AS "change_id!: String", client_id AS "client_id!: String",
219 local_seq AS "local_seq!: i64", entity_type AS "entity_type!: String",
220 entity_id AS "entity_id!: String", field, op_type AS "op_type!: String",
221 payload AS "payload!: String", base_version, created_at AS "created_at!: String",
222 server_seq
223 FROM changes WHERE server_seq IS NULL ORDER BY local_seq, created_at LIMIT ?"#,
224 limit,
225 )
226 .fetch_all(&mut *conn)
227 .await?;
228 Ok(rows.into_iter().map(ChangeRow::into_wire).collect())
229}
230
231pub(super) async fn apply_sync_response(
232 conn: &mut SqliteConnection,
233 page: ApplySyncPage,
234 expected_behavior: Option<u32>,
235) -> Result<usize> {
236 let mut applied = 0;
237 let mut tx = begin_immediate(conn).await?;
238 let current_behavior = super::super::protocol::replica_protocol(&mut tx).await?;
239 if expected_behavior.is_some_and(|expected| expected != current_behavior) {
240 bail!("error stale-sync-page replica-protocol-changed");
241 }
242 let selected = page
243 .request
244 .protocol_version
245 .context("missing selected protocol")?;
246 if selected < current_behavior {
247 bail!("error stale-sync-page replica-protocol-regressed");
248 }
249 super::super::protocol::establish_protocol(&mut tx, selected).await?;
250 let current_cursor = sync_cursor(&mut tx).await?;
251 if current_cursor != page.request.after {
252 bail!(
253 "error stale-sync-page expected_cursor={} request_cursor={}",
254 current_cursor,
255 page.request.after
256 );
257 }
258 super::update_change_server_seqs_if_missing(&mut tx, &page.response.push_acks).await?;
259 super::reconcile_acknowledged_epic_memberships(&mut tx, &page.response.push_acks).await?;
260 let existing_change_ids =
261 super::load_existing_change_ids(&mut tx, &page.response.changes).await?;
262 let mut affected_series = HashSet::new();
263 let mut affected_attachment_hashes = HashSet::new();
264 for change in &page.response.changes {
265 if existing_change_ids.contains(change.change_id.as_str()) {
266 super::verify_existing_change(&mut tx, change).await?;
267 super::update_change_server_seq(&mut tx, &change.change_id, change.server_seq).await?;
268 super::reconcile_epic_change(&mut tx, change).await?;
269 continue;
270 }
271 super::collect_attachment_liveness_hashes(&mut tx, change, &mut affected_attachment_hashes)
272 .await?;
273 if super::is_epic_change(change) {
274 let workspace_id = super::epic_change_workspace(change)?;
275 crate::epic_membership::capture_snapshot_baseline(
276 &mut tx,
277 workspace_id,
278 &change.entity_id,
279 )
280 .await?;
281 }
282 let related_mutation = matches!(
283 change.op_type.as_str(),
284 op_type::RELATED_ADD | op_type::RELATED_REMOVE
285 );
286 if related_mutation {
287 super::insert_wire_change(&mut tx, change).await?;
288 }
289 apply_remote_change(&mut tx, change).await?;
290 if change.entity_type == "recurrence_series" {
291 let workspace_id = change
292 .payload
293 .get("workspace_id")
294 .and_then(serde_json::Value::as_str)
295 .context("recurrence change missing workspace_id")?;
296 affected_series.insert((workspace_id.to_string(), change.entity_id.clone()));
297 } else if let Some(series_id) = change
298 .payload
299 .get("series_id")
300 .and_then(serde_json::Value::as_str)
301 {
302 let workspace_id = change
303 .payload
304 .get("workspace_id")
305 .and_then(serde_json::Value::as_str)
306 .context("recurrence task change missing workspace_id")?;
307 affected_series.insert((workspace_id.to_string(), series_id.to_string()));
308 }
309 if !related_mutation {
310 super::insert_wire_change(&mut tx, change).await?;
311 }
312 super::reconcile_epic_change(&mut tx, change).await?;
313 applied += 1;
314 }
315 for (workspace_id, series_id) in affected_series {
316 let workspace_id: crate::ids::WorkspaceId = workspace_id.parse()?;
317 let series_id: crate::recurrence::RecurrenceSeriesId = series_id.parse()?;
318 let workspace = crate::workspaces::workspace_for_id(&mut tx, &workspace_id).await?;
319 let at =
320 chrono::DateTime::parse_from_rfc3339(&page.attempted_at)?.with_timezone(&chrono::Utc);
321 crate::operations::recurrence::reconcile_recurrence_series_in_transaction(
322 &mut tx, &workspace, &series_id, at,
323 )
324 .await?;
325 }
326 let affected_attachment_hashes = affected_attachment_hashes.into_iter().collect::<Vec<_>>();
327 crate::attachments::lifecycle::reconcile_liveness_for_hashes_in_transaction(
328 &mut tx,
329 &affected_attachment_hashes,
330 &crate::attachments::lifecycle::SystemClock,
331 )
332 .await?;
333 let pushed = page.previous_pushed + page.response.push_acks.len() as i64;
334 let pulled = page.previous_pulled + applied;
335 set_meta(&mut tx, "sync_cursor", &page.response.cursor.to_string()).await?;
336 set_meta(&mut tx, "sync_last_success_at", &page.attempted_at).await?;
337 let pending: bool =
338 sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM changes WHERE server_seq IS NULL)")
339 .fetch_one(&mut *tx)
340 .await?;
341 let caught_up = !page.response.has_more && !pending;
342 set_meta(
343 &mut tx,
344 "sync_metadata_caught_up",
345 if caught_up { "1" } else { "0" },
346 )
347 .await?;
348 if caught_up {
349 set_meta(&mut tx, "sync_metadata_confirmed_at", &crate::ids::now()).await?;
350 }
351
352 set_meta(&mut tx, "sync_last_error", "").await?;
353 set_meta(&mut tx, "sync_blocked_protocol", "").await?;
354 set_meta(&mut tx, "sync_last_pushed", &pushed.to_string()).await?;
355 set_meta(&mut tx, "sync_last_pulled", &pulled.to_string()).await?;
356 set_meta(
357 &mut tx,
358 "sync_last_cursor",
359 &page.response.cursor.to_string(),
360 )
361 .await?;
362 tx.commit().await?;
363 Ok(applied)
364}
365
366#[cfg(test)]
367mod tests {
368 use super::super::ApplySyncPage;
369 use super::*;
370 use crate::change_log::op_type;
371 use crate::db::set_meta;
372 use crate::sync::wire::{
373 ChangeWire, MAX_PUSH_BATCH, MAX_SYNC_REQUEST_BYTES, PushAck, SYNC_PROTOCOL_VERSION,
374 SyncRequest, SyncResponse,
375 };
376 use serde_json::json;
377
378 fn budget_request() -> SyncRequest {
379 SyncRequest {
380 protocol_version: Some(SYNC_PROTOCOL_VERSION),
381 client_id: "budget-client\"\n".to_string(),
382 after: i64::MAX,
383 pull_limit: Some(crate::sync::wire::MAX_PULL_BATCH),
384 changes: (0..3)
385 .map(|index| ChangeWire {
386 change_id: format!("AAAAAAAAAAAAAAA{index}"),
387 client_id: "budget-client".to_string(),
388 local_seq: index + 1,
389 entity_type: "task".to_string(),
390 entity_id: "BBBBBBBBBBBBBBB0".to_string(),
391 field: Some("description".to_string()),
392 op_type: op_type::SET_FIELD.to_string(),
393 payload: json!({
394 "workspace_id": "0000000000000000",
395 "workspace_key": "default",
396 "value": "quoted \"text\"\n雪".repeat(8),
397 }),
398 base_version: None,
399 created_at: "2026-01-01T00:00:00Z".to_string(),
400 server_seq: None,
401 })
402 .collect(),
403 }
404 }
405
406 #[test]
407 fn push_byte_budget_selects_exact_ordered_prefix() {
408 let request = budget_request();
409 for change in &request.changes {
410 crate::sync::wire::validate_pushed_change(change).unwrap();
411 }
412 let mut prefix = request.clone();
413 prefix.changes.truncate(2);
414 let limit = serde_json::to_vec(&prefix).unwrap().len();
415 let selected = bound_push_request(request.clone(), limit).unwrap();
416 assert_eq!(
417 serde_json::to_vec(&selected).unwrap(),
418 serde_json::to_vec(&prefix).unwrap()
419 );
420 let selected = bound_push_request(request, limit - 1).unwrap();
421 assert_eq!(selected.changes.len(), 1);
422 assert_eq!(selected.changes[0].local_seq, 1);
423 assert!(serde_json::to_vec(&selected).unwrap().len() < limit);
424 }
425
426 #[test]
427 fn push_byte_budget_rejects_unfit_first_change_and_envelope() {
428 let mut request = budget_request();
429 request.changes.truncate(1);
430 let limit = serde_json::to_vec(&request).unwrap().len();
431 assert_eq!(
432 bound_push_request(request.clone(), limit)
433 .unwrap()
434 .changes
435 .len(),
436 1
437 );
438 assert!(
439 bound_push_request(request.clone(), limit - 1)
440 .unwrap_err()
441 .to_string()
442 .contains("sync-change-exceeds-request-budget")
443 );
444 request.changes.clear();
445 let limit = serde_json::to_vec(&request).unwrap().len();
446 assert!(
447 bound_push_request(request.clone(), limit)
448 .unwrap()
449 .changes
450 .is_empty()
451 );
452 assert!(
453 bound_push_request(request, limit - 1)
454 .unwrap_err()
455 .to_string()
456 .contains("sync-request-envelope-too-large")
457 );
458 }
459
460 #[test]
461 fn push_byte_budget_does_not_skip_a_change_that_does_not_fit() {
462 let mut request = budget_request();
463 let mut prefix = request.clone();
464 prefix.changes.truncate(2);
465 let limit = serde_json::to_vec(&prefix).unwrap().len();
466 request.changes[1].payload["value"] = json!("middle".repeat(limit));
467 let selected = bound_push_request(request, limit).unwrap();
468 assert_eq!(selected.changes.len(), 1);
469 assert_eq!(selected.changes[0].local_seq, 1);
470 }
471
472 #[test]
473 fn push_byte_budget_preserves_count_bound() {
474 let mut request = budget_request();
475 request.changes = vec![request.changes[0].clone(); MAX_PUSH_BATCH + 1];
476 let selected = bound_push_request(request, MAX_SYNC_REQUEST_BYTES).unwrap();
477 assert_eq!(selected.changes.len(), MAX_PUSH_BATCH);
478 }
479
480 #[tokio::test]
481 async fn related_comparison_observes_push_acknowledgement_first() {
482 let (_temp, mut conn) = crate::test_support::test_conn().await;
483 let workspace = crate::workspaces::Workspace::default();
484 let project = crate::projects::create_project(&mut conn, &workspace, "related-ack")
485 .await
486 .unwrap();
487 let task_id: crate::ids::TaskId = "AAAA000000000001".parse().unwrap();
488 let related_task_id: crate::ids::TaskId = "BBBB000000000002".parse().unwrap();
489 for id in [&task_id, &related_task_id] {
490 sqlx::query(
491 "INSERT INTO tasks(id, workspace_id, title, description, project_id, status, priority, created_at, updated_at)
492 VALUES (?, ?, 'task', '', ?, 'todo', 'none', 't', 't')",
493 )
494 .bind(id)
495 .bind(&workspace.id)
496 .bind(&project.id)
497 .execute(&mut *conn)
498 .await
499 .unwrap();
500 }
501 let local = crate::operations::set_task_related_link_in_transaction(
502 &mut conn,
503 &workspace,
504 &task_id,
505 &related_task_id,
506 true,
507 )
508 .await
509 .unwrap();
510 let local_change_id = local.change_id.unwrap();
511 set_meta(&mut conn, "sync_cursor", "3").await.unwrap();
512
513 let remote = ChangeWire {
514 change_id: "CCCC000000000003".to_string(),
515 client_id: "remote".to_string(),
516 local_seq: 1,
517 entity_type: "task".to_string(),
518 entity_id: task_id.to_string(),
519 field: Some("related".to_string()),
520 op_type: op_type::RELATED_REMOVE.to_string(),
521 payload: json!({
522 "workspace_id": workspace.id,
523 "workspace_key": workspace.key,
524 "related_task_id": related_task_id,
525 }),
526 base_version: None,
527 created_at: "2026-08-22T00:00:00Z".to_string(),
528 server_seq: Some(4),
529 };
530 let page = ApplySyncPage {
531 request: SyncRequest {
532 protocol_version: Some(SYNC_PROTOCOL_VERSION),
533 client_id: "local".to_string(),
534 after: 3,
535 pull_limit: Some(100),
536 changes: Vec::new(),
537 },
538 response: SyncResponse {
539 protocol_version: SYNC_PROTOCOL_VERSION,
540 cursor: 4,
541 has_more: false,
542 push_acks: vec![PushAck {
543 change_id: local_change_id.clone(),
544 server_seq: 5,
545 }],
546 changes: vec![remote],
547 },
548 attempted_at: "2026-08-22T00:00:01Z".to_string(),
549 previous_pushed: 0,
550 previous_pulled: 0,
551 };
552
553 apply_sync_response(&mut conn, page, None).await.unwrap();
554
555 let state: (i64, String) = sqlx::query_as(
556 "SELECT linked, last_change_id FROM task_related_links
557 WHERE workspace_id = ? AND task_a_id = ? AND task_b_id = ?",
558 )
559 .bind(&workspace.id)
560 .bind(&task_id)
561 .bind(&related_task_id)
562 .fetch_one(&mut *conn)
563 .await
564 .unwrap();
565 assert_eq!(state, (1, local_change_id.clone()));
566 let acknowledged: Option<i64> =
567 sqlx::query_scalar("SELECT server_seq FROM changes WHERE change_id = ?")
568 .bind(local_change_id)
569 .fetch_one(&mut *conn)
570 .await
571 .unwrap();
572 assert_eq!(acknowledged, Some(5));
573 }
574}