windmill/backend/windmill-api-schedule/src/lib.rs

/*
 * Author: Ruben Fiszel
 * Copyright: Windmill Labs, Inc 2022
 * This file and its contents are licensed under the AGPLv3 License.
 * Please see the included NOTICE for copyright information and
 * LICENSE-AGPL for a copy of the license.
 */

use axum::{
    extract::{Extension, Path, Query},
    routing::{delete, get, post},
    Json, Router,
};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use sql_builder::{prelude::Bind, SqlBuilder};
use sqlx::{Postgres, Transaction};
use std::str::FromStr;
use windmill_api_auth::{
    build_scope_path_predicate, check_scopes, maybe_refresh_folders, require_super_admin, ApiAuthed,
};
use windmill_audit::audit_oss::audit_log;
use windmill_audit::ActionKind;
use windmill_common::DB;
use windmill_common::{
    can_preserve_on_behalf_of,
    db::UserDB,
    error::{Error, JsonResult, Result},
    schedule::Schedule,
    trigger_history::{
        self, TriggerHistoryEvent, TriggerOperation, TriggerSource, SCHEDULE_TRIGGER_KIND,
    },
    user_drafts::{
        delete_all_drafts_for_path, delete_draft_only_for_path, fetch_draft_only_list_rows,
        overlay_or_draft_only, UserDraftItemKind, WithDraftOverlay, WithDraftQuery,
    },
    utils::{
        escape_ilike_pattern, not_found_if_none, paginate, Pagination, ScheduleType, StripPath,
    },
    worker::to_raw_value,
};
use windmill_git_sync::{handle_deployment_metadata, DeployedObject};
use windmill_queue::schedule::push_scheduled_job;

/// Resolves the permissioned_as value for a schedule.
/// When preserving, uses the provided permissioned_as value directly.
fn resolve_permissioned_as(
    permissioned_as: Option<&String>,
    preserve_permissioned_as: Option<bool>,
    authed: &ApiAuthed,
) -> String {
    if let Some(permissioned_as) = permissioned_as {
        if preserve_permissioned_as.unwrap_or(false) && can_preserve_on_behalf_of(authed) {
            return permissioned_as.clone();
        }
    }
    windmill_common::users::username_to_permissioned_as(&authed.username)
}

/// Create-time variant: applies the folder's `default_permissioned_as` rule when no
/// explicit preserved value is provided and the caller can preserve (admin / wm_deployers).
async fn resolve_permissioned_as_for_create(
    permissioned_as: Option<&String>,
    preserve_permissioned_as: Option<bool>,
    path: &str,
    authed: &ApiAuthed,
    db: &DB,
    w_id: &str,
) -> Result<String> {
    if let Some(pa) = permissioned_as {
        if preserve_permissioned_as.unwrap_or(false) && can_preserve_on_behalf_of(authed) {
            return Ok(pa.clone());
        }
    }
    if can_preserve_on_behalf_of(authed) {
        if let Some(default) =
            windmill_common::folders::resolve_folder_default_permissioned_as(db, w_id, path).await?
        {
            return Ok(default);
        }
    }
    Ok(windmill_common::users::username_to_permissioned_as(
        &authed.username,
    ))
}

fn resolve_edited_by(authed: &ApiAuthed) -> String {
    authed.username.clone()
}

/// Append this mutation to `trigger_history`, diffing the row against `before`.
///
/// Call it on the transaction that made the change, after the change: the
/// snapshot it takes is the "after" side of the diff, and the two commit or roll
/// back together.
async fn record_schedule_history(
    tx: &mut sqlx::PgConnection,
    authed: &ApiAuthed,
    w_id: &str,
    path: &str,
    operation: TriggerOperation,
    before: Option<serde_json::Value>,
) -> Result<()> {
    let after = trigger_history::snapshot_row(&mut *tx, "schedule", w_id, path).await?;
    // Nothing to describe when the row is not there after the write: the same
    // guard the trigger side needs, kept here so the two read alike.
    if after.is_none() {
        return Ok(());
    }
    trigger_history::record(
        &mut *tx,
        TriggerHistoryEvent {
            workspace_id: w_id,
            trigger_kind: SCHEDULE_TRIGGER_KIND,
            path,
            operation,
            source: TriggerSource::of_request(authed.is_session_token),
            username: Some(&authed.username),
            changes: trigger_history::summarize_changes(before.as_ref(), after.as_ref()),
        },
    )
    .await
}

pub fn workspaced_service() -> Router {
    Router::new()
        .route("/list", get(list_schedule))
        .route("/list_with_jobs", get(list_schedule_with_jobs))
        .route("/get/{*path}", get(get_schedule))
        .route("/exists/{*path}", get(exists_schedule))
        .route("/create", post(create_schedule))
        .route("/update/{*path}", post(edit_schedule))
        .route("/delete/{*path}", delete(delete_schedule))
        .route("/setenabled/{*path}", post(set_enabled))
        .route("/setdefaulthandler", post(set_default_error_handler))
    // .route("/catchup/*path", post(do_catchup).get(list_catchup))
}

pub fn global_service() -> Router {
    Router::new().route("/preview", post(preview_schedule))
}

#[derive(Deserialize)]
pub struct NewSchedule {
    pub path: String,
    pub schedule: String,
    pub timezone: String,
    pub summary: Option<String>,
    pub description: Option<String>,
    pub no_flow_overlap: Option<bool>,
    pub script_path: String,
    pub is_flow: bool,
    pub args: Option<serde_json::Value>,
    pub enabled: Option<bool>,
    pub on_failure: Option<String>,
    pub on_failure_times: Option<i32>,
    pub on_failure_exact: Option<bool>,
    pub on_failure_extra_args: Option<serde_json::Value>,
    pub on_recovery: Option<String>,
    pub on_recovery_times: Option<i32>,
    pub on_recovery_extra_args: Option<serde_json::Value>,
    pub on_success: Option<String>,
    pub on_success_extra_args: Option<serde_json::Value>,
    pub ws_error_handler_muted: Option<bool>,
    pub retry: Option<serde_json::Value>,
    pub tag: Option<String>,
    pub paused_until: Option<DateTime<Utc>>,
    pub cron_version: Option<String>,
    pub dynamic_skip: Option<String>,
    pub permissioned_as: Option<String>,
    pub preserve_permissioned_as: Option<bool>,
    #[serde(default)]
    pub labels: Option<Vec<String>>,
}

#[derive(Serialize, Deserialize)]
pub struct ErrorOrRecoveryHandler {
    pub handler_type: HandlerType,
    pub override_existing: bool,

    pub path: Option<String>,
    pub extra_args: Option<serde_json::Value>,
    pub number_of_occurence: Option<i32>,
    pub number_of_occurence_exact: Option<bool>,
    pub workspace_handler_muted: Option<bool>,
}

#[derive(Serialize, Deserialize)]
#[serde(rename_all(serialize = "lowercase", deserialize = "lowercase"))]
pub enum HandlerType {
    Error,
    Recovery,
    Success,
}

