better

backend/migrations/20251106200000_add_queue_mode_to_triggers.down.sql

@@ -1,15 +1,4 @@
 -- Add down migration script here
-DROP INDEX IF EXISTS idx_websocket_trigger_active_mode;
-DROP INDEX IF EXISTS idx_sqs_trigger_active_mode;
-DROP INDEX IF EXISTS idx_postgres_trigger_active_mode;
-DROP INDEX IF EXISTS idx_nats_trigger_active_mode;
-DROP INDEX IF EXISTS idx_mqtt_trigger_active_mode;
-DROP INDEX IF EXISTS idx_kafka_trigger_active_mode;
-DROP INDEX IF EXISTS idx_http_trigger_active_mode;
-DROP INDEX IF EXISTS idx_gcp_trigger_active_mode;
-DROP INDEX IF EXISTS idx_email_trigger_active_mode;
-
-
 ALTER TABLE websocket_trigger 
 DROP COLUMN IF EXISTS active_mode;
 

backend/migrations/20251106200000_add_queue_mode_to_triggers.up.sql

@@ -26,12 +26,3 @@ ADD COLUMN active_mode BOOLEAN NOT NULL DEFAULT TRUE;
 ALTER TABLE email_trigger 
 ADD COLUMN active_mode BOOLEAN NOT NULL DEFAULT TRUE;
 
-CREATE INDEX idx_gcp_trigger_active_mode ON gcp_trigger(workspace_id,path,active_mode);
-CREATE INDEX idx_http_trigger_active_mode ON http_trigger(workspace_id,path,active_mode);
-CREATE INDEX idx_kafka_trigger_active_mode ON kafka_trigger(workspace_id,path,active_mode);
-CREATE INDEX idx_mqtt_trigger_active_mode ON mqtt_trigger(workspace_id,path,active_mode);
-CREATE INDEX idx_nats_trigger_active_mode ON nats_trigger(workspace_id,path,active_mode);
-CREATE INDEX idx_postgres_trigger_active_mode ON postgres_trigger(workspace_id,path,active_mode);
-CREATE INDEX idx_sqs_trigger_active_mode ON sqs_trigger(workspace_id,path,active_mode);
-CREATE INDEX idx_websocket_trigger_active_mode ON websocket_trigger(workspace_id,path,active_mode);
-CREATE INDEX idx_email_trigger_active_mode ON email_trigger(workspace_id,path,active_mode);

backend/windmill-api/openapi.yaml

@@ -17542,7 +17542,7 @@ components:
         active_mode:
           type: boolean
           default: true
-          description: If set to false, each incomming event will be suspend job until ran manually or set it to true
+          description: If set to false, each incoming event will be suspend job until ran manually or set it to true
 
       required:
         - path
@@ -18725,7 +18725,7 @@ components:
         active_mode:
           type: boolean
           default: true
-          description: If set to false, each incomming event will be suspend job until ran manually or set it to true
+          description: If set to false, each incoming event will be suspend job until ran manually or set it to true
 
       required:
         - path
@@ -18755,7 +18755,7 @@ components:
         active_mode:
           type: boolean
           default: true
-          description: If set to false, each incomming event will be suspend job until ran manually or set it to true
+          description: If set to false, each incoming event will be suspend job until ran manually or set it to true
       required:
         - path
         - script_path

backend/windmill-api/src/triggers/global_handler.rs

@@ -1,16 +1,22 @@
-use crate::{db::{ApiAuthed, DB}, triggers::INACTIVE_TRIGGER_SCHEDULED_FOR_DATE};
+use crate::{db::ApiAuthed, triggers::INACTIVE_TRIGGER_SCHEDULED_FOR_DATE};
 use axum::{
     extract::{Extension, Path},
     response::Json,
 };
