Skip to main content

max / goingson

9.1 KB · 225 lines History Blame Raw
1 //! Background scheduler for automatic cloud sync.
2 //!
3 //! Checks every 60 seconds whether a sync is due based on the configured
4 //! interval. On first run, creates an initial snapshot if needed.
5 //! An SSE connection receives real-time push notifications from the server,
6 //! triggering immediate sync when another device pushes changes.
7
8 use std::sync::Arc;
9 use tauri::{Emitter, Manager};
10 use tokio::time::{interval, Duration};
11 use tokio_util::sync::CancellationToken;
12 use tracing::{debug, error, info, warn};
13
14 use crate::state::AppState;
15 use crate::sync_service;
16
17 /// How often the scheduler checks if sync is due (seconds).
18 const CHECK_INTERVAL_SECS: u64 = 60;
19
20 /// Delay before reconnecting the SSE stream after a disconnect (seconds).
21 const SSE_RECONNECT_DELAY_SECS: u64 = 5;
22
23 /// Starts the cloud sync scheduler background task.
24 pub async fn start_sync_scheduler(app: tauri::AppHandle, cancel: CancellationToken) {
25 let mut check_interval = interval(Duration::from_secs(CHECK_INTERVAL_SECS));
26
27 info!("Cloud sync scheduler started (checking every {} seconds)", CHECK_INTERVAL_SECS);
28
29 let mut consecutive_failures: u32 = 0;
30 let mut backoff_until: Option<chrono::DateTime<chrono::Utc>> = None;
31 // Flag set by SSE notification to trigger immediate sync
32 let mut sse_triggered = false;
33 // SSE stream handle — None until first successful connection
34 let mut sse_stream: Option<synckit_client::SyncNotifyStream> = None;
35
36 loop {
37 // Wait for either: timer tick, cancellation, or SSE notification
38 tokio::select! {
39 _ = cancel.cancelled() => {
40 info!("Cloud sync scheduler shutting down");
41 break;
42 }
43 _ = check_interval.tick() => {}
44 result = async {
45 if let Some(ref mut stream) = sse_stream {
46 stream.next_change().await
47 } else {
48 // No stream — sleep forever (timer or cancel will fire)
49 std::future::pending::<Option<()>>().await
50 }
51 } => {
52 match result {
53 Some(()) => {
54 debug!("SSE: received change notification, triggering immediate sync");
55 sse_triggered = true;
56 }
57 None => {
58 // Stream ended — reconnect after delay
59 debug!("SSE: stream disconnected, will reconnect");
60 sse_stream = None;
61 tokio::time::sleep(Duration::from_secs(SSE_RECONNECT_DELAY_SECS)).await;
62 }
63 }
64 }
65 }
66
67 let state: Arc<AppState> = match app.try_state::<Arc<AppState>>() {
68 Some(s) => s.inner().clone(),
69 None => {
70 debug!("Sync scheduler: app state not available yet");
71 continue;
72 }
73 };
74
75 let client: Arc<synckit_client::SyncKitClient> = match state.read_recovering() {
76 Some(c) => c,
77 None => continue,
78 };
79
80 // Must be authenticated with a non-expired token
81 if client.session_info().is_none() {
82 continue;
83 }
84 if client.is_token_expired() {
85 debug!("Sync scheduler: token expired, skipping sync");
86 client.clear_session();
87 let _ = app.emit("sync:status-changed", "logged_out");
88 sse_stream = None;
89 continue;
90 }
91
92 // Must have encryption key loaded
93 if !client.has_master_key() {
94 continue;
95 }
96
97 // Try to establish SSE connection if not connected
98 if sse_stream.is_none() {
99 match client.subscribe().await {
100 Ok(stream) => {
101 debug!("SSE: connected to push notification stream");
102 sse_stream = Some(stream);
103 }
104 Err(e) => {
105 debug!("SSE: failed to connect (will retry): {}", e);
106 }
107 }
108 }
109
110 // Check auto_sync_enabled
111 let enabled = match sync_service::get_sync_state(&state.pool, "auto_sync_enabled").await {
112 Ok(v) => v == "1",
113 Err(e) => {
114 warn!("Sync scheduler: failed to read auto_sync_enabled: {}", e);
115 continue;
116 }
117 };
118 if !enabled {
119 continue;
120 }
121
122 // Check if sync interval has elapsed (skip check if SSE-triggered)
123 if !sse_triggered {
124 let interval_minutes: u64 = match sync_service::get_sync_state(&state.pool, "sync_interval_minutes").await {
125 Ok(v) => match v.parse() {
126 Ok(mins) => mins,
127 Err(e) => {
128 warn!(value = %v, error = %e, "Failed to parse sync_interval_minutes, using default 5");
129 5
130 }
131 },
132 Err(e) => {
133 warn!(error = %e, "Failed to read sync_interval_minutes, using default 5");
134 5
135 }
136 };
137
138 let last_sync = match sync_service::get_sync_state(&state.pool, "last_sync_at").await {
139 Ok(v) => v,
140 Err(e) => {
141 warn!(error = %e, "Failed to read last_sync_at");
142 String::new()
143 }
144 };
145
146 if !last_sync.is_empty()
147 && let Ok(last) = chrono::DateTime::parse_from_rfc3339(&last_sync) {
148 let elapsed = chrono::Utc::now() - last.with_timezone(&chrono::Utc);
149 if elapsed.num_minutes() < interval_minutes as i64 {
150 continue;
151 }
152 }
153 }
154 sse_triggered = false;
155
156 // Check backoff
157 if let Some(until) = backoff_until
158 && chrono::Utc::now() < until {
159 debug!("Sync scheduler: backing off until {}", until);
160 continue;
161 }
162
163 // Create initial snapshot on first sync
164 let snapshot_done = sync_service::get_sync_state(&state.pool, "initial_snapshot_done")
165 .await
166 .unwrap_or_default();
167
168 if snapshot_done != "1" {
169 match sync_service::create_initial_snapshot(&state.pool).await {
170 Ok(count) => info!("Initial sync snapshot: {} rows", count),
171 Err(e) => {
172 error!("Failed to create initial snapshot: {}", e);
173 continue;
174 }
175 }
176 }
177
178 // Perform sync (acquire lock to prevent concurrent syncs with manual sync_now)
179 let _sync_guard = state.sync_lock.lock().await;
180 let _ = app.emit("sync:status-changed", "syncing");
181 match sync_service::perform_sync_with_blobs(&state.pool, &client, Some(&state.data_dir)).await {
182 Ok(result) => {
183 consecutive_failures = 0;
184 backoff_until = None;
185 let _ = app.emit("sync:status-changed", "idle");
186 if result.pushed > 0 || result.pulled > 0 {
187 info!("Auto-sync: pushed {}, pulled {}", result.pushed, result.pulled);
188 }
189 if result.pulled > 0 {
190 // Carry the changed tables so the UI invalidates selectively.
191 let _ = app.emit("sync:changes-applied", &result.pulled_tables);
192 }
193 }
194 Err(e) => {
195 // If the server returned 402 (payment required), stop retrying —
196 // the user needs to subscribe before sync will work.
197 let is_payment_required = e.to_string().contains("402");
198 if is_payment_required {
199 let _ = app.emit("sync:subscription-required", ());
200 let _ = app.emit("sync:status-changed", "subscription_required");
201 warn!("Auto-sync: subscription required, pausing scheduler");
202 // Back off for 1 hour — recheck after that in case user subscribes
203 backoff_until = Some(chrono::Utc::now() + chrono::Duration::hours(1));
204 } else {
205 consecutive_failures += 1;
206 // Clamp the exponent before pow: an unbounded counter (failures
207 // only reset on success) would overflow u64 at 64 consecutive
208 // failures — panic in debug, wrap-to-0 (tight retry loop) in
209 // release. min(4) caps backoff at 16m before the 15m clamp,
210 // matching email_sync_scheduler.
211 let backoff_minutes = std::cmp::min(2u64.pow(consecutive_failures.min(4)), 15);
212 backoff_until = Some(chrono::Utc::now() + chrono::Duration::minutes(backoff_minutes as i64));
213 let _ = app.emit("sync:status-changed", "error");
214 warn!("Auto-sync failed (attempt {}, backoff {}m): {}", consecutive_failures, backoff_minutes, e);
215 }
216 }
217 }
218
219 // Cleanup old changelog entries
220 if let Err(e) = sync_service::cleanup_changelog(&state.pool).await {
221 warn!("Sync changelog cleanup failed: {}", e);
222 }
223 }
224 }
225