async fn check_path_conflict<'c>(
    tx: &mut Transaction<'c, Postgres>,
    w_id: &str,
    path: &str,
) -> Result<()> {
    let exists = sqlx::query_scalar!(
        "SELECT EXISTS(SELECT 1 FROM schedule WHERE path = $1 AND workspace_id = $2)",
        path,
        w_id
    )
    .fetch_one(&mut **tx)
    .await?
    .unwrap_or(false);
    if exists {
        return Err(Error::BadRequest(format!(
            "Schedule {} already exists",
            path
        )));
    }
    return Ok(());
}

fn to_json_raw_opt(
    value: Option<&serde_json::Value>,
) -> Option<sqlx::types::Json<Box<serde_json::value::RawValue>>> {
    value.map(|v| sqlx::types::Json(to_raw_value(&v)))
}

/// Managed ducklake-maintenance schedules live under a reserved path prefix;
/// their lifecycle is owned by the workspace ducklake settings, so the
/// schedule API refuses to create/edit/delete/toggle them.
fn reject_reserved_schedule_path(path: &str) -> Result<()> {
    if path.starts_with(windmill_common::workspaces::DUCKLAKE_MAINTENANCE_PATH_PREFIX) {
        return Err(Error::BadRequest(format!(
            "Schedules under {} are managed by the workspace ducklake settings",
            windmill_common::workspaces::DUCKLAKE_MAINTENANCE_PATH_PREFIX
        )));
    }
    Ok(())
}

/// Validate that a dynamic skip handler (script or flow) exists
async fn validate_dynamic_skip<'c>(
    tx: &mut Transaction<'c, Postgres>,
    w_id: &str,
    handler_path: &str,
) -> Result<()> {
    // Check for script only (flows are not supported in the UI)
    let exists = sqlx::query_scalar!(
        "SELECT EXISTS(
            SELECT 1 FROM script
            WHERE workspace_id = $1 AND path = $2 AND archived = false AND deleted = false
        )",
        w_id,
        handler_path
    )
    .fetch_one(&mut **tx)
    .await?
    .unwrap_or(false);

    if exists {
        Ok(())
    } else {
        Err(Error::BadRequest(format!(
            "Dynamic skip handler '{}' not found. The handler must be an existing, non-archived script at schedule creation time.",
            handler_path
        )))
    }
}

async fn create_schedule(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Extension(user_db): Extension<UserDB>,
    Path(w_id): Path<String>,
    Json(ns): Json<NewSchedule>,
) -> Result<String> {
    check_scopes(&authed, || format!("schedules:write:{}", ns.path))?;
    reject_reserved_schedule_path(&ns.path)?;

    let authed = maybe_refresh_folders(&ns.path, &w_id, authed, &db).await;

    #[cfg(not(feature = "enterprise"))]
    if ns.on_recovery.is_some() {
        return Err(Error::BadRequest(
            "on_recovery is only available in enterprise version".to_string(),
        ));
    }

    #[cfg(not(feature = "enterprise"))]
    if ns.on_success.is_some() {
        return Err(Error::BadRequest(
            "on_success is only available in enterprise version".to_string(),
        ));
    }

    #[cfg(not(feature = "enterprise"))]
    if ns.on_failure_times.is_some() && ns.on_failure_times.unwrap() > 1 {
        return Err(Error::BadRequest(
            "on_failure with a number of times > 1 is only available in enterprise version"
                .to_string(),
        ));
    }

    // Check schedule for error (validate before opening the tx).
    ScheduleType::from_str(&ns.schedule, ns.cron_version.as_deref(), true)?;

    // These reads deliberately use the non-RLS `db` pool (fork-ness and
    // permissioned_as resolution must be complete regardless of the caller's
    // folder perms). Run them BEFORE opening the RLS transaction below: acquiring
    // a second pooled connection while the tx is held self-deadlocks on a
    // single-connection pool, and they don't depend on the tx.
    //
    // A git-sync/merge/create write into a fork never sets operational state:
    // force `enabled = false` so a cloned / synced / merged / UI-created schedule
    // can't fire alongside the parent's. The fork owner re-enables locally via
    // `setenabled`. Schedule analog of the trigger rule in
    // `windmill-trigger::handler::workspace_is_fork`; the read half (parent-value
    // substitution on fork export) lives in `workspaces_export.rs`.
    let target_is_fork: bool = sqlx::query_scalar!(
        "SELECT parent_workspace_id IS NOT NULL FROM workspace WHERE id = $1",
        w_id
    )
    .fetch_optional(&db)
    .await?
    .flatten()
    .unwrap_or(false);

    let resolved_edited_by = resolve_edited_by(&authed);
    let resolved_permissioned_as = resolve_permissioned_as_for_create(
        ns.permissioned_as.as_ref(),
        ns.preserve_permissioned_as,
        &ns.path,
        &authed,
        &db,
        &w_id,
    )
    .await?;
    // email is still written for backwards compat with old workers that don't know about permissioned_as
    let resolved_email = windmill_common::users::get_email_from_permissioned_as_uncached(
        &resolved_permissioned_as,
        &w_id,
        &db,
    )
    .await?;

    // Reject a forged superadmin run identity in a preserved permissioned_as
    // (the sentinel guard; the email is derived from it so it always belongs).
    windmill_common::auth::validate_on_behalf_of(
        Some(&resolved_permissioned_as),
        Some(&resolved_email),
    )?;

    let mut tx: Transaction<'_, Postgres> = user_db.begin(&authed).await?;

    check_path_conflict(&mut tx, &w_id, &ns.path).await?;
    check_flow_conflict(&mut tx, &w_id, &ns.path, ns.is_flow, &ns.script_path).await?;

    // Validate dynamic_skip if provided
    if let Some(handler_path) = &ns.dynamic_skip {
        validate_dynamic_skip(&mut tx, &w_id, handler_path).await?;
    }

    let schedule = sqlx::query_as!(
        Schedule,
        r#"
        INSERT INTO schedule (
            workspace_id, path, schedule, timezone, edited_by, script_path,
            is_flow, args, enabled, email, permissioned_as,
            on_failure, on_failure_times, on_failure_exact, on_failure_extra_args,
            on_recovery, on_recovery_times, on_recovery_extra_args,
            on_success, on_success_extra_args,
            ws_error_handler_muted, retry, summary, no_flow_overlap,
            tag, paused_until, cron_version, description, dynamic_skip, labels
        ) VALUES (
            $1, $2, $3, $4, $5, $6,
            $7, $8, $9, $10, $11,
            $12, $13, $14, $15,
            $16, $17, $18,
            $19, $20,
            $21, $22, $23, $24,
            $25, $26, $27, $28, $29, $30
        )
        RETURNING
            workspace_id,
            path,
            edited_by,
            edited_at,
            schedule,
            timezone,
            enabled,
            script_path,
            is_flow,
            args AS "args: _",
            extra_perms,
            email,
            permissioned_as,
            error,
            on_failure,
            on_failure_times,
            on_failure_exact,
            on_failure_extra_args AS "on_failure_extra_args: _",
            on_recovery,
            on_recovery_times,
            on_recovery_extra_args AS "on_recovery_extra_args: _",
            on_success,
            on_success_extra_args  AS "on_success_extra_args: _",
            ws_error_handler_muted,
            retry,
            no_flow_overlap,
            summary,
            description,
            tag,
            paused_until,
            cron_version,
            dynamic_skip,
            labels,
            late_run_streak
        "#,
        w_id,
        ns.path,
        ns.schedule,
        ns.timezone,
        resolved_edited_by,
        ns.script_path,
        ns.is_flow,
        to_json_raw_opt(ns.args.as_ref())
            as Option<sqlx::types::Json<Box<serde_json::value::RawValue>>>,
        // Default-on matches the enqueue check below (line ~410) and the trigger
        // create path (`BaseTriggerData::mode()` defaults to `Enabled`). Every
        // production caller passes `enabled` explicitly except the fork→parent
        // flows (CLI merge, UI merge, `wmill push` of a fork tarball) — which
        // either send the source's actual flag (create case) or omit `enabled`
        // entirely (update case, where `EditSchedule` lacks the field).
        // A write into a fork always lands disabled regardless of the request.
        if target_is_fork {
            false
        } else {
            ns.enabled.unwrap_or(true)
        },
        resolved_email,
        resolved_permissioned_as,
        ns.on_failure,
        ns.on_failure_times,
        ns.on_failure_exact,
        to_json_raw_opt(ns.on_failure_extra_args.as_ref())
            as Option<sqlx::types::Json<Box<serde_json::value::RawValue>>>,
        ns.on_recovery,
        ns.on_recovery_times,
        to_json_raw_opt(ns.on_recovery_extra_args.as_ref())
            as Option<sqlx::types::Json<Box<serde_json::value::RawValue>>>,
        ns.on_success,
        to_json_raw_opt(ns.on_success_extra_args.as_ref())
            as Option<sqlx::types::Json<Box<serde_json::value::RawValue>>>,
        ns.ws_error_handler_muted.unwrap_or(false),
        ns.retry,
        ns.summary,
        ns.no_flow_overlap.unwrap_or(false),
        ns.tag,
        ns.paused_until,
        ns.cron_version.clone().unwrap_or_else(|| "v2".to_string()),
        ns.description,
        ns.dynamic_skip,
        ns.labels.as_deref() as Option<&[String]>
    )
    .fetch_one(&mut *tx)
    .await
    .map_err(|e| Error::internal_err(format!("inserting schedule in {w_id}: {e:#}")))?;

    record_schedule_history(
        &mut *tx,
        &authed,
        &w_id,
        &ns.path,
        TriggerOperation::Create,
        None,
    )
    .await?;

    audit_log(
        &mut *tx,
        &authed,
        "schedule.create",
        ActionKind::Create,
        &w_id,
        Some(&ns.path.to_string()),
        Some(
            [
                Some(("schedule", ns.schedule.as_str())),
                Some(("script_path", ns.script_path.as_str())),
            ]
            .into_iter()
            .flatten()
            .collect(),
        ),
    )
    .await?;
    if let Some(on_behalf_of) = windmill_common::check_on_behalf_of_preservation(
        ns.permissioned_as.as_deref(),
        ns.preserve_permissioned_as.unwrap_or(false),
        &authed,
        &authed.username,
    ) {
        audit_log(
            &mut *tx,
            &authed,
            "schedule.on_behalf_of",
            ActionKind::Create,
            &w_id,
            Some(&ns.path),
            Some(
                [
                    ("on_behalf_of", on_behalf_of.as_str()),
                    ("action", "create"),
                ]
                .into(),
            ),
        )
        .await?;
    }

    if !target_is_fork && ns.enabled.unwrap_or(true) {
        tx = push_scheduled_job(&db, tx, &schedule, Some(&authed.clone().into()), None).await?
    }
    tx.commit().await?;

    handle_deployment_metadata(
        &authed.email,
        &authed.username,
        &db,
        &w_id,
        DeployedObject::Schedule { path: ns.path.clone() },
        Some(format!("Schedule '{}' created", ns.path.clone())),
        true,
        None,
    )
    .await?;

    Ok(ns.path.to_string())
}