-use windmill_common::{error, jobs::JobTriggerKind, utils::require_admin};
+use windmill_audit::{audit_oss::audit_log, ActionKind};
+use windmill_common::{db::UserDB, error, jobs::JobTriggerKind, utils::require_admin};
 
 pub async fn resume_suspended_trigger_jobs(
     authed: ApiAuthed,
-    Extension(db): Extension<DB>,
+    Extension(user_db): Extension<UserDB>,
     Path((w_id, trigger_kind, trigger_path)): Path<(String, JobTriggerKind, String)>,
 ) -> error::Result<Json<String>> {
     require_admin(authed.is_admin, &authed.username)?;
+
+    let mut tx = user_db.begin(&authed).await?;
+
+    // Use the date constant to identify suspended jobs
+    // This date (9999-12-31 23:59:59) is used as a marker for suspended jobs
     let scheduled_for = INACTIVE_TRIGGER_SCHEDULED_FOR_DATE.clone();
     let result = sqlx::query!(
         r#"
@@ -33,11 +39,34 @@ pub async fn resume_suspended_trigger_jobs(
         trigger_kind as _,
         trigger_path
     )
-    .execute(&db)
+    .execute(&mut *tx)
     .await?;
 
     let count = result.rows_affected();
 
+    let trigger_kind_str = format!("{:?}", trigger_kind);
+    let count_str = count.to_string();
+    audit_log(
+        &mut *tx,
+        &authed,
+        "triggers.bulk_resume",
+        ActionKind::Execute,
+        &w_id,
+        Some(&trigger_path),
+        Some(
+            [
+                ("trigger_kind", trigger_kind_str.as_str()),
+                ("jobs_count", count_str.as_str()),
+            ]
+            .iter()
+            .cloned()
+            .collect(),
+        ),
+    )
+    .await?;
+
+    tx.commit().await?;
+
     let message = format!(
         "Successfully resumed {} suspended job{} for trigger at path: {}",
         count,
@@ -49,10 +78,13 @@ pub async fn resume_suspended_trigger_jobs(
 
 pub async fn cancel_suspended_trigger_jobs(
     authed: ApiAuthed,
-    Extension(db): Extension<DB>,
+    Extension(user_db): Extension<UserDB>,
     Path((w_id, trigger_kind, trigger_path)): Path<(String, JobTriggerKind, String)>,
 ) -> error::Result<Json<String>> {
     require_admin(authed.is_admin, &authed.username)?;
+
+    let mut tx = user_db.begin(&authed).await?;
+
     let scheduled_for = INACTIVE_TRIGGER_SCHEDULED_FOR_DATE.clone();
 
     let result = sqlx::query!(
@@ -79,11 +111,34 @@ pub async fn cancel_suspended_trigger_jobs(
         trigger_kind as _,
         trigger_path
     )
-    .execute(&db)
+    .execute(&mut *tx)
     .await?;
 
     let count = result.rows_affected();
 
+    let trigger_kind_str = format!("{:?}", trigger_kind);
+    let count_str = count.to_string();
+    audit_log(
+        &mut *tx,
+        &authed,
+        "triggers.bulk_cancel",
+        ActionKind::Delete,
+        &w_id,
+        Some(&trigger_path),
+        Some(
+            [
+                ("trigger_kind", trigger_kind_str.as_str()),
+                ("jobs_count", count_str.as_str()),
+            ]
+            .iter()
+            .cloned()
+            .collect(),
+        ),
+    )
+    .await?;
+
+    tx.commit().await?;
+
     let message = format!(
         "Successfully cancelled {} suspended job{} for trigger at path: {}",
         count,

backend/windmill-api/src/triggers/mod.rs

@@ -126,7 +126,6 @@ pub struct BaseTriggerData {
     pub is_flow: bool,
     pub enabled: Option<bool>,
     pub active_mode: Option<bool>,
-    pub process_queued_jobs: Option<bool>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]

backend/windmill-api/src/triggers/trigger_helpers.rs

@@ -25,6 +25,12 @@ use windmill_common::{
 };
 use windmill_queue::{push, PushArgs, PushArgsOwned, PushIsolationLevel};
 
+/// Helper function to check if triggers should use queue mode.
+/// Queue mode suspends jobs by scheduling them for a far future date.
+fn is_queue_mode(active_mode: Option<bool>) -> bool {
+    matches!(active_mode, Some(false))
+}
+
 #[cfg(feature = "enterprise")]
 use crate::jobs::check_license_key_valid;
 use crate::{
@@ -532,9 +538,11 @@ pub async fn trigger_runnable_inner(
     });
 
     let user_db = user_db.unwrap_or_else(|| UserDB::new(db.clone()));
-    let scheduled_for = active_mode
-        .filter(|active_mode| !*active_mode)
-        .map(|_| INACTIVE_TRIGGER_SCHEDULED_FOR_DATE.clone());
+    let scheduled_for = if is_queue_mode(active_mode) {
+        Some(INACTIVE_TRIGGER_SCHEDULED_FOR_DATE.clone())
+    } else {
+        None
+    };
     let (uuid, delete_after_use, early_return) = if is_flow {
         let run_query = RunJobQuery { job_id, scheduled_for, ..Default::default() };
         let path = StripPath(runnable_path.to_string());