feat: end user email env var

backend/ee-repo-ref.txt

@@ -1 +1 @@
-e4e3056eeaf12d7fe71579c0d1f6aee828fe5ea7
+33bdef405c678616b12084cf779a68a62d1f477e

backend/migrations/20251003145612_add_end_user_to_job_perms.down.sql

@@ -0,0 +1,2 @@
+-- Add down migration script here
+ALTER TABLE job_perms DROP COLUMN end_user_email;

backend/migrations/20251003145612_add_end_user_to_job_perms.up.sql

@@ -0,0 +1,2 @@
+-- Add up migration script here
+ALTER TABLE job_perms ADD COLUMN end_user_email VARCHAR(255);

backend/tests/common/mod.rs

@@ -156,6 +156,7 @@ impl RunJob {
             None,
             None,
             false,
+            None,
         )
         .await
         .expect("push has to succeed");

backend/windmill-api/src/apps.rs

@@ -1243,6 +1243,7 @@ async fn create_app_internal<'a>(
         None,
         Some(&authed.clone().into()),
         false,
+        None,
     )
     .await?;
     tracing::info!("Pushed app dependency job {}", dependency_job_uuid);
@@ -1622,6 +1623,7 @@ async fn update_app_internal<'a>(
         None,
         Some(&authed.clone().into()),
         false,
+        None,
     )
     .await?;
     tracing::info!("Pushed app dependency job {}", dependency_job_uuid);
@@ -1908,6 +1910,8 @@ async fn execute_component(
         (email.as_str(), permissioned_as)
     };
 