async fn edit_schedule(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Extension(user_db): Extension<UserDB>,
    Path((w_id, path)): Path<(String, StripPath)>,
    Json(es): Json<EditSchedule>,
) -> Result<String> {
    let path = path.to_path();
    check_scopes(&authed, || format!("schedules:write:{}", path))?;
    reject_reserved_schedule_path(path)?;

    let authed = maybe_refresh_folders(&path, &w_id, authed, &db).await;

    // Check schedule for error
    ScheduleType::from_str(&es.schedule, es.cron_version.as_deref(), true)?;

    let resolved_edited_by = resolve_edited_by(&authed);

    // Resolved on the (non-RLS) pool before the RLS transaction opens: the lookup mid-transaction
    // would hold a second connection while `tx` is checked out.
    let resolved_permissioned_as = resolve_permissioned_as(
        es.permissioned_as.as_ref(),
        es.preserve_permissioned_as,
        &authed,
    );

    // email is still written for backwards compat with old workers that don't know about permissioned_as.
    // When permissioned_as is preserved to a different user, derive email from it.
    let resolved_email = if resolved_permissioned_as
        != windmill_common::users::username_to_permissioned_as(&authed.username)
    {
        windmill_common::users::get_email_from_permissioned_as_uncached(
            &resolved_permissioned_as,
            &w_id,
            &db,
        )
        .await?
    } else {
        authed.email.clone()
    };

    // Reject a forged superadmin run identity in a preserved permissioned_as
    // (the sentinel guard; the email is derived from it so it always belongs).
    windmill_common::auth::validate_on_behalf_of(
        Some(&resolved_permissioned_as),
        Some(&resolved_email),
    )?;

    let mut tx = user_db.begin(&authed).await?;

    // Validate dynamic_skip if provided
    if let Some(handler_path) = &es.dynamic_skip {
        validate_dynamic_skip(&mut tx, &w_id, handler_path).await?;
    }

    let before = trigger_history::snapshot_row(&mut *tx, "schedule", &w_id, path).await?;

    let schedule = sqlx::query_as!(
        Schedule,
        r#"
        UPDATE schedule SET
            schedule                = $1,
            timezone                = $2,
            args                    = $3,
            on_failure              = $4,
            on_failure_times        = $5,
            on_failure_exact        = $6,
            on_failure_extra_args   = $7,
            on_recovery             = $8,
            on_recovery_times       = $9,
            on_recovery_extra_args  = $10,
            on_success              = $11,
            on_success_extra_args   = $12,
            ws_error_handler_muted  = $13,
            retry                   = $14,
            summary                 = $15,
            no_flow_overlap         = $16,
            tag                     = $17,
            paused_until            = $18,
            path                    = $19,
            workspace_id            = $20,
            cron_version            = COALESCE($21, cron_version),
            description             = $22,
            dynamic_skip            = $23,
            email                   = $24,
            edited_by               = $25,
            edited_at               = now(),
            permissioned_as         = $26,
            labels                  = COALESCE($27, labels),
            late_run_streak            = 0,
            missed_occurrences     = 0,
            last_missed_at              = NULL
        WHERE path = $19 AND workspace_id = $20
        RETURNING
            workspace_id,
            path,
            edited_by,
            edited_at,
            schedule,
            timezone,
            enabled,
            script_path,
            is_flow,
            args AS "args: _",
            extra_perms,
            email,
            permissioned_as,
            error,
            on_failure,
            on_failure_times,
            on_failure_exact,
            on_failure_extra_args AS "on_failure_extra_args: _",
            on_recovery,
            on_recovery_times,
            on_recovery_extra_args AS "on_recovery_extra_args: _",
            on_success,
            on_success_extra_args AS "on_success_extra_args: _",
            ws_error_handler_muted,
            retry,
            no_flow_overlap,
            summary,
            description,
            tag,
            paused_until,
            cron_version,
            dynamic_skip,
            labels,
            late_run_streak
        "#,
        es.schedule,
        es.timezone,
        to_json_raw_opt(es.args.as_ref())
            as Option<sqlx::types::Json<Box<serde_json::value::RawValue>>>,
        es.on_failure,
        es.on_failure_times,
        es.on_failure_exact,
        to_json_raw_opt(es.on_failure_extra_args.as_ref())
            as Option<sqlx::types::Json<Box<serde_json::value::RawValue>>>,
        es.on_recovery,
        es.on_recovery_times,
        to_json_raw_opt(es.on_recovery_extra_args.as_ref())
            as Option<sqlx::types::Json<Box<serde_json::value::RawValue>>>,
        es.on_success,
        to_json_raw_opt(es.on_success_extra_args.as_ref())
            as Option<sqlx::types::Json<Box<serde_json::value::RawValue>>>,
        es.ws_error_handler_muted.unwrap_or(false),
        es.retry,
        es.summary,
        es.no_flow_overlap.unwrap_or(false),
        es.tag,
        es.paused_until,
        path,
        w_id,
        es.cron_version,
        es.description,
        es.dynamic_skip,
        resolved_email,
        resolved_edited_by,
        resolved_permissioned_as,
        es.labels.as_deref() as Option<&[String]>
    )
    .fetch_one(&mut *tx)
    .await
    .map_err(|e| Error::internal_err(format!("updating schedule in {w_id}: {e:#}")))?;

    // clear_schedule must come AFTER UPDATE schedule to maintain consistent lock ordering
    // (schedule row first, then v2_job_queue) and avoid deadlocks with concurrent operations
    // like set_enabled, flow updates, and worker job completions.
    clear_schedule(&mut tx, path, &w_id).await?;

    record_schedule_history(
        &mut *tx,
        &authed,
        &w_id,
        path,
        TriggerOperation::Update,
        before,
    )
    .await?;

    audit_log(
        &mut *tx,
        &authed,
        "schedule.edit",
        ActionKind::Update,
        &w_id,
        Some(&path.to_string()),
        Some(
            [Some(("schedule", es.schedule.as_str()))]
                .into_iter()
                .flatten()
                .collect(),
        ),
    )
    .await?;

    if let Some(on_behalf_of) = windmill_common::check_on_behalf_of_preservation(
        es.permissioned_as.as_deref(),
        es.preserve_permissioned_as.unwrap_or(false),
        &authed,
        &authed.username,
    ) {
        audit_log(
            &mut *tx,
            &authed,
            "schedule.on_behalf_of",
            ActionKind::Update,
            &w_id,
            Some(&path.to_string()),
            Some([("on_behalf_of", on_behalf_of.as_str()), ("action", "edit")].into()),
        )
        .await?;
    }

    if schedule.enabled {
        tx = push_scheduled_job(&db, tx, &schedule, None, None).await?;
    }
    tx.commit().await?;

    handle_deployment_metadata(
        &authed.email,
        &authed.username,
        &db,
        &w_id,
        DeployedObject::Schedule { path: path.to_string() },
        None,
        true,
        None,
    )
    .await?;

    Ok(path.to_string())
}

