mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-08-20 00:02:19 +00:00
Guillaume Bouvignies
2edba18038
* feat: Tag override for scheduled scripts * Refactor WorkerTagPicker to component * Fix test
728 lines
22 KiB
Rust
728 lines
22 KiB
Rust
/*
|
|
* 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 crate::{
|
|
db::{ApiAuthed, DB},
|
|
settings::{delete_global_setting, set_global_setting_internal},
|
|
users::maybe_refresh_folders,
|
|
utils::require_super_admin,
|
|
};
|
|
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_audit::{audit_log, ActionKind};
|
|
use windmill_common::{
|
|
db::UserDB,
|
|
error::{Error, JsonResult, Result},
|
|
schedule::Schedule,
|
|
utils::{not_found_if_none, paginate, Pagination, StripPath},
|
|
};
|
|
use windmill_queue::{self, schedule::push_scheduled_job, QueueTransaction};
|
|
|
|
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 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 ws_error_handler_muted: Option<bool>,
|
|
pub retry: Option<serde_json::Value>,
|
|
pub tag: Option<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,
|
|
}
|
|
|
|
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(());
|
|
}
|
|
|
|
async fn create_schedule(
|
|
authed: ApiAuthed,
|
|
Extension(db): Extension<DB>,
|
|
Extension(user_db): Extension<UserDB>,
|
|
Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
|
|
Path(w_id): Path<String>,
|
|
Json(ns): Json<NewSchedule>,
|
|
) -> Result<String> {
|
|
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_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(),
|
|
));
|
|
}
|
|
|
|
let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into();
|
|
|
|
cron::Schedule::from_str(&ns.schedule).map_err(|e| Error::BadRequest(e.to_string()))?;
|
|
check_path_conflict(tx.transaction_mut(), &w_id, &ns.path).await?;
|
|
check_flow_conflict(
|
|
tx.transaction_mut(),
|
|
&w_id,
|
|
&ns.path,
|
|
ns.is_flow,
|
|
&ns.script_path,
|
|
)
|
|
.await?;
|
|
|
|
let schedule = sqlx::query_as!(
|
|
Schedule,
|
|
"INSERT INTO schedule (workspace_id, path, schedule, timezone, edited_by, script_path, \
|
|
is_flow, args, enabled, email, on_failure, on_failure_times, on_failure_exact, \
|
|
on_failure_extra_args, on_recovery, on_recovery_times, on_recovery_extra_args, \
|
|
ws_error_handler_muted, retry, summary, no_flow_overlap, tag \
|
|
) VALUES ( \
|
|
$1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22 \
|
|
) RETURNING *",
|
|
w_id,
|
|
ns.path,
|
|
ns.schedule,
|
|
ns.timezone,
|
|
&authed.username,
|
|
ns.script_path,
|
|
ns.is_flow,
|
|
ns.args,
|
|
ns.enabled.unwrap_or(false),
|
|
&authed.email,
|
|
ns.on_failure,
|
|
ns.on_failure_times,
|
|
ns.on_failure_exact,
|
|
ns.on_failure_extra_args,
|
|
ns.on_recovery,
|
|
ns.on_recovery_times,
|
|
ns.on_recovery_extra_args,
|
|
ns.ws_error_handler_muted.unwrap_or(false),
|
|
ns.retry,
|
|
ns.summary,
|
|
ns.no_flow_overlap.unwrap_or(false),
|
|
ns.tag,
|
|
)
|
|
.fetch_one(&mut tx)
|
|
.await
|
|
.map_err(|e| Error::InternalErr(format!("inserting schedule in {w_id}: {e}")))?;
|
|
|
|
audit_log(
|
|
&mut tx,
|
|
&authed.username,
|
|
"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 ns.enabled.unwrap_or(true) {
|
|
tx = push_scheduled_job(&db, tx, schedule).await?
|
|
}
|
|
tx.commit().await?;
|
|
|
|
Ok(ns.path.to_string())
|
|
}
|
|
|
|
async fn edit_schedule(
|
|
authed: ApiAuthed,
|
|
Extension(db): Extension<DB>,
|
|
Extension(user_db): Extension<UserDB>,
|
|
Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
|
|
Path((w_id, path)): Path<(String, StripPath)>,
|
|
Json(es): Json<EditSchedule>,
|
|
) -> Result<String> {
|
|
let path = path.to_path();
|
|
|
|
let authed = maybe_refresh_folders(&path, &w_id, authed, &db).await;
|
|
let mut tx: QueueTransaction<'_, rsmq_async::MultiplexedRsmq> =
|
|
(rsmq, user_db.begin(&authed).await?).into();
|
|
|
|
cron::Schedule::from_str(&es.schedule).map_err(|e| Error::BadRequest(e.to_string()))?;
|
|
|
|
clear_schedule(tx.transaction_mut(), path, &w_id).await?;
|
|
let schedule = sqlx::query_as!(
|
|
Schedule,
|
|
"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, ws_error_handler_muted = $11, retry = $12, summary = $13, \
|
|
no_flow_overlap = $14, tag = $15
|
|
WHERE path = $16 AND workspace_id = $17 RETURNING *",
|
|
es.schedule,
|
|
es.timezone,
|
|
es.args,
|
|
es.on_failure,
|
|
es.on_failure_times,
|
|
es.on_failure_exact,
|
|
es.on_failure_extra_args,
|
|
es.on_recovery,
|
|
es.on_recovery_times,
|
|
es.on_recovery_extra_args,
|
|
es.ws_error_handler_muted.unwrap_or(false),
|
|
es.retry,
|
|
es.summary,
|
|
es.no_flow_overlap.unwrap_or(false),
|
|
es.tag,
|
|
path,
|
|
w_id,
|
|
)
|
|
.fetch_one(&mut tx)
|
|
.await
|
|
.map_err(|e| Error::InternalErr(format!("updating schedule in {w_id}: {e}")))?;
|
|
|
|
audit_log(
|
|
&mut tx,
|
|
&authed.username,
|
|
"schedule.edit",
|
|
ActionKind::Update,
|
|
&w_id,
|
|
Some(&path.to_string()),
|
|
Some(
|
|
[Some(("schedule", es.schedule.as_str()))]
|
|
.into_iter()
|
|
.flatten()
|
|
.collect(),
|
|
),
|
|
)
|
|
.await?;
|
|
|
|
if schedule.enabled {
|
|
tx = push_scheduled_job(&db, tx, schedule).await?;
|
|
}
|
|
tx.commit().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>,
|
|
}
|
|
|
|
async fn list_schedule(
|
|
authed: ApiAuthed,
|
|
Extension(user_db): Extension<UserDB>,
|
|
Path(w_id): Path<String>,
|
|
Query(lsq): Query<ListScheduleQuery>,
|
|
) -> JsonResult<Vec<Schedule>> {
|
|
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")
|
|
.field("*")
|
|
.order_by("edited_at", true)
|
|
.and_where("workspace_id = ?".bind(&w_id))
|
|
.offset(offset)
|
|
.limit(per_page)
|
|
.clone();
|
|
if let Some(path) = lsq.path {
|
|
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));
|
|
}
|
|
let sql = sqlb.sql().map_err(|e| Error::InternalErr(e.to_string()))?;
|
|
let rows = sqlx::query_as::<_, Schedule>(&sql)
|
|
.fetch_all(&mut *tx)
|
|
.await?;
|
|
tx.commit().await?;
|
|
Ok(Json(rows))
|
|
}
|
|
|
|
#[derive(Serialize, Deserialize, Debug)]
|
|
pub struct ScheduleWJobs {
|
|
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 args: Option<serde_json::Value>,
|
|
pub extra_perms: serde_json::Value,
|
|
pub email: String,
|
|
pub error: 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 ws_error_handler_muted: bool,
|
|
pub retry: Option<serde_json::Value>,
|
|
pub jobs: Option<Vec<serde_json::Value>>,
|
|
pub summary: Option<String>,
|
|
pub no_flow_overlap: bool,
|
|
pub tag: Option<String>,
|
|
}
|
|
|
|
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,
|
|
"SELECT schedule.*, t.jobs FROM schedule, LATERAL ( SELECT ARRAY (SELECT json_build_object('id', id, 'success', success, 'duration_ms', duration_ms) FROM completed_job WHERE
|
|
completed_job.schedule_path = schedule.path AND completed_job.workspace_id = $1 AND parent_job IS NULL AND is_skipped = False ORDER BY started_at DESC LIMIT 20) AS jobs ) t
|
|
WHERE schedule.workspace_id = $1 ORDER BY schedule.edited_at desc LIMIT $2 OFFSET $3",
|
|
w_id,
|
|
per_page as i64,
|
|
offset as i64
|
|
)
|
|
.fetch_all(&mut *tx)
|
|
.await?;
|
|
tx.commit().await?;
|
|
Ok(Json(rows))
|
|
}
|
|
|
|
// 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>,
|
|
Path((w_id, path)): Path<(String, StripPath)>,
|
|
) -> JsonResult<Schedule> {
|
|
let path = path.to_path();
|
|
let mut tx = user_db.begin(&authed).await?;
|
|
|
|
let schedule_o = windmill_queue::schedule::get_schedule_opt(&mut tx, &w_id, path).await?;
|
|
let schedule = not_found_if_none(schedule_o, "Schedule", path)?;
|
|
tx.commit().await?;
|
|
Ok(Json(schedule))
|
|
}
|
|
|
|
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 async fn preview_schedule(
|
|
Json(payload): Json<PreviewPayload>,
|
|
) -> JsonResult<Vec<DateTime<Utc>>> {
|
|
let schedule = cron::Schedule::from_str(&payload.schedule)
|
|
.map_err(|e| Error::BadRequest(e.to_string()))?;
|
|
|
|
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)
|
|
.take(5)
|
|
// Convert back to UTC for a standardised API response. The client will convert to the local timezone.
|
|
.map(|x| x.with_timezone(&Utc))
|
|
.collect();
|
|
|
|
Ok(Json(upcoming))
|
|
}
|
|
|
|
pub async fn set_enabled(
|
|
authed: ApiAuthed,
|
|
Extension(db): Extension<DB>,
|
|
Extension(user_db): Extension<UserDB>,
|
|
Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
|
|
Path((w_id, path)): Path<(String, StripPath)>,
|
|
Json(payload): Json<SetEnabled>,
|
|
) -> Result<String> {
|
|
let mut tx: QueueTransaction<'_, rsmq_async::MultiplexedRsmq> =
|
|
(rsmq, 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(tx.transaction_mut(), path, &w_id).await?;
|
|
|
|
audit_log(
|
|
&mut tx,
|
|
&authed.username,
|
|
"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).await?;
|
|
}
|
|
tx.commit().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>,
|
|
// Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
|
|
// Path((w_id, path)): Path<(String, StripPath)>,
|
|
// Json(payload): Json<SetEnabled>,
|
|
// ) -> Result<String> {
|
|
// let mut tx: QueueTransaction<'_, rsmq_async::MultiplexedRsmq> =
|
|
// (rsmq, 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(tx.transaction_mut(), path, &w_id).await?;
|
|
|
|
// audit_log(
|
|
// &mut tx,
|
|
// &authed.username,
|
|
// "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).await?;
|
|
// }
|
|
// tx.commit().await?;
|
|
|
|
// Ok(format!(
|
|
// "succesfully updated schedule at path {} to status {}",
|
|
// path, payload.enabled
|
|
// ))
|
|
// }
|
|
|
|
async fn delete_schedule(
|
|
authed: ApiAuthed,
|
|
Extension(user_db): Extension<UserDB>,
|
|
Path((w_id, path)): Path<(String, StripPath)>,
|
|
) -> Result<String> {
|
|
let mut tx = user_db.begin(&authed).await?;
|
|
let path = path.to_path();
|
|
|
|
clear_schedule(&mut tx, path, &w_id).await?;
|
|
|
|
sqlx::query!(
|
|
"DELETE FROM schedule WHERE path = $1 AND workspace_id = $2",
|
|
path,
|
|
w_id
|
|
)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
|
|
audit_log(
|
|
&mut *tx,
|
|
&authed.username,
|
|
"schedule.delete",
|
|
ActionKind::Delete,
|
|
&w_id,
|
|
Some(path),
|
|
None,
|
|
)
|
|
.await?;
|
|
|
|
tx.commit().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.email).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)
|
|
}
|
|
}
|
|
};
|
|
|
|
if let Some(value_content) = value {
|
|
set_global_setting_internal(&db, key, value_content).await?;
|
|
} else {
|
|
delete_global_setting(&db, key.as_str()).await?;
|
|
}
|
|
|
|
if payload.override_existing {
|
|
match payload.handler_type {
|
|
HandlerType::Error => {
|
|
if payload.path.is_some() {
|
|
sqlx::query!(
|
|
"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",
|
|
payload.workspace_handler_muted,
|
|
payload.path,
|
|
payload.extra_args,
|
|
payload.number_of_occurence,
|
|
payload.number_of_occurence_exact,
|
|
w_id,
|
|
)
|
|
.execute(&db)
|
|
.await?;
|
|
} else {
|
|
sqlx::query!(
|
|
"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",
|
|
w_id,
|
|
)
|
|
.execute(&db)
|
|
.await?;
|
|
}
|
|
}
|
|
HandlerType::Recovery => {
|
|
if payload.path.is_some() {
|
|
sqlx::query!(
|
|
"UPDATE schedule SET on_recovery = $1, on_recovery_extra_args = $2, on_recovery_times = $3 WHERE workspace_id = $4",
|
|
payload.path,
|
|
payload.extra_args,
|
|
payload.number_of_occurence,
|
|
w_id,
|
|
)
|
|
.execute(&db)
|
|
.await?;
|
|
} else {
|
|
sqlx::query!(
|
|
"UPDATE schedule SET on_recovery = NULL, on_recovery_extra_args = NULL, on_recovery_times = NULL WHERE workspace_id = $1",
|
|
w_id,
|
|
)
|
|
.execute(&db)
|
|
.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 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 ws_error_handler_muted: Option<bool>,
|
|
pub retry: Option<serde_json::Value>,
|
|
pub no_flow_overlap: Option<bool>,
|
|
pub tag: Option<String>,
|
|
}
|
|
|
|
pub async fn clear_schedule<'c>(
|
|
db: &mut Transaction<'c, Postgres>,
|
|
path: &str,
|
|
w_id: &str,
|
|
) -> Result<()> {
|
|
sqlx::query!(
|
|
"DELETE FROM queue WHERE schedule_path = $1 AND running = false AND workspace_id = $2",
|
|
path,
|
|
w_id
|
|
)
|
|
.execute(&mut **db)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
pub struct SetEnabled {
|
|
pub enabled: bool,
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
pub struct Catchup {
|
|
pub from: DateTime<Utc>,
|
|
pub to: Option<DateTime<Utc>>,
|
|
}
|