use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use std::sync::OnceLock; use std::time::{Duration, Instant}; use serde_json::json; use tauri::{AppHandle, Emitter}; use tokio::sync::mpsc::error::TrySendError; use tokio::sync::mpsc::{channel, Receiver, Sender}; use crate::error::AppError; use crate::services::s3_sync; use crate::settings::{self, S3SyncSettings}; const AUTO_SYNC_DEBOUNCE_MS: u64 = 1000; pub(crate) const MAX_AUTO_SYNC_WAIT_MS: u64 = 10_000; static DB_CHANGE_TX: OnceLock> = OnceLock::new(); static AUTO_SYNC_SUPPRESS_DEPTH: AtomicUsize = AtomicUsize::new(0); pub(crate) struct AutoSyncSuppressionGuard; impl AutoSyncSuppressionGuard { pub fn new() -> Self { AUTO_SYNC_SUPPRESS_DEPTH.fetch_add(1, Ordering::SeqCst); Self } } impl Drop for AutoSyncSuppressionGuard { fn drop(&mut self) { let _ = AUTO_SYNC_SUPPRESS_DEPTH.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |value| { Some(value.saturating_sub(1)) }); } } pub(crate) fn is_auto_sync_suppressed() -> bool { AUTO_SYNC_SUPPRESS_DEPTH.load(Ordering::SeqCst) > 0 } pub fn should_trigger_for_table(table: &str) -> bool { let normalized = table.trim().to_ascii_lowercase(); matches!( normalized.as_str(), "providers" | "provider_endpoints" | "mcp_servers" | "prompts" | "skills" | "skill_repos" | "settings" | "proxy_config" ) } pub(crate) fn enqueue_change_signal(tx: &Sender, table: &str) -> bool { match tx.try_send(table.to_string()) { Ok(()) => true, Err(TrySendError::Full(_)) | Err(TrySendError::Closed(_)) => false, } } pub(crate) fn auto_sync_wait_duration(started_at: Instant, now: Instant) -> Option { let max_wait = Duration::from_millis(MAX_AUTO_SYNC_WAIT_MS); let debounce = Duration::from_millis(AUTO_SYNC_DEBOUNCE_MS); let elapsed = now.saturating_duration_since(started_at); if elapsed >= max_wait { return None; } Some(debounce.min(max_wait - elapsed)) } fn should_run_auto_sync(settings: Option<&S3SyncSettings>) -> bool { let Some(sync) = settings else { return false; }; sync.enabled && sync.auto_sync } fn persist_auto_sync_error(settings: &mut S3SyncSettings, error: &AppError) { settings.status.last_error = Some(error.to_string()); settings.status.last_error_source = Some("auto".to_string()); let _ = settings::update_s3_sync_status(settings.status.clone()); } fn emit_auto_sync_status_updated(app: &AppHandle, status: &str, error: Option<&str>) { let payload = match error { Some(message) => json!({ "source": "auto", "status": status, "error": message, }), None => json!({ "source": "auto", "status": status, }), }; if let Err(err) = app.emit("s3-sync-status-updated", payload) { log::debug!("[S3] failed to emit sync status update event: {err}"); } } async fn run_auto_sync_upload( db: &crate::database::Database, app: &AppHandle, ) -> Result<(), AppError> { let mut settings = settings::get_s3_sync_settings(); if !should_run_auto_sync(settings.as_ref()) { return Ok(()); } let mut sync_settings = match settings.take() { Some(value) => value, None => return Ok(()), }; let result = s3_sync::run_with_sync_lock(s3_sync::upload(db, &mut sync_settings)).await; match result { Ok(_) => { emit_auto_sync_status_updated(app, "success", None); Ok(()) } Err(err) => { persist_auto_sync_error(&mut sync_settings, &err); emit_auto_sync_status_updated(app, "error", Some(&err.to_string())); Err(err) } } } pub fn notify_db_changed(table: &str) { if is_auto_sync_suppressed() { return; } if !should_trigger_for_table(table) { return; } let Some(tx) = DB_CHANGE_TX.get() else { return; }; let _ = enqueue_change_signal(tx, table); } pub fn start_worker(db: Arc, app: tauri::AppHandle) { if DB_CHANGE_TX.get().is_some() { return; } // Buffer size 1 is enough: we only need "dirty" signals, not every event. let (tx, rx) = channel::(1); if DB_CHANGE_TX.set(tx).is_err() { return; } tauri::async_runtime::spawn(async move { run_worker_loop(db, rx, app).await; }); } async fn run_worker_loop( db: Arc, mut rx: Receiver, app: tauri::AppHandle, ) { while let Some(first_table) = rx.recv().await { let started_at = Instant::now(); let mut merged_count = 1usize; while let Some(wait_for) = auto_sync_wait_duration(started_at, Instant::now()) { let timeout = tokio::time::timeout(wait_for, rx.recv()).await; match timeout { Ok(Some(_)) => merged_count += 1, Ok(None) => return, Err(_) => break, } } log::debug!( "[S3][AutoSync] Triggered by table={first_table}, merged_changes={merged_count}" ); if let Err(err) = run_auto_sync_upload(&db, &app).await { log::warn!("[S3][AutoSync] Upload failed: {err}"); } } } #[cfg(test)] mod tests { use super::{ auto_sync_wait_duration, enqueue_change_signal, is_auto_sync_suppressed, should_run_auto_sync, should_trigger_for_table, AutoSyncSuppressionGuard, MAX_AUTO_SYNC_WAIT_MS, }; use crate::settings::S3SyncSettings; use std::time::{Duration, Instant}; use tokio::sync::mpsc::channel; #[test] fn should_trigger_sync_for_config_tables_only() { assert!(should_trigger_for_table("providers")); assert!(should_trigger_for_table("settings")); assert!(!should_trigger_for_table("proxy_request_logs")); assert!(!should_trigger_for_table("provider_health")); } #[test] fn suppression_guard_enables_and_restores_state() { assert!(!is_auto_sync_suppressed()); { let _guard = AutoSyncSuppressionGuard::new(); assert!(is_auto_sync_suppressed()); } assert!(!is_auto_sync_suppressed()); } #[test] fn max_wait_caps_flush_latency_for_continuous_events() { let started = Instant::now(); let later = started + Duration::from_millis(MAX_AUTO_SYNC_WAIT_MS + 1); assert!(auto_sync_wait_duration(started, later).is_none()); } #[tokio::test] async fn enqueue_change_signal_drops_when_channel_is_full() { let (tx, _rx) = channel::(1); assert!(enqueue_change_signal(&tx, "providers")); assert!(!enqueue_change_signal(&tx, "providers")); } #[test] fn should_run_auto_sync_requires_enabled_and_auto_sync_flag() { assert!(!should_run_auto_sync(None)); let disabled = S3SyncSettings { enabled: false, auto_sync: true, ..S3SyncSettings::default() }; assert!(!should_run_auto_sync(Some(&disabled))); let auto_sync_off = S3SyncSettings { enabled: true, auto_sync: false, ..S3SyncSettings::default() }; assert!(!should_run_auto_sync(Some(&auto_sync_off))); let enabled = S3SyncSettings { enabled: true, auto_sync: true, ..S3SyncSettings::default() }; assert!(should_run_auto_sync(Some(&enabled))); } #[test] fn service_layer_does_not_depend_on_commands_layer() { let source = include_str!("s3_auto_sync.rs"); let needle = ["crate", "commands", ""].join("::"); assert!( !source.contains(&needle), "services layer should not depend on commands layer" ); } }