diff --git a/backend/.sqlx/query-92d00c6a1f4c40f2a23c9ba758a59597deabd7ca93653ad72d2cdc37efefc9d4.json b/backend/.sqlx/query-8e7ff45c5378c3a3406ba94dc0653afa5d28c072203617c598caefaaf1bafcfb.json similarity index 85% rename from backend/.sqlx/query-92d00c6a1f4c40f2a23c9ba758a59597deabd7ca93653ad72d2cdc37efefc9d4.json rename to backend/.sqlx/query-8e7ff45c5378c3a3406ba94dc0653afa5d28c072203617c598caefaaf1bafcfb.json index a49e26981a..e1cdc4e0c7 100644 --- a/backend/.sqlx/query-92d00c6a1f4c40f2a23c9ba758a59597deabd7ca93653ad72d2cdc37efefc9d4.json +++ b/backend/.sqlx/query-8e7ff45c5378c3a3406ba94dc0653afa5d28c072203617c598caefaaf1bafcfb.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29)", + "query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30)", "describe": { "columns": [], "parameters": { @@ -67,10 +67,11 @@ "Int4", "Varchar", "Bool", - "Bool" + "Bool", + "Varchar" ] }, "nullable": [] }, - "hash": "92d00c6a1f4c40f2a23c9ba758a59597deabd7ca93653ad72d2cdc37efefc9d4" + "hash": "8e7ff45c5378c3a3406ba94dc0653afa5d28c072203617c598caefaaf1bafcfb" } diff --git a/backend/Cargo.lock b/backend/Cargo.lock index c95e0dac69..e5979b001f 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -872,6 +872,7 @@ dependencies = [ "matchit", "memchr", "mime", + "multer", "percent-encoding", "pin-project-lite", "rustversion", @@ -1447,18 +1448,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d59ae0466b83e838b81a54256c39d5d7c20b9d7daa10510a242d9b75abd5936e" dependencies = [ "chrono", - "chrono-tz-build 0.2.1", - "phf", -] - -[[package]] -name = "chrono-tz" -version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "93698b29de5e97ad0ae26447b344c482a7284c737d9ddc5f9e52b74a336671bb" -dependencies = [ - "chrono", - "chrono-tz-build 0.3.0", + "chrono-tz-build", "phf", ] @@ -1473,17 +1463,6 @@ dependencies = [ "phf_codegen", ] -[[package]] -name = "chrono-tz-build" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0c088aee841df9c3041febbb73934cfc39708749bf96dc827e3359cd39ef11b1" -dependencies = [ - "parse-zoneinfo", - "phf", - "phf_codegen", -] - [[package]] name = "cipher" version = "0.3.0" @@ -4351,6 +4330,24 @@ dependencies = [ "syn 2.0.60", ] +[[package]] +name = "multer" +version = "3.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a15d522be0a9c3e46fd2632e272d178f56387bdb5c9fbb3a36c649062e9b5219" +dependencies = [ + "bytes", + "encoding_rs", + "futures-util", + "http 1.1.0", + "httparse", + "log", + "memchr", + "mime", + "spin 0.9.8", + "version_check", +] + [[package]] name = "multiversion" version = "0.7.4" @@ -5222,7 +5219,7 @@ dependencies = [ "atoi_simd", "bytemuck", "chrono", - "chrono-tz 0.8.6", + "chrono-tz", "dyn-clone", "either", "ethnum", @@ -5283,7 +5280,7 @@ dependencies = [ "bitflags 2.5.0", "bytemuck", "chrono", - "chrono-tz 0.8.6", + "chrono-tz", "comfy-table", "either", "hashbrown 0.14.5", @@ -5395,7 +5392,7 @@ dependencies = [ "base64 0.21.7", "bytemuck", "chrono", - "chrono-tz 0.8.6", + "chrono-tz", "either", "hashbrown 0.14.5", "hex", @@ -5475,7 +5472,7 @@ checksum = "ff48362bd1b078bbbec7e7ba9ec01fea58fee2887db22a8e3deaf78f322fa3c4" dependencies = [ "ahash 0.8.11", "bytemuck", - "chrono-tz 0.8.6", + "chrono-tz", "futures", "once_cell", "percent-encoding", @@ -5531,7 +5528,7 @@ checksum = "86eb74ea6ddfe675aa5c3f33c00dadbe2b85f0e8e3887b85db1fd5a3397267fd" dependencies = [ "atoi", "chrono", - "chrono-tz 0.8.6", + "chrono-tz", "now", "once_cell", "polars-arrow", @@ -9669,7 +9666,7 @@ dependencies = [ "candle-nn", "candle-transformers", "chrono", - "chrono-tz 0.9.0", + "chrono-tz", "cookie 0.17.0", "crc", "cron", @@ -9680,7 +9677,7 @@ dependencies = [ "hmac", "http 1.1.0", "hyper 1.3.1", - "itertools 0.12.1", + "itertools 0.10.5", "lazy_static", "magic-crypt", "mime_guess", @@ -9766,6 +9763,7 @@ dependencies = [ "aws-config", "aws-sdk-sts", "axum", + "bytes", "chrono", "cron", "git-version", @@ -9773,7 +9771,7 @@ dependencies = [ "hmac", "hyper 1.3.1", "indexmap 2.2.6", - "itertools 0.12.1", + "itertools 0.10.5", "lazy_static", "magic-crypt", "mail-send", @@ -9835,7 +9833,7 @@ version = "1.323.6" dependencies = [ "anyhow", "gosyn", - "itertools 0.12.1", + "itertools 0.10.5", "lazy_static", "regex", "windmill-parser", @@ -9857,7 +9855,7 @@ name = "windmill-parser-py" version = "1.323.6" dependencies = [ "anyhow", - "itertools 0.12.1", + "itertools 0.10.5", "rustpython-parser", "serde_json", "windmill-parser", @@ -9869,7 +9867,7 @@ version = "1.323.6" dependencies = [ "anyhow", "async-recursion", - "itertools 0.12.1", + "itertools 0.10.5", "lazy_static", "phf", "regex", @@ -9934,14 +9932,14 @@ dependencies = [ "anyhow", "async-recursion", "axum", - "bigdecimal 0.4.3", + "bigdecimal 0.3.1", "chrono", - "chrono-tz 0.9.0", + "chrono-tz", "cron", "futures-core", "hex", "hmac", - "itertools 0.12.1", + "itertools 0.10.5", "lazy_static", "prometheus", "regex", @@ -9996,7 +9994,7 @@ dependencies = [ "gcp_auth", "git-version", "hex", - "itertools 0.12.1", + "itertools 0.10.5", "jsonwebtoken", "lazy_static", "mysql_async", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index e3b09d1837..76124cb59b 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -109,7 +109,7 @@ windmill-parser-sql = { path = "./parsers/windmill-parser-sql" } windmill-parser-graphql = { path = "./parsers/windmill-parser-graphql" } windmill-api-client = { path = "./windmill-api-client" } -axum = { version = "^0.7" } +axum = { version = "^0.7", features = ["multipart"] } headers = "^0" hyper = { version = "^1", features = ["full"] } tokio = { version = "^1", features = ["full", "tracing"] } diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 4160837664..64cefd1a89 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -f59f9c1e55e5e93f9eb6c081847cc566611029d2 \ No newline at end of file +644b6f49f087790a728a7a0a82525997b1742738 \ No newline at end of file diff --git a/backend/migrations/20240430005030_add_codebases.down.sql b/backend/migrations/20240430005030_add_codebases.down.sql new file mode 100644 index 0000000000..81bb6123a2 --- /dev/null +++ b/backend/migrations/20240430005030_add_codebases.down.sql @@ -0,0 +1,2 @@ +-- Add down migration script here +ALTER TABLE script DROP COLUMN codebase; \ No newline at end of file diff --git a/backend/migrations/20240430005030_add_codebases.up.sql b/backend/migrations/20240430005030_add_codebases.up.sql new file mode 100644 index 0000000000..13af236d16 --- /dev/null +++ b/backend/migrations/20240430005030_add_codebases.up.sql @@ -0,0 +1,3 @@ +-- Add up migration script here + +ALTER TABLE script ADD COLUMN codebase VARCHAR(255); \ No newline at end of file diff --git a/backend/substitute_ee_code.sh b/backend/substitute_ee_code.sh index d7f98bf0fe..dab1c6d608 100755 --- a/backend/substitute_ee_code.sh +++ b/backend/substitute_ee_code.sh @@ -71,7 +71,7 @@ if [ "$REVERT" == "YES" ]; then ce_file="${ee_file/${EE_CODE_DIR}/.}" ce_file="${root_dirpath}/backend/${ce_file}" if [ "$REVERT_PREVIOUS" == "YES" ]; then - git checkout HEAD@{5} ${ce_file} || true + git checkout HEAD@{75} ${ce_file} || true else git restore --staged ${ce_file} || true git restore ${ce_file} || true diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 026e6bfd0e..84bbf97c6f 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -1663,6 +1663,7 @@ func main(derp string) (string, error) { .to_owned(); let result = RunJob::from(JobPayload::Code(RawCode { + hash: None, content, path: None, lock: None, @@ -1694,6 +1695,7 @@ echo "hello $msg" .to_owned(); let job = RunJob::from(JobPayload::Code(RawCode { + hash: None, content, path: None, lock: None, @@ -1722,6 +1724,7 @@ def main(): .to_owned(); let job = JobPayload::Code(RawCode { + hash: None, content, path: None, language: ScriptLang::Python3, @@ -1756,6 +1759,7 @@ def main(): .to_owned(); let job = JobPayload::Code(RawCode { + hash: None, content, path: None, language: ScriptLang::Python3, @@ -1789,6 +1793,7 @@ def main(): .to_owned(); let job = JobPayload::Code(RawCode { + hash: None, content, path: None, language: ScriptLang::Python3, @@ -3165,6 +3170,7 @@ async fn run_deployed_relative_imports(db: &Pool, script_content: Stri concurrency_key: None, visible_to_runner_only: None, no_main_func: None, + codebase: None }, ).await.unwrap(); @@ -3212,6 +3218,7 @@ async fn run_preview_relative_imports(db: &Pool, script_content: Strin let db2 = db.clone(); in_test_worker(&db, async move { let job = RunJob::from(JobPayload::Code(RawCode { + hash: None, content: script_content, path: Some("f/system/test_import".to_string()), language, diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 9aaeae5efe..ffc215359a 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -8410,6 +8410,9 @@ components: type: boolean no_main_func: type: boolean + codebase: + type: string + required: - hash - path @@ -8427,6 +8430,7 @@ components: - starred - no_main_func + NewScript: type: object properties: @@ -8501,6 +8505,8 @@ components: type: boolean no_main_func: type: boolean + codebase: + type: string required: - path - summary diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 47a6f96905..e6f65deb8e 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -19,6 +19,9 @@ use windmill_common::jobs::{ format_completed_job_result, format_result, CompletedJobWithFormattedResult, FormattedResult, ENTRYPOINT_OVERRIDE, }; + +#[cfg(all(feature = "enterprise", feature = "parquet"))] +use windmill_common::scripts::PREVIEW_IS_CODEBASE_HASH; use windmill_common::variables::get_workspace_key; use crate::db::ApiAuthed; @@ -163,6 +166,7 @@ pub fn workspaced_service() -> Router { .layer(cors.clone()), ) .route("/run/preview", post(run_preview_script)) + .route("/run/preview_bundle", post(run_bundle_preview_script)) .route("/add_batch_jobs/:n", post(add_batch_jobs)) .route("/run/preview_flow", post(run_preview_flow_job)) .route( @@ -2077,7 +2081,9 @@ enum PreviewKind { Identity, Http, Noop, + Bundle, } + #[derive(Deserialize)] struct Preview { content: Option, @@ -2420,6 +2426,7 @@ pub async fn run_workflow_as_code( let (job_payload, tag, _delete_after_use, timeout) = match job.job_kind { JobKind::Preview => ( JobPayload::Code(RawCode { + hash: None, content: job.raw_code.unwrap_or_default(), path: job.script_path, language: job.language.unwrap_or_else(|| ScriptLang::Deno), @@ -3083,6 +3090,7 @@ async fn run_preview_script( Some(PreviewKind::Identity) => JobPayload::Identity, Some(PreviewKind::Noop) => JobPayload::Noop, _ => JobPayload::Code(RawCode { + hash: None, content: preview.content.unwrap_or_default(), path: preview.path, language: preview.language.unwrap_or(ScriptLang::Deno), @@ -3117,6 +3125,136 @@ async fn run_preview_script( Ok((StatusCode::CREATED, uuid.to_string())) } +#[cfg(all(feature = "enterprise", feature = "parquet"))] +async fn run_bundle_preview_script( + authed: ApiAuthed, + Extension(db): Extension, + Extension(user_db): Extension, + Extension(rsmq): Extension>, + Path(w_id): Path, + Query(run_query): Query, + mut multipart: axum::extract::Multipart, +) -> error::Result<(StatusCode, String)> { + check_license_key_valid().await?; + + check_scopes(&authed, || format!("runscript"))?; + if authed.is_operator { + return Err(error::Error::NotAuthorized( + "Operators cannot run preview jobs for security reasons".to_string(), + )); + } + + let mut job_id = None; + let mut tx = None; + let mut uploaded = false; + while let Some(field) = multipart.next_field().await.unwrap() { + let name = field.name().unwrap().to_string(); + let data = field.bytes().await.unwrap(); + if name == "preview" { + let preview: Preview = serde_json::from_slice(&data).map_err(to_anyhow)?; + + let scheduled_for = run_query.get_scheduled_for(&db).await?; + let tag = run_query.tag.clone().or(preview.tag.clone()); + check_tag_available_for_workspace(&w_id, &tag).await?; + let ltx = + PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq.clone()); + + let args = preview.args.unwrap_or_default(); + + // hmap.insert("") + let (uuid, ntx) = push( + &db, + ltx, + &w_id, + match preview.kind { + Some(PreviewKind::Identity) => JobPayload::Identity, + Some(PreviewKind::Noop) => JobPayload::Noop, + _ => JobPayload::Code(RawCode { + hash: Some(PREVIEW_IS_CODEBASE_HASH), + content: preview.content.unwrap_or_default(), + path: preview.path, + language: preview.language.unwrap_or(ScriptLang::Deno), + lock: preview.lock, + concurrent_limit: None, // TODO(gbouv): once I find out how to store limits in the content of a script, should be easy to plug limits here + concurrency_time_window_s: None, // TODO(gbouv): same as above + cache_ttl: None, + dedicated_worker: preview.dedicated_worker, + }), + }, + args, + authed.display_username(), + &authed.email, + username_to_permissioned_as(&authed.username), + scheduled_for, + None, + None, + None, + run_query.job_id, + false, + false, + None, + true, + tag, + run_query.timeout, + None, + None, + ) + .await?; + job_id = Some(uuid); + tx = Some(ntx); + } + if name == "file" { + let id = job_id + .as_ref() + .ok_or_else(|| { + Error::BadRequest( + "script need to be passed first in the multipart upload".to_string(), + ) + })? + .to_string(); + + uploaded = true; + + if let Some(os) = windmill_common::s3_helpers::OBJECT_STORE_CACHE_SETTINGS + .read() + .await + .clone() + { + let path = windmill_common::s3_helpers::bundle(&w_id, &id); + if let Err(e) = os + .put(&object_store::path::Path::from(path.clone()), data) + .await + { + tracing::info!("Failed to put snapshot to s3 at {path}: {:?}", e); + return Err(Error::ExecutionErr(format!("Failed to put {path} to s3"))); + } + } else { + return Err(Error::BadConfig("Object store is required for snapshot script and is not configured for servers".to_string())); + } + } + // println!("Length of `{}` is {} bytes", name, data.len()); + } + if !uploaded { + return Err(Error::BadRequest("No file uploaded".to_string())); + } + if job_id.is_none() { + return Err(Error::BadRequest( + "No script found in the uploaded file".to_string(), + )); + } + + tx.unwrap().commit().await?; + + Ok((StatusCode::CREATED, job_id.unwrap().to_string())) +} + +#[cfg(not(all(feature = "enterprise", feature = "parquet")))] +async fn run_bundle_preview_script() -> error::Result<(StatusCode, String)> { + return Err(Error::BadRequest( + "bundle preview is an ee feature".to_string(), + )); +} + #[derive(Deserialize)] pub struct RunDependenciesRequest { pub raw_scripts: Vec, diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 5aee47e568..9015740904 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -25,6 +25,7 @@ use db::DB; use git_version::git_version; use reqwest::Client; use std::collections::HashMap; +use std::time::Duration; use std::{net::SocketAddr, sync::Arc}; use tokio::sync::RwLock; use tower::ServiceBuilder; @@ -104,6 +105,8 @@ lazy_static::lazy_static! { pub static ref HTTP_CLIENT: Client = reqwest::ClientBuilder::new() .user_agent("windmill/beta") + .connect_timeout(Duration::from_secs(10)) + .timeout(Duration::from_secs(30)) .danger_accept_invalid_certs(std::env::var("ACCEPT_INVALID_CERTS").is_ok()) .build().unwrap(); @@ -330,6 +333,7 @@ async fn is_up_to_date() -> Result { let error_reading_version = || anyhow::anyhow!("Error reading latest released version"); let version = HTTP_CLIENT .get("https://api.github.com/repos/windmill-labs/windmill/releases/latest") + .timeout(Duration::from_secs(10)) .send() .await .context("Impossible to reach api.github")? diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index b19481e3cf..0073d0a5f3 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -13,6 +13,9 @@ use crate::{ webhook_util::{WebhookMessage, WebhookShared}, HTTP_CLIENT, }; +#[cfg(all(feature = "enterprise", feature = "parquet"))] +use axum::extract::Multipart; + use axum::{ extract::{Extension, Path, Query}, response::IntoResponse, @@ -31,6 +34,10 @@ use std::{ }; use windmill_audit::audit_ee::audit_log; use windmill_audit::ActionKind; + +#[cfg(all(feature = "enterprise", feature = "parquet"))] +use windmill_common::error::to_anyhow; + use windmill_common::{ db::UserDB, error::{Error, JsonResult, Result}, @@ -116,6 +123,7 @@ pub fn workspaced_service() -> Router { .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/draft/*path", get(get_script_by_path_w_draft)) .route("/get/p/*path", get(get_script_by_path)) @@ -196,6 +204,7 @@ async fn list_scripts( "draft_only", "ws_error_handler_muted", "no_main_func", + "codebase IS NOT NULL as use_codebase" ]) .left() .join("favorite") @@ -312,6 +321,82 @@ fn hash_script(ns: &NewScript) -> i64 { dh.finish() as i64 } +#[cfg(not(all(feature = "enterprise", feature = "parquet")))] +async fn create_snapshot_script() -> Result<(StatusCode, String)> { + Err(Error::BadRequest("Upgrade to EE to use bundle".to_string())) +} + +#[cfg(all(feature = "enterprise", feature = "parquet"))] +async fn create_snapshot_script( + authed: ApiAuthed, + Extension(user_db): Extension, + Extension(rsmq): Extension>, + Extension(webhook): Extension, + Extension(db): Extension, + Path(w_id): Path, + mut multipart: Multipart, +) -> Result<(StatusCode, String)> { + let mut script_hash = None; + let mut tx = None; + let mut uploaded = false; + 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 = Some(serde_json::from_slice(&data).map_err(to_anyhow)?); + let (new_hash, ntx) = create_script_internal( + ns.unwrap(), + w_id.clone(), + authed.clone(), + db.clone(), + rsmq.clone(), + user_db.clone(), + webhook.clone(), + ) + .await?; + script_hash = Some(new_hash.to_string()); + tx = Some(ntx); + } + 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; + if let Some(os) = windmill_common::s3_helpers::OBJECT_STORE_CACHE_SETTINGS + .read() + .await + .clone() + { + let path = windmill_common::s3_helpers::bundle(&w_id, &hash); + if let Err(e) = os + .put(&object_store::path::Path::from(path.clone()), data) + .await + { + tracing::info!("Failed to put snapshot to s3 at {path}: {:?}", e); + return Err(Error::ExecutionErr(format!("Failed to put {path} to s3"))); + } + } else { + return Err(Error::BadConfig("Object store is required for snapshot script and is not configured for servers".to_string())); + } + } + // 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?; + return Ok((StatusCode::CREATED, format!("{}", script_hash.unwrap()))); +} + async fn create_script( authed: ApiAuthed, Extension(user_db): Extension, @@ -321,6 +406,24 @@ async fn create_script( Path(w_id): Path, Json(ns): Json, ) -> Result<(StatusCode, String)> { + let (hash, tx) = create_script_internal(ns, w_id, authed, db, rsmq, user_db, webhook).await?; + tx.commit().await?; + Ok((StatusCode::CREATED, format!("{}", hash))) +} + +async fn create_script_internal<'c>( + ns: NewScript, + w_id: String, + authed: ApiAuthed, + db: sqlx::Pool, + rsmq: Option, + user_db: UserDB, + webhook: WebhookShared, +) -> Result<( + ScriptHash, + QueueTransaction<'c, rsmq_async::MultiplexedRsmq>, +)> { + 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( @@ -328,12 +431,10 @@ async fn create_script( .to_string(), )); } - let script_path = ns.path.clone(); let hash = ScriptHash(hash_script(&ns)); let authed = maybe_refresh_folders(&ns.path, &w_id, authed, &db).await; let mut tx: QueueTransaction<'_, _> = (rsmq.clone(), user_db.begin(&authed).await?).into(); - if sqlx::query_scalar!( "SELECT 1 FROM script WHERE hash = $1 AND workspace_id = $2", hash.0, @@ -349,7 +450,6 @@ async fn create_script( .to_owned(), )); }; - let clashing_script = sqlx::query_as::<_, Script>( "SELECT * FROM script WHERE path = $1 AND archived = false AND workspace_id = $2", ) @@ -357,7 +457,6 @@ async fn create_script( .bind(&w_id) .fetch_optional(&mut tx) .await?; - struct ParentInfo { p_hashes: Vec, perms: serde_json::Value, @@ -450,13 +549,11 @@ async fn create_script( 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!({})); - let lock = if !(ns.language == ScriptLang::Python3 || ns.language == ScriptLang::Go || ns.language == ScriptLang::Bun @@ -467,23 +564,20 @@ async fn create_script( ns.lock .and_then(|e| if e.is_empty() { None } else { Some(e) }) }; - let needs_lock_gen = lock.is_none(); - let envs = ns.envs.as_ref().map(|x| x.as_slice()); let envs = if ns.envs.is_none() || ns.envs.as_ref().unwrap().is_empty() { None } else { envs }; - //::text::json is to ensure we use serde_json with preserve order sqlx::query!( "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, \ content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, \ draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, \ dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, \ - delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func) \ - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29)", + delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30)", &w_id, &hash.0, ns.path, @@ -512,11 +606,11 @@ async fn create_script( ns.timeout, ns.concurrency_key, ns.visible_to_runner_only, - ns.no_main_func + ns.no_main_func, + codebase ) .execute(&mut tx) .await?; - let p_path_opt = parent_hashes_and_perms.as_ref().map(|x| x.p_path.clone()); if let Some(ref p_path) = p_path_opt { sqlx::query!( @@ -566,7 +660,6 @@ async fn create_script( .execute(&mut tx) .await?; } - if p_hashes.is_some() && !p_hashes.unwrap().is_empty() { audit_log( &mut tx, @@ -612,7 +705,6 @@ async fn create_script( }, ); } - let permissioned_as = username_to_permissioned_as(&authed.username); if needs_lock_gen { let tag = if ns.dedicated_worker.is_some_and(|x| x) { @@ -661,7 +753,7 @@ async fn create_script( None, ) .await?; - new_tx.commit().await?; + Ok((hash, new_tx)) } else { handle_deployment_metadata( &authed.email, @@ -678,10 +770,8 @@ async fn create_script( false, ) .await?; - tx.commit().await?; + Ok((hash, tx)) } - - Ok((StatusCode::CREATED, format!("{}", hash))) } pub async fn get_hub_script_by_path( diff --git a/backend/windmill-api/src/webhook_util.rs b/backend/windmill-api/src/webhook_util.rs index 590a706cc0..27a3b489d8 100644 --- a/backend/windmill-api/src/webhook_util.rs +++ b/backend/windmill-api/src/webhook_util.rs @@ -71,6 +71,7 @@ impl WebhookShared { let (tx, mut rx) = mpsc::unbounded_channel::(); let _process = tokio::spawn(async move { let client = reqwest::Client::builder() + .connect_timeout(Duration::from_secs(5)) // TODO: investigate pool timeouts and such if TCP load is high .timeout(Duration::from_secs(5)) .build() diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index b85302d236..b242e7ef84 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -2218,6 +2218,8 @@ struct ScriptMetadata { pub visible_to_runner_only: Option, #[serde(skip_serializing_if = "Option::is_none")] pub no_main_func: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub codebase: Option, } pub fn is_none_or_false(val: &Option) -> bool { @@ -2495,6 +2497,7 @@ async fn tarball_workspace( restart_unless_cancelled: script.restart_unless_cancelled, visible_to_runner_only: script.visible_to_runner_only, no_main_func: script.no_main_func, + codebase: script.codebase, }; let metadata_str = serde_json::to_string_pretty(&metadata).unwrap(); archive diff --git a/backend/windmill-common/Cargo.toml b/backend/windmill-common/Cargo.toml index 9ac9ea67a7..e1c0af06fa 100644 --- a/backend/windmill-common/Cargo.toml +++ b/backend/windmill-common/Cargo.toml @@ -10,7 +10,7 @@ enterprise = [] prometheus = ["dep:prometheus"] flamegraph = ["dep:tracing-flame"] loki = ["dep:tracing-loki"] -parquet = ["dep:object_store", "dep:aws-config", "dep:aws-sdk-sts"] +parquet = ["dep:object_store", "dep:aws-config", "dep:aws-sdk-sts", "dep:bytes"] [lib] name = "windmill_common" @@ -47,4 +47,5 @@ prometheus = { workspace = true, optional = true } aws-config = { workspace = true, optional = true } aws-sdk-sts = { workspace = true, optional = true } indexmap.workspace = true -mail-send.workspace = true \ No newline at end of file +bytes = { workspace = true, optional = true } +mail-send.workspace = true diff --git a/backend/windmill-common/src/ee.rs b/backend/windmill-common/src/ee.rs index d3ffd98584..51ad62a8a9 100644 --- a/backend/windmill-common/src/ee.rs +++ b/backend/windmill-common/src/ee.rs @@ -1,5 +1,5 @@ use crate::ee::LicensePlan::Community; -use serde::Deserialize; +use serde::{Deserialize, Serialize}; use std::sync::Arc; use tokio::sync::RwLock; @@ -20,8 +20,9 @@ pub async fn get_license_plan() -> LicensePlan { return Community; } -#[derive(Deserialize)] -#[serde(untagged)] +#[derive(Serialize, Deserialize)] pub enum CriticalErrorChannel {} -pub async fn trigger_critical_error_channels(_error_message: String) {} +pub async fn trigger_critical_error_channels(_msg: String) { + // Implementation is not open source +} diff --git a/backend/windmill-common/src/external_ip.rs b/backend/windmill-common/src/external_ip.rs index 087c908c76..0831bb54bc 100644 --- a/backend/windmill-common/src/external_ip.rs +++ b/backend/windmill-common/src/external_ip.rs @@ -13,16 +13,19 @@ use std::time::Duration; -use reqwest::Result; - -pub async fn get_ip() -> Result { - reqwest::ClientBuilder::new() - .timeout(Duration::from_secs(3)) +pub async fn get_ip() -> anyhow::Result { + tokio::select! { + biased; + _ = tokio::time::sleep(Duration::from_secs(10)) => { + return Err(anyhow::anyhow!("Expected to get ip under 10s")) + }, + ip = reqwest::ClientBuilder::new() + .connect_timeout(Duration::from_secs(5)) + .timeout(Duration::from_secs(5)) .build()? .get("https://hub.windmill.dev/getip") - .send() - .await? - .error_for_status()? - .text() - .await + .send() => Ok(ip? + .error_for_status()? + .text().await?), + } } diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index f93609385f..f90ea4e3c7 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -345,6 +345,7 @@ pub enum JobPayload { pub struct RawCode { pub content: String, pub path: Option, + pub hash: Option, pub language: ScriptLang, pub lock: Option, pub concurrent_limit: Option, diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index f5bb5186ee..ebe666e6a9 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -11,6 +11,7 @@ use std::{ sync::{atomic::AtomicBool, Arc}, }; +#[cfg(feature = "enterprise")] use ee::CriticalErrorChannel; use error::Error; use scripts::ScriptLang; diff --git a/backend/windmill-common/src/s3_helpers.rs b/backend/windmill-common/src/s3_helpers.rs index 4be07d33d9..fbf33af0e0 100644 --- a/backend/windmill-common/src/s3_helpers.rs +++ b/backend/windmill-common/src/s3_helpers.rs @@ -177,6 +177,48 @@ pub fn build_object_store_client( } } +#[cfg(feature = "parquet")] +pub async fn attempt_fetch_bytes( + client: Arc, + path: &str, +) -> error::Result { + use object_store::path::Path; + + let object = client.get(&Path::from(path)).await; + if let Err(e) = object { + tracing::info!( + "Failed to pull bytes from object store at path {path}. Error: {:?}", + e + ); + return Err(error::Error::ExecutionErr(format!( + "Failed to pull bytes from object store: {path}" + ))); + } + + let bytes = object.unwrap().bytes().await; + if bytes.is_err() { + tracing::info!( + "Failed to read bytes from object store: {path}. Error: {:?}", + bytes.err() + ); + return Err(error::Error::ExecutionErr(format!( + "Failed to read bytes from object store: {path}" + ))); + } + let bytes = bytes.unwrap(); + + tracing::info!("{path} len: {}", bytes.len()); + + if bytes.len() == 0 { + tracing::info!("object {path} not found in bucket, bytes empty",); + return Err(error::Error::ExecutionErr(format!( + "object {path} does not exist in bucket" + ))); + } + + return Ok(bytes); +} + #[cfg(feature = "parquet")] use aws_config::{default_provider::credentials::DefaultCredentialsChain, Region}; #[cfg(feature = "parquet")] @@ -408,3 +450,7 @@ impl CredentialProvider for AwsCredentialAdapter { })) } } + +pub fn bundle(w_id: &str, hash: &str) -> String { + format!("script_bundle/{}/{}", w_id, hash) +} diff --git a/backend/windmill-common/src/scripts.rs b/backend/windmill-common/src/scripts.rs index 3543c08f78..b7ab659b5b 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -128,6 +128,8 @@ impl Display for ScriptKind { } } +pub const PREVIEW_IS_CODEBASE_HASH: i64 = -42; + #[derive(Serialize, sqlx::FromRow)] pub struct Script { pub workspace_id: String, @@ -177,6 +179,8 @@ pub struct Script { pub visible_to_runner_only: Option, #[serde(skip_serializing_if = "Option::is_none")] pub no_main_func: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub codebase: Option, } #[derive(Serialize, sqlx::FromRow)] @@ -198,6 +202,12 @@ pub struct ListableScript { pub ws_error_handler_muted: Option, #[serde(skip_serializing_if = "Option::is_none")] pub no_main_func: Option, + #[serde(skip_serializing_if = "is_false")] + pub use_codebase: bool, +} + +fn is_false(x: &bool) -> bool { + return !x; } #[derive(Serialize)] @@ -256,6 +266,7 @@ pub struct NewScript { pub concurrency_key: Option, pub visible_to_runner_only: Option, pub no_main_func: Option, + pub codebase: Option, } fn lock_deserialize<'de, D>(deserializer: D) -> Result, D::Error> diff --git a/backend/windmill-common/src/utils.rs b/backend/windmill-common/src/utils.rs index fc7157308e..007a1031a3 100644 --- a/backend/windmill-common/src/utils.rs +++ b/backend/windmill-common/src/utils.rs @@ -6,7 +6,9 @@ * LICENSE-AGPL for a copy of the license. */ -use crate::ee::{trigger_critical_error_channels, LICENSE_KEY_ID}; +#[cfg(feature = "enterprise")] +use crate::ee::trigger_critical_error_channels; +use crate::ee::LICENSE_KEY_ID; use crate::error::{to_anyhow, Error, Result}; use crate::global_settings::UNIQUE_ID_SETTING; use crate::server::Smtp; @@ -233,5 +235,6 @@ pub async fn send_email( pub async fn report_critical_error(error_message: String) -> () { tracing::error!("CRITICAL ERROR: {error_message}"); + #[cfg(feature = "enterprise")] trigger_critical_error_channels(error_message).await; } diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index f829fca51b..7f819fdbe4 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -106,11 +106,10 @@ lazy_static::lazy_static! { lazy_static::lazy_static! { pub static ref HTTP_CLIENT: Client = reqwest::ClientBuilder::new() .user_agent("windmill/beta") + .timeout(std::time::Duration::from_secs(20)) + .connect_timeout(std::time::Duration::from_secs(10)) .build().unwrap(); - pub static ref HTTP_CLIENT_WORKER: Client = reqwest::ClientBuilder::new() - .user_agent("windmill/beta") - .build().unwrap(); } @@ -2946,6 +2945,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection JobPayload::Code(RawCode { content, path, + hash, language, lock, concurrent_limit, @@ -2953,7 +2953,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection cache_ttl, dedicated_worker, }) => ( - None, + hash, path, Some((content, lock)), JobKind::Preview, diff --git a/backend/windmill-worker/loader.bun.ts b/backend/windmill-worker/loader.bun.js similarity index 97% rename from backend/windmill-worker/loader.bun.ts rename to backend/windmill-worker/loader.bun.js index 66694a948e..c3dfeeaf54 100644 --- a/backend/windmill-worker/loader.bun.ts +++ b/backend/windmill-worker/loader.bun.js @@ -27,7 +27,7 @@ const p = { const cdir = resolve("./"); const cdirNoPrivate = cdir.replace(/^\/private/, ""); // for macos const filter = new RegExp( - `^(?!\\.\/main\\.ts)(?!${cdir}\/main\\.ts)(?!(?:/private)?${cdirNoPrivate}\/wrapper\\.ts).*\\.ts$` + `^(?!\\.\/main\\.ts)(?!${cdir}\/main\\.ts)(?!(?:/private)?${cdirNoPrivate}\/wrapper\\.mjs).*\\.ts$` ); build.onResolve({ filter }, (args) => { const file_path = diff --git a/backend/windmill-worker/loader_builder.bun.ts b/backend/windmill-worker/loader_builder.bun.js similarity index 91% rename from backend/windmill-worker/loader_builder.bun.ts rename to backend/windmill-worker/loader_builder.bun.js index 111a1661ca..1ef09f1a04 100644 --- a/backend/windmill-worker/loader_builder.bun.ts +++ b/backend/windmill-worker/loader_builder.bun.js @@ -21,7 +21,7 @@ if (!bo.success) { content.replaceAll("__require", "require") ); - const dependencies: Record = {}; + const dependencies = {}; for (const i of imports) { let [_, name, version] = i.path.match(captureVersion) ?? []; if (name == undefined) { @@ -46,7 +46,7 @@ if (!bo.success) { } } } - const resolvedDeps: Record = {}; + const resolvedDeps = {}; for (const i in dependencies) { const versions = dependencies[i]; resolvedDeps[i] = @@ -61,7 +61,7 @@ if (!bo.success) { JSON.stringify({ dependencies: resolvedDeps }, null, 2) ); - function reduceIntersect(versions: string[], name: string): string { + function reduceIntersect(versions, name) { console.log( `multiple versions detected for ${name}: ${versions.join(", ")}` ); diff --git a/backend/windmill-worker/nsjail/run.bun.config.proto b/backend/windmill-worker/nsjail/run.bun.config.proto index d5de58cc51..20b0893b4e 100644 --- a/backend/windmill-worker/nsjail/run.bun.config.proto +++ b/backend/windmill-worker/nsjail/run.bun.config.proto @@ -66,7 +66,7 @@ mount { src: "{JOB_DIR}/package.json" dst: "/tmp/{LANG}/package.json" is_bind: true - mandatory: true + mandatory: false } mount { @@ -77,13 +77,6 @@ mount { } -mount { - src: "{JOB_DIR}/wrapper.ts" - dst: "/tmp/{LANG}/wrapper.ts" - is_bind: true - mandatory: false -} - mount { src: "{JOB_DIR}/wrapper.mjs" dst: "/tmp/{LANG}/wrapper.mjs" @@ -114,6 +107,13 @@ mount { mandatory: false } +mount { + src: "{JOB_DIR}/main.js" + dst: "/tmp/{LANG}/main.js" + is_bind: true + mandatory: false +} + mount { src: "{JOB_DIR}/args.json" diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 1390ed3c61..5d6f896ba1 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -35,11 +35,15 @@ use windmill_common::{ error::{self, to_anyhow, Result}, jobs::QueuedJob, }; + +#[cfg(all(feature = "enterprise", feature = "parquet"))] +use windmill_common::s3_helpers::attempt_fetch_bytes; + use windmill_parser::Typ; -const RELATIVE_BUN_LOADER: &str = include_str!("../loader.bun.ts"); +const RELATIVE_BUN_LOADER: &str = include_str!("../loader.bun.js"); -const RELATIVE_BUN_BUILDER: &str = include_str!("../loader_builder.bun.ts"); +const RELATIVE_BUN_BUILDER: &str = include_str!("../loader_builder.bun.js"); const NSJAIL_CONFIG_RUN_BUN_CONTENT: &str = include_str!("../nsjail/run.bun.config.proto"); @@ -76,7 +80,7 @@ pub async fn gen_lockfile( } else { let _ = write_file( &job_dir, - "build.ts", + "build.js", &format!( r#" {} @@ -100,7 +104,7 @@ pub async fn gen_lockfile( .current_dir(job_dir) .env_clear() .envs(common_bun_proc_envs.clone()) - .args(vec!["run", "build.ts"]) + .args(vec!["run", "build.js"]) .stdout(Stdio::piped()) .stderr(Stdio::piped()); let child_process = start_child_process(child_cmd, &*BUN_PATH).await?; @@ -370,7 +374,7 @@ try {{ }} const bo = await Bun.build({{ - entrypoints: ["{job_dir}/wrapper.ts"], + entrypoints: ["{job_dir}/wrapper.mjs"], outdir: "./", target: "node", plugins: [p], @@ -389,7 +393,7 @@ if (!bo.success) {{ } else { write_file( &job_dir, - "loader.bun.ts", + "loader.bun.js", &format!( r#" import {{ plugin }} from "bun"; @@ -450,9 +454,46 @@ pub async fn generate_wrapper_mjs( Ok(()) } +#[cfg(all(feature = "enterprise", feature = "parquet"))] +pub async fn pull_codebase(w_id: &str, id: &str, job_dir: &str) -> Result<()> { + let path = windmill_common::s3_helpers::bundle(&w_id, &id); + let bun_cache_path = format!("{}/{}", BUN_CACHE_DIR, path); + let dst = format!("{job_dir}/main.js"); + let dirs_splitted = bun_cache_path.split("/").collect_vec(); + tokio::fs::create_dir_all(dirs_splitted[..dirs_splitted.len() - 1].join("/")).await?; + if tokio::fs::metadata(&bun_cache_path).await.is_ok() { + tracing::info!("loading {bun_cache_path} from cache"); + tokio::fs::symlink(&bun_cache_path, dst).await?; + } else if let Some(os) = windmill_common::s3_helpers::OBJECT_STORE_CACHE_SETTINGS + .read() + .await + .clone() + { + let bytes = attempt_fetch_bytes(os, &path).await?; + if *windmill_common::worker::CLOUD_HOSTED { + tokio::fs::write(dst, &bytes).await?; + } else { + tokio::fs::write(&bun_cache_path, &bytes).await?; + tokio::fs::symlink(bun_cache_path, dst).await?; + } + + // extract_tar(bytes, job_dir).await?; + } + + return Ok(()); +} + +#[cfg(not(all(feature = "enterprise", feature = "parquet")))] +pub async fn pull_codebase(_w_id: &str, _id: &str, _job_dir: &str) -> Result<()> { + return Err(error::Error::ExecutionErr( + "codebase is an EE feature".to_string(), + )); +} + #[tracing::instrument(level = "trace", skip_all)] pub async fn handle_bun_job( requirements_o: Option, + codebase: Option, mem_peak: &mut i32, canceled_by: &mut Option, job: &QueuedJob, @@ -465,13 +506,20 @@ pub async fn handle_bun_job( envs: HashMap, shared_mount: &str, ) -> error::Result> { - let _ = write_file(job_dir, "main.ts", inner_content).await?; + if !codebase.is_some() { + let _ = write_file(job_dir, "main.ts", inner_content).await?; + } else { + let _ = write_file(job_dir, "package.json", r#"{ "type": "module" }"#).await?; + } let common_bun_proc_envs: HashMap = get_common_bun_proc_envs(&base_internal_url).await; - let annotation = get_annotation(inner_content); + let mut annotation = get_annotation(inner_content); + if codebase.is_some() { + annotation.nodejs_mode = true + } let main_override = get_main_override(job.args.as_ref()); #[cfg(not(feature = "enterprise"))] @@ -481,7 +529,9 @@ pub async fn handle_bun_job( )); } - if let Some(reqs) = requirements_o { + if let Some(codebase) = codebase.as_ref() { + pull_codebase(&job.workspace_id, codebase, job_dir).await?; + } else if let Some(reqs) = requirements_o { let splitted = reqs.split(BUN_LOCKB_SPLIT).collect::>(); if splitted.len() != 2 { return Err(error::Error::ExecutionErr( @@ -547,10 +597,11 @@ pub async fn handle_bun_job( // } } - let main_code = remove_pinned_imports(inner_content)?; - let _ = write_file(job_dir, "main.ts", &main_code).await?; + let _ = write_file(job_dir, "main.ts", &remove_pinned_imports(inner_content)?).await?; - let init_logs = if annotation.nodejs_mode { + let init_logs = if codebase.is_some() { + "\n\n--- NODE SNAPSHOT EXECUTION ---\n".to_string() + } else if annotation.nodejs_mode { "\n\n--- NODE CODE EXECUTION ---\n".to_string() } else { "\n\n--- BUN CODE EXECUTION ---\n".to_string() @@ -581,11 +632,17 @@ pub async fn handle_bun_job( // we cannot use Bun.read and Bun.write because it results in an EBADF error on cloud let main_name = main_override.unwrap_or("main".to_string()); + let main_import = if codebase.is_some() { + "./main.js" + } else { + "./main.ts" + }; + let wrapper_content: String = format!( r#" -import {{ {main_name} }} from "./main.ts"; +import {{ {main_name} }} from "{main_import}"; -const fs = require('fs/promises'); +import * as fs from "fs/promises"; const args = await fs.readFile('args.json', {{ encoding: 'utf8' }}).then(JSON.parse) .then(({{ {spread} }}) => [ {spread} ]) @@ -596,7 +653,7 @@ BigInt.prototype.toJSON = function () {{ {dates} async function run() {{ - let res: any = await {main_name}(...args); + let res = await {main_name}(...args); const res_json = JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value); await fs.writeFile("result.json", res_json); process.exit(0); @@ -614,7 +671,7 @@ try {{ }} "#, ); - write_file(job_dir, "wrapper.ts", &wrapper_content).await?; + write_file(job_dir, "wrapper.mjs", &wrapper_content).await?; Ok(()) as error::Result<()> }; @@ -633,15 +690,19 @@ try {{ }; let write_loader_f = async { - build_loader( - job_dir, - base_internal_url, - &client.get_token().await, - &job.workspace_id, - &job.script_path(), - annotation.nodejs_mode, - ) - .await + if !codebase.is_some() { + build_loader( + job_dir, + base_internal_url, + &client.get_token().await, + &job.workspace_id, + &job.script_path(), + annotation.nodejs_mode, + ) + .await + } else { + Ok(()) + } }; let (reserved_variables, _, _) = tokio::try_join!( @@ -650,7 +711,7 @@ try {{ write_loader_f )?; - if annotation.nodejs_mode { + if annotation.nodejs_mode && !codebase.is_some() { generate_wrapper_mjs( job_dir, &job.workspace_id, @@ -703,8 +764,18 @@ try {{ "run.config.proto", "--", &NODE_PATH, + "--experimental-default-type=module", "/tmp/nodejs/wrapper.mjs", ] + } else if codebase.is_some() { + vec![ + "--config", + "run.config.proto", + "--", + &BUN_PATH, + "run", + "/tmp/bun/wrapper.mjs", + ] } else { vec![ "--config", @@ -715,8 +786,8 @@ try {{ "-i", "--prefer-offline", "-r", - "/tmp/bun/loader.bun.ts", - "/tmp/bun/wrapper.ts", + "/tmp/bun/loader.bun.js", + "/tmp/bun/wrapper.mjs", ] }; nsjail_cmd @@ -741,28 +812,33 @@ try {{ .envs(envs) .envs(reserved_variables) .envs(common_bun_proc_envs) - .args(vec![&script_path]) + .args(vec!["--experimental-default-type=module", &script_path]) .stdout(Stdio::piped()) .stderr(Stdio::piped()); bun_cmd } else { - let script_path = format!("{job_dir}/wrapper.ts"); + let script_path = format!("{job_dir}/wrapper.mjs"); let mut bun_cmd = Command::new(&*BUN_PATH); + let args = if codebase.is_some() { + vec!["run", &script_path] + } else { + vec![ + "run", + "-i", + "--prefer-offline", + "-r", + "./loader.bun.js", + &script_path, + ] + }; bun_cmd .current_dir(job_dir) .env_clear() .envs(envs) .envs(reserved_variables) .envs(common_bun_proc_envs) - .args(vec![ - "run", - "-i", - "--prefer-offline", - "-r", - "./loader.bun.ts", - &script_path, - ]) + .args(args) .stdout(Stdio::piped()) .stderr(Stdio::piped()); bun_cmd @@ -824,6 +900,7 @@ use std::sync::Arc; #[cfg(feature = "enterprise")] pub async fn start_worker( requirements_o: Option, + codebase: Option, db: &sqlx::Pool, inner_content: &str, base_internal_url: &str, @@ -840,7 +917,13 @@ pub async fn start_worker( let mut logs = "".to_string(); let mut mem_peak: i32 = 0; let mut canceled_by: Option = None; - let _ = write_file(job_dir, "main.ts", inner_content).await?; + tracing::info!("Starting worker {w_id};{script_path} (codebase: {codebase:?}"); + if !codebase.is_some() { + let _ = write_file(job_dir, "main.ts", inner_content).await?; + } else { + let _ = write_file(job_dir, "package.json", r#"{ "type": "module" }"#).await?; + } + let common_bun_proc_envs: HashMap = get_common_bun_proc_envs(&base_internal_url).await; @@ -867,8 +950,10 @@ pub async fn start_worker( ) .await; let context_envs = build_envs_map(context.to_vec()).await; - let annotation = get_annotation(inner_content); - if let Some(reqs) = requirements_o { + + if let Some(codebase) = codebase.as_ref() { + pull_codebase(w_id, codebase, job_dir).await?; + } else if let Some(reqs) = requirements_o { let splitted = reqs.split(BUN_LOCKB_SPLIT).collect::>(); if splitted.len() != 2 { return Err(error::Error::ExecutionErr( @@ -959,9 +1044,14 @@ pub async fn start_worker( "" }; + let main_import = if codebase.is_some() { + "./main.js" + } else { + "./main.ts" + }; let wrapper_content: String = format!( r#" -import {{ main }} from "./main.ts"; +import {{ main }} from "{main_import}"; import {{ createInterface }} from "node:readline" BigInt.prototype.toJSON = function () {{ @@ -980,7 +1070,7 @@ for await (const line of createInterface({{ input: process.stdin }})) {{ }} try {{ let {{ {spread} }} = JSON.parse(line) - let res: any = await main(...[ {spread} ]); + let res = await main(...[ {spread} ]); console.log("wm_res[success]:" + JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value)); }} catch (e) {{ console.log("wm_res[error]:" + JSON.stringify({{ message: e.message, name: e.name, stack: e.stack, line: line }})); @@ -988,20 +1078,22 @@ for await (const line of createInterface({{ input: process.stdin }})) {{ }} "#, ); - write_file(job_dir, "wrapper.ts", &wrapper_content).await?; + write_file(job_dir, "wrapper.mjs", &wrapper_content).await?; } - build_loader( - job_dir, - base_internal_url, - token, - w_id, - script_path, - annotation.nodejs_mode, - ) - .await?; + if !codebase.is_some() { + build_loader( + job_dir, + base_internal_url, + token, + w_id, + script_path, + annotation.nodejs_mode, + ) + .await?; + } - if annotation.nodejs_mode { + if annotation.nodejs_mode && !codebase.is_some() { generate_wrapper_mjs( job_dir, w_id, @@ -1026,7 +1118,7 @@ for await (const line of createInterface({{ input: process.stdin }})) {{ envs, context, common_bun_proc_envs, - vec![&script_path], + vec!["--experimental-default-type=module", &script_path], killpill_rx, job_completed_tx, token, @@ -1050,8 +1142,8 @@ for await (const line of createInterface({{ input: process.stdin }})) {{ "-i", "--prefer-offline", "-r", - "./loader.bun.ts", - &format!("{job_dir}/wrapper.ts"), + "./loader.bun.js", + &format!("{job_dir}/wrapper.mjs"), ], killpill_rx, job_completed_tx, diff --git a/backend/windmill-worker/src/global_cache.rs b/backend/windmill-worker/src/global_cache.rs index e9553943e3..a1887d60f8 100644 --- a/backend/windmill-worker/src/global_cache.rs +++ b/backend/windmill-worker/src/global_cache.rs @@ -71,50 +71,10 @@ pub async fn build_tar_and_push( Ok(()) } -#[cfg(all(feature = "enterprise", feature = "parquet"))] -pub async fn attempt_fetch_bytes( - client: Arc, - path: &str, -) -> error::Result { - use object_store::path::Path; - - let object = client.get(&Path::from(path.clone())).await; - if let Err(e) = object { - tracing::info!( - "Failed to pull bytes from object store at path {path}. Error: {:?}", - e - ); - return Err(error::Error::ExecutionErr(format!( - "Failed to pull bytes from object store: {path}" - ))); - } - - let bytes = object.unwrap().bytes().await; - if bytes.is_err() { - tracing::info!( - "Failed to read bytes from object store: {path}. Error: {:?}", - bytes.err() - ); - return Err(error::Error::ExecutionErr(format!( - "Failed to read bytes from object store: {path}" - ))); - } - let bytes = bytes.unwrap(); - - tracing::info!("{path} len: {}", bytes.len()); - - if bytes.len() == 0 { - tracing::info!("object {path} not found in bucket, bytes empty",); - return Err(error::Error::ExecutionErr(format!( - "object {path} does not exist in bucket" - ))); - } - - return Ok(bytes); -} - #[cfg(all(feature = "enterprise", feature = "parquet"))] pub async fn pull_from_tar(client: Arc, folder: String) -> error::Result<()> { + use windmill_common::s3_helpers::attempt_fetch_bytes; + let folder_name = folder.split("/").last().unwrap(); tracing::info!("Attempting to pull piptar {folder_name} from bucket"); @@ -125,7 +85,7 @@ pub async fn pull_from_tar(client: Arc, folder: String) -> erro // tracing::info!("B: {target} {folder}"); - extract_pip_tar(bytes, &folder).await.map_err(|e| { + extract_tar(bytes, &folder).await.map_err(|e| { tracing::error!("Failed to extract piptar {folder_name}. Error: {:?}", e); e })?; @@ -139,7 +99,7 @@ pub async fn pull_from_tar(client: Arc, folder: String) -> erro } #[cfg(all(feature = "enterprise", feature = "parquet"))] -pub async fn extract_pip_tar(tar: bytes::Bytes, folder: &str) -> error::Result<()> { +pub async fn extract_tar(tar: bytes::Bytes, folder: &str) -> error::Result<()> { use bytes::Buf; use tokio::fs::{self}; @@ -149,14 +109,14 @@ pub async fn extract_pip_tar(tar: bytes::Bytes, folder: &str) -> error::Result<( let mut ar = tar::Archive::new(tar.reader()); if let Err(e) = ar.unpack(folder) { - tracing::info!("Failed to untar piptar. Error: {:?}", e); + tracing::info!("Failed to untar to {folder}. Error: {:?}", e); fs::remove_dir_all(&folder).await?; return Err(error::Error::ExecutionErr(format!( "Failed to untar piptar {folder}" ))); } tracing::info!( - "Finished extracting pip tar {folder}. Took {}ms", + "Finished extracting tar to {folder}. Took {}ms", start.elapsed().as_millis(), ); Ok(()) diff --git a/backend/windmill-worker/src/go_executor.rs b/backend/windmill-worker/src/go_executor.rs index 53562107af..129389ec2a 100644 --- a/backend/windmill-worker/src/go_executor.rs +++ b/backend/windmill-worker/src/go_executor.rs @@ -120,7 +120,7 @@ async fn load_cache(bin_path: &str, _hash: &str) -> (bool, String) { .await .clone() { - use crate::global_cache::attempt_fetch_bytes; + use windmill_common::s3_helpers::attempt_fetch_bytes; if let Ok(mut x) = attempt_fetch_bytes(os, &hash_to_os_path(_hash)).await { if let Err(e) = write_binary_file(bin_path, &mut x).await { diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index b6abef7fcb..2d57c63264 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -41,7 +41,7 @@ use windmill_common::{ flows::{FlowModule, FlowModuleValue, FlowValue}, get_latest_deployed_hash_for_path, jobs::{JobKind, JobPayload, QueuedJob}, - scripts::{get_full_hub_script_by_path, ScriptHash, ScriptLang}, + scripts::{get_full_hub_script_by_path, ScriptHash, ScriptLang, PREVIEW_IS_CODEBASE_HASH}, users::{SUPERADMIN_NOTIFICATION_EMAIL, SUPERADMIN_SECRET_EMAIL, SUPERADMIN_SYNC_EMAIL}, utils::{rd_string, StripPath}, worker::{ @@ -198,6 +198,7 @@ pub const DENO_CACHE_DIR_NPM: &str = concatcp!(ROOT_CACHE_DIR, "deno/npm"); pub const GO_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "go"); pub const BUN_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "bun"); + pub const HUB_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "hub"); pub const GO_BIN_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "gobin"); pub const POWERSHELL_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "powershell"); @@ -1038,7 +1039,7 @@ pub async fn run_worker path.to_string(), SpawnWorker::Script { path, .. } => path.to_string(), }; + let path2 = path.clone(); let w_id = w_id.to_string(); - let (content, lock, language, envs) = match sw { + let (content, lock, language, envs, codebase) = match sw { SpawnWorker::Script { path, hash } => { let q = if let Some(hash) = hash { get_script_content_by_hash(&hash, &w_id, &db).await.map( - |r: ContentReqLangEnvs| Some((r.content, r.lockfile, r.language, r.envs)), + |r: ContentReqLangEnvs| { + Some((r.content, r.lockfile, r.language, r.envs, r.codebase)) + }, ) } else { - sqlx::query_as::<_, (String, Option, Option, Option>)>( - "SELECT content, lock, language, envs FROM script WHERE path = $1 AND workspace_id = $2 AND + sqlx::query_as::<_, (String, Option, Option, Option>, bool, Option)>( + "SELECT content, lock, language, envs, codebase IS NOT NULL, hash FROM script WHERE path = $1 AND workspace_id = $2 AND created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND deleted = false AND lock IS not NULL AND lock_error_logs IS NULL)", ) @@ -2045,6 +2062,7 @@ async fn spawn_dedicated_worker( .fetch_optional(&db) .await .map_err(|e| Error::InternalErr(format!("expected content and lock: {e}"))) + .map(|x| x.map(|y| (y.0, y.1, y.2, y.3, if y.4 { y.5.map(|z| z.to_string()) } else { None }))) }; if let Ok(q) = q { if let Some(wp) = q { @@ -2063,7 +2081,9 @@ async fn spawn_dedicated_worker( return None; } } - SpawnWorker::RawScript { content, lock, lang, .. } => (content, lock, Some(lang), None), + SpawnWorker::RawScript { content, lock, lang, .. } => { + (content, lock, Some(lang), None, None) + } }; match language { @@ -2118,6 +2138,7 @@ async fn spawn_dedicated_worker( Some(ScriptLang::Bun) => { crate::bun_executor::start_worker( lock, + codebase, &db, &content, &base_internal_url, @@ -2176,6 +2197,7 @@ async fn queue_init_bash_maybe<'c, R: rsmq_async::RsmqConnection + Send + 'c>( tx, "admins", windmill_common::jobs::JobPayload::Code(windmill_common::jobs::RawCode { + hash: None, content: content.clone(), path: Some(format!("init_script_{worker_name}")), language: ScriptLang::Bash, @@ -2982,6 +3004,7 @@ struct ContentReqLangEnvs { lockfile: Option, language: Option, envs: Option>, + codebase: Option, } async fn get_hub_script_content_and_requirements( @@ -3018,6 +3041,7 @@ async fn get_hub_script_content_and_requirements( lockfile: script.lockfile, language: Some(script.language), envs: None, + codebase: None, }) } @@ -3050,16 +3074,27 @@ async fn get_script_content_by_hash( Option, Option, Option>, + bool, ), >( - "SELECT content, lock, language, envs FROM script WHERE hash = $1 AND workspace_id = $2", + "SELECT content, lock, language, envs, codebase IS NOT NULL FROM script WHERE hash = $1 AND workspace_id = $2", ) .bind(script_hash.0) .bind(w_id) .fetch_optional(db) .await? .ok_or_else(|| Error::InternalErr(format!("expected content and lock")))?; - Ok(ContentReqLangEnvs { content: r.0, lockfile: r.1, language: r.2, envs: r.3 }) + Ok(ContentReqLangEnvs { + content: r.0, + lockfile: r.1, + language: r.2, + envs: r.3, + codebase: if r.4 { + Some(script_hash.to_string()) + } else { + None + }, + }) } #[tracing::instrument(level = "trace", skip_all)] @@ -3075,35 +3110,48 @@ async fn handle_code_execution_job( worker_name: &str, column_order: &mut Option>, ) -> error::Result> { - let ContentReqLangEnvs { content: inner_content, lockfile: requirements_o, language, envs } = - match job.job_kind { - JobKind::Preview => ContentReqLangEnvs { - content: job - .raw_code - .clone() - .unwrap_or_else(|| "no raw code".to_owned()), - lockfile: job.raw_lock.clone(), - language: job.language.to_owned(), - envs: None, + let ContentReqLangEnvs { + content: inner_content, + lockfile: requirements_o, + language, + envs, + codebase, + } = match job.job_kind { + JobKind::Preview => ContentReqLangEnvs { + content: job + .raw_code + .clone() + .unwrap_or_else(|| "no raw code".to_owned()), + lockfile: job.raw_lock.clone(), + language: job.language.to_owned(), + envs: None, + codebase: if job + .script_hash + .is_some_and(|y| y.0 == PREVIEW_IS_CODEBASE_HASH) + { + Some(job.id.to_string()) + } else { + None }, - JobKind::Script_Hub => { - get_hub_script_content_and_requirements(job.script_path.clone(), db).await? - } - JobKind::Script => { - get_script_content_by_hash( - &job.script_hash.unwrap_or(ScriptHash(0)), - &job.workspace_id, - db, - ) - .await? - } - JobKind::DeploymentCallback => { - get_script_content_by_path(job.script_path.clone(), &job.workspace_id, db).await? - } - _ => unreachable!( - "handle_code_execution_job should never be reachable with a non-code execution job" - ), - }; + }, + JobKind::Script_Hub => { + get_hub_script_content_and_requirements(job.script_path.clone(), db).await? + } + JobKind::Script => { + get_script_content_by_hash( + &job.script_hash.unwrap_or(ScriptHash(0)), + &job.workspace_id, + db, + ) + .await? + } + JobKind::DeploymentCallback => { + get_script_content_by_path(job.script_path.clone(), &job.workspace_id, db).await? + } + _ => unreachable!( + "handle_code_execution_job should never be reachable with a non-code execution job" + ), + }; if language == Some(ScriptLang::Postgresql) { return do_postgresql( @@ -3306,6 +3354,7 @@ mount {{ Some(ScriptLang::Bun) => { handle_bun_job( requirements_o, + codebase, mem_peak, canceled_by, job, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index a0b2789774..92cb019089 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -3124,6 +3124,7 @@ fn raw_script_to_payload( ) -> JobPayloadWithTag { JobPayloadWithTag { payload: JobPayload::Code(RawCode { + hash: None, path, content: content.clone(), language: language.clone(), diff --git a/cli/codebase.ts b/cli/codebase.ts new file mode 100644 index 0000000000..05b3c0138f --- /dev/null +++ b/cli/codebase.ts @@ -0,0 +1,21 @@ +import { Codebase, SyncOptions } from "./conf.ts"; +import { log } from "./deps.ts"; +import { digestDir } from "./utils.ts"; + +export type SyncCodebase = Codebase & { digest: string }; +export async function listSyncCodebases( + options: SyncOptions +): Promise { + const res: SyncCodebase[] = []; + const nb_codebase = options?.codebases?.length ?? 0; + if (nb_codebase > 0) { + log.info(`Found ${nb_codebase} codebases:`); + } + for (const codebase of options?.codebases ?? []) { + const digest = await digestDir(codebase.relative_path); + log.info(`Codebase ${codebase.relative_path}, digest: ${digest}`); + res.push({ ...codebase, digest }); + } + + return res; +} diff --git a/cli/conf.ts b/cli/conf.ts index 30439bf217..a0c713481c 100644 --- a/cli/conf.ts +++ b/cli/conf.ts @@ -20,6 +20,12 @@ export interface SyncOptions { extraIncludes?: string[]; excludes?: string[]; defaultTs?: "bun" | "deno"; + codebases?: Codebase[]; +} + +export interface Codebase { + relative_path: string; + includes: string[]; } export async function readConfigFile(): Promise { diff --git a/cli/main.ts b/cli/main.ts index ea1206b44c..93f6001ae4 100644 --- a/cli/main.ts +++ b/cli/main.ts @@ -3,7 +3,9 @@ import { CompletionsCommand, DenoLandProvider, UpgradeCommand, + colors, log, + yamlStringify, } from "./deps.ts"; import flow from "./flow.ts"; import app from "./apps.ts"; @@ -58,6 +60,18 @@ let command: any = new Command() "Specify headers to use for all requests. e.g: \"HEADERS='h1: v1, h2: v2'\"" ) .version(VERSION) + .command("init", "Bootstrap a windmill project with a wmill.yaml file") + .action(async () => { + if (await Deno.stat("wmill.yaml").catch(() => null)) { + log.error(colors.red("wmill.yaml already exists")); + return; + } + await Deno.writeTextFile( + "wmill.yaml", + yamlStringify({ defaultTs: "bun", includes: [], excludes: [] }) + ); + log.info(colors.green("wmill.yaml created")); + }) .command("app", app) .command("flow", flow) .command("script", script) @@ -70,6 +84,7 @@ let command: any = new Command() .command("schedule", schedule) .command("dev", dev) .command("sync", sync) + .command("version", "Show version information") .action(async (opts) => { console.log("CLI build against " + VERSION); diff --git a/cli/metadata.ts b/cli/metadata.ts index bf65713339..b448dd0559 100644 --- a/cli/metadata.ts +++ b/cli/metadata.ts @@ -1,6 +1,6 @@ // deno-lint-ignore-file no-explicit-any import { GlobalOptions } from "./types.ts"; -import { colors, encodeHex, log, yamlParse, yamlStringify } from "./deps.ts"; +import { colors, log, yamlParse, yamlStringify } from "./deps.ts"; import { ScriptMetadata, defaultScriptMetadata, @@ -25,6 +25,7 @@ import { ScriptLanguage } from "./script_common.ts"; import { inferContentTypeFromFilePath } from "./script_common.ts"; import { GlobalDeps } from "./script.ts"; import { yamlOptions } from "./sync.ts"; +import { generateHash } from "./utils.ts"; export async function generateAllMetadata() {} @@ -457,7 +458,11 @@ export async function replaceLock(o?: { lock?: string | string[] }) { export async function parseMetadataFile( scriptPath: string, generateMetadataIfMissing: - | (GlobalOptions & { path: string; workspaceRemote: Workspace }) + | (GlobalOptions & { + path: string; + workspaceRemote: Workspace; + schemaOnly?: boolean; + }) | undefined, globalDeps: GlobalDeps ): Promise<{ isJson: boolean; payload: any; path: string }> { @@ -509,7 +514,6 @@ export async function parseMetadataFile( false, globalDeps ); - scriptInitialMetadata = yamlParse( await Deno.readTextFile(metadataFilePath) ) as ScriptMetadata; @@ -552,12 +556,6 @@ export async function readLockfile(): Promise { } } -async function generateHash(content: string): Promise { - const messageBuffer = new TextEncoder().encode(content); - const hashBuffer = await crypto.subtle.digest("SHA-256", messageBuffer); - return encodeHex(hashBuffer); -} - export async function checkifMetadataUptodate( path: string, requirement: string diff --git a/cli/script.ts b/cli/script.ts index a0cada5706..9148fb4d97 100644 --- a/cli/script.ts +++ b/cli/script.ts @@ -7,7 +7,6 @@ import { Confirm, JobService, log, - NewScript, readAll, Script, ScriptService, @@ -29,6 +28,7 @@ import { } from "./script_common.ts"; import { elementsToMap, + findCodebase, readDirRecursiveWithIgnore, yamlOptions, } from "./sync.ts"; @@ -39,6 +39,7 @@ import { mergeConfigWithConfigFile, readConfigFile, } from "./conf.ts"; +import { SyncCodebase, listSyncCodebases } from "./codebase.ts"; export interface ScriptFile { parent_hash?: string; @@ -71,8 +72,18 @@ async function push(opts: PushOptions, filePath: string) { } await requireLogin(opts); - const globalDeps = await findGlobalDeps(); - await handleFile(filePath, workspace, [], undefined, opts, globalDeps); + const codebases = await listSyncCodebases(opts as SyncOptions); + + const globalDeps = await findGlobalDeps(codebases); + await handleFile( + filePath, + workspace, + [], + undefined, + opts, + globalDeps, + codebases + ); log.info(colors.bold.underline.green(`Script ${filePath} pushed`)); } @@ -81,7 +92,9 @@ export async function handleScriptMetadata( workspace: Workspace, alreadySynced: string[], message: string | undefined, - globalDeps: GlobalDeps + globalDeps: GlobalDeps, + codebases: SyncCodebase[], + opts: GlobalOptions ): Promise { if ( path.endsWith(".script.json") || @@ -94,8 +107,9 @@ export async function handleScriptMetadata( workspace, alreadySynced, message, - undefined, - globalDeps + opts, + globalDeps, + codebases ); } else { return false; @@ -108,7 +122,8 @@ export async function handleFile( alreadySynced: string[], message: string | undefined, opts: (GlobalOptions & { defaultTs?: "bun" | "deno" }) | undefined, - globalDeps: GlobalDeps + globalDeps: GlobalDeps, + codebases: SyncCodebase[] ): Promise { if ( !path.includes(".inline_script.") && @@ -123,10 +138,33 @@ export async function handleFile( const remotePath = path .substring(0, path.indexOf(".")) .replaceAll("\\", "/"); + + const codebase = findCodebase(path, codebases); + + let bundleContent: string | undefined = undefined; + if (codebase) { + const esbuild = await import("npm:esbuild"); + log.info(`Starting building the bundle for ${path}`); + const out = await esbuild.build({ + entryPoints: [path], + format: "esm", + bundle: true, + write: false, + }); + bundleContent = out.outputFiles[0].text; + log.info(`Finished building the bundle for ${path}`); + } const typed = ( await parseMetadataFile( remotePath, - opts ? { ...opts, path, workspaceRemote: workspace } : undefined, + opts + ? { + ...opts, + path, + workspaceRemote: workspace, + schemaOnly: codebase ? true : undefined, + } + : undefined, globalDeps ) )?.payload; @@ -146,6 +184,29 @@ export async function handleFile( } const content = await Deno.readTextFile(path); + const requestBodyCommon = { + content, + description: typed?.description ?? "", + language: language, + path: remotePath.replaceAll("\\", "/"), + summary: typed?.summary ?? "", + kind: typed?.kind, + lock: typed?.lock, + schema: typed?.schema, + tag: typed?.tag, + ws_error_handler_muted: typed?.ws_error_handler_muted, + dedicated_worker: typed?.dedicated_worker, + cache_ttl: typed?.cache_ttl, + concurrency_time_window_s: typed?.concurrency_time_window_s, + concurrent_limit: typed?.concurrent_limit, + deployment_message: message, + restart_unless_cancelled: typed?.restart_unless_cancelled, + visible_to_runner_only: typed?.visible_to_runner_only, + no_main_func: typed?.no_main_func, + priority: typed?.priority, + codebase: codebase?.digest, + }; + if (remote) { if (content === remote.content) { if ( @@ -182,67 +243,85 @@ export async function handleFile( log.info( colors.yellow.bold(`Creating script with a parent ${remotePath}`) ); - await ScriptService.createScript({ - workspace: workspaceId, - requestBody: { - content, - description: typed?.description ?? "", - language: language, - path: remotePath.replaceAll("\\", "/"), - summary: typed?.summary ?? "", - kind: typed?.kind, - lock: typed?.lock, - parent_hash: remote.hash, - schema: typed?.schema, - tag: typed?.tag, - ws_error_handler_muted: typed?.ws_error_handler_muted, - dedicated_worker: typed?.dedicated_worker, - cache_ttl: typed?.cache_ttl, - concurrency_time_window_s: typed?.concurrency_time_window_s, - concurrent_limit: typed?.concurrent_limit, - deployment_message: message, - restart_unless_cancelled: typed?.restart_unless_cancelled, - visible_to_runner_only: typed?.visible_to_runner_only, - no_main_func: typed?.no_main_func, - priority: typed?.priority, - }, - }); + const body = { + ...requestBodyCommon, + parent_hash: remote.hash, + }; + await createScript(bundleContent, workspaceId, body, workspace); } else { log.info( colors.yellow.bold(`Creating script without parent ${remotePath}`) ); - // no parent hash - await ScriptService.createScript({ - workspace: workspaceId, - requestBody: { - content, - description: typed?.description ?? "", - language: language, - path: remotePath.replaceAll("\\", "/"), - summary: typed?.summary ?? "", - kind: typed?.kind, - lock: typed?.lock, - parent_hash: undefined, - schema: typed?.schema, - tag: typed?.tag, - ws_error_handler_muted: typed?.ws_error_handler_muted, - dedicated_worker: typed?.dedicated_worker, - cache_ttl: typed?.cache_ttl, - concurrency_time_window_s: typed?.concurrency_time_window_s, - concurrent_limit: typed?.concurrent_limit, - deployment_message: message, - restart_unless_cancelled: typed?.restart_unless_cancelled, - visible_to_runner_only: typed?.visible_to_runner_only, - no_main_func: typed?.no_main_func, - priority: typed?.priority, - }, - }); + + const body = { + ...requestBodyCommon, + parent_hash: undefined, + }; + await createScript(bundleContent, workspaceId, body, workspace); } return true; } return false; } +async function createScript( + bundleContent: string | undefined, + workspaceId: string, + body: { + parent_hash: string | undefined; + content: string; + description: any; + language: ScriptLanguage; + path: string; + summary: any; + kind: any; + lock: any; + schema: any; + tag: any; + ws_error_handler_muted: any; + dedicated_worker: any; + cache_ttl: any; + concurrency_time_window_s: any; + concurrent_limit: any; + deployment_message: string | undefined; + restart_unless_cancelled: any; + visible_to_runner_only: any; + no_main_func: any; + priority: any; + }, + workspace: Workspace +) { + if (!bundleContent) { + // no parent hash + await ScriptService.createScript({ + workspace: workspaceId, + requestBody: body, + }); + } else { + const form = new FormData(); + form.append("script", JSON.stringify(body)); + form.append("file", bundleContent); + + const url = + workspace.remote + + "api/w/" + + workspace.workspaceId + + "/scripts/create_snapshot"; + const req = await fetch(url, { + method: "POST", + headers: { Authorization: `Bearer ${workspace.token}` }, + body: form, + }); + if (req.status != 201) { + throw Error( + `Script snapshot creation was not successful: ${req.status} - ${ + req.statusText + } - ${await req.text()}` + ); + } + } +} + export async function findContentFile(filePath: string) { const candidates = filePath.endsWith("script.json") ? exts.map((x) => filePath.replace(".script.json", x)) @@ -588,10 +667,12 @@ export type GlobalDeps = { pkgs: Record; reqs: Record; }; -export async function findGlobalDeps(): Promise { +export async function findGlobalDeps( + codebases: SyncCodebase[] +): Promise { const pkgs: { [key: string]: string } = {}; const reqs: { [key: string]: string } = {}; - const els = await FSFSElement(Deno.cwd()); + const els = await FSFSElement(Deno.cwd(), codebases); for await (const entry of readDirRecursiveWithIgnore((p, isDir) => { p = "/" + p; return ( @@ -614,7 +695,7 @@ async function generateMetadata( schemaOnly?: boolean; yes?: boolean; } & SyncOptions, - scriptPath?: string + scriptPath: string | undefined ) { if (scriptPath == "") { scriptPath = undefined; @@ -626,8 +707,9 @@ async function generateMetadata( const workspace = await resolveWorkspace(opts); await requireLogin(opts); opts = await mergeConfigWithConfigFile(opts); + const codebases = await listSyncCodebases(opts); - const globalDeps = await findGlobalDeps(); + const globalDeps = await findGlobalDeps(codebases); if (scriptPath) { // read script metadata file await generateMetadataInternal( @@ -641,7 +723,7 @@ async function generateMetadata( } else { const ignore = await ignoreF(opts); const elems = await elementsToMap( - await FSFSElement(Deno.cwd()), + await FSFSElement(Deno.cwd(), codebases), (p, isD) => { return ( (!isD && !exts.some((ext) => p.endsWith(ext))) || diff --git a/cli/sync.ts b/cli/sync.ts index 08f72f283d..3dd677c355 100644 --- a/cli/sync.ts +++ b/cli/sync.ts @@ -43,8 +43,9 @@ import { import { handleFile } from "./script.ts"; import { deepEqual } from "./utils.ts"; -import { SyncOptions, mergeConfigWithConfigFile } from "./conf.ts"; +import { Codebase, SyncOptions, mergeConfigWithConfigFile } from "./conf.ts"; import { removePathPrefix } from "./types.ts"; +import { SyncCodebase, listSyncCodebases } from "./codebase.ts"; type DynFSElement = { isDirectory: boolean; @@ -54,8 +55,53 @@ type DynFSElement = { getChildren(): AsyncIterable; }; -export async function FSFSElement(p: string): Promise { - function _internal_element(localP: string, isDir: boolean): DynFSElement { +export function findCodebase( + path: string, + codebases: SyncCodebase[] +): SyncCodebase | undefined { + for (const c of codebases) { + if (typeof c.includes == "string") { + c.includes = [c.includes]; + } + for (const r of c.includes) { + if (minimatch(path, r)) { + return c; + } + } + } +} +function addCodebaseDigestIfRelevant( + path: string, + content: string, + codebases: SyncCodebase[] +): string { + const isScript = path.endsWith(".script.yaml"); + if (isScript) { + const c = findCodebase(path, codebases); + if (c) { + const parsed: any = yamlParse(content); + if (parsed && typeof parsed == "object") { + parsed["codebase"] = c.digest; + return yamlStringify(parsed, yamlOptions); + } else { + throw Error( + `Expected local yaml ${path} to be an object, found: ${content} instead` + ); + } + } + } + return content; +} + +export async function FSFSElement( + p: string, + codebases: SyncCodebase[] +): Promise { + function _internal_element( + localP: string, + isDir: boolean, + codebases: SyncCodebase[] + ): DynFSElement { return { isDirectory: isDir, path: localP.substring(p.length + 1), @@ -63,7 +109,11 @@ export async function FSFSElement(p: string): Promise { if (!isDir) return []; try { for await (const e of Deno.readDir(localP)) { - yield _internal_element(path.join(localP, e.name), e.isDirectory); + yield _internal_element( + path.join(localP, e.name), + e.isDirectory, + codebases + ); } } catch (e) { log.warning(`Error reading dir: ${localP}, ${e}`); @@ -74,11 +124,12 @@ export async function FSFSElement(p: string): Promise { // }, async getContentText(): Promise { const content = await Deno.readTextFile(localP); - return content; + + return addCodebaseDigestIfRelevant(localP, content, codebases); }, }; } - return _internal_element(p, (await Deno.stat(p)).isDirectory); + return _internal_element(p, (await Deno.stat(p)).isDirectory, codebases); } function prioritizeName(name: string): string { @@ -727,6 +778,8 @@ async function pull(opts: GlobalOptions & SyncOptions) { const workspace = await resolveWorkspace(opts); await requireLogin(opts); + const codebases = await listSyncCodebases(opts); + log.info( colors.gray( "Computing the files to update locally to match remote (taking wmill.yaml into account)" @@ -748,8 +801,8 @@ async function pull(opts: GlobalOptions & SyncOptions) { !opts.json ); const local = !opts.stateful - ? await FSFSElement(Deno.cwd()) - : await FSFSElement(path.join(Deno.cwd(), ".wmill")); + ? await FSFSElement(Deno.cwd(), codebases) + : await FSFSElement(path.join(Deno.cwd(), ".wmill"), []); const changes = await compareDynFSElement( remote, local, @@ -924,6 +977,7 @@ function removeSuffix(str: string, suffix: string) { async function push(opts: GlobalOptions & SyncOptions) { opts = await mergeConfigWithConfigFile(opts); + const codebases = await listSyncCodebases(opts); if (opts.raw) { log.info("--raw is now the default, you can remove it as a flag"); } @@ -962,7 +1016,7 @@ async function push(opts: GlobalOptions & SyncOptions) { !opts.json ); - const local = await FSFSElement(path.join(Deno.cwd(), "")); + const local = await FSFSElement(path.join(Deno.cwd(), ""), codebases); const changes = await compareDynFSElement( local, remote, @@ -991,10 +1045,11 @@ async function push(opts: GlobalOptions & SyncOptions) { ) { return; } + log.info(colors.gray(`Applying changes to files ...`)); const alreadySynced: string[] = []; - const globalDeps = await findGlobalDeps(); + const globalDeps = await findGlobalDeps(codebases); for await (const change of changes) { const stateTarget = path.join(Deno.cwd(), ".wmill", change.path); @@ -1012,7 +1067,9 @@ async function push(opts: GlobalOptions & SyncOptions) { workspace, alreadySynced, opts.message, - globalDeps + globalDeps, + codebases, + opts ) ) { if (opts.stateful && stateExists) { @@ -1026,7 +1083,8 @@ async function push(opts: GlobalOptions & SyncOptions) { alreadySynced, opts.message, opts, - globalDeps + globalDeps, + codebases ) ) { if (opts.stateful && stateExists) { @@ -1067,7 +1125,8 @@ async function push(opts: GlobalOptions & SyncOptions) { alreadySynced, opts.message, opts, - globalDeps + globalDeps, + codebases ) ) { continue; diff --git a/cli/utils.ts b/cli/utils.ts index 080337d39d..185fe50144 100644 --- a/cli/utils.ts +++ b/cli/utils.ts @@ -2,7 +2,7 @@ // eslint-disable-next-line @typescript-eslint/ban-ts-comment // @ts-nocheck This file is copied from a JS project, so it's not type-safe. -import { log } from "./deps.ts"; +import { log, encodeHex } from "./deps.ts"; export function deepEqual(a: T, b: T): boolean { if (a === b) return true; @@ -92,3 +92,28 @@ export function getHeaders(): Record | undefined { return undefined; } } + +export async function digestDir(path: string) { + const hashes: string = []; + for await (const e of Deno.readDir(path)) { + const npath = path + "/" + e.name; + if (e.isFile) { + hashes.push(await generateHashFromBuffer(await Deno.readFile(npath))); + } else if (e.isDirectory && !e.isSymlink) { + hashes.push(await digestDir(npath)); + } + } + return await generateHash(hashes.join("")); +} + +export async function generateHash(content: string): Promise { + const messageBuffer = new TextEncoder().encode(content); + return await generateHashFromBuffer(messageBuffer); +} + +export async function generateHashFromBuffer( + content: BufferSource +): Promise { + const hashBuffer = await crypto.subtle.digest("SHA-256", content); + return encodeHex(hashBuffer); +} diff --git a/frontend/src/lib/components/ArgInput.svelte b/frontend/src/lib/components/ArgInput.svelte index e4ec4165d1..d7be76dab6 100644 --- a/frontend/src/lib/components/ArgInput.svelte +++ b/frontend/src/lib/components/ArgInput.svelte @@ -550,6 +550,7 @@ {#if properties && Object.keys(properties).length > 0}
Now +
diff --git a/frontend/src/lib/components/Dev.svelte b/frontend/src/lib/components/Dev.svelte index b568dd37a1..7a6d9103c0 100644 --- a/frontend/src/lib/components/Dev.svelte +++ b/frontend/src/lib/components/Dev.svelte @@ -159,6 +159,8 @@ path: string language: Preview['language'] lock?: string + isCodebase?: boolean + tag?: string } let currentScript: LastEditScript | undefined = undefined @@ -177,13 +179,25 @@ let lockChanges = false let timeout: NodeJS.Timeout | undefined = undefined + let loadingCodebaseButton = false + let lastBundleCommandId = '' + const el = (event) => { // sendUserToast(`Received message from parent ${event.data.type}`, true) if (event.data.type == 'runTest') { runTest() + event.preventDefault() } else if (event.data.type == 'replaceScript') { mode = 'script' replaceScript(event.data) + } else if (event.data.type == 'testBundle') { + if (event.data.id == lastBundleCommandId) { + testBundle(event.data.file) + } else { + sendUserToast(`Bundle received ${lastBundleCommandId} was obsolete, ignoring`, true) + } + } else if (event.data.type == 'testBundleError') { + sendUserToast(event.data.error, true) } else if (event.data.type == 'replaceFlow') { mode = 'flow' lockChanges = true @@ -223,6 +237,47 @@ window.parent?.postMessage({ type: 'refresh' }, '*') }) + async function testBundle(file: string) { + testJobLoader?.abstractRun(async () => { + try { + const form = new FormData() + form.append( + 'preview', + JSON.stringify({ + content: currentScript?.content, + kind: 'bundle', + path: currentScript?.path, + args, + language: currentScript?.language, + tag: currentScript?.tag + }) + ) + form.append('file', file) + + const url = '/api/w/' + workspace + '/jobs/run/preview_bundle' + + const req = await fetch(url, { + method: 'POST', + body: form, + headers: { + Authorization: 'Bearer ' + token + } + }) + if (req.status != 201) { + throw Error( + `Script snapshot creation was not successful: ${req.status} - ${ + req.statusText + } - ${await req.text()}` + ) + } + return await req.text() + } catch (e) { + sendUserToast(`Failed to send bundle ${e}`, true) + throw Error(e) + } + }) + loadingCodebaseButton = false + } onDestroy(() => { window.removeEventListener('message', el) if (socket && socket.readyState === WebSocket.OPEN) { @@ -261,15 +316,21 @@ if (!currentScript) { return } - //@ts-ignore - testJobLoader.runPreview( - currentScript.path, - currentScript.content, - currentScript.language, - args, - undefined, - useLock ? currentScript.lock : undefined - ) + if (currentScript.isCodebase) { + loadingCodebaseButton = true + lastBundleCommandId = Math.random().toString(36).substring(7) + window.parent?.postMessage({ type: 'testBundle', id: lastBundleCommandId }, '*') + } else { + //@ts-ignore + testJobLoader.runPreview( + currentScript.path, + currentScript.content, + currentScript.language, + args, + currentScript.tag, + useLock ? currentScript.lock : undefined + ) + } } else { flowPreviewButtons?.openPreview() } @@ -288,10 +349,11 @@ }) } - function onKeyDown(event: KeyboardEvent) { - if ((event.ctrlKey || event.metaKey) && event.key == 'Enter') { - event.preventDefault() - runTest() + async function onKeyDown(event: KeyboardEvent) { + if ((event.ctrlKey || event.metaKey) && event.code === 'KeyC') { + document.execCommand('copy') + } else if ((event.ctrlKey || event.metaKey) && event.code === 'KeyX') { + document.execCommand('cut') } } @@ -445,7 +507,9 @@
{currentScript?.path ?? 'Not editing a script'} - {currentScript?.language ?? ''} + {currentScript?.language ?? ''}{currentScript?.isCodebase ? ' (codebase)' : ''}
@@ -482,7 +546,7 @@ +
+ {/each} + {:else if value != undefined} + List is not an array + {/if} + +
+ +
+ + {(value ?? []).length} item{(value ?? []).length != 1 ? 's' : ''} + + {/if} + + {:else if inputCat == 'resource-object'} + + {:else if inputCat == 'object'} + {#if properties && Object.keys(properties).length > 0} +
+
{:else} -
- {#if Array.isArray(value)} - {#each value ?? [] as v, i} -
- {#if itemsType?.type == 'number'} - - {:else if itemsType?.type == 'string' && itemsType?.contentEncoding == 'base64'} - fileChanged(x, (val) => (value[i] = val))} - multiple={false} - /> - {:else if Array.isArray(itemsType?.enum)} - - {:else} - - {/if} - -
- {/each} - {:else if value != undefined} - List is not an array - {/if} -
-
- -
- - {(value ?? []).length} item{(value ?? []).length != 1 ? 's' : ''} - - {/if} - - {:else if inputCat == 'resource-object'} - - {:else if inputCat == 'object'} - {#if properties && Object.keys(properties).length > 0} -
- { + dispatch('focus') + }} + use:autosize + style="min-height: 5px;" + class="col-span-10 {valid && error == '' + ? '' + : 'border !border-red-700 !border-opacity-70 focus:!border-red-700 focus:!border-opacity-30'}" + placeholder={defaultValue ? JSON.stringify(defaultValue, null, 4) : ''} + bind:value={rawValue} /> -
- {:else} -