feat: end user email env var (#6750)

* feat: end user email env var

* nit

* nits

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

backend/windmill-queue/src/jobs.rs

@@ -446,6 +446,7 @@ pub async fn push_init_job<'c>(
         None,
         None,
         false,
+        None,
     )
     .await?;
     inner_tx.commit().await?;
@@ -500,6 +501,7 @@ pub async fn push_periodic_bash_job<'c>(
         None,
         None,
         false,
+        None,
     )
     .await?;
     inner_tx.commit().await?;
@@ -1371,6 +1373,7 @@ async fn restart_job_if_perpetual_inner(
             queued_job.priority,
             None,
             false,
+            None,
         )
         .await?;
         tx.commit().await?;
@@ -1858,6 +1861,7 @@ pub async fn push_error_handler<'a, 'c, T: Serialize + Send + Sync>(
         priority,
         None,
         false,
+        None,
     )
     .await?;
     tx.commit().await?;
@@ -1913,6 +1917,7 @@ pub struct MiniPulledJob {
     pub trigger: Option<String>,
     pub trigger_kind: Option<JobTriggerKind>,
     pub visible_to_owner: bool,
+    pub permissioned_as_end_user_email: Option<String>,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
@@ -1989,6 +1994,7 @@ impl MiniPulledJob {
                 None
             },
             visible_to_owner: job.visible_to_owner.clone(),
+            permissioned_as_end_user_email: None,
         }
     }
     pub fn is_flow(&self) -> bool {
@@ -2189,7 +2195,8 @@ pub async fn get_mini_pulled_job<'c>(
         script_entrypoint_override,
         trigger,
         trigger_kind as \"trigger_kind: JobTriggerKind\",
-        visible_to_owner
+        visible_to_owner,
+        NULL as permissioned_as_end_user_email
         FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id WHERE v2_job_queue.id = $1",
         job_id,
     )
@@ -3147,6 +3154,7 @@ pub async fn push<'c, 'd>(
     _priority_override: Option<i16>,
     authed: Option<&Authed>,
     running: bool, // whether the job is already running: only set this to true if you don't want the job to be picked up by a worker from the queue. It will also set started_at to now.
+    end_user_email: Option<String>,
 ) -> Result<(Uuid, Transaction<'c, Postgres>), Error> {
     #[cfg(feature = "cloud")]
     if *CLOUD_HOSTED {
@@ -4233,8 +4241,8 @@ pub async fn push<'c, 'd>(
             INSERT INTO v2_job_runtime (id, ping) VALUES ($1, null)
         ),
         inserted_job_perms AS (
-            INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id) 
-            values ($1, $32, $33, $34, $35, $36, $37, $2) 
+            INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id, end_user_email) 
+            values ($1, $32, $33, $34, $35, $36, $37, $2, $41) 
             ON CONFLICT (job_id) DO UPDATE SET email = $32, username = $33, is_admin = $34, is_operator = $35, folders = $36, groups = $37, workspace_id = $2
         )
         INSERT INTO v2_job_queue
@@ -4284,6 +4292,7 @@ pub async fn push<'c, 'd>(
         root_job,
         trigger_kind as Option<JobTriggerKind>,
         running,
+        end_user_email,
     )
     .execute(&mut *tx)
     .warn_after_seconds(1)
@@ -4697,7 +4706,7 @@ pub async fn get_same_worker_job(
                     v2_job.raw_flow,
                     pj.runnable_path as parent_runnable_path,
                     p.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 v2_job_queue
                     INNER JOIN v2_job ON v2_job.id = v2_job_queue.id
                     LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id

backend/windmill-queue/src/schedule.rs

@@ -331,6 +331,7 @@ pub async fn push_scheduled_job<'c>(
         None,
         push_authed,
         false,
+        None,
     )
     .await?;
 

backend/windmill-worker/src/ai_executor.rs

@@ -855,6 +855,7 @@ pub async fn run_agent(
                                     job_priority,
                                     job_perms.as_ref(),
                                     true,
+                                    None,
                                 )
                                 .await?;
 

backend/windmill-worker/src/bun_executor.rs

@@ -25,7 +25,11 @@ use crate::{
     DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, NODE_BIN_PATH, NODE_PATH, NPM_CONFIG_REGISTRY,
     NPM_PATH, NSJAIL_PATH, PATH_ENV, PROXY_ENVS, TZ_ENV,
 };
-use windmill_common::{client::AuthedClient, s3_helpers::BundleFormat, scripts::{id_to_codebase_info, CodebaseInfo}};
+use windmill_common::{
+    client::AuthedClient,
+    s3_helpers::BundleFormat,
+    scripts::{id_to_codebase_info, CodebaseInfo},
+};
 
 #[cfg(windows)]
 use crate::SYSTEM_ROOT;
@@ -910,7 +914,6 @@ pub async fn handle_bun_job(
     let common_bun_proc_envs: HashMap<String, String> =
         get_common_bun_proc_envs(Some(&base_internal_url)).await;
 
-
     let main_override = job.script_entrypoint_override.as_deref();
     let apply_preprocessor =
         job.flow_step_id.as_deref() != Some("preprocessor") && job.preprocessed == Some(false);
@@ -1621,11 +1624,11 @@ pub async fn start_worker(
         None,
         None,
         None,
+        None,
     )
     .await;
     let context_envs = build_envs_map(context.to_vec()).await;
 
-
     let mut format = BundleFormat::Cjs;
     if let Some(codebase) = codebase.as_ref() {
         let pulled_codebase = pull_codebase(w_id, codebase, job_dir).await?;

backend/windmill-worker/src/common.rs

@@ -452,6 +452,7 @@ pub async fn get_reserved_variables(
         Some(get_root_job_id(job).to_string()),
         Some(job.scheduled_for.clone()),
         job.runnable_id,
+        job.permissioned_as_end_user_email.clone(),
     )
     .await
     .to_vec();

backend/windmill-worker/src/deno_executor.rs

@@ -556,6 +556,7 @@ pub async fn start_worker(
         None,
         None,
         None,
+        None,
     )
     .await;
     let context_envs = build_envs_map(context.to_vec()).await;

backend/windmill-worker/src/python_executor.rs

@@ -2173,6 +2173,7 @@ pub async fn start_worker(
         None,
         None,
         None,
+        None,
     )
     .await
     .to_vec();
@@ -2294,6 +2295,7 @@ for line in sys.stdin:
         None,
         None,
         None,
+        None,
     )
     .await;
 

backend/windmill-worker/src/worker_flow.rs

@@ -3153,6 +3153,7 @@ async fn push_next_flow_job(
             new_job_priority_override,
             job_perms.as_ref(),
             false,
+            None,
         )
         .warn_after_seconds(2)
         .await?;

backend/windmill-worker/src/worker_lockfiles.rs

@@ -833,6 +833,7 @@ pub async fn trigger_dependents_to_recompute_dependencies(
             None,
             None,
             false,
+            None,
         )
         .await?;
         tracing::info!(