+    let end_user_email = opt_authed.as_ref().map(|a| a.email.clone());
+
     let (uuid, tx) = push(
         &db,
         tx,
@@ -1937,6 +1941,7 @@ async fn execute_component(
         None,
         None,
         false,
+        end_user_email,
     )
     .await?;
     tx.commit().await?;

backend/windmill-api/src/flows.rs

@@ -545,6 +545,7 @@ async fn create_flow(
         None,
         Some(&authed.clone().into()),
         false,
+        None,
     )
     .await?;
 
@@ -1012,6 +1013,7 @@ async fn update_flow(
         None,
         Some(&authed.clone().into()),
         false,
+        None,
     )
     .await?;
     sqlx::query!(

backend/windmill-api/src/jobs.rs

@@ -4103,6 +4103,7 @@ pub async fn run_flow_by_path_inner(
         None,
         push_authed.as_ref(),
         false,
+        None,
     )
     .await?;
 
@@ -4219,6 +4220,7 @@ pub async fn restart_flow(
         completed_job.priority,
         Some(&authed.clone().into()),
         false,
+        None,
     )
     .await?;
     tx.commit().await?;
@@ -4321,6 +4323,7 @@ pub async fn run_script_by_path_inner(
         None,
         push_authed.as_ref(),
         false,
+        None,
     )
     .await?;
     tx.commit().await?;
@@ -4473,6 +4476,7 @@ pub async fn run_workflow_as_code(
         None,
         push_authed.as_ref(),
         false,
+        None,
     )
     .await?;
 
@@ -5016,6 +5020,7 @@ pub async fn run_wait_result_job_by_path_get(
         None,
         push_authed.as_ref(),
         false,
+        None,
     )
     .await?;
     tx.commit().await?;
@@ -5168,6 +5173,7 @@ pub async fn run_wait_result_script_by_path_internal(
         None,
         push_authed.as_ref(),
         false,
+        None,
     )
     .await?;
     tx.commit().await?;
@@ -5284,6 +5290,7 @@ pub async fn run_wait_result_script_by_hash(
         None,
         push_authed.as_ref(),
         false,
+        None,
     )
     .await?;
     tx.commit().await?;
@@ -5595,6 +5602,7 @@ pub async fn run_wait_result_flow_by_path_internal(
         None,
         push_authed.as_ref(),
         false,
+        None,
     )
     .await?;
 
@@ -5686,6 +5694,7 @@ async fn run_preview_script(
         None,
         Some(&authed.clone().into()),
         false,
+        None,
     )
     .await?;
     tx.commit().await?;
@@ -5802,6 +5811,7 @@ async fn run_bundle_preview_script(
                 None,
                 Some(&authed.clone().into()),
                 false,
+                None,
             )
             .await?;
             job_id = Some(uuid);
@@ -5939,6 +5949,7 @@ async fn run_dependencies_job(
         None,
         Some(&authed.clone().into()),
         false,
+        None,
     )
     .await?;
     tx.commit().await?;
@@ -6006,6 +6017,7 @@ async fn run_flow_dependencies_job(
         None,
         Some(&authed.clone().into()),
         false,
+        None,
     )
     .await?;
     tx.commit().await?;
@@ -6349,6 +6361,7 @@ async fn run_preview_flow_job(
         None,
         Some(&authed.clone().into()),
         false,
+        None,
     )
     .await?;
     tx.commit().await?;
@@ -6522,6 +6535,7 @@ async fn run_dynamic_select(
         None,
         Some(&authed.clone().into()),
         false,
+        None,
     )
     .await?;
     tx.commit().await?;
@@ -6649,6 +6663,7 @@ pub async fn run_job_by_hash_inner(
         None,
         push_authed.as_ref(),
         false,
+        None,
     )
     .await?;
     tx.commit().await?;

backend/windmill-api/src/resources.rs

@@ -515,7 +515,8 @@ pub async fn transform_json_value<'c>(
     match v {
         Value::String(y) if y.starts_with("$var:") => {
             let path = y.strip_prefix("$var:").unwrap();
-            let userdb_authed = UserDbWithOptAuthed { authed: authed, user_db: user_db.clone(), db: db.clone() };
+            let userdb_authed =
+                UserDbWithOptAuthed { authed: authed, user_db: user_db.clone(), db: db.clone() };
 
             let v = crate::variables::get_value_internal(
                 &userdb_authed,
@@ -616,6 +617,7 @@ pub async fn transform_json_value<'c>(
                 job.root_job.map(|x| x.to_string()),
                 Some(job.scheduled_for.clone()),
                 None,
+                None,
             )
             .await;
 

backend/windmill-api/src/scripts.rs

@@ -41,7 +41,12 @@ 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, s3_helpers::upload_artifact_to_store, 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::{
@@ -418,7 +423,12 @@ async fn create_snapshot_script(
             uploaded = true;
 
             let path = windmill_common::s3_helpers::bundle(&w_id, &hash);
-            upload_artifact_to_store(&path, data, &windmill_common::worker::ROOT_STANDALONE_BUNDLE_DIR).await?;
+            upload_artifact_to_store(
+                &path,
+                data,
+                &windmill_common::worker::ROOT_STANDALONE_BUNDLE_DIR,
+            )
+            .await?;
         }
         // println!("Length of `{}` is {} bytes", name, data.len());
     }
@@ -438,7 +448,6 @@ 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>,
@@ -980,6 +989,7 @@ async fn create_script_internal<'c>(
             None,
             Some(&authed.clone().into()),
             false,
+            None,
         )
         .await?;
         Ok((hash, new_tx, None))

backend/windmill-api/src/triggers/trigger_helpers.rs

@@ -684,7 +684,8 @@ pub async fn trigger_runnable_and_wait_for_raw_result_with_error_ctx(
         error_handler_path,
         error_handler_args,
         trigger_path,
-    ).await?;
+    )
+    .await?;
 
     if !success {
         Err(windmill_common::error::Error::internal_err(format!(
@@ -857,6 +858,7 @@ async fn trigger_script_with_retry_and_error_handler(
         None,
         push_authed.as_ref(),
         false,
+        None,
     )
     .await?;
     tx.commit().await?;

backend/windmill-api/src/variables.rs

@@ -83,6 +83,7 @@ async fn list_contextual_variables(
             Some("017e0ad5-f499-73b6-5488-92a61c5196dd".to_string()),
             Some(chrono::offset::Utc::now()),
             Some(ScriptHash(1234567890)),
+            None,
         )
         .await
         .to_vec(),
@@ -843,11 +844,11 @@ pub async fn get_value_internal<'a, 'e, A: sqlx::Acquire<'e, Database = Postgres
         } else if !value.is_empty() {
             let mc = build_crypt(&db, &w_id).await?;
             decrypt(&mc, value).map_err(|e| {
-                    Error::internal_err(format!(
-                        "Error decrypting variable {}: {}",
-                        variable.path, e
-                    ))
-                })?
+                Error::internal_err(format!(
+                    "Error decrypting variable {}: {}",
+                    variable.path, e
+                ))
+            })?
         } else {
             "".to_string()
         }
@@ -884,11 +885,8 @@ pub async fn get_variable_or_self(path: String, db: &DB, w_id: &str) -> Result<S
         if record.is_secret {
             let mc = build_crypt(db, w_id).await?;
             value = decrypt(&mc, value).map_err(|e| {
-                    Error::internal_err(format!(
-                        "Error decrypting variable {}: {}",
-                        path, e
-                    ))
-                })?;
+                Error::internal_err(format!("Error decrypting variable {}: {}", path, e))
+            })?;
         }
 
         Ok(value)

backend/windmill-common/src/variables.rs

@@ -165,11 +165,11 @@ pub async fn get_secret_value_as_admin(
         if !value.is_empty() {
             let mc = build_crypt(db, w_id).await?;
             decrypt(&mc, value).map_err(|e| {
-                    crate::error::Error::internal_err(format!(
-                        "Error decrypting variable {}: {}",
-                        variable.path, e
-                    ))
-                })?
+                crate::error::Error::internal_err(format!(
+                    "Error decrypting variable {}: {}",
+                    variable.path, e
+                ))
+            })?
         } else {
             "".to_string()
         }
