Skip to main content

max / goingson

3.8 KB · 114 lines History Blame Raw
1 //! Push local changes to the remote sync server.
2
3 use chrono::Utc;
4 use goingson_core::CoreError;
5 use sqlx::SqlitePool;
6 use synckit_client::{ChangeEntry, ChangeOp, DeviceId, Hlc, SyncKitClient};
7 use tracing::{debug, warn};
8 use uuid::Uuid;
9
10 use super::PUSH_BATCH_LIMIT;
11 use super::hlc::assign_pending_hlcs;
12
13 /// A pending sync_changelog row with its id:
14 /// (id, table_name, op, row_id, timestamp, data, hlc_wall, hlc_counter).
15 type PendingRow = (i64, String, String, String, String, Option<String>, Option<i64>, Option<i64>);
16
17 #[tracing::instrument(skip_all)]
18 pub async fn push_changes(
19 pool: &SqlitePool,
20 client: &SyncKitClient,
21 device_id: Uuid,
22 ) -> Result<i64, CoreError> {
23 // Stamp any unpushed changes with their HLC before reading them for the wire.
24 assign_pending_hlcs(pool, device_id).await?;
25
26 let mut total_pushed: i64 = 0;
27
28 loop {
29 let rows: Vec<PendingRow> = sqlx::query_as(
30 "SELECT id, table_name, op, row_id, timestamp, data, hlc_wall, hlc_counter \
31 FROM sync_changelog WHERE pushed = 0 ORDER BY id ASC LIMIT ?"
32 )
33 .bind(PUSH_BATCH_LIMIT)
34 .fetch_all(pool)
35 .await
36 .map_err(CoreError::database)?;
37
38 if rows.is_empty() {
39 break;
40 }
41
42 let row_count = rows.len() as i64;
43 let mut pushed_ids: Vec<i64> = Vec::new();
44 let changes: Vec<ChangeEntry> = rows
45 .into_iter()
46 .filter_map(|(id, table, op, row_id, timestamp, data, hlc_wall, hlc_counter)| {
47 // Always mark as pushed (even unknown ops) to avoid infinite re-fetch
48 pushed_ids.push(id);
49
50 let ts = chrono::DateTime::parse_from_rfc3339(&timestamp)
51 .map(|dt| dt.with_timezone(&Utc))
52 .unwrap_or(chrono::DateTime::UNIX_EPOCH);
53
54 let json_data = data.and_then(|d| serde_json::from_str(&d).ok());
55
56 let change_op = match ChangeOp::from_str_opt(&op) {
57 Some(o) => o,
58 None => {
59 warn!("Skipping changelog entry with unknown op: {}", op);
60 return None;
61 }
62 };
63
64 // Stamped by assign_pending_hlcs above; fall back to a
65 // timestamp-derived HLC if a row was somehow left unstamped.
66 let hlc = match (hlc_wall, hlc_counter) {
67 (Some(wall), Some(counter)) => Hlc { wall_ms: wall, counter: counter as u32, node: DeviceId::new(device_id) },
68 _ => Hlc::from_legacy(ts.timestamp_millis(), DeviceId::new(device_id)),
69 };
70
71 Some(ChangeEntry {
72 table,
73 op: change_op,
74 row_id,
75 timestamp: ts,
76 hlc,
77 data: json_data,
78 extra: Default::default(),
79 })
80 })
81 .collect();
82
83 let count = changes.len() as i64;
84 let is_last_batch = row_count < PUSH_BATCH_LIMIT;
85
86 client
87 .push(DeviceId::new(device_id), changes)
88 .await
89 .map_err(|e| CoreError::sync(format!("push failed: {}", e)))?;
90
91 // Mark pushed entries in a single transaction
92 let mut tx = pool.begin().await.map_err(CoreError::database)?;
93 for id in &pushed_ids {
94 sqlx::query("UPDATE sync_changelog SET pushed = 1 WHERE id = ?")
95 .bind(id)
96 .execute(&mut *tx)
97 .await
98 .map_err(CoreError::database)?;
99 }
100 tx.commit().await.map_err(CoreError::database)?;
101
102 total_pushed += count;
103
104 if is_last_batch {
105 break;
106 }
107 }
108
109 if total_pushed > 0 {
110 debug!("Pushed {} changes", total_pushed);
111 }
112 Ok(total_pushed)
113 }
114