#[derive(Deserialize)]
pub struct ListScheduleQuery {
    pub page: Option<usize>,
    pub per_page: Option<usize>,
    pub path: Option<String>,
    pub is_flow: Option<bool>,
    // filter by matching a subset of the args using base64 encoded json subset
    pub args: Option<String>,
    pub path_start: Option<String>,
    // exact match on schedule path
    pub schedule_path: Option<String>,
    // filter on description (pattern match)
    pub description: Option<String>,
    // filter on summary (pattern match)
    pub summary: Option<String>,
    pub broad_filter: Option<String>,
    pub label: Option<String>,
    /// When true, append per-user draft-only rows; picker callers leave it off
    /// to stay deployed-only. See list synthesis in scripts.rs.
    pub include_draft_only: Option<bool>,
}

#[derive(sqlx::FromRow, Serialize, Deserialize, Debug, Clone)]
pub struct ScheduleLight {
    pub workspace_id: String,
    pub path: String,
    pub edited_by: String,
    pub edited_at: DateTime<chrono::Utc>,
    pub schedule: String,
    pub timezone: String,
    pub enabled: bool,
    pub script_path: String,
    pub is_flow: bool,
    pub summary: Option<String>,
    pub extra_perms: serde_json::Value,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub labels: Option<Vec<String>>,
    /// `Some(true)` only on synthesized draft-only rows; `None` on deployed rows.
    #[serde(skip_serializing_if = "Option::is_none")]
    #[sqlx(default)]
    pub draft_only: Option<bool>,
    /// True when the authed user has a per-user draft at this path (drives the
    /// `*` suffix on the schedules page).
    #[serde(skip_serializing_if = "Option::is_none")]
    #[sqlx(default)]
    pub is_draft: Option<bool>,
    /// Labels inherited from the parent folder, computed at read time.
    #[sqlx(default)]
    #[serde(skip_serializing_if = "Option::is_none")]
    pub inherited_labels: Option<Vec<String>>,
}
async fn list_schedule(
    authed: ApiAuthed,
    Extension(user_db): Extension<UserDB>,
    Extension(db): Extension<DB>,
    Path(w_id): Path<String>,
    Query(lsq): Query<ListScheduleQuery>,
) -> JsonResult<Vec<ScheduleLight>> {
    let mut tx = user_db.begin(&authed).await?;
    let (per_page, offset) = paginate(Pagination { per_page: lsq.per_page, page: lsq.page });
    let mut sqlb = SqlBuilder::select_from("schedule")
        .fields(&[
            "workspace_id",
            "path",
            "edited_by",
            "edited_at",
            "schedule",
            "timezone",
            "enabled",
            "script_path",
            "is_flow",
            "summary",
            "extra_perms",
            "labels",
            "folder_labels(workspace_id, path) as inherited_labels",
        ])
        // Scalar EXISTS flags the authed user's per-user draft; see resources.rs.
        .field(
            &"EXISTS(SELECT 1 FROM draft WHERE draft.workspace_id = schedule.workspace_id \
              AND draft.path = schedule.path AND draft.typ = 'trigger_schedule' \
              AND draft.email = ?) as is_draft"
                .bind(&authed.email),
        )
        .order_by("edited_at", true)
        .and_where("workspace_id = ?".bind(&w_id))
        // managed ducklake-maintenance schedules are edited from the
        // workspace ducklake settings, not the schedules UI/CLI.
        // starts_with, not LIKE: the prefix contains `_` which LIKE treats as
        // a wildcard, and a user folder like `ducklake-maintenance` must not
        // be swept up.
        .and_where(
            "NOT starts_with(path, ?)"
                .bind(&windmill_common::workspaces::DUCKLAKE_MAINTENANCE_PATH_PREFIX),
        )
        .offset(offset)
        .limit(per_page)
        .clone();
    if let Some(path) = lsq.path.as_ref() {
        sqlb.and_where_eq("script_path", "?".bind(path));
    }
    if let Some(is_flow) = lsq.is_flow {
        sqlb.and_where_eq("is_flow", "?".bind(&is_flow));
    }
    if let Some(args) = &lsq.args {
        if let Ok(v) = serde_json::from_str::<serde_json::Value>(args) {
            sqlb.and_where("args @> ?".bind(&v.to_string()));
        } else {
            sqlb.and_where("FALSE");
        }
    }
    if let Some(path_start) = &lsq.path_start {
        sqlb.and_where_like_left("path", path_start);
    }
    if let Some(schedule_path) = &lsq.schedule_path {
        sqlb.and_where_eq("path", "?".bind(schedule_path));
    }
    if let Some(description) = &lsq.description {
        let pat = format!("%{}%", escape_ilike_pattern(description));
        sqlb.and_where("description ILIKE ?".bind(&pat));
    }
    if let Some(summary) = &lsq.summary {
        let pat = format!("%{}%", escape_ilike_pattern(summary));
        sqlb.and_where("summary ILIKE ?".bind(&pat));
    }
    if let Some(broad_filter) = &lsq.broad_filter {
        let pat = format!("%{}%", escape_ilike_pattern(broad_filter));
        sqlb.and_where(
            "(path ILIKE ? OR script_path ILIKE ? OR description ILIKE ? OR summary ILIKE ? OR schedule ILIKE ?)"
                .bind(&pat).bind(&pat).bind(&pat).bind(&pat).bind(&pat)
        );
    }
    if let Some(label) = &lsq.label {
        for l in label.split(',') {
            sqlb.and_where(
                "(labels @> ARRAY[?] OR folder_labels(workspace_id, path) @> ARRAY[?])"
                    .bind(&l.trim())
                    .bind(&l.trim()),
            );
        }
    }
    let sql = sqlb.sql().map_err(|e| Error::internal_err(e.to_string()))?;
    let mut rows = sqlx::query_as::<_, ScheduleLight>(&sql)
        .fetch_all(&mut *tx)
        .await?;
    tx.commit().await?;

    // Append the authed user's draft-only schedules; see scripts.rs.
    if lsq.include_draft_only.unwrap_or(false)
        && offset == 0
        && lsq.path.is_none()
        && lsq.is_flow.is_none()
        && lsq.args.is_none()
        && lsq.path_start.is_none()
        && lsq.schedule_path.is_none()
        && lsq.description.is_none()
        && lsq.summary.is_none()
        && lsq.broad_filter.is_none()
        && lsq.label.is_none()
        && (!authed.is_operator
            || windmill_common::workspaces::operator_can_draft(
                &db,
                &w_id,
                UserDraftItemKind::TriggerSchedule,
            )
            .await?)
    {
        let draft_only_rows = fetch_draft_only_list_rows(
            &db,
            &w_id,
            &authed.email,
            UserDraftItemKind::TriggerSchedule,
        )
        .await?;

        for row in draft_only_rows {
            let v: serde_json::Value =
                serde_json::from_str(row.value.0.get()).unwrap_or(serde_json::Value::Null);
            // Schedule editor's draft mirrors NewSchedule: { path, schedule, timezone, script_path, is_flow, enabled?, summary?, labels? }
            let path = v
                .get("path")
                .and_then(|s| s.as_str())
                .unwrap_or("")
                .to_string();
            if path.is_empty() {
                continue;
            }
            let schedule = v
                .get("schedule")
                .and_then(|x| x.as_str())
                .unwrap_or("")
                .to_string();
            let timezone = v
                .get("timezone")
                .and_then(|x| x.as_str())
                .unwrap_or("UTC")
                .to_string();
            let script_path = v
                .get("script_path")
                .and_then(|x| x.as_str())
                .unwrap_or("")
                .to_string();
            let is_flow = v.get("is_flow").and_then(|x| x.as_bool()).unwrap_or(false);
            let enabled = v.get("enabled").and_then(|x| x.as_bool()).unwrap_or(true);
            let summary = v
                .get("summary")
                .and_then(|x| x.as_str())
                .map(|s| s.to_string());
            let labels = v.get("labels").and_then(|x| {
                x.as_array().map(|arr| {
                    arr.iter()
                        .filter_map(|s| s.as_str().map(|s| s.to_string()))
                        .collect::<Vec<_>>()
                })
            });

            rows.push(ScheduleLight {
                workspace_id: w_id.clone(),
                path,
                edited_by: String::new(),
                edited_at: row.created_at,
                schedule,
                timezone,
                enabled,
                script_path,
                is_flow,
                summary,
                extra_perms: serde_json::Value::Object(serde_json::Map::new()),
                labels,
                // No deployed row to inherit folder labels from.
                inherited_labels: None,
                draft_only: Some(true),
                // Synthesized rows are the authed user's draft.
                is_draft: Some(true),
            });
        }
    }

    let allowed = build_scope_path_predicate(&authed, "schedules", "read");
    rows.retain(|r| allowed(&r.path));

    Ok(Json(rows))
}