@@ -219,6 +219,7 @@ pub async fn get_reserved_variables(
     root_job_id: Option<String>,
     scheduled_for: Option<chrono::DateTime<Utc>>,
     runnable_id: Option<ScriptHash>,
+    end_user_email: Option<String>,
 ) -> Vec<ContextualVariable> {
     let state_path = {
         let trigger = if schedule_path.is_some() {
@@ -389,6 +390,12 @@ pub async fn get_reserved_variables(
         description: "Hash of the script. Useful as cache key for cache that should be runnable specific.".to_string(),
         is_custom: false,
     },
+    ContextualVariable {
+        name: "WM_END_USER_EMAIL".to_string(),
+        value: end_user_email.unwrap_or_else(|| "".to_string()),
+        description: "Email of the end user that executed the current script. Only available when triggered from an app.".to_string(),
+        is_custom: false,
+    },
 ].into_iter().chain(custom_envs.into_iter().map(|(name, value)| ContextualVariable {
     name,
     value,

backend/windmill-common/src/worker.rs

@@ -395,7 +395,7 @@ fn format_pull_query(peek: String) -> String {
             j.timeout, j.flow_step_id, j.cache_ttl, j.priority, j.raw_code, j.raw_lock, j.raw_flow,
             j.script_entrypoint_override, j.preprocessed, pj.runnable_path as parent_runnable_path,
             COALESCE(p.email, j.permissioned_as_email) as permissioned_as_email, p.username as permissioned_as_username, p.is_admin as permissioned_as_is_admin,
-            p.is_operator as permissioned_as_is_operator, p.groups as permissioned_as_groups, p.folders as permissioned_as_folders
+            p.is_operator as permissioned_as_is_operator, p.groups as permissioned_as_groups, p.folders as permissioned_as_folders, p.end_user_email as permissioned_as_end_user_email
         FROM q, j
             LEFT JOIN v2_job_status f USING (id)
             LEFT JOIN job_perms p ON p.job_id = j.id