use std::collections::HashMap; use windmill_common::error::Result; use windmill_common::DB; use serde::Serialize; use crate::ServiceName; #[cfg(feature = "native_trigger")] use crate::{ decrypt_oauth_data, list_native_triggers, update_native_trigger_error, update_native_trigger_service_config, External, NativeTrigger, }; /// The one `native_trigger.error` an existence check owns. Finding the trigger on the service /// may only clear this value: every other failure recorded there — a webhook left aimed at a /// pre-rename path, a channel that failed to renew — is one existence cannot disprove. pub const EXTERNAL_TRIGGER_MISSING_ERROR: &str = "Trigger no longer exists on external service"; #[derive(Debug, Serialize)] pub struct TriggerSyncInfo { pub external_id: String, pub script_path: String, pub action: SyncAction, } #[derive(Debug, Serialize)] pub enum SyncAction { ErrorSet(String), ErrorCleared, ConfigUpdated, } #[derive(Debug, Serialize)] pub struct SyncError { pub resource_path: String, pub error_message: String, pub error_type: String, } #[derive(Debug)] pub struct BackgroundSyncResult { pub workspaces_processed: usize, pub total_synced: usize, pub total_errors: usize, pub service_results: HashMap, } #[derive(Debug)] pub struct ServiceSyncResult { pub synced_triggers: Vec, pub errors: Vec, } #[allow(unused_variables, unused_mut)] pub async fn sync_all_triggers(db: &DB) -> Result { tracing::info!("Starting native triggers sync"); let mut service_results: HashMap = HashMap::new(); let mut total_synced = 0; let mut total_errors = 0; let mut workspaces_processed = 0; // Sync all registered services // Each service only syncs workspaces that have the corresponding integration configured #[cfg(feature = "native_trigger")] { use crate::github::GitHub; use crate::google::Google; use crate::nextcloud::NextCloud; // Nextcloud sync let (service_name, result) = sync_service_triggers(db, NextCloud).await; total_synced += result.synced_triggers.len(); total_errors += result.errors.len(); service_results.insert(service_name, result); // Google sync (handles both Drive and Calendar triggers) let (service_name, result) = sync_service_triggers(db, Google).await; total_synced += result.synced_triggers.len(); total_errors += result.errors.len(); service_results.insert(service_name, result); // GitHub sync (verifies webhooks still exist) let (service_name, result) = sync_service_triggers(db, GitHub).await; total_synced += result.synced_triggers.len(); total_errors += result.errors.len(); service_results.insert(service_name, result); } // Count unique workspaces processed across all services for result in service_results.values() { workspaces_processed += result .synced_triggers .iter() .map(|t| &t.external_id) .collect::>() .len(); } let result = BackgroundSyncResult { workspaces_processed, total_synced, total_errors, service_results }; tracing::info!( "Completed native triggers sync: {} updated, {} errors", result.total_synced, result.total_errors ); Ok(result) } #[cfg(feature = "native_trigger")] async fn sync_service_triggers( db: &DB, handler: T, ) -> (ServiceName, ServiceSyncResult) { let mut all_synced_triggers = Vec::new(); let mut all_errors = Vec::new(); // Use the integration service for lookup (e.g., GoogleDrive/GoogleCalendar -> Google) let integration_service = T::SERVICE_NAME.integration_service(); // Only sync workspaces that have the corresponding integration configured let workspaces_with_integration = match sqlx::query_scalar!( r#" SELECT wi.workspace_id FROM workspace_integrations wi JOIN workspace w ON w.id = wi.workspace_id WHERE wi.service_name = $1 AND wi.oauth_data IS NOT NULL AND w.deleted = false "#, integration_service as ServiceName ) .fetch_all(db) .await { Ok(workspaces) => workspaces, Err(e) => { tracing::error!( "Error querying workspaces with {} integration: {:#}", T::SERVICE_NAME.as_str(), e ); all_errors.push(SyncError { resource_path: "database".to_string(), error_message: format!("Failed to query workspaces: {}", e), error_type: "database_error".to_string(), }); return ( T::SERVICE_NAME, ServiceSyncResult { synced_triggers: Vec::new(), errors: all_errors }, ); } }; if workspaces_with_integration.is_empty() { tracing::debug!( "No workspaces with {} integration configured, skipping sync", T::SERVICE_NAME.as_str() ); return ( T::SERVICE_NAME, ServiceSyncResult { synced_triggers: Vec::new(), errors: Vec::new() }, ); } tracing::info!( "Found {} workspaces with {} integration configured", workspaces_with_integration.len(), T::SERVICE_NAME.as_str() ); for workspace_id in workspaces_with_integration { let sync_result = sync_workspace_triggers::(db, &workspace_id, &handler).await; match sync_result { Ok((synced_triggers, errors)) => { all_synced_triggers.extend(synced_triggers); all_errors.extend(errors); } Err(e) => { tracing::error!( "Error syncing {} triggers for workspace {}: {:#}", T::SERVICE_NAME.as_str(), workspace_id, e ); all_errors.push(SyncError { resource_path: format!("workspace:{}", workspace_id), error_message: format!("Failed to sync workspace: {}", e), error_type: "workspace_sync_error".to_string(), }); } } } ( T::SERVICE_NAME, ServiceSyncResult { synced_triggers: all_synced_triggers, errors: all_errors }, ) } #[cfg(feature = "native_trigger")] pub async fn sync_workspace_triggers( db: &DB, workspace_id: &str, handler: &T, ) -> Result<(Vec, Vec)> { tracing::info!( "Syncing {} triggers for workspace '{}'", T::SERVICE_NAME.as_str(), workspace_id ); let windmill_triggers = list_native_triggers( db, workspace_id, T::SERVICE_NAME, None, None, None, None, windmill_api_auth::ScopePathFilter::AllowAll, ) .await?; if windmill_triggers.is_empty() { tracing::info!( "No {} triggers found for workspace '{}'", T::SERVICE_NAME.as_str(), workspace_id ); return Ok((Vec::new(), Vec::new())); } let mut synced = Vec::new(); let mut errors = Vec::new(); // Use the integration service for OAuth lookup (e.g., GoogleDrive/GoogleCalendar -> Google) let integration_service = T::SERVICE_NAME.integration_service(); let oauth_data = { match decrypt_oauth_data(db, workspace_id, integration_service).await { Ok(oauth_data) => oauth_data, Err(e) => { tracing::error!( "Failed to get workspace integration OAuth data for {}: {}", workspace_id, e ); errors.push(SyncError { resource_path: format!("workspace:{}", workspace_id), error_message: format!("Failed to get workspace integration OAuth data: {}", e), error_type: "oauth_error".to_string(), }); return Ok((Vec::new(), errors)); } } }; handler .maintain_triggers( db, workspace_id, &windmill_triggers, &oauth_data, &mut synced, &mut errors, ) .await; tracing::info!( "Sync completed for {} in workspace '{}'. Updated: {}, Errors: {}", T::SERVICE_NAME.as_str(), workspace_id, synced.len(), errors.len() ); Ok((synced, errors)) } /// Reusable reconciliation logic for services with real external state (e.g. Nextcloud). /// Compares external triggers with DB triggers: sets errors for missing ones, /// clears errors and updates config for existing ones. #[cfg(feature = "native_trigger")] pub async fn reconcile_with_external_state( db: &DB, workspace_id: &str, service_name: ServiceName, windmill_triggers: &[NativeTrigger], external_triggers: &[(String, serde_json::Value)], synced: &mut Vec, errors: &mut Vec, ) { // Build a map of external trigger IDs to their config let mut external_trigger_map: HashMap = HashMap::new(); for (external_id, config) in external_triggers { external_trigger_map.insert(external_id.clone(), config); } for trigger in windmill_triggers { if !external_trigger_map.contains_key(&trigger.external_id) { // Trigger no longer exists on external service - set error let error_msg = EXTERNAL_TRIGGER_MISSING_ERROR.to_string(); if trigger.error.as_deref() != Some(&error_msg) { tracing::info!( "Trigger (external_id: '{}', script_path: '{}') no longer exists in external service, setting error", trigger.external_id, trigger.script_path ); match update_native_trigger_error( db, workspace_id, service_name, &trigger.external_id, Some(&error_msg), ) .await { Ok(()) => { synced.push(TriggerSyncInfo { external_id: trigger.external_id.clone(), script_path: trigger.script_path.clone(), action: SyncAction::ErrorSet(error_msg), }); } Err(e) => { tracing::error!( "Failed to update error for trigger (external_id: '{}'): {}", trigger.external_id, e ); errors.push(SyncError { resource_path: format!("workspace:{}", workspace_id), error_message: format!( "Failed to update error for trigger (external_id: '{}'): {}", trigger.external_id, e ), error_type: "database_update_error".to_string(), }); } } } } else { // Trigger exists on external service let external_service_config = external_trigger_map.get(&trigger.external_id).unwrap(); // Only clear the error this reconciler itself sets. Existence says nothing about the // failures other paths record — a webhook left aimed at a pre-rename path is present // here yet still broken, and wiping its error would erase the only signal the user has. if trigger.error.as_deref() == Some(EXTERNAL_TRIGGER_MISSING_ERROR) { tracing::info!( "Trigger (external_id: '{}', script_path: '{}') exists on external service, clearing error", trigger.external_id, trigger.script_path ); match update_native_trigger_error( db, workspace_id, service_name, &trigger.external_id, None, ) .await { Ok(()) => { synced.push(TriggerSyncInfo { external_id: trigger.external_id.clone(), script_path: trigger.script_path.clone(), action: SyncAction::ErrorCleared, }); } Err(e) => { tracing::error!( "Failed to clear error for trigger (external_id: '{}'): {}", trigger.external_id, e ); errors.push(SyncError { resource_path: format!("workspace:{}", workspace_id), error_message: format!( "Failed to clear error for trigger (external_id: '{}'): {}", trigger.external_id, e ), error_type: "database_update_error".to_string(), }); } } } // Compare service_config and update if different let stored_service_config = trigger .service_config .clone() .unwrap_or(serde_json::Value::Null); if **external_service_config != stored_service_config { tracing::info!( "Trigger (external_id: '{}', script_path: '{}') config differs from external service, updating local config", trigger.external_id, trigger.script_path ); match update_native_trigger_service_config( db, workspace_id, service_name, &trigger.external_id, external_service_config, None, ) .await { Ok(()) => { synced.push(TriggerSyncInfo { external_id: trigger.external_id.clone(), script_path: trigger.script_path.clone(), action: SyncAction::ConfigUpdated, }); } Err(e) => { tracing::error!( "Failed to update config for trigger (external_id: '{}'): {}", trigger.external_id, e ); errors.push(SyncError { resource_path: format!("workspace:{}", workspace_id), error_message: format!( "Failed to update config for trigger (external_id: '{}'): {}", trigger.external_id, e ), error_type: "database_update_error".to_string(), }); } } } else { tracing::info!( "Trigger (external_id: '{}', script_path: '{}') config is the same as external service, no update needed", trigger.external_id, trigger.script_path ); } } } }