#[derive(Serialize, Deserialize, Debug)]
pub struct ScheduleWJobs {
    pub path: String,
    pub jobs: Option<Vec<serde_json::Value>>,
    pub late_run_streak: i32,
    pub missed_occurrences: i32,
    pub last_missed_at: Option<chrono::DateTime<chrono::Utc>>,
}

async fn list_schedule_with_jobs(
    authed: ApiAuthed,
    Extension(user_db): Extension<UserDB>,
    Path(w_id): Path<String>,
    Query(pagination): Query<Pagination>,
) -> JsonResult<Vec<ScheduleWJobs>> {
    let mut tx = user_db.begin(&authed).await?;
    let (per_page, offset) = paginate(pagination);
    let rows = sqlx::query_as!(ScheduleWJobs,
        // Query plan:
        // - use of the `ix_completed_job_workspace_id_started_at_new_2` index first, then;
        // - use of the `ix_v2_job_root_by_path` index; hence the `parent_job IS NULL` clause.
        // - both `workspace_id = $1` checks are required to hit both indexes.
        "SELECT
            schedule.path, t.jobs, schedule.late_run_streak, schedule.missed_occurrences, schedule.last_missed_at FROM schedule,
            LATERAL(SELECT ARRAY(
                SELECT json_build_object('id', id, 'success', status = 'success', 'duration_ms', duration_ms)
                FROM v2_job_completed c JOIN v2_job j USING (id)
                WHERE trigger_kind = 'schedule'
                    AND trigger = schedule.path
                    AND c.workspace_id = $1
                    AND j.workspace_id = $1
                    AND parent_job IS NULL AND runnable_path = schedule.script_path
                    AND status <> 'skipped'
                ORDER BY completed_at DESC
                LIMIT 20
            ) AS jobs) t
        WHERE workspace_id = $1 AND NOT starts_with(schedule.path, $4)
        ORDER BY edited_at DESC
        LIMIT $2 OFFSET $3",
        w_id,
        per_page as i64,
        offset as i64,
        windmill_common::workspaces::DUCKLAKE_MAINTENANCE_PATH_PREFIX
    )
    .fetch_all(&mut *tx)
    .await?;
    tx.commit().await?;
    let allowed = build_scope_path_predicate(&authed, "schedules", "read");
    Ok(Json(
        rows.into_iter().filter(|r| allowed(&r.path)).collect(),
    ))
}

