From b7d4c634566fb0f00aae6519adb0e35541fc6a3f Mon Sep 17 00:00:00 2001 From: HugoCasa Date: Fri, 28 Nov 2025 16:49:46 +0100 Subject: [PATCH] UI and http triggers --- backend/windmill-api/openapi.yaml | 65 ++- backend/windmill-api/src/jobs.rs | 163 ++++--- .../src/triggers/global_handler.rs | 241 ++++++++-- backend/windmill-api/src/triggers/handler.rs | 151 +++++- .../windmill-api/src/triggers/http/handler.rs | 13 +- .../src/triggers/trigger_helpers.rs | 111 +++-- backend/windmill-queue/src/jobs.rs | 12 +- backend/windmill-worker/src/worker.rs | 7 + .../src/lib/components/DropdownV2Inner.svelte | 4 + frontend/src/lib/components/RunsPage.svelte | 10 +- .../lib/components/common/modal/Modal2.svelte | 4 +- .../src/lib/components/runs/JobsLoader.svelte | 6 +- .../src/lib/components/runs/RunsFilter.svelte | 7 +- .../triggers/TriggerActiveMode.svelte | 146 ------ .../triggers/TriggerEditorToolbar.svelte | 70 ++- .../triggers/TriggerSuspendedJobsModal.svelte | 453 +++++++++++------- .../email/EmailTriggerEditorInner.svelte | 3 - .../triggers/gcp/GcpTriggerEditorInner.svelte | 3 - .../triggers/http/RouteEditorInner.svelte | 30 +- .../kafka/KafkaTriggerEditorInner.svelte | 3 - .../mqtt/MqttTriggerEditorInner.svelte | 3 - .../nats/NatsTriggerEditorInner.svelte | 3 - .../PostgresTriggerEditorInner.svelte | 2 - .../triggers/sqs/SqsTriggerEditorInner.svelte | 3 - frontend/src/lib/components/triggers/utils.ts | 8 +- .../WebsocketTriggerEditorInner.svelte | 3 - frontend/src/lib/utils.ts | 1 + 27 files changed, 974 insertions(+), 551 deletions(-) delete mode 100644 frontend/src/lib/components/triggers/TriggerActiveMode.svelte diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 228883b2a1..eaa031f060 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -8638,6 +8638,20 @@ paths: type: string style: simple explode: false + requestBody: + description: Optional list of job IDs to reassign + required: false + content: + application/json: + schema: + type: object + properties: + job_ids: + type: array + items: + type: string + format: uuid + description: Optional list of specific job UUIDs to reassign. If not provided, all suspended jobs for the trigger will be reassigned. responses: "200": description: confirmation message @@ -8668,6 +8682,20 @@ paths: type: string style: simple explode: false + requestBody: + description: Optional list of job IDs to cancel + required: false + content: + application/json: + schema: + type: object + properties: + job_ids: + type: array + items: + type: string + format: uuid + description: Optional list of specific job UUIDs to cancel. If not provided, all suspended jobs for the trigger will be canceled. responses: "200": description: confirmation message @@ -10287,6 +10315,39 @@ paths: schema: type: string + /w/{workspace}/http_triggers/update_status/{path}: + post: + summary: update http trigger suspended mode and enabled status + operationId: updateHttpTriggerStatus + tags: + - http_trigger + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - $ref: "#/components/parameters/Path" + requestBody: + required: true + content: + application/json: + schema: + type: object + properties: + suspended_mode: + type: boolean + description: Whether the trigger is in suspended mode (queues jobs instead of running them) + enabled: + type: boolean + description: Whether the trigger is enabled + required: + - suspended_mode + - enabled + responses: + "200": + description: http trigger status updated + content: + text/plain: + schema: + type: string + /w/{workspace}/websocket_triggers/create: post: summary: create websocket trigger @@ -17344,7 +17405,7 @@ components: - args JobTriggerKind: - description: trigger kind (schedule, http, websocket...) + description: job trigger kind (schedule, http, websocket...) type: string enum: - webhook @@ -17359,8 +17420,6 @@ components: - mqtt - sqs - gcp - - poll - - cli TriggerExtraProperty: type: object diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index c8c4b3c555..b4b0f63f81 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -3900,6 +3900,7 @@ async fn batch_rerun_handle_job( let result = push_flow_job_by_path_into_queue( authed.clone(), db.clone(), + None, user_db.clone(), w_id.clone(), StripPath(job.script_path.clone()), @@ -3908,7 +3909,7 @@ async fn batch_rerun_handle_job( None, ) .await; - if let Ok((uuid, _)) = result { + if let Ok((uuid, _, _)) = result { return Ok(uuid.to_string()); } } @@ -3917,6 +3918,7 @@ async fn batch_rerun_handle_job( push_script_job_by_path_into_queue( authed.clone(), db.clone(), + None, user_db.clone(), w_id.clone(), StripPath(job.script_path.clone()), @@ -3925,6 +3927,7 @@ async fn batch_rerun_handle_job( None, ) .await + .map(|r| r.0) } else { run_job_by_hash_inner( authed.clone(), @@ -3937,8 +3940,9 @@ async fn batch_rerun_handle_job( None, ) .await + .map(|r| r.0) }; - if let Ok((uuid, _)) = result { + if let Ok(uuid) = result { return Ok(uuid.to_string()); } } @@ -4056,17 +4060,18 @@ pub async fn run_flow_by_path( ) .await?; - let (uuid, _) = push_flow_job_by_path_into_queue( - authed, db, user_db, w_id, flow_path, run_query, args, None, + let (uuid, _, _) = push_flow_job_by_path_into_queue( + authed, db, None, user_db, w_id, flow_path, run_query, args, None, ) .await?; Ok((StatusCode::CREATED, uuid.to_string())) } -pub async fn run_flow( +pub async fn run_flow<'c>( authed: &ApiAuthed, db: &DB, + tx_o: Option>, user_db: UserDB, w_id: &str, flow_path: &str, @@ -4074,7 +4079,11 @@ pub async fn run_flow( run_query: RunJobQuery, args: PushArgsOwned, trigger: Option, -) -> error::Result<(Uuid, Option)> { +) -> error::Result<( + Uuid, + Option, + Option>, +)> { let FlowVersionInfo { version, tag, @@ -4092,22 +4101,30 @@ pub async fn run_flow( check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?; let scheduled_for = run_query.get_scheduled_for(&db).await?; - let (email, permissioned_as, push_authed, tx) = - if let Some(on_behalf_of_email) = on_behalf_of_email.as_ref() { - ( - on_behalf_of_email, - username_to_permissioned_as(&edited_by), - None, - PushIsolationLevel::IsolatedRoot(db.clone()), - ) - } else { - ( - &authed.email, - username_to_permissioned_as(&authed.username), - Some(authed.clone().into()), - PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into()), - ) - }; + let return_tx = tx_o.is_some(); + + let (email, permissioned_as, push_authed, tx) = if let Some(tx) = tx_o { + ( + &authed.email, + username_to_permissioned_as(&authed.username), + Some(authed.clone().into()), + PushIsolationLevel::Transaction(tx), + ) + } else if let Some(on_behalf_of_email) = on_behalf_of_email.as_ref() { + ( + on_behalf_of_email, + username_to_permissioned_as(&edited_by), + None, + PushIsolationLevel::IsolatedRoot(db.clone()), + ) + } else { + ( + &authed.email, + username_to_permissioned_as(&authed.username), + Some(authed.clone().into()), + PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into()), + ) + }; let (uuid, mut tx) = push( &db, @@ -4166,9 +4183,13 @@ pub async fn run_flow( .await?; } - tx.commit().await?; - - Ok((uuid, early_return)) + // If we were given a transaction, return it; otherwise commit it + if return_tx { + Ok((uuid, early_return, Some(tx))) + } else { + tx.commit().await?; + Ok((uuid, early_return, None)) + } } pub async fn run_flow_and_wait_result( @@ -4182,9 +4203,10 @@ pub async fn run_flow_and_wait_result( args: PushArgsOwned, trigger: Option, ) -> error::Result { - let (uuid, early_return) = run_flow( + let (uuid, early_return, _) = run_flow( authed, db, + None, user_db, w_id, flow_path, @@ -4198,16 +4220,21 @@ pub async fn run_flow_and_wait_result( run_wait_result(&db, uuid, w_id, early_return, &authed.username).await } -pub async fn push_flow_job_by_path_into_queue( +pub async fn push_flow_job_by_path_into_queue<'c>( authed: ApiAuthed, db: DB, + tx_o: Option>, user_db: UserDB, w_id: String, flow_path: StripPath, run_query: RunJobQuery, args: PushArgsOwned, trigger: Option, -) -> error::Result<(Uuid, Option)> { +) -> error::Result<( + Uuid, + Option, + Option>, +)> { #[cfg(feature = "enterprise")] check_license_key_valid().await?; @@ -4223,6 +4250,7 @@ pub async fn push_flow_job_by_path_into_queue( run_flow( &authed, &db, + tx_o, user_db, &w_id, flow_path, @@ -4293,9 +4321,10 @@ pub async fn run_flow_by_version_inner( let flow_version_info = get_flow_version_info_from_version(&db, version, &w_id, &flow_path).await?; - run_flow( + let (uuid, early_return, _) = run_flow( &authed, &db, + None, user_db, &w_id, &flow_path, @@ -4304,7 +4333,9 @@ pub async fn run_flow_by_version_inner( args, trigger, ) - .await + .await?; + + Ok((uuid, early_return)) } #[cfg(not(feature = "enterprise"))] @@ -4425,9 +4456,10 @@ pub async fn run_script_by_path( ) .await?; - let (uuid, _) = push_script_job_by_path_into_queue( + let (uuid, _, _) = push_script_job_by_path_into_queue( authed, db, + None, user_db, w_id, script_path, @@ -4440,16 +4472,21 @@ pub async fn run_script_by_path( Ok((StatusCode::CREATED, uuid.to_string())) } -pub async fn push_script_job_by_path_into_queue( +pub async fn push_script_job_by_path_into_queue<'c>( authed: ApiAuthed, db: DB, + tx_o: Option>, user_db: UserDB, w_id: String, script_path: StripPath, run_query: RunJobQuery, args: PushArgsOwned, trigger: Option, -) -> error::Result<(Uuid, Option)> { +) -> error::Result<( + Uuid, + Option, + Option>, +)> { #[cfg(feature = "enterprise")] check_license_key_valid().await?; @@ -4470,22 +4507,30 @@ pub async fn push_script_job_by_path_into_queue( let tag = run_query.tag.clone().or(tag); check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?; - let (email, permissioned_as, push_authed, tx) = - if let Some(on_behalf_of) = on_behalf_of.as_ref() { - ( - on_behalf_of.email.as_str(), - on_behalf_of.permissioned_as.clone(), - None, - PushIsolationLevel::IsolatedRoot(db.clone()), - ) - } else { - ( - authed.email.as_str(), - username_to_permissioned_as(&authed.username), - Some(authed.clone().into()), - PushIsolationLevel::Isolated(user_db, authed.clone().into()), - ) - }; + let return_tx = tx_o.is_some(); + + let (email, permissioned_as, push_authed, tx) = if let Some(tx) = tx_o { + ( + authed.email.as_str(), + username_to_permissioned_as(&authed.username), + Some(authed.clone().into()), + PushIsolationLevel::Transaction(tx), + ) + } else if let Some(on_behalf_of) = on_behalf_of.as_ref() { + ( + on_behalf_of.email.as_str(), + on_behalf_of.permissioned_as.clone(), + None, + PushIsolationLevel::IsolatedRoot(db.clone()), + ) + } else { + ( + authed.email.as_str(), + username_to_permissioned_as(&authed.username), + Some(authed.clone().into()), + PushIsolationLevel::Isolated(user_db, authed.clone().into()), + ) + }; let (uuid, tx) = push( &db, @@ -4524,9 +4569,14 @@ pub async fn push_script_job_by_path_into_queue( run_query.suspended_mode, ) .await?; - tx.commit().await?; - Ok((uuid, delete_after_use)) + // If we were given a transaction, return it; otherwise commit it + if return_tx { + Ok((uuid, delete_after_use, Some(tx))) + } else { + tx.commit().await?; + Ok((uuid, delete_after_use, None)) + } } #[derive(Deserialize)] @@ -5652,9 +5702,10 @@ pub async fn stream_job( let uuid = match runnable_id { RunnableId::ScriptId(ScriptId::ScriptPath(script_path)) | RunnableId::HubScript(script_path) => { - push_script_job_by_path_into_queue( + let (uuid, _, _) = push_script_job_by_path_into_queue( authed.clone(), db.clone(), + None, user_db, w_id.clone(), StripPath(script_path), @@ -5662,8 +5713,8 @@ pub async fn stream_job( args, None, ) - .await? - .0 + .await?; + uuid } RunnableId::ScriptId(ScriptId::ScriptHash(script_hash)) => { run_job_by_hash_inner( @@ -5683,6 +5734,7 @@ pub async fn stream_job( push_flow_job_by_path_into_queue( authed.clone(), db.clone(), + None, user_db, w_id.clone(), StripPath(flow_path), @@ -6738,9 +6790,10 @@ async fn run_dynamic_select( let push_args = PushArgsOwned { extra: None, args: script_args.clone() }; - let (uuid, _) = push_script_job_by_path_into_queue( + let (uuid, _, _) = push_script_job_by_path_into_queue( authed.clone(), db.clone(), + None, user_db.clone(), w_id.clone(), StripPath(path), diff --git a/backend/windmill-api/src/triggers/global_handler.rs b/backend/windmill-api/src/triggers/global_handler.rs index 831982c4fa..5df941e350 100644 --- a/backend/windmill-api/src/triggers/global_handler.rs +++ b/backend/windmill-api/src/triggers/global_handler.rs @@ -1,10 +1,10 @@ use crate::{ db::{ApiAuthed, DB}, - triggers::{trigger_helpers::trigger_runnable_inner, TriggerForReassignment, Trigger, TriggerCrud}, + triggers::{trigger_helpers::trigger_runnable_inner, TriggerCrud, TriggerForReassignment}, }; #[cfg(feature = "http_trigger")] -use crate::triggers::http::{handler::HttpTrigger, HttpConfig}; +use crate::triggers::http::handler::HttpTrigger; #[cfg(feature = "mqtt_trigger")] use crate::triggers::mqtt::{MqttConfig, MqttTrigger}; @@ -13,7 +13,7 @@ use crate::triggers::mqtt::{MqttConfig, MqttTrigger}; use crate::triggers::postgres::{PostgresConfig, PostgresTrigger}; #[cfg(feature = "websocket")] -use crate::triggers::websocket::{WebsocketConfig, WebsocketTrigger}; +use crate::triggers::websocket::WebsocketTrigger; #[cfg(all(feature = "smtp", feature = "enterprise", feature = "private"))] use crate::triggers::email::{EmailConfig, EmailTrigger}; @@ -33,6 +33,7 @@ use axum::{ extract::{Extension, Path}, response::Json, }; +use serde::{Deserialize, Serialize}; use serde_json::value::RawValue; use std::collections::HashMap; use uuid::Uuid; @@ -41,13 +42,21 @@ use windmill_common::{db::UserDB, error, jobs::JobTriggerKind, triggers::Trigger struct JobWithArgs { id: Uuid, args: Option>>>, + created_at: chrono::DateTime, } -pub async fn reassign_suspended_jobs( +#[derive(Deserialize, Serialize, Default)] +pub struct ReassignJobsBody { + #[serde(default)] + pub job_ids: Option>, +} + +pub async fn resume_suspended_trigger_jobs( authed: ApiAuthed, Extension(db): Extension, Extension(user_db): Extension, Path((w_id, trigger_kind, trigger_path)): Path<(String, JobTriggerKind, String)>, + Json(body): Json, ) -> error::Result> { let mut tx = user_db.clone().begin(&authed).await?; @@ -185,70 +194,206 @@ pub async fn reassign_suspended_jobs( } }; - let jobs = sqlx::query_as!(JobWithArgs, - "SELECT id, args as \"args: _\" FROM v2_job WHERE workspace_id = $1 AND kind = 'unassigned'::JOB_KIND AND trigger_kind = $2 AND trigger = $3", - w_id, - trigger_kind as _, - trigger_path, - ).fetch_all(&mut *tx).await?; + let jobs = if let Some(job_ids) = body.job_ids.as_ref() { + if job_ids.is_empty() { + vec![] + } else { + sqlx::query_as!( + JobWithArgs, + "SELECT id, args as \"args: _\", created_at FROM v2_job + WHERE workspace_id = $1 + AND kind = 'unassigned'::JOB_KIND + AND trigger_kind = $2 + AND trigger = $3 + AND id = ANY($4)", + w_id, + trigger_kind as _, + trigger_path, + job_ids as _, + ) + .fetch_all(&mut *tx) + .await? + } + } else { + sqlx::query_as!( + JobWithArgs, + "SELECT id, args as \"args: _\", created_at FROM v2_job + WHERE workspace_id = $1 + AND kind = 'unassigned'::JOB_KIND + AND trigger_kind = $2 + AND trigger = $3", + w_id, + trigger_kind as _, + trigger_path, + ) + .fetch_all(&mut *tx) + .await? + }; let trigger_metadata = TriggerMetadata::new(Some(trigger_path.clone()), trigger_kind); let l = jobs.len(); for job in jobs { - trigger_runnable_inner( - &db, - Some(user_db.clone()), - authed.clone(), - &w_id, - &trigger.script_path, - trigger.is_flow, - windmill_queue::PushArgsOwned { - extra: None, - args: job.args.map(|a| a.0).unwrap_or_default(), - }, - trigger.retry.as_ref(), - trigger.error_handler_path.as_deref(), - trigger.error_handler_args.as_ref(), - trigger_path.clone(), - None, - trigger_metadata.clone(), - None, - ) - .await?; + // If job was created before trigger was edited, simply update it to unsuspend + // instead of deleting and repushing + if job.created_at > trigger.edited_at { + let job_kind = if trigger.is_flow { + windmill_common::jobs::JobKind::Flow + } else { + windmill_common::jobs::JobKind::Script + }; - // Delete the unassigned job from all related tables - sqlx::query!("DELETE FROM v2_job_queue WHERE id = $1", job.id) + sqlx::query!( + "UPDATE v2_job SET kind = $1 WHERE id = $2", + job_kind as _, + job.id + ) .execute(&mut *tx) .await?; - sqlx::query!("DELETE FROM v2_job_runtime WHERE id = $1", job.id) + // Update the job to unsuspend it and set the correct kind + sqlx::query!( + "UPDATE v2_job_queue SET scheduled_for = now() WHERE id = $1", + job.id + ) .execute(&mut *tx) .await?; + } else { + // Job was created after trigger edit - delete and repush with new configuration + // Pass the transaction to trigger_runnable_inner so everything is in the same transaction + let (_uuid, _delete_after_use, _early_return, tx_o) = trigger_runnable_inner( + &db, + Some(tx), + Some(user_db.clone()), + authed.clone(), + &w_id, + &trigger.script_path, + trigger.is_flow, + windmill_queue::PushArgsOwned { + extra: None, + args: job.args.map(|a| a.0).unwrap_or_default(), + }, + trigger.retry.as_ref(), + trigger.error_handler_path.as_deref(), + trigger.error_handler_args.as_ref(), + trigger_path.clone(), + None, + trigger_metadata.clone(), + None, + ) + .await?; - sqlx::query!("DELETE FROM job_perms WHERE job_id = $1", job.id) - .execute(&mut *tx) - .await?; + tx = match tx_o { + Some(tx) => tx, + None => { + return Err(error::Error::internal_err( + "Transaction should be returned when passed in".to_string(), + )); + } + }; - sqlx::query!("DELETE FROM concurrency_key WHERE job_id = $1", job.id) - .execute(&mut *tx) - .await?; + // Delete the unassigned job from all related tables + sqlx::query!("DELETE FROM v2_job_queue WHERE id = $1", job.id) + .execute(&mut *tx) + .await?; - sqlx::query!("DELETE FROM debounce_key WHERE job_id = $1", job.id) - .execute(&mut *tx) - .await?; + sqlx::query!("DELETE FROM v2_job_runtime WHERE id = $1", job.id) + .execute(&mut *tx) + .await?; - sqlx::query!("DELETE FROM debounce_stale_data WHERE job_id = $1", job.id) - .execute(&mut *tx) - .await?; + sqlx::query!("DELETE FROM job_perms WHERE job_id = $1", job.id) + .execute(&mut *tx) + .await?; - sqlx::query!("DELETE FROM v2_job WHERE id = $1", job.id) - .execute(&mut *tx) - .await?; + sqlx::query!("DELETE FROM concurrency_key WHERE job_id = $1", job.id) + .execute(&mut *tx) + .await?; + + sqlx::query!("DELETE FROM debounce_key WHERE job_id = $1", job.id) + .execute(&mut *tx) + .await?; + + sqlx::query!("DELETE FROM debounce_stale_data WHERE job_id = $1", job.id) + .execute(&mut *tx) + .await?; + + sqlx::query!("DELETE FROM v2_job WHERE id = $1", job.id) + .execute(&mut *tx) + .await?; + } } tx.commit().await?; Ok(Json(format!("Reassigned {} jobs", l))) } + +pub async fn cancel_suspended_trigger_jobs( + authed: ApiAuthed, + Extension(db): Extension, + Extension(user_db): Extension, + Path((w_id, trigger_kind, trigger_path)): Path<(String, JobTriggerKind, String)>, + Json(body): Json, +) -> error::Result> { + let mut tx = user_db.clone().begin(&authed).await?; + + // Get the list of job IDs to cancel + let jobs_to_cancel = if let Some(job_ids) = body.job_ids.as_ref() { + if job_ids.is_empty() { + vec![] + } else { + sqlx::query_scalar!( + "SELECT id FROM v2_job + WHERE workspace_id = $1 + AND kind = 'unassigned'::JOB_KIND + AND trigger_kind = $2 + AND trigger = $3 + AND id = ANY($4)", + w_id, + trigger_kind as _, + trigger_path, + job_ids as _, + ) + .fetch_all(&mut *tx) + .await? + } + } else { + sqlx::query_scalar!( + "SELECT id FROM v2_job + WHERE workspace_id = $1 + AND kind = 'unassigned'::JOB_KIND + AND trigger_kind = $2 + AND trigger = $3", + w_id, + trigger_kind as _, + trigger_path, + ) + .fetch_all(&mut *tx) + .await? + }; + + let count = jobs_to_cancel.len(); + + if count > 0 { + // Use the cancel_jobs helper from windmill_queue + for job_id in &jobs_to_cancel { + let (returned_tx, _) = windmill_queue::cancel_job( + &authed.username, + Some("canceled by trigger management".to_string()), + *job_id, + &w_id, + tx, + &db, + false, + false, + ) + .await?; + tx = returned_tx; + } + } + + tx.commit().await?; + + Ok(Json(format!("Canceled {} jobs", count))) +} diff --git a/backend/windmill-api/src/triggers/handler.rs b/backend/windmill-api/src/triggers/handler.rs index 76a2de6755..b65342a6b6 100644 --- a/backend/windmill-api/src/triggers/handler.rs +++ b/backend/windmill-api/src/triggers/handler.rs @@ -37,8 +37,7 @@ pub struct TriggerForReassignment { pub email: String, pub edited_at: DateTime, pub error_handler_path: Option, - pub error_handler_args: - Option>>, + pub error_handler_args: Option>>, pub retry: Option>, } @@ -428,7 +427,8 @@ pub fn trigger_routes() -> Router { .route("/update/*path", post(update_trigger::)) .route("/delete/*path", delete(delete_trigger::)) .route("/exists/*path", get(exists_trigger::)) - .route("/setenabled/*path", post(set_enabled_trigger::)); + .route("/setenabled/*path", post(set_enabled_trigger::)) + .route("/update_status/*path", post(update_trigger_status::)); if T::SUPPORTS_TEST_CONNECTION { router = router.route("/test", post(test_connection::)); @@ -651,6 +651,123 @@ struct SetEnabledPayload { enabled: bool, } +#[derive(Deserialize)] +struct UpdateStatusPayload { + suspended_mode: bool, + enabled: bool, +} + +async fn update_trigger_status( + Extension(handler): Extension>, + authed: ApiAuthed, + Extension(user_db): Extension, + Extension(db): Extension, + Path((workspace_id, path)): Path<(String, StripPath)>, + Json(payload): Json, +) -> Result { + let path = path.to_path(); + check_scopes(&authed, || format!("{}:write", T::scope_domain_name()))?; + + let mut tx = user_db.begin(&authed).await?; + + let updated = if T::SUPPORTS_SERVER_STATE { + sqlx::query(&format!( + r#" + UPDATE + {} + SET + suspended_mode = $1, + enabled = $2, + email = $3, + edited_by = $4, + edited_at = now(), + server_id = NULL, + error = NULL + WHERE + workspace_id = $5 AND + path = $6 + "#, + T::TABLE_NAME + )) + .bind(payload.suspended_mode) + .bind(payload.suspended_mode || payload.enabled) + .bind(&authed.email) + .bind(&authed.username) + .bind(&workspace_id) + .bind(path) + .execute(&mut *tx) + .await? + .rows_affected() + } else { + sqlx::query(&format!( + r#" + UPDATE + {} + SET + suspended_mode = $1, + enabled = $2, + email = $3, + edited_by = $4, + edited_at = now() + WHERE + workspace_id = $5 AND + path = $6 + "#, + T::TABLE_NAME + )) + .bind(payload.suspended_mode) + .bind(payload.suspended_mode || payload.enabled) + .bind(&authed.email) + .bind(&authed.username) + .bind(&workspace_id) + .bind(path) + .execute(&mut *tx) + .await? + .rows_affected() + }; + + if updated == 0 { + return Err(Error::NotFound(format!( + "Trigger not found at path: {}", + path + ))); + } + + handler.set_enabled_extra_action(&mut *tx).await?; + + tx.commit().await?; + + handle_deployment_metadata( + &authed.email, + &authed.username, + &db, + &workspace_id, + T::get_deployed_object(path.to_owned()), + Some(format!( + "{} trigger '{}' status updated", + T::DEPLOYMENT_NAME, + path + )), + true, + ) + .await?; + + Ok(format!( + "Trigger '{}' updated: {} mode, {}", + path, + if payload.suspended_mode { + "suspended" + } else { + "active" + }, + if payload.enabled { + "enabled" + } else { + "disabled" + } + )) +} + async fn set_enabled_trigger( Extension(handler): Extension>, authed: ApiAuthed, @@ -825,21 +942,21 @@ pub fn generate_trigger_routers() -> Router { ); } - // { - // use crate::triggers::global_handler::{ - // cancel_suspended_trigger_jobs, resume_suspended_trigger_jobs, - // }; + { + use crate::triggers::global_handler::{ + cancel_suspended_trigger_jobs, resume_suspended_trigger_jobs, + }; - // router = router - // .route( - // "/trigger/:trigger_kind/resume_suspended_trigger_job/*trigger_path", - // post(resume_suspended_trigger_jobs), - // ) - // .route( - // "/trigger/:trigger_kind/cancel_suspended_trigger_job/*trigger_path", - // post(cancel_suspended_trigger_jobs), - // ); - // } + router = router + .route( + "/trigger/:trigger_kind/resume_suspended_trigger_job/*trigger_path", + post(resume_suspended_trigger_jobs), + ) + .route( + "/trigger/:trigger_kind/cancel_suspended_trigger_job/*trigger_path", + post(cancel_suspended_trigger_jobs), + ); + } router } diff --git a/backend/windmill-api/src/triggers/http/handler.rs b/backend/windmill-api/src/triggers/http/handler.rs index 0bbb2bc076..055ff1223a 100644 --- a/backend/windmill-api/src/triggers/http/handler.rs +++ b/backend/windmill-api/src/triggers/http/handler.rs @@ -41,7 +41,7 @@ use windmill_common::{ db::UserDB, error::{Error, Result}, jobs::JobTriggerKind, - triggers::{TriggerMetadata, TriggerKind}, + triggers::{TriggerKind, TriggerMetadata}, utils::{not_found_if_none, require_admin, StripPath}, worker::CLOUD_HOSTED, }; @@ -1050,7 +1050,7 @@ async fn route_job( .map_err(|e| e.into_response())?; let trigger_info = TriggerMetadata::new(Some(trigger.path.clone()), JobTriggerKind::Http); - if !trigger.suspended_mode { + if trigger.suspended_mode { let _ = trigger_runnable( &db, Some(user_db), @@ -1064,7 +1064,7 @@ async fn route_job( trigger.error_handler_args.as_ref(), format!("http_trigger/{}", trigger.path), None, - trigger.suspended_mode, + true, trigger_info, ) .await @@ -1073,7 +1073,7 @@ async fn route_job( return Ok(( StatusCode::OK, format!( - "Trigger: {} in inactive mode, incoming request has been queued", + "Trigger {} is in suspended mode, jobs are added to the queue but suspended", &trigger.path ), ) @@ -1084,8 +1084,9 @@ async fn route_job( match trigger.request_type { RequestType::SyncSse => { // Trigger the job (always async when streaming) - let (uuid, _, _) = trigger_runnable_inner( + let (uuid, _, _, _) = trigger_runnable_inner( &db, + None, Some(user_db.clone()), authed.clone(), &trigger.workspace_id, @@ -1157,7 +1158,7 @@ async fn route_job( trigger.error_handler_args.as_ref(), format!("http_trigger/{}", trigger.path), None, - trigger.suspended_mode, + false, trigger_info, ) .await diff --git a/backend/windmill-api/src/triggers/trigger_helpers.rs b/backend/windmill-api/src/triggers/trigger_helpers.rs index e8d50aa692..d3135c168b 100644 --- a/backend/windmill-api/src/triggers/trigger_helpers.rs +++ b/backend/windmill-api/src/triggers/trigger_helpers.rs @@ -1,6 +1,5 @@ use anyhow::Context; use axum::response::IntoResponse; -use chrono::{DateTime, Utc}; use http::StatusCode; use serde::Deserialize; use serde_json::value::RawValue; @@ -505,8 +504,9 @@ pub trait TriggerJobArgs { } #[allow(dead_code)] -pub async fn trigger_runnable_inner( +pub async fn trigger_runnable_inner<'c>( db: &DB, + tx_o: Option>, user_db: Option, authed: ApiAuthed, workspace_id: &str, @@ -520,7 +520,12 @@ pub async fn trigger_runnable_inner( job_id: Option, trigger: TriggerMetadata, suspended_mode: Option, -) -> Result<(Uuid, Option, Option)> { +) -> Result<( + Uuid, + Option, + Option, + Option>, +)> { let error_handler_args = error_handler_args.map(|args| { let args = args .0 @@ -531,12 +536,13 @@ pub async fn trigger_runnable_inner( }); let user_db = user_db.unwrap_or_else(|| UserDB::new(db.clone())); - let (uuid, delete_after_use, early_return) = if is_flow { + let (uuid, delete_after_use, early_return, tx_out) = if is_flow { let run_query = RunJobQuery { job_id, suspended_mode, ..Default::default() }; let path = StripPath(runnable_path.to_string()); - let (uuid, early_return) = push_flow_job_by_path_into_queue( + let (uuid, early_return, tx_out) = push_flow_job_by_path_into_queue( authed, db.clone(), + tx_o, user_db, workspace_id.to_string(), path, @@ -545,10 +551,11 @@ pub async fn trigger_runnable_inner( Some(trigger), ) .await?; - (uuid, None, early_return) + (uuid, None, early_return, tx_out) } else { - let (uuid, delete_after_use) = trigger_script_internal( + let (uuid, delete_after_use, tx_out) = trigger_script_internal( db, + tx_o, user_db, authed, workspace_id, @@ -563,10 +570,10 @@ pub async fn trigger_runnable_inner( suspended_mode, ) .await?; - (uuid, delete_after_use, None) + (uuid, delete_after_use, None, tx_out) }; - Ok((uuid, delete_after_use, early_return)) + Ok((uuid, delete_after_use, early_return, tx_out)) } #[allow(dead_code)] @@ -588,6 +595,7 @@ pub async fn trigger_runnable( ) -> Result { let uuid = trigger_runnable_inner( db, + None, user_db, authed, workspace_id, @@ -623,8 +631,9 @@ pub async fn trigger_runnable_and_wait_for_result( trigger: TriggerMetadata, ) -> Result { let username = authed.username.clone(); - let (uuid, delete_after_use, early_return) = trigger_runnable_inner( + let (uuid, delete_after_use, early_return, _) = trigger_runnable_inner( db, + None, user_db, authed, workspace_id, @@ -666,8 +675,9 @@ pub async fn trigger_runnable_and_wait_for_raw_result( trigger: TriggerMetadata, ) -> Result<(Box, bool)> { let username = authed.username.clone(); - let (uuid, delete_after_use, early_return) = trigger_runnable_inner( + let (uuid, delete_after_use, early_return, _) = trigger_runnable_inner( db, + None, user_db, authed, workspace_id, @@ -743,8 +753,9 @@ pub async fn trigger_runnable_and_wait_for_raw_result_with_error_ctx( } } -async fn trigger_script_internal( +async fn trigger_script_internal<'c>( db: &DB, + tx_o: Option>, user_db: UserDB, authed: ApiAuthed, workspace_id: &str, @@ -757,13 +768,18 @@ async fn trigger_script_internal( job_id: Option, trigger: TriggerMetadata, suspended_mode: Option, -) -> Result<(Uuid, Option)> { +) -> Result<( + Uuid, + Option, + Option>, +)> { if retry.is_none() && error_handler_path.is_none() { let run_query = RunJobQuery { job_id, suspended_mode, ..Default::default() }; let path = StripPath(script_path.to_string()); - push_script_job_by_path_into_queue( + let (uuid, delete_after_use, tx_out) = push_script_job_by_path_into_queue( authed, db.clone(), + tx_o, user_db, workspace_id.to_string(), path, @@ -771,10 +787,12 @@ async fn trigger_script_internal( args, Some(trigger), ) - .await + .await?; + Ok((uuid, delete_after_use, tx_out)) } else { - trigger_script_with_retry_and_error_handler( + let (uuid, delete_after_use, tx_out) = trigger_script_with_retry_and_error_handler( db, + tx_o, user_db, authed, workspace_id, @@ -788,12 +806,14 @@ async fn trigger_script_internal( trigger, suspended_mode, ) - .await + .await?; + Ok((uuid, delete_after_use, tx_out)) } } -async fn trigger_script_with_retry_and_error_handler( +async fn trigger_script_with_retry_and_error_handler<'c>( db: &DB, + tx_o: Option>, user_db: UserDB, authed: ApiAuthed, workspace_id: &str, @@ -806,7 +826,11 @@ async fn trigger_script_with_retry_and_error_handler( job_id: Option, trigger: TriggerMetadata, suspended_mode: Option, -) -> Result<(Uuid, Option)> { +) -> Result<( + Uuid, + Option, + Option>, +)> { #[cfg(feature = "enterprise")] check_license_key_valid().await?; @@ -830,22 +854,30 @@ async fn trigger_script_with_retry_and_error_handler( check_tag_available_for_workspace(&db, &workspace_id, &tag, &authed).await?; - let (email, permissioned_as, push_authed, tx) = - if let Some(on_behalf_of) = on_behalf_of.as_ref() { - ( - on_behalf_of.email.as_str(), - on_behalf_of.permissioned_as.clone(), - None, - PushIsolationLevel::IsolatedRoot(db.clone()), - ) - } else { - ( - authed.email.as_str(), - username_to_permissioned_as(&authed.username), - Some(authed.clone().into()), - PushIsolationLevel::Isolated(user_db, authed.clone().into()), - ) - }; + let return_tx = tx_o.is_some(); + + let (email, permissioned_as, push_authed, tx) = if let Some(tx) = tx_o { + ( + authed.email.as_str(), + username_to_permissioned_as(&authed.username), + Some(authed.clone().into()), + PushIsolationLevel::Transaction(tx), + ) + } else if let Some(on_behalf_of) = on_behalf_of.as_ref() { + ( + on_behalf_of.email.as_str(), + on_behalf_of.permissioned_as.clone(), + None, + PushIsolationLevel::IsolatedRoot(db.clone()), + ) + } else { + ( + authed.email.as_str(), + username_to_permissioned_as(&authed.username), + Some(authed.clone().into()), + PushIsolationLevel::Isolated(user_db, authed.clone().into()), + ) + }; let push_args = PushArgs { args: &args.args, extra: args.extra }; @@ -921,7 +953,12 @@ async fn trigger_script_with_retry_and_error_handler( suspended_mode, ) .await?; - tx.commit().await?; - Ok((uuid, delete_after_use)) + // If we were given a transaction, return it; otherwise commit it + if return_tx { + Ok((uuid, delete_after_use, Some(tx))) + } else { + tx.commit().await?; + Ok((uuid, delete_after_use, None)) + } } diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 637fe3165c..58e9780485 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -5279,10 +5279,10 @@ pub async fn push<'c, 'd>( root_job }; - let (job_kind, suspend, suspend_until) = if suspended_mode.unwrap_or(false) { - (JobKind::Unassigned, Some(1), Some(Utc::now() + chrono::Duration::days(30))) + let (job_kind, scheduled_for_o) = if suspended_mode.unwrap_or(false) { + (JobKind::Unassigned, Some(Utc::now() + chrono::Duration::days(30))) } else { - (job_kind, None, None) + (job_kind, scheduled_for_o) }; sqlx::query!( @@ -5305,8 +5305,8 @@ pub async fn push<'c, 'd>( ON CONFLICT (job_id) DO UPDATE SET email = EXCLUDED.email, username = EXCLUDED.username, is_admin = EXCLUDED.is_admin, is_operator = EXCLUDED.is_operator, folders = EXCLUDED.folders, groups = EXCLUDED.groups, workspace_id = EXCLUDED.workspace_id, end_user_email = EXCLUDED.end_user_email ) INSERT INTO v2_job_queue - (workspace_id, id, running, scheduled_for, started_at, tag, priority, suspend, suspend_until) - VALUES ($2, $1, $28, COALESCE($29, now()), CASE WHEN $27 OR $40 THEN now() END, $30, $31, $42, $43)", + (workspace_id, id, running, scheduled_for, started_at, tag, priority) + VALUES ($2, $1, $28, COALESCE($29, now()), CASE WHEN $27 OR $40 THEN now() END, $30, $31)", job_id, workspace_id, raw_code, @@ -5352,8 +5352,6 @@ pub async fn push<'c, 'd>( trigger_kind as Option, running, end_user_email, - suspend, - suspend_until, ) .execute(&mut *tx) .warn_after_seconds(1) diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 8d7caeda55..8207169016 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -2634,6 +2634,13 @@ pub async fn handle_queued_job( return Err(Error::ExecutionErr(e.to_string())); } + match job.kind { + JobKind::Unassigned => { + return Err(Error::ExecutionError("Suspended job was not handled by the user within 30 days, job will not be executed.".to_string())); + } + _ => {} + } + #[cfg(any(not(feature = "enterprise"), feature = "sqlx"))] match conn { Connection::Sql(db) => { diff --git a/frontend/src/lib/components/DropdownV2Inner.svelte b/frontend/src/lib/components/DropdownV2Inner.svelte index bbf97a35ec..7515cc1450 100644 --- a/frontend/src/lib/components/DropdownV2Inner.svelte +++ b/frontend/src/lib/components/DropdownV2Inner.svelte @@ -4,6 +4,7 @@ import { twMerge } from 'tailwind-merge' import type { MenubarMenuElements } from '@melt-ui/svelte' import type { Item } from '$lib/utils' + import Tooltip from './Tooltip.svelte' interface Props { aiId?: string @@ -52,6 +53,9 @@

{item.displayName}

+ {#if item.tooltip} + {item.tooltip} + {/if} {@render item.extra?.()} {/each} diff --git a/frontend/src/lib/components/RunsPage.svelte b/frontend/src/lib/components/RunsPage.svelte index 86bcac405d..d5ffc19c36 100644 --- a/frontend/src/lib/components/RunsPage.svelte +++ b/frontend/src/lib/components/RunsPage.svelte @@ -12,7 +12,8 @@ ScriptService, FlowService, type ExtendedJobs, - OpenAPI + OpenAPI, + type JobTriggerKind } from '$lib/gen' import { sendUserToast } from '$lib/toast' @@ -54,7 +55,6 @@ import ToggleButton from '$lib/components/common/toggleButton-v2/ToggleButton.svelte' import Select from '$lib/components/select/Select.svelte' import AnimatedPane from '$lib/components/splitPanes/AnimatedPane.svelte' - import type { JobTriggerType } from '$lib/components/triggers/utils' let { perPage = $bindable() }: { perPage: number } = $props() @@ -115,8 +115,8 @@ ? JSON.parse(decodeURIComponent(page.url.searchParams.get('result') ?? '{}')) : undefined ) - let jobTriggerKind: JobTriggerType | undefined = $state( - (page.url.searchParams.get('job_trigger_kind') as JobTriggerType) ?? undefined + let jobTriggerKind: JobTriggerKind | undefined = $state( + (page.url.searchParams.get('job_trigger_kind') as JobTriggerKind) ?? undefined ) // Handled on the main page @@ -167,7 +167,7 @@ resultFilter = page.url.searchParams.get('result') ? JSON.parse(decodeURIComponent(page.url.searchParams.get('result') ?? '{}')) : undefined - jobTriggerKind = (page.url.searchParams.get('job_trigger_kind') as JobTriggerType) ?? undefined + jobTriggerKind = (page.url.searchParams.get('job_trigger_kind') as JobTriggerKind) ?? undefined // Handled on the main page minTs = page.url.searchParams.get('min_ts') ?? undefined diff --git a/frontend/src/lib/components/common/modal/Modal2.svelte b/frontend/src/lib/components/common/modal/Modal2.svelte index 0fbefa59be..bec6efb3b4 100644 --- a/frontend/src/lib/components/common/modal/Modal2.svelte +++ b/frontend/src/lib/components/common/modal/Modal2.svelte @@ -50,7 +50,7 @@ {#if isOpen}
@@ -59,7 +59,7 @@ css?.popup?.style || '' }`} class={twMerge( - 'max-h-screen-80 max-w-screen-80 rounded-lg relative bg-surface pt-2 px-4 pb-4', + 'max-h-screen-80 max-w-screen-80 rounded-lg relative bg-surface p-4', css?.popup?.class, 'wm-modal-form-popup' )} diff --git a/frontend/src/lib/components/runs/JobsLoader.svelte b/frontend/src/lib/components/runs/JobsLoader.svelte index 4da600b91d..2868390c9f 100644 --- a/frontend/src/lib/components/runs/JobsLoader.svelte +++ b/frontend/src/lib/components/runs/JobsLoader.svelte @@ -8,7 +8,8 @@ ConcurrencyGroupsService, type ObscuredJob, CancelablePromise, - CancelError + CancelError, + type JobTriggerKind } from '$lib/gen' import { sendUserToast } from '$lib/toast' @@ -16,7 +17,6 @@ import { tweened, type Tweened } from 'svelte/motion' import { subtractDaysFromDateString } from '$lib/utils' - import type { JobTriggerType } from '../triggers/utils' import { CancelablePromiseUtils } from '$lib/cancelable-promise-utils' interface Props { @@ -31,7 +31,7 @@ showFutureJobs?: boolean argFilter: string | undefined resultFilter?: string | undefined - jobTriggerKind?: JobTriggerType | undefined + jobTriggerKind?: JobTriggerKind | undefined schedulePath?: string | undefined jobKindsCat?: string | undefined minTs?: string | undefined diff --git a/frontend/src/lib/components/runs/RunsFilter.svelte b/frontend/src/lib/components/runs/RunsFilter.svelte index 0df73f1b42..2c7ea5251b 100644 --- a/frontend/src/lib/components/runs/RunsFilter.svelte +++ b/frontend/src/lib/components/runs/RunsFilter.svelte @@ -17,7 +17,8 @@ import DropdownSelect from '../DropdownSelect.svelte' import TooltipV2 from '$lib/components/meltComponents/Tooltip.svelte' import TextInput from '../text_input/TextInput.svelte' - import { jobTriggerTypes, triggerDisplayNamesMap, type JobTriggerType } from '../triggers/utils' + import { jobTriggerKinds, triggerDisplayNamesMap } from '../triggers/utils' + import type { JobTriggerKind } from '$lib/gen' interface Props { // Filters @@ -31,7 +32,7 @@ argFilter: string argError: string resultFilter: string - jobTriggerKind: JobTriggerType | undefined + jobTriggerKind: JobTriggerKind | undefined resultError: string jobKindsCat: string user?: string | null @@ -964,7 +965,7 @@ {`Filter by what kind of trigger started the run.`} +
+ + Created At + Script/Flow Path + + + {#if loading} + + {#each new Array(3) as _} + + {#each new Array(4) as _} + + + + {/each} + + {/each} + + {:else if queuedJobs.length === 0} +
+

No suspended jobs found.

+
+ {:else} + + {#each queuedJobs as job} + + +
+ toggleJobSelection(job.id)} + disabled={processingAction} + class="cursor-pointer" + /> +
+ + + {displayDate(job.created_at)} + + + +
{job.script_path || '-'}
+ + {#if editedAt && job.created_at && new Date(editedAt) > new Date(job.created_at)} + Outdated + Trigger was edited after these jobs were created and suspended. They will be + reassigned to match the new trigger configuration, in particular the + runnable, retry and error handling settings. + + + {/if} + + + {/each} + + {/if} +
- -{/if} + + +
+ {#if queuedJobs.length === 0} + + + + {:else if hasSelectedJobs} + + + + {:else} + + + {/if} +
+ + diff --git a/frontend/src/lib/components/triggers/email/EmailTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/email/EmailTriggerEditorInner.svelte index 7ef99e6031..7135b65c85 100644 --- a/frontend/src/lib/components/triggers/email/EmailTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/email/EmailTriggerEditorInner.svelte @@ -25,7 +25,6 @@ import Tab from '$lib/components/common/tabs/Tab.svelte' import TriggerRetriesAndErrorHandler from '../TriggerRetriesAndErrorHandler.svelte' import { saveEmailTriggerFromCfg } from './utils' - import TriggerActiveMode from '../TriggerActiveMode.svelte' let { useDrawer = true, @@ -316,8 +315,6 @@ {/if} - - - - {/if} ({})) @@ -253,6 +253,7 @@ error_handler_args = defaultValues?.error_handler_args ?? {} retry = defaultValues?.retry ?? undefined errorHandlerSelected = getHandlerType(error_handler_path ?? '') + editedAt = undefined } finally { clearTimeout(loader) drawerLoading = false @@ -293,7 +294,8 @@ error_handler_args = cfg?.error_handler_args ?? {} retry = cfg?.retry errorHandlerSelected = getHandlerType(error_handler_path ?? '') - suspended_mode = cfg?.suspended_mode ?? true + suspended_mode = cfg?.suspended_mode ?? false + editedAt = cfg?.edited_at ?? undefined } async function loadTrigger(defaultConfig?: Partial): Promise { @@ -377,7 +379,22 @@ workspace: $workspaceStore ?? '', requestBody: { enabled: newEnabled } }) - sendUserToast(`${newEnabled ? 'enabled' : 'disabled'} HTTP trigger ${initialPath}`) + sendUserToast(`${newEnabled ? 'Enabled' : 'Disabled'} HTTP trigger ${initialPath}`) + } + } + + async function handleToggleSuspendedMode(newSuspendedMode: boolean, newEnabled: boolean = true) { + suspended_mode = newSuspendedMode + enabled = newSuspendedMode || newEnabled + if (!trigger?.draftConfig) { + await HttpTriggerService.updateHttpTriggerStatus({ + workspace: $workspaceStore ?? '', + path: initialPath, + requestBody: { suspended_mode: newSuspendedMode, enabled: newSuspendedMode || newEnabled } + }) + sendUserToast( + `${newSuspendedMode ? 'Suspended' : newEnabled ? 'Resumed' : 'Disabled'} HTTP trigger ${initialPath}` + ) } } @@ -610,8 +627,6 @@ {/if} - - {/if} {/if} {/snippet} diff --git a/frontend/src/lib/components/triggers/kafka/KafkaTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/kafka/KafkaTriggerEditorInner.svelte index b051266c85..3dab6d0891 100644 --- a/frontend/src/lib/components/triggers/kafka/KafkaTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/kafka/KafkaTriggerEditorInner.svelte @@ -19,7 +19,6 @@ import Tabs from '$lib/components/common/tabs/Tabs.svelte' import Tab from '$lib/components/common/tabs/Tab.svelte' import TriggerRetriesAndErrorHandler from '../TriggerRetriesAndErrorHandler.svelte' - import TriggerActiveMode from '../TriggerActiveMode.svelte' interface Props { useDrawer?: boolean @@ -386,8 +385,6 @@ {/if} - - {/if} {/if} - - {/if} - - {/if} -
{#snippet badge()} {#if isEditor} diff --git a/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte index 95842ee422..7ac8fad636 100644 --- a/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte @@ -24,7 +24,6 @@ import Tabs from '$lib/components/common/tabs/Tabs.svelte' import Tab from '$lib/components/common/tabs/Tab.svelte' import TriggerRetriesAndErrorHandler from '../TriggerRetriesAndErrorHandler.svelte' - import TriggerActiveMode from '../TriggerActiveMode.svelte' interface Props { useDrawer?: boolean @@ -387,8 +386,6 @@ {/if}
- - {/if} - -