/* * 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::Multipart; use windmill_api_auth::{ auth::{list_tokens_internal, AuthCache, TruncatedTokenWithEmail}, build_scope_path_predicate, check_scopes, maybe_refresh_folders, require_owner_of_path, ApiAuthed, }; use windmill_common::{ user_drafts::{overlay_or_draft_only, DraftUserRef, UserDraftItemKind, WithDraftOverlay}, utils::{BulkDeleteRequest, WithStarredInfoQuery, HTTP_CLIENT}, webhook::{WebhookMessage, WebhookShared}, workspaces::{check_deploy_rules, RuleCheckResult}, DB, }; use windmill_queue::schedule::clear_schedule; use axum::{ extract::{Extension, Path, Query}, response::IntoResponse, routing::{delete, get, post}, Json, Router, }; use futures::future::try_join_all; use http::header; use hyper::StatusCode; use itertools::Itertools; use quick_cache::sync::Cache; use serde::{Deserialize, Serialize}; use serde_json::json; use sql_builder::prelude::*; use sqlx::{FromRow, Postgres, Transaction}; use std::{collections::HashMap, sync::Arc}; use windmill_audit::audit_oss::{audit_log, AuditAuthorable}; use windmill_audit::ActionKind; use windmill_dep_map::process_relative_imports; use windmill_dep_map::scoped_dependency_map::ScopedDependencyMap; use windmill_common::{ assets::{ clear_script_triggers, clear_static_asset_usage, clear_static_asset_usage_by_script_hash, derive_pipeline_asset_trigger_refs, insert_script_trigger, parse_duration_secs, parse_pipeline_annotations, replace_static_asset_usage, trigger_spec_to_row, AssetUsageKind, ScriptTriggerKind, TriggerSpec, }, error::{self, to_anyhow}, min_version::{MIN_VERSION_SUPPORTS_DEBOUNCING, MIN_VERSION_SUPPORTS_DEBOUNCING_V2}, runnable_settings::{ min_version_supports_runnable_settings_v0, RunnableSettings, RunnableSettingsTrait, }, scripts::{hash_script, ScriptRunnableSettingsHandle, ScriptRunnableSettingsInline}, utils::{paginate_without_limits, WarnAfterExt}, worker::CLOUD_HOSTED, }; use windmill_object_store::upload_artifact_to_store; use windmill_common::{ db::UserDB, error::{Error, JsonResult, Result}, jobs::JobPayload, schedule::Schedule, schema::should_validate_schema, scripts::{ to_i64, HubScript, ListScriptQuery, ListableScript, NewScript, Schema, Script, ScriptHash, ScriptHistory, ScriptHistoryUpdate, ScriptKind, ScriptLang, ScriptModule, ScriptWithStarred, }, triggers::MovedNativeTrigger, users::username_to_permissioned_as, utils::{not_found_if_none, query_elems_from_hub, require_admin, Pagination, StripPath}, worker::to_raw_value, HUB_BASE_URL, }; use windmill_git_sync::{handle_deployment_metadata, DeployedObject}; use windmill_parser_ts::remove_pinned_imports; use windmill_queue::{ schedule::push_scheduled_job, PushIsolationLevel, WMDEBUG_FORCE_NO_LEGACY_DEBOUNCING_COMPAT, }; const MAX_HASH_HISTORY_LENGTH_STORED: usize = 20; pub fn global_service() -> Router { Router::new() .route("/hub/top", get(get_top_hub_scripts)) .route("/hub/get/{*path}", get(get_hub_script_by_path)) .route("/hub/get_full/{*path}", get(get_full_hub_script_by_path)) .route("/hub/pick/{*path}", get(pick_hub_script_by_path)) } pub fn global_unauthed_service() -> Router { Router::new() .route( "/tokened_raw/{workspace}/{token}/{*path}", get(get_tokened_raw_script_by_path), ) .route("/empty_ts/{*path}", get(get_empty_ts_script_by_path)) } pub fn workspaced_service() -> Router { Router::new() .route("/list", get(list_scripts)) .route("/list_search", get(list_search_scripts)) .route("/create", post(create_script)) .route("/create_snapshot", post(create_snapshot_script)) .route("/archive/p/{*path}", post(archive_script_by_path)) .route("/get/p/{*path}", get(get_script_by_path)) .route("/list_tokens/{*path}", get(list_tokens)) .route("/raw/p/{*path}", get(raw_script_by_path)) .route("/raw_unpinned/p/{*path}", get(raw_script_by_path_unpinned)) .route("/exists/p/{*path}", get(exists_script_by_path)) .route("/archive/h/{hash}", post(archive_script_by_hash)) .route("/delete/h/{hash}", post(delete_script_by_hash)) .route("/delete/p/{*path}", post(delete_script_by_path)) .route("/delete_bulk", delete(delete_scripts_bulk)) .route("/get/h/{hash}", get(get_script_by_hash)) .route("/raw/h/{hash}", get(raw_script_by_hash)) .route("/deployment_status/h/{hash}", get(get_deployment_status)) .route("/list_paths", get(list_paths)) .route( "/toggle_workspace_error_handler/p/{*path}", post(toggle_workspace_error_handler), ) .route("/history/p/{*path}", get(get_script_history)) .route("/get_latest_version/{*path}", get(get_latest_version)) .route( "/list_paths_from_workspace_runnable/{*path}", get(list_paths_from_workspace_runnable), ) .route( "/history_update/h/{hash}/p/{*path}", post(update_script_history), ) .route("/list_dedicated_with_deps", get(list_dedicated_with_deps)) // Temporary raw script storage for CLI lock generation .route("/raw_temp/store", post(store_raw_script_temp)) .route("/raw_temp/diff", post(diff_raw_scripts_with_deployed)) // CI test results .route("/ci_test_results/{kind}/{*path}", get(get_ci_test_results)) .route("/ci_test_results_batch", post(get_ci_test_results_batch)) // Save-time schema-contract check (pipelines gap #2b) .route("/check_schema_contracts", post(check_schema_contracts)) } #[derive(Serialize, FromRow)] pub struct SearchScript { path: String, content: String, } async fn list_search_scripts( authed: ApiAuthed, Path(w_id): Path, Extension(user_db): Extension, ) -> JsonResult> { let mut tx = user_db.begin(&authed).await?; let n = 10000; let allowed = build_scope_path_predicate(&authed, "scripts", "read"); let rows = sqlx::query_as!( SearchScript, "SELECT path, content from script WHERE workspace_id = $1 AND archived = false LIMIT $2", &w_id, n ) .fetch_all(&mut *tx) .await? .into_iter() .filter(|r| allowed(&r.path)) .collect::>(); tx.commit().await?; Ok(Json(rows)) } async fn list_scripts( authed: ApiAuthed, Extension(user_db): Extension, Extension(db): Extension, Path(w_id): Path, Query(pagination): Query, Query(lq): Query, ) -> JsonResult> { let (per_page, offset) = paginate_without_limits(pagination); let mut sqlb = SqlBuilder::select_from("script as o") .fields(&[ "hash", "o.path", "summary", "o.created_at as created_at", "archived", "extra_perms", if !lq.without_description.unwrap_or(false) { "description" } else { "NULL as description" }, "CASE WHEN lock_error_logs IS NOT NULL THEN true ELSE false END as has_deploy_errors", "language", "favorite.path IS NOT NULL as starred", "tag", "ws_error_handler_muted", "auto_kind", "codebase IS NOT NULL as use_codebase", "kind", "o.labels", "draft.email IS NOT NULL as is_draft", // Canonical reference for the draft-feature comments; flows/apps point here. // Per-path draft owners as a JSON array (`Json>`); NULL -> None, // never an empty array. LEFT JOIN `usr` keeps orphaned drafts (user left workspace) // visible with `username = None`. A superadmin authoring in a workspace they are not // a member of has no `usr` row, so fall back to their instance-derived username // (`password.username`), or their email when derivation is disabled — this keeps the // raw email out of the payload whenever a derived username exists. The genuine // NULL-email legacy row stays None (no `usr`/`password` match, `d.email` is NULL). "(SELECT json_agg(json_build_object('username', COALESCE(u.username, p.username, CASE WHEN p.email IS NOT NULL THEN d.email END)) ORDER BY COALESCE(u.username, p.username, CASE WHEN p.email IS NOT NULL THEN d.email END) NULLS LAST) \ FROM draft d \ LEFT JOIN usr u ON u.workspace_id = d.workspace_id AND u.email = d.email \ LEFT JOIN password p ON p.email = d.email AND p.super_admin = true \ WHERE d.workspace_id = o.workspace_id AND d.path = o.path AND d.typ = 'script') as draft_users", "folder_labels(o.workspace_id, o.path) as inherited_labels" ]) .left() .join("favorite") .on( "favorite.favorite_kind = 'script' AND favorite.workspace_id = o.workspace_id AND favorite.path = o.path AND favorite.usr = ?" .bind(&authed.username), ) .left() .join("draft") .on( "draft.path = o.path AND draft.workspace_id = o.workspace_id AND draft.typ = 'script' AND draft.email = ?" .bind(&authed.email), ) .order_desc("favorite.path IS NOT NULL") .order_by("created_at", lq.order_desc.unwrap_or(true)) .and_where("o.workspace_id = ?".bind(&w_id)) .offset(offset) .limit(per_page) .clone(); let lowercased_kinds: Option> = lq .kinds .map(|x| x.split(",").map(&str::to_lowercase).collect()); if (!lq.include_without_main.unwrap_or(false) && lowercased_kinds .as_ref() .map(|x| !x.contains(&"preprocessor".to_string())) .unwrap_or(true)) || authed.is_operator { // only include scripts that have a runnable entrypoint. Use a // deny-list: anything that isn't a 'lib' (library script without // main) is callable, including future `auto_kind` values. sqlb.and_where("(o.auto_kind IS NULL OR o.auto_kind <> 'lib')"); } if lq.show_archived.unwrap_or(false) { sqlb.and_where_eq( "o.ctid", "(SELECT ctid FROM script WHERE path = o.path AND workspace_id = ? ORDER BY created_at DESC LIMIT 1)" .bind(&w_id), ); sqlb.and_where_eq("archived", true); } else { sqlb.and_where_eq("archived", false); } if let Some(ps) = &lq.path_start { sqlb.and_where_like_left("o.path", ps); } if let Some(p) = &lq.path_exact { sqlb.and_where_eq("o.path", "?".bind(p)); } if let Some(cb) = &lq.created_by { sqlb.and_where_eq("created_by", "?".bind(cb)); } if let Some(ph) = &lq.first_parent_hash { sqlb.and_where_eq("parent_hashes[1]", &ph.0); } if let Some(ph) = &lq.last_parent_hash { sqlb.and_where_eq("parent_hashes[array_upper(parent_hashes, 1)]", &ph.0); } if let Some(ph) = &lq.parent_hash { sqlb.and_where_eq("any(parent_hashes)", &ph.0); } if let Some(it) = &lq.is_template { sqlb.and_where_eq("is_template", it); } if let Some(dw) = &lq.dedicated_worker { sqlb.and_where_eq("dedicated_worker", dw); } if let Some(label) = &lq.label { for l in label.split(',') { sqlb.and_where( "(o.labels @> ARRAY[?] OR folder_labels(o.workspace_id, o.path) @> ARRAY[?])" .bind(&l.trim()) .bind(&l.trim()), ); } } if authed.is_operator { sqlb.and_where_eq("kind", quote("script")); } else if let Some(lowercased_kinds) = lowercased_kinds { let safe_kinds = lowercased_kinds .into_iter() .map(sql_builder::quote) .collect_vec(); if safe_kinds.len() > 0 { sqlb.and_where_in("kind", safe_kinds.as_slice()); } } if lq.starred_only.unwrap_or(false) { sqlb.and_where_is_not_null("favorite.path"); } if lq.with_deployment_msg.unwrap_or(false) { sqlb.join("deployment_metadata dm") .left() .on("dm.script_hash = o.hash") .fields(&["dm.deployment_msg"]); } if let Some(languages) = &lq.languages { sqlb.and_where_in( "language", &languages .iter() .map(|language| quote(language.as_str())) .collect_vec(), ); } let sql = sqlb.sql().map_err(|e| Error::internal_err(e.to_string()))?; let mut tx = user_db.begin(&authed).await?; let allowed = build_scope_path_predicate(&authed, "scripts", "read"); let mut rows = sqlx::query_as::<_, ListableScript>(&sql) .fetch_all(&mut *tx) .await? .into_iter() .filter(|r| allowed(&r.path)) .collect::>(); tx.commit().await?; // Canonical reference for draft-only synthesis; the other kinds point here. // Append the authed user's drafts at paths with no deployed script. Gated on // `include_draft_only` so picker callers stay deployed-only (home page opts in); // skipped past page 0 or under any narrowing filter to keep pagination clean. if lq.include_draft_only.unwrap_or(false) && !authed.is_operator && offset == 0 && lq.path_start.is_none() && lq.path_exact.is_none() && lq.created_by.is_none() && lq.first_parent_hash.is_none() && lq.last_parent_hash.is_none() && lq.parent_hash.is_none() && lq.is_template.is_none() && lq.dedicated_worker.is_none() && lq.label.is_none() && lq.languages.is_none() && !lq.starred_only.unwrap_or(false) && !lq.show_archived.unwrap_or(false) { // `(email = $2 OR email IS NULL)` surfaces the user's own draft-only rows plus // legacy NULL-email workspace rows; `DISTINCT ON (path)` ordered `email IS NULL` // last collapses a path holding both to the owned row. let draft_only_rows = sqlx::query!( r#"SELECT DISTINCT ON (path) path, value as "value!: sqlx::types::Json>", created_at FROM draft WHERE workspace_id = $1 AND typ = 'script' AND (email = $2 OR email IS NULL) AND NOT EXISTS ( SELECT 1 FROM script s WHERE s.workspace_id = draft.workspace_id AND s.path = draft.path ) ORDER BY path, (email IS NULL)"#, &w_id, &authed.email, ) .fetch_all(&db) .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); let language: ScriptLang = v .get("language") .and_then(|x| serde_json::from_value(x.clone()).ok()) .unwrap_or_default(); let kind: ScriptKind = v .get("kind") .and_then(|x| serde_json::from_value(x.clone()).ok()) .unwrap_or(ScriptKind::Script); // Scripts bind the Path widget to `script.path`, so the typed path // round-trips through the draft JSON's own `path` (no `draft_path` field). let draft_path = v .get("path") .and_then(|s| s.as_str()) .filter(|s| !s.is_empty() && *s != row.path.as_str()) .map(|s| s.to_string()); // A draft-only pipeline node (`// pipeline`) has no deployed row to carry // auto_kind, so compute it from the draft content — mirroring the create // path — so the home page folds it into its pipeline like a deployed member. // Otherwise fall back to the `auto_kind` the frontend saved into the draft // (e.g. `lib` for scripts without a `main`); the content-derived pipeline // annotation keeps priority since it mirrors the deploy-time computation. let auto_kind = v .get("content") .and_then(|s| s.as_str()) .filter(|c| parse_pipeline_annotations(c).in_pipeline) .map(|_| "pipeline".to_string()) .or_else(|| { v.get("auto_kind") .and_then(|s| s.as_str()) .map(|s| s.to_string()) }); rows.push(ListableScript { hash: ScriptHash(0), path: row.path, summary: v .get("summary") .and_then(|s| s.as_str()) .unwrap_or("") .to_string(), created_at: row.created_at, archived: false, extra_perms: serde_json::Value::Object(serde_json::Map::new()), language, starred: false, tag: v.get("tag").and_then(|s| s.as_str()).map(|s| s.to_string()), description: v .get("description") .and_then(|s| s.as_str()) .map(|s| s.to_string()), draft_only: Some(true), has_deploy_errors: false, ws_error_handler_muted: None, auto_kind, use_codebase: false, deployment_msg: None, kind, labels: None, // Synthesized rows have no deployed row to inherit folder labels from. inherited_labels: None, is_draft: true, draft_path, // Synthesized rows are the authed user's own draft (single-user case). draft_users: Some(sqlx::types::Json(vec![DraftUserRef { username: Some(authed.username.clone()), }])), }); } } Ok(Json(rows)) } #[derive(Deserialize)] struct TopHubScriptsQuery { limit: Option, app: Option, kind: Option, } async fn get_top_hub_scripts( Query(query): Query, Extension(db): Extension, ) -> impl IntoResponse { let mut query_params = vec![]; if let Some(query_limit) = query.limit { query_params.push(("limit", query_limit.to_string().clone())); } if let Some(query_app) = query.app { query_params.push(("app", query_app.to_string().clone())); } if let Some(query_kind) = query.kind { query_params.push(("kind", query_kind.to_string().clone())); } let (status_code, headers, response) = query_elems_from_hub( &HTTP_CLIENT, &format!("{}/scripts/top", **HUB_BASE_URL.load()), Some(query_params), &db, ) .await?; Ok::<_, Error>((status_code, headers, response)) } /// Re-point the webhooks of the native triggers a rename carried onto the new path. /// /// Runs after the deploy transaction commits — repointing a webhook is not undoable — and off the /// request, because it waits on a third-party service that may be slow or gone, and a deploy that /// already committed must not look like it failed. The rename itself marked these rows /// `REREGISTRATION_PENDING`, so nothing is lost silently if this never finishes. fn reregister_moved_native_triggers( db: &DB, authed: &ApiAuthed, w_id: &str, moved: Vec, ) { if moved.is_empty() { return; } #[cfg(feature = "native_trigger")] { let (db, authed, w_id) = (db.clone(), authed.clone(), w_id.to_string()); tokio::spawn(async move { windmill_native_triggers::rename::reregister_triggers_after_rename( &db, &authed, &w_id, &moved, ) .await; }); } #[cfg(not(feature = "native_trigger"))] let _ = (db, authed, w_id, moved); } async fn create_snapshot_script( authed: ApiAuthed, Extension(user_db): Extension, Extension(webhook): Extension, Extension(db): Extension, Path(w_id): Path, Query(query): Query, mut multipart: Multipart, ) -> Result<(StatusCode, String)> { // TODO: Check for debouncing here as well. let mut script_hash = None; let mut tx = None; let mut uploaded = false; let mut handle_deployment_metadata = None; let mut moved_native_triggers = Vec::new(); while let Some(field) = multipart.next_field().await.unwrap() { let name = field.name().unwrap().to_string(); let data = field.bytes().await.unwrap(); if name == "script" { let ns: NewScript = Some(serde_json::from_slice(&data).map_err(to_anyhow)?).unwrap(); let is_tar = ns.codebase.as_ref().is_some_and(|x| x.ends_with(".tar")); let use_esm = ns.codebase.as_ref().is_some_and(|x| x.contains(".esm")); let (new_hash, ntx, hdm, moved) = create_script_internal( ns, w_id.clone(), authed.clone(), db.clone(), user_db.clone(), webhook.clone(), query.skip_if_noop, ) .await?; let mut nh = new_hash.to_string(); if use_esm { nh = format!("{nh}.esm"); } if is_tar { nh = format!("{nh}.tar"); } script_hash = Some(nh); tx = Some(ntx); handle_deployment_metadata = hdm; moved_native_triggers = moved; } if name == "file" { let hash = script_hash.as_ref().ok_or_else(|| { Error::BadRequest( "script need to be passed first in the multipart upload".to_string(), ) })?; uploaded = true; let path = windmill_object_store::bundle(&w_id, &hash); upload_artifact_to_store( &path, data, &windmill_common::worker::ROOT_STANDALONE_BUNDLE_DIR, ) .await?; } // println!("Length of `{}` is {} bytes", name, data.len()); } if !uploaded { return Err(Error::BadRequest("No file uploaded".to_string())); } if script_hash.is_none() { return Err(Error::BadRequest( "No script found in the uploaded file".to_string(), )); } tx.unwrap().commit().await?; reregister_moved_native_triggers(&db, &authed, &w_id, moved_native_triggers); if let Some(hdm) = handle_deployment_metadata { hdm.handle(&db).await?; } return Ok((StatusCode::CREATED, format!("{}", script_hash.unwrap()))); } async fn list_paths_from_workspace_runnable( authed: ApiAuthed, Extension(user_db): Extension, Path((w_id, path)): Path<(String, StripPath)>, ) -> JsonResult> { let mut tx = user_db.begin(&authed).await?; let runnables = sqlx::query_scalar!( r#"SELECT importer_path FROM dependency_map WHERE workspace_id = $1 AND imported_path = $2"#, w_id, path.to_path(), ) .fetch_all(&mut *tx) .await?; tx.commit().await?; Ok(Json(runnables)) } #[derive(Deserialize, Default, Clone, Copy)] struct CreateScriptQuery { /// When set by the caller (currently only the CLI), the backend /// short-circuits deploys whose content, lockfile, and metadata are /// identical to the parent: no new version is inserted and the git /// sync / promotion callbacks are suppressed. #[serde(default)] skip_if_noop: bool, } async fn create_script( authed: ApiAuthed, Extension(user_db): Extension, Extension(webhook): Extension, Extension(db): Extension, Path(w_id): Path, Query(query): Query, Json(ns): Json, ) -> Result<(StatusCode, String)> { if let RuleCheckResult::Blocked(msg) = check_deploy_rules( &w_id, AuditAuthorable::username(&authed), &authed.groups, authed.is_admin, &db, ) .await? { return Err(Error::PermissionDenied(msg)); } let script_path = ns.path.clone(); let email = authed.email.clone(); let username = authed.username.clone(); let authed_for_triggers = authed.clone(); let (hash, tx, hdm, moved_native_triggers) = create_script_internal( ns, w_id.clone(), authed, db.clone(), user_db, webhook, query.skip_if_noop, ) .await?; tx.commit().await?; reregister_moved_native_triggers(&db, &authed_for_triggers, &w_id, moved_native_triggers); if let Some(hdm) = hdm { // Only a script that needed no lock generation is deployed and runnable by // now; one that did gets its CI tests from the dependency job instead, so // they don't run against a version whose lock does not exist yet — and // don't run twice. let ready_to_test = matches!(hdm, PostCommitDeploy::Full { .. }); hdm.handle(&db).await?; let db2 = db.clone(); if ready_to_test { tokio::spawn(async move { if let Err(e) = windmill_dep_map::ci_tests::trigger_ci_tests_for_item( &db2, &w_id, &script_path, "script", &email, &username, ) .await { tracing::error!(%e, "error triggering CI tests after script deploy"); } }); } } Ok((StatusCode::CREATED, format!("{}", hash))) } /// What a script deploy still has to do once its transaction has committed. enum PostCommitDeploy { /// Everything, for a deploy with no dependency job to hand it to. Full { email: String, created_by: String, w_id: String, obj: DeployedObject, deployment_message: Option, renamed_from: Option, }, /// Only the path a rename left behind; the dependency job handles the rest. /// See `tally_rename_vacated_path`. VacatedPath { w_id: String, obj: DeployedObject }, } impl PostCommitDeploy { async fn handle(self, db: &DB) -> Result<()> { match self { PostCommitDeploy::Full { email, created_by, w_id, obj, deployment_message, renamed_from, } => { handle_deployment_metadata( &email, &created_by, &db, &w_id, obj, deployment_message, false, renamed_from.as_deref(), ) .await } PostCommitDeploy::VacatedPath { w_id, obj } => { windmill_git_sync::tally_rename_vacated_path(db, &w_id, obj).await } } } } /// Returns true when `ns` is effectively identical to its `parent`: same /// path, content, lockfile, and all persisted metadata. Used by the deploy /// path to short-circuit no-op re-deploys and suppress the follow-on git /// sync / promotion callbacks. /// /// ## How drift is prevented /// /// `NewScript` grows over time. If a new field is added but not wired in /// here, a real change would be silently classified as a no-op and /// dropped. The exhaustive destructure below turns that into a *compile /// error*: every field of `NewScript` must either feed into a comparison /// or be explicitly opted out via `field: _` with a one-line rationale. /// /// When adding a field to `NewScript`: /// 1. Add it to the destructure here. The compiler will refuse to build /// until you do. /// 2. Either add a `!=` comparison against `parent` below, or bind it to /// `_` and write why it shouldn't affect the no-op decision. /// 3. Don't forget `impl Hash for NewScript` in windmill-types. async fn is_noop_deploy_against_parent( ns: &NewScript, parent: &Script, resolved_on_behalf_of: Option<&str>, db: &DB, ) -> Result { if parent.archived || parent.deleted { return Ok(false); } // Compile-time drift guard — see the docstring above. The `ns.foo` bindings // below shadow each field as `&T`; every binding must be referenced in a // comparison (the compiler will warn on any that isn't), and every ignored // field (`_`) carries its rationale inline. let NewScript { path, // version-identity field — always differs between child and parent by design parent_hash: _, summary, description, content, schema, is_template, lock, language, kind, tag, envs, concurrency_settings, debouncing_settings, cache_ttl, cache_ignore_s3_path, dedicated_worker, ws_error_handler_muted, priority, timeout, delete_after_use, delete_after_secs, restart_unless_cancelled, // per-deploy audit field — does not change what the script *is* deployment_message: _, visible_to_runner_only, // derived from `content` at deploy time; content equality implies equality here auto_kind: _, codebase, has_preprocessor, // both halves are folded into `resolved_on_behalf_of` before the comparison below, // which is the identity that would actually be stored on_behalf_of_email: _, on_behalf_of: _, // caller-intent flag (permission preservation), not script state preserve_on_behalf_of: _, assets, modules, // caller-intent flag (auto-resolve parent), not script state auto_parent: _, labels, // caller-intent flag (preserve user drafts on CLI/git-sync deploys); // transient, never persisted, does not change what the script *is* skip_draft_deletion: _, } = ns; if path != &parent.path { return Ok(false); } if content != &parent.content { return Ok(false); } // A dbt lock is DERIVED, never supplied: the request carries none — the CLI // sends `lock: undefined` and this route discards any anyway — while a // deployed parent carries what its dependency job wrote. Comparing them makes // every unchanged push a new version, a fresh dependency job and the sync // activity `skip_if_noop` exists to prevent. The parent must actually hold // one, or a deploy whose dependency job failed could never be retried by // pushing the same project again. if matches!(language, ScriptLang::Dbt) { if normalize_optional_text(parent.lock.as_deref()).is_empty() { return Ok(false); } } else if normalize_optional_text(lock.as_deref()) != normalize_optional_text(parent.lock.as_deref()) { return Ok(false); } if summary != &parent.summary { return Ok(false); } if description != &parent.description { return Ok(false); } if language != &parent.language { return Ok(false); } if std::mem::discriminant(kind.as_ref().unwrap_or(&ScriptKind::Script)) != std::mem::discriminant(&parent.kind) { return Ok(false); } if tag != &parent.tag { return Ok(false); } if envs.as_deref().unwrap_or_default() != parent.envs.as_deref().unwrap_or_default() { return Ok(false); } if labels != &parent.labels { return Ok(false); } if cache_ttl != &parent.cache_ttl || cache_ignore_s3_path != &parent.cache_ignore_s3_path || dedicated_worker != &parent.dedicated_worker || ws_error_handler_muted != &parent.ws_error_handler_muted || priority != &parent.priority || timeout != &parent.timeout || delete_after_use != &parent.delete_after_use || delete_after_secs != &parent.delete_after_secs || restart_unless_cancelled != &parent.restart_unless_cancelled || visible_to_runner_only != &parent.visible_to_runner_only || has_preprocessor != &parent.has_preprocessor || is_template.unwrap_or(false) != parent.is_template.unwrap_or(false) { return Ok(false); } if normalize_optional_text(codebase.as_deref()) != normalize_optional_text(parent.codebase.as_deref()) { return Ok(false); } if resolved_on_behalf_of != parent.on_behalf_of.as_deref() { return Ok(false); } // Both of a dbt script's derived fields are compared as they WOULD BE STORED, // not as they arrived: the schema comes from the descriptor and the clients // cannot derive it (`windmill-parser-wasm` has no dbt arm), so they send the // previous version's or none at all. Comparing what they sent makes every // unchanged push differ. let dbt_schema = matches!(language, ScriptLang::Dbt) .then(|| windmill_parser_yaml::dbt_arg_schema(content).ok()) .flatten() .and_then(|v| serde_json::value::to_raw_value(&v).ok()) .map(|v| Schema(sqlx::types::Json(v))); let effective_schema = if matches!(language, ScriptLang::Dbt) { dbt_schema.as_ref() } else { schema.as_ref() }; if !schema_opt_eq(effective_schema, parent.schema.as_ref()) { return Ok(false); } if !json_serialize_eq(assets, &parent.assets) { return Ok(false); } if !modules_eq(modules.as_ref(), parent.modules.as_ref()) { return Ok(false); } let (parent_debouncing, parent_concurrency) = windmill_common::runnable_settings::prefetch_cached_from_handle( parent.runnable_settings.runnable_settings_handle, db, ) .await?; if concurrency_settings != &parent_concurrency { return Ok(false); } if debouncing_settings != &parent_debouncing { return Ok(false); } Ok(true) } /// Treats `None` and `Some("")` as equivalent — matches how the insert path /// normalizes optional text fields like `lock` and `codebase`. fn normalize_optional_text(s: Option<&str>) -> &str { s.unwrap_or("") } fn schema_opt_eq(a: Option<&Schema>, b: Option<&Schema>) -> bool { match (a, b) { (None, None) => true, (Some(x), Some(y)) => { match ( serde_json::from_str::(x.0.get()), serde_json::from_str::(y.0.get()), ) { (Ok(xv), Ok(yv)) => xv == yv, _ => false, } } _ => false, } } fn json_serialize_eq(a: &T, b: &T) -> bool { match (serde_json::to_value(a), serde_json::to_value(b)) { (Ok(av), Ok(bv)) => av == bv, _ => false, } } fn modules_eq( a: Option<&HashMap>, b: Option<&HashMap>, ) -> bool { match (a, b) { (None, None) => true, (Some(x), Some(y)) => { if x.len() != y.len() { return false; } x.iter().all(|(k, v)| match y.get(k) { Some(ov) => { v.content == ov.content && v.language == ov.language && v.lock == ov.lock } None => false, }) } _ => false, } } async fn create_script_internal<'c>( mut ns: NewScript, w_id: String, authed: ApiAuthed, db: sqlx::Pool, user_db: UserDB, webhook: WebhookShared, skip_if_noop: bool, ) -> Result<( ScriptHash, Transaction<'c, Postgres>, Option, Vec, )> { if authed.is_operator { return Err(Error::NotAuthorized( "Operators cannot create scripts for security reasons".to_string(), )); } check_scopes(&authed, || format!("scripts:write:{}", ns.path))?; // Normalize positive-only settings so a `<= 0` value (e.g. a CLI-pushed `0`) persists as // disabled rather than as a zero-slot concurrency cap or a 0-second timeout. Deserialization // already normalizes the concurrency fields; re-applying here also covers `timeout` and any // NewScript built in-process rather than from a request body. ns.timeout = windmill_common::runnable_settings::none_if_non_positive(ns.timeout); ns.concurrency_settings = ns.concurrency_settings.normalized(); guard_script_from_debounce_data(&ns).await?; let codebase = ns.codebase.as_ref(); #[cfg(not(feature = "enterprise"))] if ns.ws_error_handler_muted.is_some_and(|val| val) { return Err(Error::BadRequest( "Muting the error handler for certain script is only available in enterprise version" .to_string(), )); } if *CLOUD_HOSTED { let nb_scripts = sqlx::query_scalar!("SELECT COUNT(*) FROM script WHERE workspace_id = $1", &w_id) .fetch_one(&db) .await?; if nb_scripts.unwrap_or(0) >= 5000 { return Err(Error::BadRequest( "You have reached the maximum number of scripts (5000) on cloud. Check your usage in Workspace Settings > General > Cloud Quotas. Contact support@windmill.dev to increase the limit" .to_string(), )); } if ns.summary.len() > 300 { return Err(Error::BadRequest( "Summary must be less than 300 characters on cloud".to_string(), )); } if ns.description.len() > 3000 { return Err(Error::BadRequest( "Description must be less than 3000 characters on cloud".to_string(), )); } } let script_path = ns.path.clone(); // Caller-intent: CLI / git-sync deploys ask us to preserve any existing // user draft at this path instead of wiping it as part of the deploy. let skip_draft_deletion = ns.skip_draft_deletion.unwrap_or(false); let hash = ScriptHash(hash_script(&ns)); let authed = maybe_refresh_folders(&ns.path, &w_id, authed, &db).await; let mut tx: Transaction<'_, Postgres> = user_db.begin(&authed).await?; // Apply folder default_permissioned_as the first time a script is deployed // at this path. Check inside the transaction to avoid TOCTOU with concurrent deploys. let explicit_preserve = (ns.on_behalf_of_email.is_some() || ns.on_behalf_of.is_some()) && ns.preserve_on_behalf_of.unwrap_or(false) && windmill_common::can_preserve_on_behalf_of(&authed); if !explicit_preserve && windmill_common::can_preserve_on_behalf_of(&authed) { let path_already_exists = sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM script WHERE path = $1 AND workspace_id = $2)", &ns.path, &w_id ) .fetch_one(&mut *tx) .await? .unwrap_or(false); if !path_already_exists { if let Some((default_email, default_permissioned_as)) = windmill_common::folders::resolve_folder_default_on_behalf_of(&db, &w_id, &ns.path) .await? { ns.on_behalf_of_email = Some(default_email); ns.on_behalf_of = Some(default_permissioned_as); ns.preserve_on_behalf_of = Some(true); } } } let authed_principal = windmill_common::users::username_to_permissioned_as(&authed.username); // Resolved here rather than at the INSERT so the no-op check below compares the identity // that would actually be stored: the parent row holds only the principal, while a // preserving push may name that same principal by address alone. let resolved_on_behalf_of = windmill_common::resolve_on_behalf_of( ns.on_behalf_of_email.as_deref(), ns.on_behalf_of.as_deref(), ns.preserve_on_behalf_of.unwrap_or(false), &authed, &w_id, &db, ) .await?; // Written beside the principal only while a worker that still reads it may be live. let legacy_on_behalf_of_email = windmill_common::legacy_on_behalf_of_email(resolved_on_behalf_of.as_deref(), &w_id, &db) .await?; if sqlx::query_scalar!( "SELECT 1 FROM script WHERE hash = $1 AND workspace_id = $2", hash.0, &w_id ) .fetch_optional(&mut *tx) .await? .is_some() { return Err(Error::BadRequest( "A script with same hash (hence same path, description, summary, content) already \ exists!" .to_owned(), )); }; // When auto_parent is set, serialize concurrent creates for the same (workspace, path) // so the clashing_script query always sees the latest committed head. if ns.auto_parent.unwrap_or(false) { sqlx::query_scalar!( "SELECT pg_advisory_xact_lock(hashtext($1 || '/' || $2))", &w_id, &ns.path ) .fetch_one(&mut *tx) .await?; } let clashing_script = sqlx::query_as::<_, Script>(&format!( "SELECT {} FROM script WHERE path = $1 AND archived = false AND workspace_id = $2", windmill_common::scripts::SCRIPT_COLUMNS, )) .bind(&ns.path) .bind(&w_id) .fetch_optional(&mut *tx) .await?; struct ParentInfo { p_hashes: Vec, perms: serde_json::Value, p_path: String, } // When auto_parent is set, resolve parent_hash to the current head for this path // within the transaction. The advisory lock above ensures the second concurrent // request waits until the first commits, so this query sees the updated head. if ns.auto_parent.unwrap_or(false) { if let Some(ref cs) = clashing_script { ns.parent_hash = Some(cs.hash.clone()); } else { ns.parent_hash = None; } } let parent_hashes_and_perms: Option = match (&ns.parent_hash, clashing_script) { (None, None) => Ok(None), (None, Some(s)) => Err(Error::BadRequest(format!( "Path conflict for {} with non-archived hash {}", &ns.path, &s.hash ))), (Some(p_hash), o) => { // Lock the parent row to prevent concurrent updates with the same parent_hash // This ensures linear lineage - only one script can have a given parent at a time if sqlx::query_scalar!( "SELECT 1 FROM script WHERE hash = $1 AND workspace_id = $2 FOR UPDATE", p_hash.0, &w_id ) .fetch_optional(&mut *tx) .await? .is_none() { return Err(Error::BadRequest( "The parent hash does not seem to exist".to_owned(), )); }; let clashing_hash_o = sqlx::query_scalar!( "SELECT hash FROM script WHERE parent_hashes[1] = $1 AND workspace_id = $2", p_hash.0, &w_id ) .fetch_optional(&mut *tx) .await?; if let Some(clashing_hash) = clashing_hash_o { return Err(Error::BadRequest(format!( "A script with hash {} with same parent_hash has been found. However, the \ lineage must be linear: no 2 scripts can have the same parent", ScriptHash(clashing_hash) ))); }; let ScriptWithStarred { script: ps, .. } = get_script_by_hash_internal(&mut tx, &w_id, p_hash, None).await?; // No-op detection (opt-in via `?skip_if_noop=true`, currently only // passed by the CLI): when the new script is effectively identical // to its parent (same path, content, lockfile, and metadata), skip // creating a new version entirely. Returning `None` for the // deployment-metadata handle also suppresses the follow-on git // sync / promotion callbacks — the whole point is that idempotent // CLI pushes must not produce phantom commits on the downstream // git repository. if skip_if_noop && is_noop_deploy_against_parent(&ns, &ps, resolved_on_behalf_of.as_deref(), &db) .await? { tracing::info!( workspace_id = %w_id, path = %ns.path, parent_hash = %p_hash.0, "Skipping no-op script deploy (identical to parent)" ); return Ok((p_hash.clone(), tx, None, Vec::new())); } if ps.path != ns.path { // A rename writes the source as much as the destination, and only the destination // is scope-checked above. `require_owner_of_path` answers whether the *user* owns // the source, never what their token is scoped to — so without this a path-scoped // token could move a script it has no say over, taking its native triggers along // and re-registering them under that token's identity. check_scopes(&authed, || format!("scripts:write:{}", ps.path))?; require_owner_of_path(&authed, &ps.path)?; } let ph = { let v = ps.parent_hashes.map(|x| x.0).unwrap_or_default(); let mut v: Vec = v .into_iter() .take(MAX_HASH_HISTORY_LENGTH_STORED - 1) .collect(); v.insert(0, p_hash.0); v }; let r: Result> = match o { Some(clashing_script) if clashing_script.path == ns.path && clashing_script.hash.0 != p_hash.0 => { Err(Error::BadRequest(format!( "Path conflict for {} with non-archived hash {}", &ns.path, &clashing_script.hash ))) } Some(_) | None => Ok(Some(ParentInfo { p_hashes: ph, perms: ps.extra_perms, p_path: ps.path, })), }; sqlx::query!( "UPDATE script SET archived = true WHERE hash = $1 AND workspace_id = $2", p_hash.0, &w_id ) .execute(&mut *tx) .await?; r } }?; let p_hashes = parent_hashes_and_perms.as_ref().map(|v| &v.p_hashes[..]); let extra_perms = parent_hashes_and_perms .as_ref() .map(|v| v.perms.clone()) .unwrap_or(json!({})); // A dbt lock names a manifest digest and engine versions that only a // dependency job can determine, and that job is also what publishes the // script's manifest graph. Honouring a supplied one would skip it, leaving // the script with no graph — including on the UI paths that round-trip an // existing lock, like rename and unarchive. if matches!(ns.language, ScriptLang::Dbt) { ns.lock = None; // A codebase is a bundle of JS/TS sources; a dbt script's project is // its modules. Accepting one takes the branch below that stands in for // lock generation, which would suppress the very job that parses the // project and publishes the graph. if ns.codebase.is_some() { return Err(Error::BadRequest( "a dbt script has no codebase: its project is its modules, the \ `