// SELECT id, title AS item_title, t.tag_array
// FROM   items i, LATERAL (  -- this is an implicit CROSS JOIN
//    SELECT ARRAY (
//       SELECT t.title
//       FROM   items_tags it
//       JOIN   tags       t  ON t.id = it.tag_id
//       WHERE  it.item_id = i.id
//       ) AS tag_array
//    ) t;

async fn get_schedule(
    authed: ApiAuthed,
    Extension(user_db): Extension<UserDB>,
    Extension(db): Extension<DB>,
    Path((w_id, path)): Path<(String, StripPath)>,
    Query(q): Query<WithDraftQuery>,
) -> JsonResult<WithDraftOverlay> {
    let path = path.to_path();
    check_scopes(&authed, || format!("schedules:read:{}", path))?;
    let mut tx = user_db.begin(&authed).await?;

    let schedule_o = windmill_queue::schedule::get_schedule_opt(&mut *tx, &w_id, path).await?;
    tx.commit().await?;
    let overlay = overlay_or_draft_only(
        &db,
        &w_id,
        &authed.email,
        UserDraftItemKind::TriggerSchedule,
        path,
        q.get_draft,
        schedule_o,
        || Error::NotFound(format!("Schedule not found at path {path}")),
    )
    .await?;
    Ok(Json(overlay))
}

async fn exists_schedule(
    Extension(db): Extension<DB>,
    Path((w_id, path)): Path<(String, StripPath)>,
) -> JsonResult<bool> {
    let mut tx = db.begin().await?;
    let res = windmill_queue::schedule::exists_schedule(&mut tx, w_id, path).await?;
    tx.commit().await?;
    Ok(Json(res))
}

#[derive(Deserialize)]
pub struct PreviewPayload {
    pub schedule: String,
    pub timezone: String,
    pub cron_version: Option<String>,
}

pub async fn preview_schedule(
    Json(payload): Json<PreviewPayload>,
) -> JsonResult<Vec<DateTime<Utc>>> {
    let schedule =
        ScheduleType::from_str(&payload.schedule, payload.cron_version.as_deref(), true)?;

    let tz =
        chrono_tz::Tz::from_str(&payload.timezone).map_err(|e| Error::BadRequest(e.to_string()))?;

    let upcoming: Vec<DateTime<Utc>> = schedule.upcoming(tz, 5)?;

    Ok(Json(upcoming))
}

pub async fn set_enabled(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Extension(user_db): Extension<UserDB>,
    Path((w_id, path)): Path<(String, StripPath)>,
    Json(payload): Json<SetEnabled>,
) -> Result<String> {
    let mut tx = user_db.begin(&authed).await?;
    let path = path.to_path();
    check_scopes(&authed, || format!("schedules:write:{}", path))?;
    reject_reserved_schedule_path(path)?;

    // Block enabling a schedule in a fork when an ancestor has the same path
    // (regardless of its enabled flag), unless force=true. Two enabled crons
    // fire in lockstep; even when the ancestor is currently disabled the user
    // is likely to re-enable it later, at which point both fire — better to
    // surface that risk at every fork-side enable. There's no namespacing fix
    // for schedules (Phase 3 doesn't help cron); the user has to confirm or
    // point the script at fork-only side effects.
    if payload.enabled && !payload.force {
        if let Some(ancestor_id) = windmill_common::workspaces::nearest_fork_ancestor_having(
            &mut *tx, "schedule", &w_id, path,
        )
        .await?
        {
            return Err(Error::BadRequest(format!(
                "fork-conflict:schedule:{}",
                ancestor_id
            )));
        }
    }
    let before = trigger_history::snapshot_row(&mut *tx, "schedule", &w_id, path).await?;

    // email is still written for backwards compat with old workers that don't know about permissioned_as
    let schedule_o = sqlx::query_as!(
        Schedule,
        r#"
        UPDATE schedule SET
            enabled = $1,
            email = $2,
            late_run_streak = 0,
            missed_occurrences = 0,
            last_missed_at = NULL
        WHERE path = $3 AND workspace_id = $4
        RETURNING
            workspace_id,
            path,
            edited_by,
            edited_at,
            schedule,
            timezone,
            enabled,
            script_path,
            is_flow,
            args AS "args: _",
            extra_perms,
            email,
            permissioned_as,
            error,
            on_failure,
            on_failure_times,
            on_failure_exact,
            on_failure_extra_args AS "on_failure_extra_args: _",
            on_recovery,
            on_recovery_times,
            on_recovery_extra_args AS "on_recovery_extra_args: _",
            on_success,
            on_success_extra_args AS "on_success_extra_args: _",
            ws_error_handler_muted,
            retry,
            no_flow_overlap,
            summary,
            description,
            tag,
            paused_until,
            cron_version,
            dynamic_skip,
            labels,
            late_run_streak
        "#,
        payload.enabled,
        authed.email,
        path,
        w_id
    )
    .fetch_optional(&mut *tx)
    .await?;

    let schedule = not_found_if_none(schedule_o, "Schedule", path)?;

    clear_schedule(&mut tx, path, &w_id).await?;

    record_schedule_history(
        &mut *tx,
        &authed,
        &w_id,
        path,
        if payload.enabled {
            TriggerOperation::Enable
        } else {
            TriggerOperation::Disable
        },
        before,
    )
    .await?;

    audit_log(
        &mut *tx,
        &authed,
        "schedule.setenabled",
        ActionKind::Update,
        &w_id,
        Some(path),
        Some([("enabled", payload.enabled.to_string().as_ref())].into()),
    )
    .await?;

    if payload.enabled {
        tx = push_scheduled_job(&db, tx, &schedule, None, None).await?;
    }
    tx.commit().await?;

    handle_deployment_metadata(
        &authed.email,
        &authed.username,
        &db,
        &w_id,
        DeployedObject::Schedule { path: path.to_string() },
        None,
        true,
        None,
    )
    .await?;

    Ok(format!(
        "succesfully updated schedule at path {} to status {}",
        path, payload.enabled
    ))
}

// pub async fn do_catchup(
//     authed: ApiAuthed,
//     Extension(db): Extension<DB>,
//     Extension(user_db): Extension<UserDB>,
// //     Path((w_id, path)): Path<(String, StripPath)>,
//     Json(payload): Json<SetEnabled>,
// ) -> Result<String> {
//     let mut tx: QueueTransaction<'_, rsmq_async::MultiplexedRsmq> =
//         (user_db.begin(&authed).await?).into();
//     let path = path.to_path();
//     let schedule_o = sqlx::query_as!(
//         Schedule,
//         "UPDATE schedule SET enabled = $1, email = $2 WHERE path = $3 AND workspace_id = $4 RETURNING *",
//         &payload.enabled,
//         authed.email,
//         path,
//         w_id
//     )
//     .fetch_optional(&mut *tx)
//     .await?;

