Skip to main content

max / audiofiles

17.5 KB · 465 lines History Blame Raw
1 //! audiofiles SyncKit integration: cloud sync for sample metadata, VFS, tags, and collections.
2 //!
3 //! Provides [`SyncManager`] as the public API, callable from the GUI thread.
4 //! All async work runs on an internal tokio runtime handle.
5
6 pub mod auth;
7 pub mod error;
8 pub mod scheduler;
9 pub mod service;
10
11 use std::path::PathBuf;
12 use std::sync::Arc;
13
14 use tracing::instrument;
15
16 use parking_lot::Mutex;
17 use synckit_client::SyncKitClient;
18 use tokio::runtime::Handle;
19 use tokio::sync::mpsc;
20
21 use error::Result;
22 use scheduler::SyncCommand;
23
24 /// Sync engine entry point, owned by the app and shared with the GUI.
25 pub struct SyncManager {
26 client: Arc<SyncKitClient>,
27 db_path: PathBuf,
28 content_dir: PathBuf,
29 runtime: Handle,
30 status: Arc<Mutex<SyncStatus>>,
31 command_tx: Mutex<Option<mpsc::UnboundedSender<SyncCommand>>>,
32 /// Sender for cancelling the in-flight OAuth flow. Set in `start_auth`,
33 /// taken (and triggered) in `cancel_auth`. Closing this channel makes the
34 /// spawned auth-await task return early without mutating sync state.
35 auth_cancel_tx: Mutex<Option<tokio::sync::oneshot::Sender<()>>>,
36 }
37
38 /// Observable sync status, read by the GUI each frame.
39 #[derive(Debug, Clone)]
40 pub struct SyncStatus {
41 pub state: SyncState,
42 pub last_sync_at: Option<String>,
43 pub pending_changes: i64,
44 pub last_error: Option<String>,
45 pub device_id: Option<String>,
46 pub auto_sync_enabled: bool,
47 pub sync_interval_minutes: u32,
48 /// Set to true when remote changes were pulled — GUI should reload VFS/contents.
49 pub needs_refresh: bool,
50 /// Subscription status for blob sync tier (populated async).
51 pub subscription: Option<synckit_client::SubscriptionStatus>,
52 /// Pricing-formula constants (fetched once from server, used to quote
53 /// prices locally as the user adjusts the cap slider).
54 pub pricing: Option<synckit_client::AppPricing>,
55 }
56
57 impl Default for SyncStatus {
58 fn default() -> Self {
59 Self {
60 state: SyncState::Disconnected,
61 last_sync_at: None,
62 pending_changes: 0,
63 last_error: None,
64 device_id: None,
65 auto_sync_enabled: false,
66 sync_interval_minutes: 15,
67 needs_refresh: false,
68 subscription: None,
69 pricing: None,
70 }
71 }
72 }
73
74 /// High-level sync state for UI display.
75 #[derive(Debug, Clone, PartialEq)]
76 pub enum SyncState {
77 /// Not configured or not authenticated.
78 Disconnected,
79 /// OAuth flow in progress.
80 Authenticating,
81 /// Authenticated but encryption not set up. `has_server_key` indicates whether
82 /// the user has previously set up encryption on another device.
83 NeedsEncryption { has_server_key: bool },
84 /// Fully configured and idle.
85 Ready,
86 /// Sync cycle in progress.
87 Syncing,
88 }
89
90 impl SyncManager {
91 /// Create a new SyncManager. Call [`start_scheduler`] after construction.
92 ///
93 /// `content_dir` is the content-addressed sample store root used for blob sync.
94 pub fn new(config: SyncKitConfig, db_path: PathBuf, content_dir: PathBuf, runtime: Handle) -> Self {
95 let client = Arc::new(SyncKitClient::new(config));
96 let status = Arc::new(Mutex::new(SyncStatus::default()));
97
98 Self {
99 client,
100 db_path,
101 content_dir,
102 runtime,
103 status,
104 command_tx: Mutex::new(None),
105 auth_cancel_tx: Mutex::new(None),
106 }
107 }
108
109 /// Read the current sync status (cheap mutex read).
110 pub fn status(&self) -> SyncStatus {
111 self.status.lock().clone()
112 }
113
114 /// Clear the needs_refresh flag after the GUI has reloaded.
115 pub fn clear_needs_refresh(&self) {
116 self.status.lock().needs_refresh = false;
117 }
118
119 /// Dismiss the surfaced `last_error`. The error banner in the sync panel
120 /// calls this when the user clicks Dismiss or Retry — Retry re-runs the
121 /// action and clears the previous error in the same gesture.
122 pub fn clear_last_error(&self) {
123 self.status.lock().last_error = None;
124 }
125
126 /// Start the OAuth2 PKCE authentication flow.
127 /// Opens the callback server and returns the auth URL. A background task
128 /// automatically awaits the callback and completes authentication, or
129 /// terminates early if [`cancel_auth`] fires.
130 #[instrument(skip_all)]
131 pub fn start_auth(&self) -> Result<String> {
132 self.status.lock().state = SyncState::Authenticating;
133 let session = auth::start_auth(&self.client)?;
134 let auth_url = session.auth_url.clone();
135
136 // Install a fresh cancel channel for this flow. Dropping any prior
137 // sender quietly cancels its waiter — start_auth being re-entrant is
138 // a corner case but shouldn't leak old senders.
139 let (cancel_tx, cancel_rx) = tokio::sync::oneshot::channel::<()>();
140 *self.auth_cancel_tx.lock() = Some(cancel_tx);
141
142 // Spawn background task to await callback and auto-complete auth
143 let client = self.client.clone();
144 let status = self.status.clone();
145 let db_path = self.db_path.clone();
146 let expected_state = session.expected_state.clone();
147 let code_verifier = session.code_verifier.clone();
148 let port = session.port;
149 let mut code_rx = session.code_rx;
150
151 self.runtime.spawn(async move {
152 // Race the callback against the cancel signal. If cancel wins, the
153 // task exits without mutating state — `cancel_auth` already set
154 // state to Disconnected at the call site, and any late callback
155 // result would otherwise transition state back to Ready /
156 // NeedsEncryption.
157 tokio::select! {
158 _ = cancel_rx => {
159 return;
160 }
161 recv = &mut code_rx => {
162 match recv {
163 Ok(result) => {
164 if result.state != expected_state {
165 let mut s = status.lock();
166 s.last_error = Some("CSRF state mismatch".to_string());
167 s.state = SyncState::Disconnected;
168 return;
169 }
170
171 match client
172 .authenticate_with_code(&result.code, &code_verifier, port, "__internal__")
173 .await
174 {
175 Ok(_) => {
176 let has_key = client.try_load_key_from_keychain().unwrap_or(false);
177 if has_key {
178 let mut s = status.lock();
179 s.state = SyncState::Ready;
180 load_sync_settings_into_status(&db_path, &mut s);
181 } else {
182 let has_server_key =
183 client.has_server_key().await.unwrap_or(false);
184 status.lock().state =
185 SyncState::NeedsEncryption { has_server_key };
186 }
187 }
188 Err(e) => {
189 let mut s = status.lock();
190 s.state = SyncState::Disconnected;
191 s.last_error = Some(format!("Auth failed: {e}"));
192 }
193 }
194 }
195 Err(_) => {
196 // Callback server timed out or was dropped
197 let mut s = status.lock();
198 if s.state == SyncState::Authenticating {
199 s.state = SyncState::Disconnected;
200 s.last_error = Some("Authentication timed out".to_string());
201 }
202 }
203 }
204 }
205 }
206 });
207
208 Ok(auth_url)
209 }
210
211 /// Cancel an in-flight OAuth flow. Returns to [`SyncState::Disconnected`]
212 /// immediately; the background callback-await task aborts on its next poll.
213 /// No-op when there is no active flow.
214 #[instrument(skip_all)]
215 pub fn cancel_auth(&self) {
216 if let Some(tx) = self.auth_cancel_tx.lock().take() {
217 // Send may fail if the task already terminated — that's fine.
218 let _ = tx.send(());
219 }
220 let mut s = self.status.lock();
221 if s.state == SyncState::Authenticating {
222 s.state = SyncState::Disconnected;
223 s.last_error = None;
224 }
225 }
226
227
228 /// Set up encryption (new or existing, depending on `is_new`).
229 #[instrument(skip_all)]
230 pub fn setup_encryption(&self, password: String, is_new: bool) {
231 let client = self.client.clone();
232 let status = self.status.clone();
233 let db_path = self.db_path.clone();
234
235 self.runtime.spawn(async move {
236 let result = if is_new {
237 client.setup_encryption_new(&password).await
238 } else {
239 client.setup_encryption_existing(&password).await
240 };
241
242 match result {
243 Ok(()) => {
244 let mut s = status.lock();
245 s.state = SyncState::Ready;
246 load_sync_settings_into_status(&db_path, &mut s);
247 }
248 Err(e) => {
249 status.lock().last_error =
250 Some(format!("Encryption setup failed: {e}"));
251 }
252 }
253 });
254 }
255
256 /// Trigger an immediate sync cycle.
257 pub fn sync_now(&self) {
258 if let Some(tx) = self.command_tx.lock().as_ref() {
259 let _ = tx.send(SyncCommand::SyncNow);
260 }
261 }
262
263 /// Trigger a targeted download of one cloud-only sample blob. Returns
264 /// `true` if the request was queued; `false` if the scheduler isn't
265 /// running yet (in which case the caller should fall back to a full sync
266 /// or surface "sync not ready").
267 pub fn download_sample(&self, hash: &str) -> bool {
268 if let Some(tx) = self.command_tx.lock().as_ref() {
269 tx.send(SyncCommand::DownloadOne { hash: hash.to_string() }).is_ok()
270 } else {
271 false
272 }
273 }
274
275 /// Update auto-sync settings.
276 #[instrument(skip_all)]
277 pub fn update_settings(&self, auto_sync: Option<bool>, interval: Option<u32>) {
278 if let Ok(conn) = rusqlite::Connection::open(&self.db_path) {
279 if let Some(enabled) = auto_sync {
280 let _ = service::set_sync_state(
281 &conn,
282 "auto_sync_enabled",
283 if enabled { "1" } else { "0" },
284 );
285 }
286 if let Some(mins) = interval {
287 let _ = service::set_sync_state(
288 &conn,
289 "sync_interval_minutes",
290 &mins.to_string(),
291 );
292 }
293 }
294
295 let mut s = self.status.lock();
296 if let Some(enabled) = auto_sync {
297 s.auto_sync_enabled = enabled;
298 }
299 if let Some(mins) = interval {
300 s.sync_interval_minutes = mins;
301 }
302 }
303
304 /// Disconnect: clear status, reset to disconnected.
305 pub fn disconnect(&self) {
306 let mut s = self.status.lock();
307 s.state = SyncState::Disconnected;
308 s.last_error = None;
309 s.device_id = None;
310 }
311
312 /// Try to restore a previous session from keychain on startup.
313 #[instrument(skip_all)]
314 pub fn try_restore_session(&self) {
315 if self.client.session_info().is_some() {
316 let has_key = self.client.try_load_key_from_keychain().unwrap_or(false);
317 if has_key {
318 let mut s = self.status.lock();
319 s.state = SyncState::Ready;
320 load_sync_settings_into_status(&self.db_path, &mut s);
321 }
322 }
323 }
324
325 /// Fetch the pricing formula from the server (async, no JWT needed).
326 /// Stored on the status so UI code can quote a price for any cap locally.
327 pub fn fetch_pricing(&self) {
328 let client = self.client.clone();
329 let status = self.status.clone();
330 self.runtime.spawn(async move {
331 match client.get_app_pricing().await {
332 Ok(pricing) => {
333 status.lock().pricing = Some(pricing);
334 }
335 Err(e) => {
336 tracing::debug!("Failed to fetch app pricing: {e}");
337 }
338 }
339 });
340 }
341
342 /// Fetch subscription status from the server (async, result goes to status.subscription).
343 /// On error (404, network, etc.) treats the user as unsubscribed so the UI can show the
344 /// subscribe CTA instead of spinning forever.
345 pub fn fetch_subscription_status(&self) {
346 let client = self.client.clone();
347 let status = self.status.clone();
348 self.runtime.spawn(async move {
349 let sub = match client.get_subscription_status().await {
350 Ok(sub) => sub,
351 Err(e) => {
352 tracing::debug!("Failed to fetch subscription status, treating as inactive: {e}");
353 synckit_client::SubscriptionStatus::default()
354 }
355 };
356 status.lock().subscription = Some(sub);
357 });
358 }
359
360 /// Create a Stripe checkout session at the chosen storage cap and open
361 /// it in the browser. Polls for subscription activation after opening.
362 pub fn subscribe(&self, cap_bytes: i64, interval: synckit_client::BillingInterval) {
363 let client = self.client.clone();
364 let status = self.status.clone();
365 self.runtime.spawn(async move {
366 match client.create_subscription_checkout(cap_bytes, interval).await {
367 Ok(resp) => {
368 if let Err(e) = open::that(&resp.checkout_url) {
369 tracing::warn!("Failed to open browser: {e}");
370 }
371 // Poll for subscription activation (5s intervals, up to 10 minutes)
372 for _ in 0..120 {
373 tokio::time::sleep(std::time::Duration::from_secs(5)).await;
374 if let Ok(sub) = client.get_subscription_status().await {
375 if sub.active {
376 status.lock().subscription = Some(sub);
377 tracing::info!("Subscription activated");
378 break;
379 }
380 }
381 }
382 }
383 Err(e) => {
384 tracing::error!("Failed to create checkout: {e}");
385 }
386 }
387 });
388 }
389
390 /// Queue a storage-cap change that applies at the next billing cycle.
391 pub fn queue_cap_change(&self, cap_bytes: i64) {
392 let client = self.client.clone();
393 let status = self.status.clone();
394 self.runtime.spawn(async move {
395 match client.queue_storage_cap_change(cap_bytes).await {
396 Ok(sub) => {
397 status.lock().subscription = Some(sub);
398 tracing::info!(cap_bytes, "Storage cap change queued");
399 }
400 Err(e) => {
401 tracing::error!("Failed to queue cap change: {e}");
402 }
403 }
404 });
405 }
406
407 /// Spawn the background sync scheduler task.
408 #[instrument(skip_all)]
409 pub fn start_scheduler(&self) {
410 let (tx, rx) = mpsc::unbounded_channel();
411 *self.command_tx.lock() = Some(tx);
412
413 let client = self.client.clone();
414 let db_path = self.db_path.clone();
415 let content_dir = self.content_dir.clone();
416 let status = self.status.clone();
417
418 self.runtime.spawn(scheduler::run_scheduler(
419 client, db_path, content_dir, status, rx,
420 ));
421 }
422 }
423
424 /// Load sync settings from the database into a SyncStatus.
425 #[instrument(skip_all)]
426 fn load_sync_settings_into_status(db_path: &PathBuf, s: &mut SyncStatus) {
427 let conn = match rusqlite::Connection::open(db_path) {
428 Ok(c) => c,
429 Err(e) => {
430 tracing::warn!("Failed to open DB for sync settings: {e}");
431 return;
432 }
433 };
434
435 s.auto_sync_enabled = service::get_sync_state(&conn, "auto_sync_enabled")
436 .inspect_err(|e| tracing::warn!("Failed to read auto_sync_enabled: {e}"))
437 .unwrap_or_default()
438 == "1";
439 s.sync_interval_minutes = service::get_sync_state(&conn, "sync_interval_minutes")
440 .inspect_err(|e| tracing::warn!("Failed to read sync_interval_minutes: {e}"))
441 .unwrap_or_else(|_| "15".to_string())
442 .parse()
443 .unwrap_or(15);
444 s.pending_changes = service::count_pending_changes(&conn)
445 .inspect_err(|e| tracing::warn!("Failed to count pending changes: {e}"))
446 .unwrap_or(0);
447 s.last_sync_at = {
448 let v = service::get_sync_state(&conn, "last_sync_at")
449 .inspect_err(|e| tracing::warn!("Failed to read last_sync_at: {e}"))
450 .unwrap_or_default();
451 if v.is_empty() { None } else { Some(v) }
452 };
453 s.device_id = {
454 let v = service::get_sync_state(&conn, "device_id")
455 .inspect_err(|e| tracing::warn!("Failed to read device_id: {e}"))
456 .unwrap_or_default();
457 if v.is_empty() { None } else { Some(v) }
458 };
459 }
460
461 // Re-export for convenience
462 pub use synckit_client::SyncKitConfig;
463 pub use synckit_client::{AppPricing, BillingInterval, PriceQuote};
464 pub use synckit_client::validate_api_key;
465