fix encrypted values passed from apps in flows

backend/windmill-worker/src/common.rs

@@ -213,6 +213,25 @@ pub fn parse_npm_config(s: &str) -> (String, Option<String>) {
     return (url, token_opt);
 }
 
+#[async_recursion]
+pub async fn get_root_job_id(job: &Uuid, db: &Pool<Postgres>) -> anyhow::Result<Uuid> {
+    let njob = sqlx::query_scalar!(
+        "SELECT flow_innermost_root_job FROM v2_job WHERE id = $1",
+        job
+    )
+    .fetch_optional(db)
+    .await?
+    .flatten();
+    if let Some(root_job) = njob {
+        if root_job == *job {
+            return Ok(job.to_owned());
+        }
+        get_root_job_id(&root_job, db).await
+    } else {
+        Ok(job.to_owned())
+    }
+}
+
 #[async_recursion]
 pub async fn transform_json_value(
     name: &str,
@@ -252,8 +271,10 @@ pub async fn transform_json_value(
         }
         Value::String(y) if y.starts_with("$encrypted:") => {
             let encrypted = y.strip_prefix("$encrypted:").unwrap();
-            let mc =
-                build_crypt_with_key_suffix(&db, &job.workspace_id, &job.id.to_string()).await?;
+
+            let root_job_id = get_root_job_id(&job.root_job.unwrap_or_else(|| job.id), db).await?;
+            let mc = build_crypt_with_key_suffix(&db, &job.workspace_id, &root_job_id.to_string())
+                .await?;
             decrypt(&mc, encrypted.to_string()).and_then(|x| {
                 serde_json::from_str(&x).map_err(|e| Error::internal_err(e.to_string()))
             })