//     let schedule = not_found_if_none(schedule_o, "Schedule", path)?;

//     clear_schedule(&mut tx, path, &w_id).await?;

//     audit_log(
//         &mut *tx,
//         &authed,
//         "schedule.setenabled",
//         ActionKind::Update,
//         &w_id,
//         Some(path),
//         Some([("enabled", payload.enabled.to_string().as_ref())].into()),
//     )
//     .await?;

//     if payload.enabled {
//         tx = push_scheduled_job(&db, tx, &schedule, None).await?;
//     }
//     tx.commit().await?;

//     Ok(format!(
//         "succesfully updated schedule at path {} to status {}",
//         path, payload.enabled
//     ))
// }

async fn delete_schedule(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Extension(user_db): Extension<UserDB>,
    Path((w_id, path)): Path<(String, StripPath)>,
) -> Result<String> {
    let path = path.to_path();
    check_scopes(&authed, || format!("schedules:write:{}", path))?;
    reject_reserved_schedule_path(path)?;
    let mut tx = user_db.begin(&authed).await?;

    clear_schedule(&mut tx, path, &w_id).await?;
    let exists = sqlx::query_scalar!(
        "SELECT 1 FROM schedule WHERE path = $1 AND workspace_id = $2",
        path,
        w_id
    )
    .fetch_optional(&mut *tx)
    .await?
    .flatten();

    if exists.is_none() {
        drop(tx);
        if delete_draft_only_for_path(
            &db,
            &w_id,
            UserDraftItemKind::TriggerSchedule,
            path,
            &authed.email,
        )
        .await?
        {
            return Ok(format!("Draft-only schedule {} deleted", path));
        }
        return Err(windmill_common::error::Error::NotFound(format!(
            "Schedule {} not found",
            path
        )));
    }

    // Capture row for trashbin before deleting
    let trash_data: Option<serde_json::Value> = sqlx::query_scalar(
        "SELECT jsonb_build_object('row', to_jsonb(t)) FROM schedule t WHERE path = $1 AND workspace_id = $2",
    )
    .bind(path)
    .bind(&w_id)
    .fetch_optional(&mut *tx)
    .await?;

    let del = sqlx::query_scalar!(
        "DELETE FROM schedule WHERE path = $1 AND workspace_id = $2 RETURNING 1",
        path,
        w_id
    )
    .fetch_optional(&mut *tx)
    .await?
    .flatten();

    if del.is_none() {
        return Err(windmill_common::error::Error::NotAuthorized(format!(
            "Not authorized to delete schedule {}",
            path
        )));
    }

    if let Some(data) = trash_data {
        windmill_common::trashbin::move_to_trash(
            &mut *tx,
            &w_id,
            "schedule",
            path,
            data,
            &authed.username,
        )
        .await?;
    }

    // No diff: the row is gone, and the trashbin above already keeps its full
    // contents for a restore.
    trigger_history::record(
        &mut *tx,
        TriggerHistoryEvent {
            workspace_id: &w_id,
            trigger_kind: SCHEDULE_TRIGGER_KIND,
            path,
            operation: TriggerOperation::Delete,
            source: TriggerSource::of_request(authed.is_session_token),
            username: Some(&authed.username),
            changes: None,
        },
    )
    .await?;

    audit_log(
        &mut *tx,
        &authed,
        "schedule.delete",
        ActionKind::Delete,
        &w_id,
        Some(path),
        None,
    )
    .await?;

    tx.commit().await?;

    // Schedule gone for everyone: wipe ALL users' drafts at this path; see scripts.rs.
    delete_all_drafts_for_path(&db, &w_id, UserDraftItemKind::TriggerSchedule, path).await?;

    handle_deployment_metadata(
        &authed.email,
        &authed.username,
        &db,
        &w_id,
        DeployedObject::Schedule { path: path.to_string() },
        Some(format!("Schedule '{}' deleted", path)),
        true,
        None,
    )
    .await?;

    Ok(format!("schedule {} deleted", path))
}

