feat: support esm mode for codebase bundles (#6709)

* rawAppsS3

* make fn common

* all

* merge

* nit

* fix ingress

* all

* all

* all

* all

Author: rubenfiszel

backend/windmill-api/src/jobs.rs

@@ -18,6 +18,7 @@ use quick_cache::sync::Cache;
 use serde_json::value::RawValue;
 use serde_json::Value;
 use sqlx::Pool;
+use windmill_common::s3_helpers::{upload_artifact_to_store, BundleFormat};
 use std::collections::HashMap;
 use std::hash::{DefaultHasher, Hash, Hasher};
 use std::ops::{Deref, DerefMut};
@@ -41,7 +42,6 @@ use windmill_common::DYNAMIC_INPUT_CACHE;
 #[cfg(all(feature = "enterprise", feature = "smtp"))]
 use windmill_common::{email_oss::send_email_html, server::load_smtp_config};
 
-use windmill_common::scripts::PREVIEW_IS_CODEBASE_HASH;
 use windmill_common::variables::get_workspace_key;
 
 use crate::triggers::trigger_helpers::ScriptId;
@@ -3526,6 +3526,7 @@ struct Preview {
     tag: Option<String>,
     dedicated_worker: Option<bool>,
     lock: Option<String>,
+    format: Option<String>
 }
 
 #[derive(Deserialize)]
@@ -5590,6 +5591,7 @@ async fn run_wait_result_preview_script(
     return result;
 }
 
+
 async fn run_bundle_preview_script(
     authed: ApiAuthed,
     Extension(db): Extension<DB>,
@@ -5598,7 +5600,6 @@ async fn run_bundle_preview_script(
     Query(run_query): Query<RunJobQuery>,
     mut multipart: axum::extract::Multipart,
 ) -> error::Result<(StatusCode, String)> {
-    use windmill_common::scripts::PREVIEW_IS_TAR_CODEBASE_HASH;
 
     if authed.is_operator {
         return Err(error::Error::NotAuthorized(
@@ -5610,13 +5611,15 @@ async fn run_bundle_preview_script(
     let mut tx = None;
     let mut uploaded = false;
     let mut is_tar = false;
+    let mut format = BundleFormat::Cjs;
 
     while let Some(field) = multipart.next_field().await.unwrap() {
         let name = field.name().unwrap().to_string();
         let data = field.bytes().await;
         let data = data.map_err(to_anyhow)?;
         if name == "preview" {
             let preview: Preview = serde_json::from_slice(&data).map_err(to_anyhow)?;
+            format = preview.format.and_then(|s| BundleFormat::from_string(&s)).unwrap_or(BundleFormat::Cjs);
 
             let scheduled_for = run_query.get_scheduled_for(&db).await?;
             let tag = run_query.tag.clone().or(preview.tag.clone());
@@ -5637,11 +5640,7 @@ async fn run_bundle_preview_script(
                 ltx,
                 &w_id,
                 JobPayload::Code(RawCode {
-                    hash: if is_tar {
-                        Some(PREVIEW_IS_TAR_CODEBASE_HASH)
-                    } else {
-                        Some(PREVIEW_IS_CODEBASE_HASH)
-                    },
+                    hash: Some(windmill_common::scripts::codebase_to_hash(is_tar, format == BundleFormat::Esm)),
                     content: preview.content.unwrap_or_default(),
                     path: preview.path,
                     language: preview.language.unwrap_or(ScriptLang::Deno),
@@ -5690,52 +5689,17 @@ async fn run_bundle_preview_script(
 
             // tracing::info!("is_tar 2: {is_tar}");
 
+            if format == BundleFormat::Esm {
+                id = format!("{}.esm", id);
+            }
             if is_tar {
                 id = format!("{}.tar", id);
             }
 
             uploaded = true;
 
-            #[cfg(all(feature = "enterprise", feature = "parquet"))]
-            let object_store = windmill_common::s3_helpers::get_object_store().await;
-
-            #[cfg(not(all(feature = "enterprise", feature = "parquet")))]
-            let object_store: Option<()> = None;
-
-            if &windmill_common::utils::MODE_AND_ADDONS.mode
-                == &windmill_common::utils::Mode::Standalone
-                && object_store.is_none()
-            {
-                std::fs::create_dir_all(
-                    windmill_common::worker::ROOT_STANDALONE_BUNDLE_DIR.clone(),
-                )?;
-                windmill_common::worker::write_file_bytes(
-                    &windmill_common::worker::ROOT_STANDALONE_BUNDLE_DIR,
-                    &id,
-                    &data,
-                )?;
-            } else {
-                #[cfg(not(all(feature = "enterprise", feature = "parquet")))]
-                {
-                    return Err(Error::ExecutionErr("codebase is an EE feature".to_string()));
-                }
-
-                #[cfg(all(feature = "enterprise", feature = "parquet"))]
-                if let Some(os) = object_store {
-                    check_license_key_valid().await?;
-
-                    let path = windmill_common::s3_helpers::bundle(&w_id, &id);
-                    if let Err(e) = os
-                        .put(&object_store::path::Path::from(path.clone()), data.into())
-                        .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()));
-                }
-            }
+            let path = windmill_common::s3_helpers::bundle(&w_id, &id);
+            upload_artifact_to_store(&path, data, &windmill_common::worker::ROOT_STANDALONE_BUNDLE_DIR).await?;
         }
         // println!("Length of `{}` is {} bytes", name, data.len());
     }

backend/windmill-api/src/scripts.rs

@@ -41,11 +41,7 @@ use windmill_audit::ActionKind;
 use windmill_worker::process_relative_imports;
 
 use windmill_common::{
-    assets::{clear_asset_usage, insert_asset_usage, AssetUsageKind, AssetWithAltAccessType},
-    error::to_anyhow,
-    scripts::hash_script,
-    utils::WarnAfterExt,
-    worker::CLOUD_HOSTED,
+    assets::{clear_asset_usage, insert_asset_usage, AssetUsageKind, AssetWithAltAccessType}, error::to_anyhow, s3_helpers::upload_artifact_to_store, scripts::hash_script, utils::WarnAfterExt, worker::CLOUD_HOSTED
 };
 
 use windmill_common::{
@@ -421,45 +417,8 @@ async fn create_snapshot_script(
 
             uploaded = true;
 
-            #[cfg(all(feature = "enterprise", feature = "parquet"))]
-            let object_store = windmill_common::s3_helpers::get_object_store().await;
-
-            #[cfg(not(all(feature = "enterprise", feature = "parquet")))]
-            let object_store: Option<()> = None;
-
-            if &windmill_common::utils::MODE_AND_ADDONS.mode
-                == &windmill_common::utils::Mode::Standalone
-                && object_store.is_none()
-            {
-                std::fs::create_dir_all(
-                    windmill_common::worker::ROOT_STANDALONE_BUNDLE_DIR.clone(),
-                )?;
-                windmill_common::worker::write_file_bytes(
-                    &windmill_common::worker::ROOT_STANDALONE_BUNDLE_DIR,
-                    &hash,
-                    &data,
-                )?;
-            } else {
-                #[cfg(not(all(feature = "enterprise", feature = "parquet")))]
-                {
-                    return Err(Error::ExecutionErr("codebase is an EE feature".to_string()));
-                }
-
-                #[cfg(all(feature = "enterprise", feature = "parquet"))]
-                if let Some(os) = object_store {
-                    let path = windmill_common::s3_helpers::bundle(&w_id, &hash);
-
-                    if let Err(e) = os
-                        .put(&object_store::path::Path::from(path.clone()), data.into())
-                        .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()));
-                }
-            }
+            let path = windmill_common::s3_helpers::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());
     }
@@ -479,6 +438,7 @@ async fn create_snapshot_script(
     return Ok((StatusCode::CREATED, format!("{}", script_hash.unwrap())));
 }
 
+
 async fn list_paths_from_workspace_runnable(
     authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,

backend/windmill-common/src/s3_helpers.rs

@@ -526,6 +526,67 @@ pub async fn build_object_store_client(
     }
 }
 
+
+#[derive(PartialEq)]
+pub enum BundleFormat {
+    Esm,
+    Cjs,
+}
+
+impl BundleFormat {
+   pub fn from_string(s: &str) -> Option<Self> {
+        match s {
+            "esm" => Some(Self::Esm),
+            "cjs" => Some(Self::Cjs),
+            _ => None,
+        }
+    }
+}
+
+pub async fn upload_artifact_to_store(path: &str, data: bytes::Bytes, standalone_dir: &str) -> error::Result<()> {
+    #[cfg(all(feature = "enterprise", feature = "parquet"))]
+    let object_store = crate::s3_helpers::get_object_store().await;
+    #[cfg(not(all(feature = "enterprise", feature = "parquet")))]
+    let object_store: Option<()> = None;
+    Ok(if &crate::utils::MODE_AND_ADDONS.mode
+        == &crate::utils::Mode::Standalone
+        && object_store.is_none()
+    {
+        let path = format!("{}/{}", standalone_dir, path);
+        tracing::info!("Writing file to path {path}");
+
+        let split_path = path.split("/").collect::<Vec<&str>>();
+        std::fs::create_dir_all(
+            split_path[..split_path.len() - 1].join("/"),
+        )?;
+
+        crate::worker::write_file_bytes(
+            &path,
+            &data,
+        )?;
+    } else {
+        #[cfg(not(all(feature = "enterprise", feature = "parquet")))]
+        {
+            return Err(error::Error::ExecutionErr("codebase is an EE feature".to_string()));
+        }
+
+        #[cfg(all(feature = "enterprise", feature = "parquet"))]
+        if let Some(os) = object_store {
+
+            if let Err(e) = os
+                .put(&object_store::path::Path::from(path), data.into())
+                .await
+            {
+                tracing::info!("Failed to put snapshot to s3 at {path}: {:?}", e);
+                return Err(error::Error::ExecutionErr(format!("Failed to put {path} to s3")));
+            }
+        } else {
+            return Err(error::Error::BadConfig("Object store is required for snapshot script and is not configured for servers".to_string()));
+        }
+    })
+}
+
+
 #[cfg(feature = "parquet")]
 pub async fn attempt_fetch_bytes(
     client: Arc<dyn ObjectStore>,

backend/windmill-common/src/scripts.rs

@@ -211,9 +211,53 @@ impl Display for ScriptKind {
     }
 }
 
-pub const PREVIEW_IS_CODEBASE_HASH: i64 = -42;
-pub const PREVIEW_IS_TAR_CODEBASE_HASH: i64 = -43;
+const PREVIEW_IS_CODEBASE_HASH: i64 = -42;
+const PREVIEW_IS_TAR_CODEBASE_HASH: i64 = -43;
+const PREVIEW_IS_ESM_CODEBASE_HASH: i64 = -44;
+const PREVIEW_IS_TAR_ESM_CODEBASE_HASH: i64 = -45;
 
+pub fn is_special_codebase_hash(hash: i64) -> bool {
+    hash == PREVIEW_IS_CODEBASE_HASH || hash == PREVIEW_IS_TAR_CODEBASE_HASH || hash == PREVIEW_IS_ESM_CODEBASE_HASH || hash == PREVIEW_IS_TAR_ESM_CODEBASE_HASH
+}
+
+pub fn codebase_to_hash(is_tar: bool, is_esm: bool) -> i64 {
+    if is_tar {
+        if is_esm {
+            PREVIEW_IS_TAR_ESM_CODEBASE_HASH
+        } else {
+            PREVIEW_IS_TAR_CODEBASE_HASH
+        }
+    } else {
+        if is_esm {
+            PREVIEW_IS_ESM_CODEBASE_HASH
+        } else {
+            PREVIEW_IS_CODEBASE_HASH
+        }
+    }
+}
+
+
+pub fn hash_to_codebase_id(job_id: &str, hash: i64) -> Option<String> {
+    match hash {
+        PREVIEW_IS_CODEBASE_HASH => Some(job_id.to_string()),
+        PREVIEW_IS_TAR_CODEBASE_HASH => Some(format!("{}.tar", job_id)),
+        PREVIEW_IS_ESM_CODEBASE_HASH => Some(format!("{}.esm", job_id)),
+        PREVIEW_IS_TAR_ESM_CODEBASE_HASH => Some(format!("{}.esm.tar", job_id)),
+        _ => None,
+    }
+}
+
+
+pub struct CodebaseInfo {
+    pub is_tar: bool,
+    pub is_esm: bool,
+}
+
+pub fn id_to_codebase_info(id: &str) -> CodebaseInfo {
+    let is_tar = id.ends_with(".tar");
+    let is_esm = id.contains(".esm");
+    CodebaseInfo { is_tar, is_esm }
+}
 #[derive(Serialize, sqlx::FromRow)]
 pub struct Script {
     pub workspace_id: String,

backend/windmill-common/src/worker.rs

@@ -258,7 +258,7 @@ lazy_static::lazy_static! {
     // Features flags:
     pub static ref DISABLE_FLOW_SCRIPT: bool = std::env::var("DISABLE_FLOW_SCRIPT").ok().is_some_and(|x| x == "1" || x == "true");
 
-    pub static ref ROOT_STANDALONE_BUNDLE_DIR: String = format!("{}/.windmill/standalone_bundle/", std::env::var("HOME").unwrap_or_else(|_| "/root".to_string()));
+    pub static ref ROOT_STANDALONE_BUNDLE_DIR: String = format!("{}/.windmill/standalone_bundle", std::env::var("HOME").unwrap_or_else(|_| "/root".to_string()));
 }
 
 pub const ROOT_CACHE_NOMOUNT_DIR: &str = concatcp!(TMP_DIR, "/cache_nomount/");
@@ -470,9 +470,8 @@ pub fn write_file(dir: &str, path: &str, content: &str) -> error::Result<File> {
     Ok(file)
 }
 
-pub fn write_file_bytes(dir: &str, path: &str, content: &Bytes) -> error::Result<File> {
-    let path = format!("{}/{}", dir, path);
-    let mut file = File::create(&path)?;
+pub fn write_file_bytes(path: &str, content: &Bytes) -> error::Result<File> {
+    let mut file = File::create(path)?;
     file.write_all(content)?;
     file.flush()?;
     Ok(file)