async fn set_default_error_handler(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Path(w_id): Path<String>,
    Json(payload): Json<ErrorOrRecoveryHandler>,
) -> Result<()> {
    require_super_admin(&db, &authed).await?;
    let (key, value) = match payload.handler_type {
        HandlerType::Error => {
            let key = format!("default_error_handler_{}", w_id);
            if let Some(payload_path) = payload.path.as_ref() {
                let value = serde_json::json!({
                    "wsErrorHandlerMuted": payload.workspace_handler_muted,
                    "errorHandlerPath": payload_path,
                    "errorHandlerExtraArgs": payload.extra_args,
                    "failedTimes": payload.number_of_occurence,
                    "failedExact": payload.number_of_occurence_exact,
                });
                (key, Some(value))
            } else {
                (key, None)
            }
        }
        HandlerType::Recovery => {
            let key = format!("default_recovery_handler_{}", w_id);
            if let Some(payload_path) = payload.path.as_ref() {
                let value = serde_json::json!({
                    "recoveryHandlerPath": payload_path,
                    "recoveryHandlerExtraArgs": payload.extra_args,
                    "recoveredTimes": payload.number_of_occurence,
                });
                (key, Some(value))
            } else {
                (key, None)
            }
        }
        HandlerType::Success => {
            let key = format!("default_success_handler_{}", w_id);
            if let Some(payload_path) = payload.path.as_ref() {
                let value = serde_json::json!({
                    "successHandlerPath": payload_path,
                    "successHandlerExtraArgs": payload.extra_args,
                });
                (key, Some(value))
            } else {
                (key, None)
            }
        }
    };

    if let Some(value_content) = value.clone() {
        windmill_api_settings::set_global_setting_internal(&db, key.clone(), value_content).await?;
    } else {
        windmill_api_settings::delete_global_setting(&db, key.as_str()).await?;
    }
    windmill_audit::instance_config::audit_setting_write(&db, &authed, &key, value.as_ref(), None)
        .await?;

    if payload.override_existing {
        // The rewrite and its history rows go in one transaction: on separate
        // connections a concurrent edit could interleave, leaving the
        // id-ordered drawer showing the wrong latest change, and a failed
        // insert would leave the schedules rewritten with nothing recording it.
        let mut tx = db.begin().await?;
        let updated_schedules: Vec<String>;
        match payload.handler_type {
            HandlerType::Error => {
                if payload.path.is_some() {
                    updated_schedules = sqlx::query_scalar!(
                        "UPDATE schedule SET ws_error_handler_muted = $1, on_failure = $2, on_failure_extra_args = $3, on_failure_times = $4, on_failure_exact = $5 WHERE workspace_id = $6 RETURNING path",
                        payload.workspace_handler_muted,
                        payload.path,
                        payload.extra_args,
                        payload.number_of_occurence,
                        payload.number_of_occurence_exact,
                        w_id,
                    )
                    .fetch_all(&mut *tx)
                    .await?;
                } else {
                    updated_schedules = sqlx::query_scalar!(
                        "UPDATE schedule SET ws_error_handler_muted = false, on_failure = NULL, on_failure_extra_args = NULL, on_failure_times = NULL, on_failure_exact = NULL WHERE workspace_id = $1 RETURNING path",
                        w_id,
                    )
                    .fetch_all(&mut *tx)
                    .await?;
                }
            }
            HandlerType::Recovery => {
                if payload.path.is_some() {
                    updated_schedules = sqlx::query_scalar!(
                        "UPDATE schedule SET on_recovery = $1, on_recovery_extra_args = $2, on_recovery_times = $3 WHERE workspace_id = $4 RETURNING path",
                        payload.path,
                        payload.extra_args,
                        payload.number_of_occurence,
                        w_id,
                    )
                    .fetch_all(&mut *tx)
                    .await?;
                } else {
                    updated_schedules = sqlx::query_scalar!(
                        "UPDATE schedule SET on_recovery = NULL, on_recovery_extra_args = NULL, on_recovery_times = NULL WHERE workspace_id = $1 RETURNING path",
                        w_id,
                    )
                    .fetch_all(&mut *tx)
                    .await?;
                }
            }
            HandlerType::Success => {
                if payload.path.is_some() {
                    updated_schedules = sqlx::query_scalar!(
                        "UPDATE schedule SET on_success = $1, on_success_extra_args = $2 WHERE workspace_id = $3 RETURNING path",
                        payload.path,
                        payload.extra_args,
                        w_id,
                    )
                    .fetch_all(&mut *tx)
                    .await?;
                } else {
                    updated_schedules = sqlx::query_scalar!(
                        "UPDATE schedule SET on_success = NULL, on_success_extra_args = NULL WHERE workspace_id = $1 RETURNING path",
                        w_id,
                    )
                    .fetch_all(&mut *tx)
                    .await?;
                }
            }
        }
        // One row per schedule the workspace-wide override rewrote, so a handler
        // that appeared on a schedule nobody edited is traceable. Every column
        // the UPDATE above wrote, not just the handler path: the mute flag and
        // the occurrence thresholds are what someone auditing a surprise
        // notification change most needs. No `old` side and no
        // already-had-this-value filter — the UPDATE rewrites the whole
        // workspace unconditionally, so these rows record the write rather than
        // a delta.
        // Built from the same values the branch that ran actually bound: a reset
        // (`payload.path` absent) hardcodes NULL / false in SQL while the request
        // still carries the form's other fields, so reading them here would name
        // values the write never produced.
        let cleared = payload.path.is_none();
        let handler_path = payload.path.clone();
        let extra_args = (!cleared).then(|| payload.extra_args.clone()).flatten();
        let times = (!cleared).then_some(payload.number_of_occurence).flatten();
        let handler_fields = match payload.handler_type {
            HandlerType::Error => serde_json::json!({
                "on_failure": { "new": handler_path },
                "on_failure_extra_args": { "new": extra_args },
                "on_failure_times": { "new": times },
                "on_failure_exact": {
                    "new": (!cleared).then_some(payload.number_of_occurence_exact).flatten()
                },
                "ws_error_handler_muted": {
                    "new": !cleared && payload.workspace_handler_muted.unwrap_or(false)
                },
            }),
            HandlerType::Recovery => serde_json::json!({
                "on_recovery": { "new": handler_path },
                "on_recovery_extra_args": { "new": extra_args },
                "on_recovery_times": { "new": times },
            }),
            HandlerType::Success => serde_json::json!({
                "on_success": { "new": handler_path },
                "on_success_extra_args": { "new": extra_args },
            }),
        };
        trigger_history::record_bulk(
            &mut tx,
            &w_id,
            SCHEDULE_TRIGGER_KIND,
            &updated_schedules,
            TriggerOperation::Update,
            TriggerSource::of_request(authed.is_session_token),
            Some(&authed.username),
            Some(handler_fields),
        )
        .await?;

        tx.commit().await?;

        for updated_schedule_path in updated_schedules {
            // managed ducklake-maintenance rows get the handler update (their
            // failures should reach workspace handlers) but must not be
            // pushed into git-sync as deployed schedules
            if updated_schedule_path
                .starts_with(windmill_common::workspaces::DUCKLAKE_MAINTENANCE_PATH_PREFIX)
            {
                continue;
            }
            handle_deployment_metadata(
                &authed.email,
                &authed.username,
                &db,
                &w_id,
                DeployedObject::Schedule { path: updated_schedule_path },
                None,
                true,
                None,
            )
            .await?;
        }
    }
    Ok(())
}

async fn check_flow_conflict<'c>(
    tx: &mut Transaction<'c, Postgres>,
    w_id: &str,
    path: &str,
    is_flow: bool,
    script_path: &str,
) -> Result<()> {
    if path != script_path || !is_flow {
        let exists_flow = sqlx::query_scalar!(
            "SELECT EXISTS (SELECT 1 FROM flow WHERE path = $1 AND workspace_id = $2)",
            path,
            w_id
        )
        .fetch_one(&mut **tx)
        .await?
        .unwrap_or(false);
        if exists_flow {
            return Err(Error::BadRequest(format!(
                "The path is the same as a flow, it can only trigger that flow.
            However the provided path is: {script_path} and is_flow is {is_flow}"
            )));
        };
    }
    Ok(())
}

#[derive(Deserialize)]
pub struct EditSchedule {
    pub schedule: String,
    pub timezone: String,
    pub args: Option<serde_json::Value>,
    pub summary: Option<String>,
    pub description: Option<String>,
    pub on_failure: Option<String>,
    pub on_failure_times: Option<i32>,
    pub on_failure_exact: Option<bool>,
    pub on_failure_extra_args: Option<serde_json::Value>,
    pub on_recovery: Option<String>,
    pub on_recovery_times: Option<i32>,
    pub on_recovery_extra_args: Option<serde_json::Value>,
    pub on_success: Option<String>,
    pub on_success_extra_args: Option<serde_json::Value>,
    pub ws_error_handler_muted: Option<bool>,
    pub retry: Option<serde_json::Value>,
    pub no_flow_overlap: Option<bool>,
    pub tag: Option<String>,
    pub paused_until: Option<DateTime<Utc>>,
    pub cron_version: Option<String>,
    pub dynamic_skip: Option<String>,
    pub permissioned_as: Option<String>,
    pub preserve_permissioned_as: Option<bool>,
    #[serde(default)]
    pub labels: Option<Vec<String>>,
}

pub use windmill_queue::schedule::clear_schedule;

#[derive(Deserialize)]
pub struct SetEnabled {
    pub enabled: bool,
    /// Bypass the fork-conflict warning when enabling a schedule in a fork
    /// while an ancestor workspace has the same path. The frontend sets this
    /// after the user confirms the duplicate-firing dialog.
    #[serde(default)]
    pub force: bool,
}

// #[derive(Deserialize)]
// pub struct Catchup {
//     pub from: DateTime<Utc>,
//     pub to: Option<DateTime<Utc>>,
// }