Merge branch 'main' into migration/queue-params

backend/Cargo.lock

@@ -10834,6 +10834,7 @@ dependencies = [
  "bigdecimal",
  "chrono",
  "chrono-tz 0.10.0",
+ "const_format",
  "cron",
  "futures-core",
  "hex",

backend/migrations/20241004112544_create_params_args_result_tables.down.sql

@@ -0,0 +1,3 @@
+DROP TABLE job_params;
+DROP TABLE job_args;
+DROP TABLE completed_jobs_result;

backend/migrations/20241004112544_create_params_args_result_tables.up.sql

@@ -0,0 +1,25 @@
+-- Add up migration script here
+-- Add down migration script here
+-- Add up migration script here
+CREATE TABLE job_params (
+  id UUID PRIMARY KEY,
+  raw_code TEXT,
+  raw_flow jsonb NULL,
+  tag VARCHAR(50),
+  workspace_id VARCHAR(50)
+);
+
+-- Add up migration script here
+CREATE TABLE job_args (
+  id UUID PRIMARY KEY,
+  args JSONB,
+  tag VARCHAR(50),
+  workspace_id VARCHAR(50)
+);
+
+CREATE TABLE completed_jobs_result (
+  id UUID PRIMARY KEY,
+  result JSONB,
+  tag VARCHAR(50),
+  workspace_id VARCHAR(50)
+);

backend/src/monitor.rs

@@ -19,6 +19,7 @@ use tokio::{
     sync::{mpsc, RwLock},
 };
 
+use uuid::Uuid;
 #[cfg(feature = "embedding")]
 use windmill_api::embeddings::update_embeddings_db;
 use windmill_api::{
@@ -32,7 +33,7 @@ use windmill_common::{
     auth::JWT_SECRET,
     ee::CriticalErrorChannel,
     error,
-    flow_status::FlowStatusModule,
+    flow_status::{FlowStatusModule, ParsedFlowStatusGetter as _},
     global_settings::{
         BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ERROR_CHANNELS_SETTING,
         DEFAULT_TAGS_PER_WORKSPACE_SETTING, DEFAULT_TAGS_WORKSPACES_SETTING,
@@ -1420,7 +1421,7 @@ async fn handle_zombie_flows(
                 }
             );
             report_critical_error(reason.clone(), db.clone()).await;
-            cancel_zombie_flow_job(db, flow, &rsmq, reason).await?;
+            cancel_zombie_flow_job(db, &flow.id, &flow.workspace_id, &rsmq, reason).await?;
         }
     }
 
@@ -1436,11 +1437,17 @@ async fn handle_zombie_flows(
     .fetch_all(db)
     .await?;
 
+    #[derive(sqlx::FromRow, Debug)]
+    struct InQueueJobResult {
+        id: uuid::Uuid,
+        workspace_id: String,
+    }
+
     for flow in flows2 {
-        let in_queue = sqlx::query_as::<_, QueuedJob>(
-            "SELECT * FROM queue WHERE id = $1 AND running = true AND canceled = false",
+        let in_queue = sqlx::query_as!(InQueueJobResult,
+            "SELECT id, workspace_id FROM queue WHERE id = $1 AND running = true AND canceled = false",
+            flow.parent_flow_id
         )
-        .bind(flow.parent_flow_id)
         .fetch_optional(db)
         .await?;
         if let Some(job) = in_queue {
@@ -1450,7 +1457,7 @@ async fn handle_zombie_flows(
                 job.workspace_id,
                 flow.last_ping
             );
-            cancel_zombie_flow_job(db, job, &rsmq,
+            cancel_zombie_flow_job(db, &job.id, &job.workspace_id, &rsmq,
                 format!("Flow {} cancelled as one of the parallel branch {} was unable to make the last transition ", flow.parent_flow_id, flow.job_id))
                 .await?;
         } else {
@@ -1462,21 +1469,22 @@ async fn handle_zombie_flows(
 
 async fn cancel_zombie_flow_job(
     db: &Pool<Postgres>,
-    flow: QueuedJob,
+    job_id: &Uuid,
+    workspace_id: &str,
     rsmq: &Option<MultiplexedRsmq>,
     message: String,
 ) -> Result<(), error::Error> {
     let tx = db.begin().await.unwrap();
     tracing::error!(
         "zombie flow detected: {} in workspace {}. Cancelling it.",
-        flow.id,
-        flow.workspace_id
+        job_id,
+        workspace_id
     );
     let (ntx, _) = cancel_job(
         "monitor",
         Some(message),
-        flow.id,
-        flow.workspace_id.as_str(),
+        *job_id,
+        workspace_id,
         tx,
         db,
         rsmq.clone(),

backend/windmill-api/src/inputs.rs

@@ -21,11 +21,7 @@ use std::{
     vec,
 };
 use windmill_common::{
-    db::UserDB,
-    error::JsonResult,
-    jobs::JobKind,
-    scripts::to_i64,
-    utils::{not_found_if_none, paginate, Pagination},
+    db::UserDB, error::JsonResult, jobs::JobKind, query_scalar_with_fallback, scripts::to_i64, utils::{not_found_if_none, paginate, Pagination}
 };
 pub fn workspaced_service() -> Router {
     Router::new()
@@ -178,6 +174,7 @@ struct GetArgs {
     input: Option<bool>,
     allow_large: Option<bool>,
 }
+
 async fn get_args_from_history_or_saved_input(
     authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
@@ -185,6 +182,7 @@ async fn get_args_from_history_or_saved_input(
     Path((w_id, job_or_input_id)): Path<(String, Uuid)>,
 ) -> JsonResult<Option<Value>> {
     let mut tx = user_db.begin(&authed).await?;
+
     let result_o = if let Some(input) = g.input {
         if input {
             sqlx::query_scalar!(
@@ -196,24 +194,20 @@ async fn get_args_from_history_or_saved_input(
                 .fetch_optional(&mut *tx)
                 .await?
         } else {
-            sqlx::query_scalar!(
+            query_scalar_with_fallback!(tx, 
+                "SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM job_args WHERE id = $1 AND workspace_id = $2", 
                 "SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM completed_job WHERE id = $1 AND workspace_id = $2",
-                job_or_input_id,
+                job_or_input_id, 
                 w_id,
-                g.allow_large.unwrap_or(true)
-            )
-            .fetch_optional(&mut *tx)
-            .await?
+                g.allow_large.unwrap_or(true))?
         }
     } else {
-        sqlx::query_scalar!(
-        "SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM completed_job WHERE id = $1 AND workspace_id = $2 UNION ALL SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM input WHERE id = $1 AND workspace_id = $2",
-        job_or_input_id,
-        w_id,
-        g.allow_large.unwrap_or(true)
-    )
-    .fetch_optional(&mut *tx)
-    .await?
+        query_scalar_with_fallback!(tx, 
+                 "SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM job_args WHERE id = $1 AND workspace_id = $2 UNION ALL SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM input WHERE id = $1 AND workspace_id = $2", 
+                "SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM completed_job WHERE id = $1 AND workspace_id = $2 UNION ALL SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM input WHERE id = $1 AND workspace_id = $2",
+                job_or_input_id,
+                w_id,
+                g.allow_large.unwrap_or(true))?
     };
 
     tx.commit().await?;

backend/windmill-api/src/jobs.rs

@@ -314,9 +314,14 @@ async fn get_result_by_id(
     Path((w_id, flow_id, node_id)): Path<(String, Uuid, String)>,
     Query(JsonPath { json_path, .. }): Query<JsonPath>,
 ) -> windmill_common::error::JsonResult<Box<JsonRawValue>> {
-    let res =
-        windmill_queue::get_result_by_id(db.clone(), w_id.clone(), flow_id, node_id, json_path)
-            .await?;
+    let res = windmill_queue::get_result_by_id(
+        db.clone(),
+        &w_id,
+        flow_id,
+        &node_id,
+        json_path.as_deref(),
+    )
+    .await?;
 
     log_job_view(&db, Some(&authed), &w_id, &flow_id).await?;
 
@@ -1349,6 +1354,8 @@ async fn cancel_jobs(
 ) -> error::JsonResult<Vec<Uuid>> {
     let mut uuids = vec![];
     let mut tx = db.begin().await?;
+
+    let result = serde_json::json!({"error": { "message": format!("Job canceled: cancel all by {username}"), "name": "Canceled", "reason": "cancel all", "canceler": username}});
     let trivial_jobs =  sqlx::query!("INSERT INTO completed_job AS cj
                    ( workspace_id
                    , id
@@ -1412,10 +1419,19 @@ async fn cancel_jobs(
                    , tag
                    , priority FROM queue 
         WHERE id = any($2) AND running = false AND parent_job IS NULL AND workspace_id = $3 AND schedule_path IS NULL FOR UPDATE SKIP LOCKED
-        ON CONFLICT (id) DO NOTHING RETURNING id", username, &jobs, w_id, serde_json::json!({"error": { "message": format!("Job canceled: cancel all by {username}"), "name": "Canceled", "reason": "cancel all", "canceler": username}}))
+        ON CONFLICT (id) DO NOTHING RETURNING id", username, &jobs, w_id, &result)
         .fetch_all(&mut *tx)
         .await?.into_iter().map(|x| x.id).collect::<Vec<Uuid>>();
 
+    sqlx::query!(
+        "INSERT INTO completed_jobs_result(id, result, tag, workspace_id) SELECT id, $1, tag, $2 FROM completed_job  WHERE id = any($3) ON CONFLICT (id) DO NOTHING",
+        result,
+        w_id,
+        &trivial_jobs,
+    )
+    .execute(&mut *tx)
+    .await?;
+
     sqlx::query!(
         "DELETE FROM queue WHERE id = any($1) AND workspace_id = $2",
         &trivial_jobs,
@@ -1423,6 +1439,7 @@ async fn cancel_jobs(
     )
     .execute(&mut *tx)
     .await?;
+
     tx.commit().await?;
 
     // sqlx::query!(
@@ -1436,18 +1453,10 @@ async fn cancel_jobs(
             continue;
         }
         let rsmq = rsmq.clone();
-        match tokio::time::timeout(tokio::time::Duration::from_secs(5), async move {
+        if let Ok(result) = tokio::time::timeout(tokio::time::Duration::from_secs(5), async move {
             let tx = db.begin().await?;
             let (tx, _) = windmill_queue::cancel_job(
-                username,
-                None,
-                job_id.clone(),
-                w_id,
-                tx,
-                db,
-                rsmq,
-                false,
-                false,
+                username, None, job_id, w_id, tx, db, rsmq, false, false,
             )
             .await?;
             tx.commit().await?;
@@ -1455,20 +1464,19 @@ async fn cancel_jobs(
         })
         .await
         {
-            Ok(result) => match result {
+            match result {
                 Ok(_) => {
                     uuids.push(job_id);
                 }
                 Err(e) => {
                     tracing::error!("Failed to cancel job {:?}: {:?}", job_id, e);
                 }
-            },
-            Err(_) => {
-                tracing::error!(
-                    "Timeout while trying to cancel job {:?} after 5 seconds",
-                    job_id
-                );
             }
+        } else {
+            tracing::error!(
+                "Timeout while trying to cancel job {:?} after 5 seconds",
+                job_id
+            );
         }
     }
 
@@ -3283,7 +3291,7 @@ async fn run_wait_result(
     };
 
     let fast_poll_duration = *WAIT_RESULT_FAST_POLL_DURATION_SECS as u64 * 1000;
-    let mut accumulated_delay = 0 as u64;
+    let mut accumulated_delay = 0_u64;
 
     loop {
         if let Some(node_id_for_empty_return) = node_id_for_empty_return.as_ref() {
@@ -3400,6 +3408,16 @@ async fn delete_job_metadata_after_use(db: &DB, job_uuid: Uuid) -> Result<(), Er
     )
     .execute(db)
     .await?;
+
+    sqlx::query!(
+        "UPDATE job_args
+        SET args = '{}'::jsonb
+        WHERE id = $1",
+        job_uuid,
+    )
+    .execute(db)
+    .await?;
+
     sqlx::query!(
         "UPDATE job_logs
         SET logs = '##DELETED##'
@@ -4384,11 +4402,8 @@ async fn add_batch_jobs(
                 tx = PushIsolationLevel::Transaction(ntx);
                 uuids.push(uuid);
             }
-            match tx {
-                PushIsolationLevel::Transaction(tx) => {
-                    tx.commit().await?;
-                }
-                _ => (),
+            if let PushIsolationLevel::Transaction(tx) = tx {
+                tx.commit().await?;
             }
             return Ok(Json(uuids));
         }

backend/windmill-api/src/resources.rs

@@ -31,7 +31,6 @@ use windmill_audit::ActionKind;
 use windmill_common::{
     db::UserDB,
     error::{Error, JsonResult, Result},
-    jobs::QueuedJob,
     utils::{not_found_if_none, paginate, require_admin, Pagination, StripPath},
     variables,
 };
@@ -536,11 +535,29 @@ pub async fn transform_json_value<'c>(
         }
         Value::String(y) if y.starts_with("$") && job_id.is_some() => {
             let mut tx = authed_transaction_or_default(authed, user_db.clone(), db).await?;
-            let job = sqlx::query_as::<_, QueuedJob>(
-                "SELECT * FROM queue WHERE id = $1 AND workspace_id = $2",
+
+            #[derive(sqlx::FromRow, Debug)]
+            struct QueuedJobLite {
+                pub id: Uuid,
+                pub workspace_id: String,
+                pub parent_job: Option<Uuid>,
+                pub created_by: String,
+                pub email: String,
+                pub permissioned_as: String,
+                pub script_path: Option<String>,
+                pub schedule_path: Option<String>,
+                pub root_job: Option<Uuid>,
+                pub flow_step_id: Option<String>,
+                pub scheduled_for: chrono::DateTime<chrono::Utc>,
+            }
+
+            let job = sqlx::query_as!(
+                QueuedJobLite,
+                "SELECT id, workspace_id,parent_job, created_by, email, permissioned_as, script_path, schedule_path, root_job, flow_step_id, scheduled_for
+                 FROM queue WHERE id = $1 AND workspace_id = $2",
+                job_id.unwrap(),
+                workspace
             )
-            .bind(job_id.unwrap())
-            .bind(workspace)
             .fetch_optional(&mut *tx)
             .await?;
             tx.commit().await?;

backend/windmill-common/src/flow_status.rs

@@ -45,6 +45,33 @@ pub struct FlowStatus {
     pub restarted_from: Option<RestartedFrom>,
 }
 
+pub trait FlowStatusGetter {
+    fn get_raw_flow_status(&self) -> Option<&sqlx::types::Json<Box<serde_json::value::RawValue>>>;
+}
+
+#[macro_export]
+macro_rules! impl_flow_status_getter {
+    ($struct_name:ident) => {
+        impl FlowStatusGetter for $struct_name {
+            fn get_raw_flow_status(
+                &self,
+            ) -> Option<&sqlx::types::Json<Box<serde_json::value::RawValue>>> {
+                self.flow_status.as_ref()
+            }
+        }
+    };
+}
+
+pub trait ParsedFlowStatusGetter {
+    fn parse_flow_status(&self) -> Option<FlowStatus>;
+}
+impl<I: FlowStatusGetter> ParsedFlowStatusGetter for I {
+    fn parse_flow_status(&self) -> Option<FlowStatus> {
+        self.get_raw_flow_status()
+            .and_then(|v| serde_json::from_str::<FlowStatus>((**v).get()).ok())
+    }
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone, Default)]
 #[serde(default)]
 pub struct RetryStatus {

backend/windmill-common/src/flows.rs

@@ -125,6 +125,34 @@ pub struct FlowValue {
     pub concurrency_key: Option<String>,
 }
 
+pub trait FlowValueGetter {
+    fn get_raw_flow_value(&self) -> Option<&sqlx::types::Json<Box<serde_json::value::RawValue>>>;
+}
+
+#[macro_export]
+macro_rules! impl_flow_value_getter {
+    ($struct_name:ident) => {
+        impl FlowValueGetter for $struct_name {
+            fn get_raw_flow_value(
+                &self,
+            ) -> Option<&sqlx::types::Json<Box<serde_json::value::RawValue>>> {
+                self.raw_flow.as_ref()
+            }
+        }
+    };
+}
+
+pub trait ParsedFlowValueGetter {
+    fn parse_raw_flow(&self) -> Option<FlowValue>;
+}
+
+impl<I: FlowValueGetter> ParsedFlowValueGetter for I {
+    fn parse_raw_flow(&self) -> Option<FlowValue> {
+        self.get_raw_flow_value()
+            .and_then(|v| serde_json::from_str::<FlowValue>((**v).get()).ok())
+    }
+}
+
 #[derive(Deserialize, Serialize, Debug, Clone)]
 pub struct StopAfterIf {
     pub expr: String,

backend/windmill-common/src/jobs.rs

@@ -15,9 +15,9 @@ pub const PREPROCESSOR_FAKE_ENTRYPOINT: &str = "__WM_PREPROCESSOR";
 
 use crate::{
     error::{self, to_anyhow, Error},
-    flow_status::{FlowStatus, RestartedFrom},
-    flows::{FlowValue, Retry},
-    get_latest_deployed_hash_for_path,
+    flow_status::{FlowStatusGetter, RestartedFrom},
+    flows::{FlowValue, FlowValueGetter, Retry},
+    get_latest_deployed_hash_for_path, impl_flow_status_getter, impl_flow_value_getter,
     scripts::{ScriptHash, ScriptLang},
     worker::{to_raw_value, TMP_DIR},
 };
@@ -119,10 +119,7 @@ pub struct QueuedJob {
 
 impl QueuedJob {
     pub fn script_path(&self) -> &str {
-        self.script_path
-            .as_ref()
-            .map(String::as_str)
-            .unwrap_or("tmp/main")
+        self.script_path.as_deref().unwrap_or("tmp/main")
     }
     pub fn is_flow(&self) -> bool {
         matches!(
@@ -139,22 +136,11 @@ impl QueuedJob {
             self.script_path()
         )
     }
-
-    pub fn parse_raw_flow(&self) -> Option<FlowValue> {
-        self.raw_flow.as_ref().and_then(|v| {
-            let str = (**v).get();
-            // tracing::error!("raw_flow: {}", str);
-            return serde_json::from_str::<FlowValue>(str).ok();
-        })
-    }
-
-    pub fn parse_flow_status(&self) -> Option<FlowStatus> {
-        self.flow_status
-            .as_ref()
-            .and_then(|v| serde_json::from_str::<FlowStatus>((**v).get()).ok())
-    }
 }
 
+impl_flow_status_getter!(QueuedJob);
+impl_flow_value_getter!(QueuedJob);
+
 impl Default for QueuedJob {
     fn default() -> Self {
         Self {
@@ -266,23 +252,13 @@ impl CompletedJob {
     pub fn json_result(&self) -> Option<serde_json::Value> {
         self.result
             .as_ref()
-            .map(|r| serde_json::from_str(r.get()).ok())
-            .flatten()
-    }
-
-    pub fn parse_raw_flow(&self) -> Option<FlowValue> {
-        self.raw_flow
-            .as_ref()
-            .and_then(|v| serde_json::from_str::<FlowValue>((**v).get()).ok())
-    }
-
-    pub fn parse_flow_status(&self) -> Option<FlowStatus> {
-        self.flow_status
-            .as_ref()
-            .and_then(|v| serde_json::from_str::<FlowStatus>((**v).get()).ok())
+            .and_then(|r| serde_json::from_str(r.get()).ok())
     }
 }
 
+impl_flow_status_getter!(CompletedJob);
+impl_flow_value_getter!(CompletedJob);
+
 #[derive(sqlx::FromRow)]
 pub struct BranchResults {
     pub result: sqlx::types::Json<Box<RawValue>>,

backend/windmill-common/src/lib.rs

@@ -6,6 +6,7 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
+pub mod macros;
 use std::{
     net::SocketAddr,
     sync::{atomic::AtomicBool, Arc},

backend/windmill-common/src/macros.rs

@@ -0,0 +1,57 @@
+#[macro_export]
+macro_rules! fetch_one_with_fallback {
+    ($db:expr, $query_method:ident, $row_type:ty, $query:literal, $table:literal || $fallback_table:literal, $( $param:expr ),* ) => {{
+        let primary_query = sqlx::$query_method::<_, $row_type>(const_format::formatcp!($query, $table))
+            $(.bind($param))*
+            .fetch_one($db)
+            .await;
+
+        if let Err(sqlx::Error::RowNotFound) = primary_query {
+            tracing::info!("Data not found in job_params, falling back to fetching from $fallback_table");
+            sqlx::$query_method::<_, $row_type>(const_format::formatcp!($query, $fallback_table))
+                $(.bind($param))*
+                .fetch_one($db)
+                .await
+        } else {
+            primary_query
+        }
+    }};
+}
+
+#[macro_export]
+macro_rules! fetch_optional_with_fallback {
+    ($db:expr, $query_method:ident, $row_type:ty, $query:literal, $table:literal || $fallback_table:literal, $( $param:expr ),* ) => {{
+        let primary_query = sqlx::$query_method::<_, $row_type>(const_format::formatcp!($query, $table))
+            $(.bind($param))*
+            .fetch_optional($db)
+            .await;
+
+        if let Ok(None) = primary_query {
+            tracing::info!("Data not found in job_params, falling back to fetching from $fallback_table");
+            sqlx::$query_method::<_, $row_type>(const_format::formatcp!($query, $fallback_table))
+                $(.bind($param))*
+                .fetch_optional($db)
+                .await
+        } else {
+            primary_query
+        }
+    }};
+}
+
+#[macro_export]
+macro_rules! query_scalar_with_fallback {
+    ($tx:expr, $query:literal, $fallback_query:literal,$( $param:expr ),* ) => {{
+        let primary_query = sqlx::query_scalar!($query,$($param),*)
+            .fetch_optional(&mut *$tx)
+            .await;
+
+        if let Ok(None) = primary_query {
+            tracing::info!("Data not found in job_args, falling back to fetching from queue");
+            sqlx::query_scalar!($fallback_query, $($param),*)
+                .fetch_optional(&mut *$tx)
+                .await
+        } else {
+            primary_query
+        }
+    }};
+}

backend/windmill-common/src/worker.rs

@@ -91,7 +91,7 @@ lazy_static::lazy_static! {
 }
 
 pub async fn make_suspended_pull_query(wc: &WorkerConfig) {
-    if wc.worker_tags.len() == 0 {
+    if wc.worker_tags.is_empty() {
         tracing::error!("Empty tags in worker tags, skipping");
         return;
     }
@@ -123,7 +123,7 @@ pub async fn make_suspended_pull_query(wc: &WorkerConfig) {
 pub async fn make_pull_query(wc: &WorkerConfig) {
     let mut queries = vec![];
     for tags in wc.priority_tags_sorted.iter() {
-        if tags.tags.len() == 0 {
+        if tags.tags.is_empty() {
             tracing::error!("Empty tags in priority tags, skipping");
             continue;
         }

backend/windmill-queue/Cargo.toml

@@ -42,4 +42,5 @@ async-recursion.workspace = true
 bigdecimal.workspace = true
 axum.workspace = true
 serde_urlencoded.workspace = true
-regex.workspace = true
+regex.workspace = true
+const_format.workspace = true

backend/windmill-queue/src/jobs.rs

@@ -6,8 +6,6 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
-use std::{borrow::Borrow, collections::HashMap, sync::Arc, vec};
-
 use anyhow::Context;
 use async_recursion::async_recursion;
 use axum::{
@@ -31,12 +29,14 @@ use serde_json::{json, value::RawValue};
 use sqlx::{types::Json, FromRow, Pool, Postgres, Transaction};
 #[cfg(feature = "benchmark")]
 use std::time::Instant;
+use std::{borrow::Borrow, collections::HashMap, sync::Arc, vec};
 use tokio::{sync::RwLock, time::sleep};
 use tracing::{instrument, Instrument};
 use ulid::Ulid;
 use uuid::Uuid;
 use windmill_audit::audit_ee::{audit_log, AuditAuthor};
 use windmill_audit::ActionKind;
+use windmill_common::{fetch_optional_with_fallback, flows::FlowValueGetter};
 
 use windmill_common::{
     add_time,
@@ -44,12 +44,15 @@ use windmill_common::{
     db::{Authed, UserDB},
     error::{self, to_anyhow, Error},
     flow_status::{
-        BranchAllStatus, FlowCleanupModule, FlowStatus, FlowStatusModule, FlowStatusModuleWParent,
-        Iterator, JobResult, RestartedFrom, RetryStatus, MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL,
+        BranchAllStatus, FlowCleanupModule, FlowStatus, FlowStatusGetter, FlowStatusModule,
+        FlowStatusModuleWParent, Iterator, JobResult, ParsedFlowStatusGetter, RestartedFrom,
+        RetryStatus, MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL,
     },
     flows::{
         add_virtual_items_if_necessary, FlowModule, FlowModuleValue, FlowValue, InputTransform,
+        ParsedFlowValueGetter,
     },
+    impl_flow_status_getter, impl_flow_value_getter,
     jobs::{
         get_payload_tag_from_prefixed_path, CompletedJob, JobKind, JobPayload, QueuedJob, RawCode,
         ENTRYPOINT_OVERRIDE, PREPROCESSOR_FAKE_ENTRYPOINT,
@@ -142,6 +145,14 @@ pub struct CanceledBy {
     pub reason: Option<String>,
 }
 
+#[derive(Debug, sqlx::FromRow)]
+pub struct CompletedSubFlow {
+    pub id: Uuid,
+    pub flow_status: Option<sqlx::types::Json<Box<RawValue>>>,
+}
+
+impl_flow_status_getter!(CompletedSubFlow);
+
 pub async fn cancel_single_job<'c>(
     username: &str,
     reason: Option<String>,
@@ -222,26 +233,23 @@ pub async fn cancel_job<'c>(
     force_cancel: bool,
     require_anonymous: bool,
 ) -> error::Result<(Transaction<'c, Postgres>, Option<Uuid>)> {
-    let job = get_queued_job_tx(id, &w_id, &mut tx).await?;
-
-    if job.is_none() {
+    let Some(mut job) = get_queued_job_tx(id, w_id, &mut tx).await? else {
         return Ok((tx, None));
-    }
+    };
 
-    if require_anonymous && job.as_ref().unwrap().created_by != "anonymous" {
+    if require_anonymous && job.created_by != "anonymous" {
         return Err(Error::BadRequest(
             "You are not logged in and this job was not created by an anonymous user like you so you cannot cancel it".to_string(),
         ));
     }
 
-    let mut job = job.unwrap();
     if force_cancel {
         // if force canceling a flow step, make sure we force cancel from the highest parent
         loop {
             if job.parent_job.is_none() {
                 break;
             }
-            match get_queued_job_tx(job.parent_job.unwrap(), &w_id, &mut tx).await? {
+            match get_queued_job_tx(job.parent_job.unwrap(), w_id, &mut tx).await? {
                 Some(j) => {
                     job = j;
                 }
@@ -587,6 +595,8 @@ pub async fn add_completed_job<
 
     let mem_peak = mem_peak.max(queued_job.mem_peak.unwrap_or(0));
     add_time!(bench, "add_completed_job query START");
+
+    // On conflict (when id already exists), update the success and result fields.
     let _duration: i64 = sqlx::query_scalar!(
         "INSERT INTO completed_job AS cj
                    ( workspace_id
@@ -623,8 +633,8 @@ pub async fn add_completed_job<
             VALUES ($1, $2, $3, $4, $5, COALESCE($6, now()), (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE($6, now()))))*1000, $7, $8, $9,\
                     $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29)
          ON CONFLICT (id) DO UPDATE SET success = $7, result = $11 RETURNING duration_ms",
-        queued_job.workspace_id,
-        queued_job.id,
+        &queued_job.workspace_id,
+        &queued_job.id,
         queued_job.parent_job,
         queued_job.created_by,
         queued_job.created_at,
@@ -633,7 +643,7 @@ pub async fn add_completed_job<
         queued_job.script_hash.map(|x| x.0),
         queued_job.script_path,
         &queued_job.args as &Option<Json<HashMap<String, Box<RawValue>>>>,
-        result as Json<&T>,
+        &result as &Json<&T>,
         queued_job.raw_code,
         queued_job.raw_lock,
         canceled_by.is_some(),
@@ -650,16 +660,32 @@ pub async fn add_completed_job<
         queued_job.email,
         queued_job.visible_to_owner,
         if mem_peak > 0 { Some(mem_peak) } else { None },
-        queued_job.tag,
+        &queued_job.tag,
         queued_job.priority,
     )
     .fetch_one(&mut tx)
     .await
     .map_err(|e| Error::InternalErr(format!("Could not add completed job {job_id}: {e:#}")))?;
-    // tracing::error!("2 {:?}", start.elapsed());
-
     add_time!(bench, "add_completed_job query END");
 
+    add_time!(bench, "completed_jobs_result query START");
+    sqlx::query!(
+        "INSERT INTO completed_jobs_result(id, result, tag, workspace_id) VALUES($1, $2, $3, $4) ON CONFLICT (id) DO UPDATE SET result = $2",
+        queued_job.id,
+        &result as &Json<&T>,
+        queued_job.tag,
+        queued_job.workspace_id,
+    )
+    .execute(&mut tx)
+    .await
+    .map_err(|e| {
+        Error::InternalErr(format!(
+            "Could not add completed job result {job_id}: {e:#}"
+        ))
+    })?;
+
+    add_time!(bench, "completed_jobs_result query END");
+
     if !queued_job.is_flow_step {
         if _duration > 500
             && (queued_job.job_kind == JobKind::Script || queued_job.job_kind == JobKind::Preview)
@@ -864,10 +890,10 @@ pub async fn add_completed_job<
     tx.commit().await?;
     tracing::info!(
         %job_id,
-        root_job = ?queued_job.root_job.map(|x| x.to_string()).unwrap_or_else(|| String::new()),
+        root_job = ?queued_job.root_job.map(|x| x.to_string()).unwrap_or_default(),
         path = &queued_job.script_path(),
         job_kind = ?queued_job.job_kind,
-        started_at = ?queued_job.started_at.map(|x| x.to_string()).unwrap_or_else(|| String::new()),
+        started_at = ?queued_job.started_at.map(|x| x.to_string()).unwrap_or_default(),
         duration = ?_duration,
         permissioned_as = ?queued_job.permissioned_as,
         email = ?queued_job.email,
@@ -2267,55 +2293,48 @@ pub struct ResultWithId {
 
 pub async fn get_result_by_id(
     db: Pool<Postgres>,
-    w_id: String,
+    w_id: &str,
     flow_id: Uuid,
-    node_id: String,
-    json_path: Option<String>,
+    node_id: &str,
+    json_path: Option<&str>,
 ) -> error::Result<Box<RawValue>> {
-    match get_result_by_id_from_running_flow(
-        &db,
-        w_id.as_str(),
-        &flow_id,
-        node_id.as_str(),
-        json_path.clone(),
-    )
-    .await
-    {
+    #[derive(sqlx::FromRow, Debug)]
+    struct RunningFlowJobResult {
+        pub id: Uuid,
+        pub flow_status: Option<Json<Box<RawValue>>>,
+    }
+    impl_flow_status_getter!(RunningFlowJobResult);
+
+    match get_result_by_id_from_running_flow(&db, w_id, &flow_id, node_id, json_path).await {
         Ok(res) => Ok(res),
         Err(_) => {
-            let running_flow_job =sqlx::query_as::<_, QueuedJob>(
-                "SELECT * FROM queue WHERE COALESCE((SELECT root_job FROM queue WHERE id = $1), $1) = id AND workspace_id = $2"
+            let running_flow_job =sqlx::query_as::<_, RunningFlowJobResult>(
+                "SELECT id, flow_status FROM queue WHERE COALESCE((SELECT root_job FROM queue WHERE id = $1), $1) = id AND workspace_id = $2"
             ).bind(flow_id)
             .bind(&w_id)
             .fetch_optional(&db).await?;
+
             match running_flow_job {
                 Some(job) => {
                     let restarted_from = windmill_common::utils::not_found_if_none(
                         job.parse_flow_status()
-                            .map(|status| status.restarted_from)
-                            .flatten(),
+                            .and_then(|status| status.restarted_from),
                         "Id not found in the result's mapping of the root job and root job had no restarted from information",
                         format!("parent: {}, root: {}, id: {}", flow_id, job.id, node_id),
                     )?;
 
                     get_result_by_id_from_original_flow(
                         &db,
-                        w_id.as_str(),
+                        w_id,
                         &restarted_from.flow_job_id,
-                        node_id.as_str(),
-                        json_path.clone(),
+                        node_id,
+                        json_path,
                     )
                     .await
                 }
                 None => {
-                    get_result_by_id_from_original_flow(
-                        &db,
-                        w_id.as_str(),
-                        &flow_id,
-                        node_id.as_str(),
-                        json_path.clone(),
-                    )
-                    .await
+                    get_result_by_id_from_original_flow(&db, w_id, &flow_id, node_id, json_path)
+                        .await
                 }
             }
         }
@@ -2334,7 +2353,7 @@ pub async fn get_result_by_id_from_running_flow(
     w_id: &str,
     flow_id: &Uuid,
     node_id: &str,
-    json_path: Option<String>,
+    json_path: Option<&str>,
 ) -> error::Result<Box<RawValue>> {
     let flow_job_result = sqlx::query_as::<_, FlowJobResult>(
         "SELECT leaf_jobs->$1::text as leaf_jobs, parent_job FROM queue WHERE COALESCE((SELECT root_job FROM queue WHERE id = $2), $2) = id AND workspace_id = $3")
@@ -2352,8 +2371,7 @@ pub async fn get_result_by_id_from_running_flow(
 
     let job_result = flow_job_result
         .leaf_jobs
-        .map(|x| serde_json::from_str(x.get()).ok())
-        .flatten();
+        .and_then(|x| serde_json::from_str(x.get()).ok());
 
     if job_result.is_none() && flow_job_result.parent_job.is_some() {
         let parent_job = flow_job_result.parent_job.unwrap();
@@ -2378,9 +2396,9 @@ pub async fn get_result_by_id_from_running_flow(
 async fn get_completed_flow_node_result_rec(
     db: &Pool<Postgres>,
     w_id: &str,
-    subflows: Vec<CompletedJob>,
+    subflows: &[CompletedSubFlow],
     node_id: &str,
-    json_path: Option<String>,
+    json_path: Option<&str>,
 ) -> error::Result<Option<Box<RawValue>>> {
     for subflow in subflows {
         let flow_status = subflow.parse_flow_status().ok_or_else(|| {
@@ -2397,7 +2415,7 @@ async fn get_completed_flow_node_result_rec(
                     db,
                     w_id,
                     JobResult::SingleJob(leaf_job_uuid),
-                    json_path.clone(),
+                    json_path,
                 )
                 .await
                 .map(Some),
@@ -2405,7 +2423,7 @@ async fn get_completed_flow_node_result_rec(
                     db,
                     w_id,
                     JobResult::ListJob(jobs),
-                    json_path.clone(),
+                    json_path,
                 )
                 .await
                 .map(Some),
@@ -2416,10 +2434,11 @@ async fn get_completed_flow_node_result_rec(
                 ))),
             };
         } else {
-            let subflows = sqlx::query_as::<_, CompletedJob>(
-                "SELECT *, null as labels FROM completed_job WHERE parent_job = $1 AND workspace_id = $2 AND flow_status IS NOT NULL",
+            let subflows = sqlx::query_as::<_, CompletedSubFlow>(
+                "SELECT id, flow_status as labels FROM completed_job WHERE parent_job = $1 AND workspace_id = $2 AND flow_status IS NOT NULL",
             ).bind(subflow.id).bind(w_id).fetch_all(db).await?;
-            match get_completed_flow_node_result_rec(db, w_id, subflows, node_id, json_path.clone())
+
+            match get_completed_flow_node_result_rec(db, w_id, &subflows, node_id, json_path)
                 .await?
             {
                 Some(res) => return Ok(Some(res)),
@@ -2436,10 +2455,10 @@ async fn get_result_by_id_from_original_flow(
     w_id: &str,
     completed_flow_id: &Uuid,
     node_id: &str,
-    json_path: Option<String>,
+    json_path: Option<&str>,
 ) -> error::Result<Box<RawValue>> {
-    let flow_job = sqlx::query_as::<_, CompletedJob>(
-        "SELECT *, null as labels FROM completed_job WHERE id = $1 AND workspace_id = $2",
+    let flow_job = sqlx::query_as::<_, CompletedSubFlow>(
+        "SELECT id, flow_status FROM completed_job WHERE id = $1 AND workspace_id = $2",
     )
     .bind(completed_flow_id)
     .bind(w_id)
@@ -2452,7 +2471,7 @@ async fn get_result_by_id_from_original_flow(
         format!("root: {}, id: {}", completed_flow_id, node_id),
     )?;
 
-    match get_completed_flow_node_result_rec(db, w_id, vec![flow_job], node_id, json_path).await? {
+    match get_completed_flow_node_result_rec(db, w_id, &[flow_job], node_id, json_path).await? {
         Some(res) => Ok(res),
         None => Err(error::Error::NotFound(format!(
             "Flow result by id not found going top-down from {}, (id: {})",
@@ -2465,31 +2484,33 @@ async fn extract_result_from_job_result(
     db: &Pool<Postgres>,
     w_id: &str,
     job_result: JobResult,
-    json_path: Option<String>,
+    json_path: Option<&str>,
 ) -> error::Result<Box<RawValue>> {
     match job_result {
         JobResult::ListJob(job_ids) => match json_path {
             Some(json_path) => {
-                let mut parts = json_path.split(".");
+                let mut parts = json_path.split('.');
 
-                let Some(idx) = parts.next().map(|x| x.parse::<usize>().ok()).flatten() else {
+                let Some(ref idx) = parts.next().and_then(|x| x.parse::<usize>().ok()) else {
                     return Ok(to_raw_value(&serde_json::Value::Null));
                 };
-                let Some(job_id) = job_ids.get(idx).cloned() else {
+                let Some(job_id) = job_ids.get(*idx) else {
                     return Ok(to_raw_value(&serde_json::Value::Null));
                 };
-                Ok(sqlx::query_as::<_, ResultR>(
-                    "SELECT result #> $3 as result FROM completed_job WHERE id = $1 AND workspace_id = $2",
-                )
-                .bind(job_id)
-                .bind(w_id)
-                .bind(
-                    parts.map(|x| x.to_string()).collect::<Vec<_>>()
-                )
-                .fetch_optional(db)
-                .await?
-                .map(|r| r.result.map(|x| x.0))
-                .flatten()
+
+                let parts = parts.map(|x| x.to_string()).collect_vec();
+
+                Ok(fetch_optional_with_fallback!(
+                    db,
+                    query_as,
+                    ResultR,
+                    "SELECT result #> $3 as result FROM {} WHERE id = $1 AND workspace_id = $2",
+                    "completed_jobs_result" || "completed_job",
+                    *job_id,
+                    w_id,
+                    &parts
+                )?
+                .and_then(|r| r.result.map(|x| x.0))
                 .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null)))
             }
             None => {
@@ -2503,6 +2524,7 @@ async fn extract_result_from_job_result(
                 .into_iter()
                 .filter_map(|x| x.result.map(|y| (x.id, y)))
                 .collect::<HashMap<Uuid, Json<Box<RawValue>>>>();
+
                 let result = job_ids
                     .into_iter()
                     .map(|id| {
@@ -2510,25 +2532,31 @@ async fn extract_result_from_job_result(
                             .map(|x| x.0.clone())
                             .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null))
                     })
-                    .collect::<Vec<_>>();
+                    .collect_vec();
                 Ok(to_raw_value(&result))
             }
         },
-        JobResult::SingleJob(x) => Ok(sqlx::query_as::<_, ResultR>(
-            "SELECT result #> $3 as result FROM completed_job WHERE id = $1 AND workspace_id = $2",
-        )
-        .bind(x)
-        .bind(w_id)
-        .bind(
-            json_path
+        // ici
+        JobResult::SingleJob(x) => {
+            let path = json_path
                 .map(|x| x.split(".").map(|x| x.to_string()).collect::<Vec<_>>())
-                .unwrap_or_default(),
-        )
-        .fetch_optional(db)
-        .await?
-        .map(|r| r.result.map(|x| x.0))
-        .flatten()
-        .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null))),
+                .unwrap_or_default();
+
+            let res = fetch_optional_with_fallback!(
+                db,
+                query_as,
+                ResultR,
+                "SELECT result #> $3 as result FROM {} WHERE id = $1 AND workspace_id = $2",
+                "completed_jobs_result" || "completed_job",
+                x,
+                w_id,
+                &path
+            )?
+            .and_then(|r| r.result.map(|x| x.0))
+            .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null));
+
+            Ok(res)
+        }
     }
 }
 
@@ -2643,7 +2671,7 @@ pub struct PushArgsOwned {
     pub args: HashMap<String, Box<RawValue>>,
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct PushArgs<'c> {
     pub extra: Option<HashMap<String, Box<RawValue>>>,
     pub args: &'c HashMap<String, Box<RawValue>>,
@@ -2754,7 +2782,7 @@ fn restructure_cloudevents_metadata(
         .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null));
     let str = data.to_string();
 
-    let wrap_body = str.len() > 0 && str.chars().next().unwrap() != '{';
+    let wrap_body = !str.is_empty() && !str.starts_with('{');
 
     if wrap_body {
         let args = serde_json::from_str::<Option<Box<RawValue>>>(&str)
@@ -2784,7 +2812,7 @@ impl PushArgsOwned {
             extra.insert("raw_string".to_string(), to_raw_value(&str));
         }
 
-        let wrap_body = force_wrap_body || str.len() > 0 && str.chars().next().unwrap() != '{';
+        let wrap_body = force_wrap_body || !str.is_empty() && !str.starts_with('{');
 
         if wrap_body {
             let args = serde_json::from_str::<Option<Box<RawValue>>>(&str)
@@ -3895,6 +3923,10 @@ pub async fn push<'c, 'd, R: rsmq_async::RsmqConnection + Send + 'c>(
     };
 
     tracing::debug!("Pushing job {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}");
+
+    let raw_flow = raw_flow.map(Json);
+    let args = Json(args);
+
     let uuid = sqlx::query_scalar!(
         "INSERT INTO queue
             (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, 
@@ -3913,12 +3945,12 @@ pub async fn push<'c, 'd, R: rsmq_async::RsmqConnection + Send + 'c>(
         scheduled_for_o,
         script_hash,
         script_path.clone(),
-        raw_code,
+        raw_code.clone(),
         raw_lock,
-        Json(args) as Json<PushArgs>,
+        args.clone() as Json<PushArgs>,
         job_kind.clone() as JobKind,
         schedule_path,
-        raw_flow.map(Json) as Option<Json<FlowValue>>,
+        raw_flow.clone() as Option<Json<FlowValue>>,
         flow_status.map(Json) as Option<Json<FlowStatus>>,
         is_flow_step,
         language as Option<ScriptLang>,
@@ -3927,7 +3959,7 @@ pub async fn push<'c, 'd, R: rsmq_async::RsmqConnection + Send + 'c>(
         email,
         visible_to_owner,
         root_job,
-        tag,
+        tag.clone(),
         concurrent_limit,
         if concurrent_limit.is_some() {  concurrency_time_window_s } else { None },
         custom_timeout,
@@ -3939,6 +3971,34 @@ pub async fn push<'c, 'd, R: rsmq_async::RsmqConnection + Send + 'c>(
     .await
     .map_err(|e| Error::InternalErr(format!("Could not insert into queue {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}: {e:#}")))?;
 
+    // insert into args queue
+    sqlx::query!(
+        r#"
+        INSERT INTO job_args (id, workspace_id, args, tag)
+        VALUES ($1, $2, $3, $4)
+        "#,
+        uuid,
+        workspace_id,
+        args as Json<PushArgs>,
+        &tag
+    )
+    .execute(&mut tx)
+    .await?;
+
+    sqlx::query!(
+        r#"
+        INSERT INTO job_params (id, workspace_id, raw_code, raw_flow, tag)
+        VALUES ($1, $2, $3, $4, $5)
+        "#,
+        uuid,
+        workspace_id,
+        raw_code,
+        raw_flow as Option<Json<FlowValue>>,
+        tag
+    )
+    .execute(&mut tx)
+    .await?;
+
     tracing::debug!("Pushed {job_id}");
     // TODO: technically the job isn't queued yet, as the transaction can be rolled back. Should be solved when moving these metrics to the queue abstraction.
     #[cfg(feature = "prometheus")]
@@ -4059,11 +4119,8 @@ pub async fn push<'c, 'd, R: rsmq_async::RsmqConnection + Send + 'c>(
 }
 
 pub fn canceled_job_to_result(job: &QueuedJob) -> serde_json::Value {
-    let reason = job
-        .canceled_reason
-        .as_deref()
-        .unwrap_or_else(|| "no reason given");
-    let canceler = job.canceled_by.as_deref().unwrap_or_else(|| "unknown");
+    let reason = job.canceled_reason.as_deref().unwrap_or("no reason given");
+    let canceler = job.canceled_by.as_deref().unwrap_or("unknown");
     serde_json::json!({"message": format!("Job canceled: {reason} by {canceler}"), "name": "Canceled", "reason": reason, "canceler": canceler})
 }
 
@@ -4086,8 +4143,19 @@ async fn restarted_flows_resolution(
     ),
     Error,
 > {
-    let completed_job = sqlx::query_as::<_, CompletedJob>(
-        "SELECT *, null as labels FROM completed_job WHERE id = $1 and workspace_id = $2",
+    #[derive(Debug, sqlx::FromRow)]
+    struct QueryResults {
+        pub script_path: Option<String>,
+        pub priority: Option<i16>,
+        pub raw_flow: Option<sqlx::types::Json<Box<RawValue>>>,
+        pub flow_status: Option<sqlx::types::Json<Box<RawValue>>>,
+    }
+
+    impl_flow_status_getter!(QueryResults);
+    impl_flow_value_getter!(QueryResults);
+
+    let completed_job = sqlx::query_as::<_, QueryResults>(
+        "SELECT script_path, priority, raw_flow, flow_status FROM completed_job WHERE id = $1 and workspace_id = $2",
     )
     .bind(completed_flow_id)
     .bind(workspace_id)
@@ -4120,10 +4188,9 @@ async fn restarted_flows_resolution(
         if flow_value_if_any
             .clone()
             .map(|fv| {
-                fv.modules
+                !fv.modules
                     .iter()
-                    .find(|flow_value_module| flow_value_module.id == module.id())
-                    .is_none()
+                    .any(|flow_value_module| flow_value_module.id == module.id())
             })
             .unwrap_or(false)
         {
@@ -4239,7 +4306,7 @@ async fn restarted_flows_resolution(
             truncated_modules.push(FlowStatusModule::WaitingForPriorSteps { id: module.id() });
         } else {
             // else we simply "transfer" the module from the completed flow to the new one if it's a success
-            step_n = step_n + 1;
+            step_n += 1;
             match module.clone() {
                 FlowStatusModule::Success { .. } => Ok(truncated_modules.push(module)),
                 _ => Err(Error::InternalErr(format!(

backend/windmill-worker/src/worker_flow.rs

@@ -29,20 +29,20 @@ use sqlx::FromRow;
 use tokio::sync::mpsc::Sender;
 use tracing::instrument;
 use uuid::Uuid;
-use windmill_common::add_time;
 use windmill_common::auth::JobPerms;
 #[cfg(feature = "benchmark")]
 use windmill_common::bench::BenchmarkIter;
 use windmill_common::db::Authed;
 use windmill_common::flow_status::{
-    ApprovalConditions, FlowStatusModuleWParent, Iterator, JobResult,
+    ApprovalConditions, FlowStatusModuleWParent, Iterator, JobResult, ParsedFlowStatusGetter,
 };
-use windmill_common::flows::add_virtual_items_if_necessary;
+use windmill_common::flows::{add_virtual_items_if_necessary, ParsedFlowValueGetter};
 use windmill_common::jobs::{
     script_hash_to_tag_and_limits, script_path_to_payload, BranchResults, JobPayload, QueuedJob,
     RawCode, ENTRYPOINT_OVERRIDE,
 };
 use windmill_common::worker::to_raw_value;
+use windmill_common::{add_time, fetch_one_with_fallback, fetch_optional_with_fallback};
 use windmill_common::{
     error::{self, to_anyhow, Error},
     flow_status::{
@@ -61,6 +61,19 @@ type DB = sqlx::Pool<sqlx::Postgres>;
 
 use windmill_queue::{canceled_job_to_result, get_queued_job_tx, push, QueueTransaction};
 
+async fn get_args_from_job_id(db: &DB, id: &Uuid) -> Result<RowArgs, Error> {
+    let args = fetch_one_with_fallback!(
+        db,
+        query_as,
+        RowArgs,
+        "SELECT args FROM {} WHERE id = $1",
+        "job_args" || "queue",
+        id
+    );
+
+    args.map_err(|e| Error::InternalErr(format!("retrieval of args from state: {e:#}")))
+}
+
 // #[instrument(level = "trace", skip_all)]
 pub async fn update_flow_status_after_job_completion<
     R: rsmq_async::RsmqConnection + Send + Sync + Clone,
@@ -229,10 +242,10 @@ pub async fn update_flow_status_after_job_completion_internal<
         })?;
 
         let old_status = serde_json::from_str::<FlowStatus>(old_status_json.flow_status.get())
-            .or_else(|e| {
-                Err(Error::InternalErr(format!(
+            .map_err(|e| {
+                Error::InternalErr(format!(
                     "requiring status to be parsable as FlowStatus: {e:?}"
-                )))
+                ))
             })?;
 
         let current_module = if let Some(x) = old_status_json.current_module {
@@ -289,7 +302,7 @@ pub async fn update_flow_status_after_job_completion_internal<
 
         // 0 length flows are not failure steps
         let is_failure_step =
-            old_status.step >= old_status.modules.len() as i32 && old_status.modules.len() > 0;
+            old_status.step >= old_status.modules.len() as i32 && !old_status.modules.is_empty();
 
         let (mut stop_early, mut skip_if_stop_early, continue_on_error) = if let Some(se) =
             stop_early_override
@@ -302,14 +315,13 @@ pub async fn update_flow_status_after_job_completion_internal<
             };
 
             let is_flow = if let Some(step) = step {
-                sqlx::query_scalar!(
-                    "SELECT raw_flow->'modules'->($1)->'value'->>'type' = 'flow' FROM queue WHERE id = $2",
+                fetch_one_with_fallback!(db,
+                    query_scalar,
+                    Option<bool>,
+                    "SELECT raw_flow->'modules'->($1)->'value'->>'type' = 'flow' FROM {} WHERE id = $2",
+                    "job_params" || "queue",
                     step as i32,
-                    &flow
-                )
-                    .fetch_one(db)
-                    .await
-                    .map_err(|e| {
+                    &flow).map_err(|e| {
                         Error::InternalErr(format!("error during retrieval of step's type: {e:#}"))
                     })?
                     .unwrap_or(false)
@@ -342,19 +354,9 @@ pub async fn update_flow_status_after_job_completion_internal<
                         }
                         _ => None,
                     };
-                    let args = sqlx::query_as::<_, RowArgs>(
-                        "SELECT
-                                        args
-                                    FROM queue
-                                    WHERE id = $2",
-                    )
-                    .bind(old_status.step)
-                    .bind(flow)
-                    .fetch_one(db)
-                    .await
-                    .map_err(|e| {
-                        Error::InternalErr(format!("retrieval of args from state: {e:#}"))
-                    })?;
+
+                    let args = get_args_from_job_id(db, &flow).await?;
+
                     compute_bool_from_expr(
                         expr.to_string(),
                         Marc::new(args.args.unwrap_or_default().0),
@@ -442,7 +444,7 @@ pub async fn update_flow_status_after_job_completion_internal<
                             None
                         };
 
-                        let nindex = if let Some(position) = position { 
+                        let nindex = if let Some(position) = position {
                             sqlx::query_scalar!(
                             "UPDATE queue
                             SET flow_status = JSONB_SET(
@@ -489,7 +491,7 @@ pub async fn update_flow_status_after_job_completion_internal<
                             None
                         };
 
-                        let nindex = if let Some(position) = position { 
+                        let nindex = if let Some(position) = position {
                             sqlx::query_scalar!(
                                 "UPDATE queue
                                 SET flow_status = JSONB_SET(
@@ -658,7 +660,7 @@ pub async fn update_flow_status_after_job_completion_internal<
                 flow_jobs_success,
                 flow_jobs,
                 ..
-            } if branch.to_owned() < len - 1 && (success || skip_branch_failure) => {
+            } if *branch < len - 1 && (success || skip_branch_failure) => {
                 if let Some(jobs) = flow_jobs {
                     set_success_in_flow_job_success(
                         flow_jobs_success,
@@ -870,19 +872,7 @@ pub async fn update_flow_status_after_job_completion_internal<
                     .as_ref()
                     .and_then(|m| m.stop_after_all_iters_if.as_ref().map(|x| x.expr.clone()))
                 {
-                    let args = sqlx::query_as::<_, RowArgs>(
-                        "SELECT
-                            args
-                        FROM queue
-                        WHERE id = $2",
-                    )
-                    .bind(old_status.step)
-                    .bind(flow)
-                    .fetch_one(db)
-                    .await
-                    .map_err(|e| {
-                        Error::InternalErr(format!("retrieval of args from state: {e:#}"))
-                    })?;
+                    let args = get_args_from_job_id(db, &flow).await?;
 
                     let should_stop = compute_bool_from_expr(
                         expr.to_string(),
@@ -1244,7 +1234,7 @@ fn get_module(flow_job: &QueuedJob, module_step: &Step) -> Option<FlowModule> {
     if let Some(raw_flow) = raw_flow {
         match module_step {
             Step::PreprocessorStep => raw_flow.preprocessor_module.map(|x| *x.clone()),
-            Step::Step(i) => raw_flow.modules.get(*i).map(|x| x.clone()),
+            Step::Step(i) => raw_flow.modules.get(*i).cloned(),
             Step::FailureStep => raw_flow.failure_module.map(|x| *x.clone()),
         }
     } else {
@@ -1269,8 +1259,7 @@ async fn compute_skip_branchall_failure<'c>(
             .map(|p| {
                 BRANCHALL_INDEX_RE
                     .captures(&p)
-                    .map(|x| x.get(1).unwrap().as_str().parse::<i32>().ok())
-                    .flatten()
+                    .and_then(|x| x.get(1).unwrap().as_str().parse::<i32>().ok())
                     .ok_or(Error::InternalErr(format!(
                         "could not parse branchall index from path: {p}"
                     )))
@@ -1291,14 +1280,14 @@ async fn compute_skip_branchall_failure<'c>(
 }
 
 async fn has_failure_module<'c>(flow: Uuid, db: &DB) -> Result<bool, Error> {
-    sqlx::query_scalar::<_, Option<bool>>(
-        "SELECT raw_flow->'failure_module' != 'null'::jsonb
-        FROM queue
-        WHERE id = $1",
+    fetch_one_with_fallback!(
+        db,
+        query_scalar,
+        Option<bool>,
+        "SELECT raw_flow->'failure_module' != 'null'::jsonb FROM {} WHERE id = $1",
+        "job_params" || "queue",
+        flow
     )
-    .bind(flow)
-    .fetch_one(db)
-    .await
     .map_err(|e| {
         Error::InternalErr(format!(
             "error during retrieval of has_failure_module: {e:#}"
@@ -1610,6 +1599,7 @@ pub struct ResumeRow {
 
 #[derive(FromRow)]
 pub struct RawArgs {
+    #[allow(dead_code)]
     pub args: Option<Json<HashMap<String, Box<RawValue>>>>,
 }
 
@@ -1638,6 +1628,7 @@ fn potentially_crash_for_testing() {
 lazy_static::lazy_static! {
     pub static ref EHM: HashMap<String, Box<RawValue>> = HashMap::new();
 }
+
 // #[async_recursion]
 // #[instrument(level = "trace", skip_all)]
 async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
@@ -1725,7 +1716,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
                     flow_job.workspace_id.as_str(),
                     flow_job.id
                 ).fetch_all(db).await?;
-                if overlapping.len() > 0 {
+                if !overlapping.is_empty() {
                     let overlapping_str = overlapping
                         .iter()
                         .map(|x| x.to_string())
@@ -1804,8 +1795,8 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
         // else pass the last job result. Either from the function arg if it's set, or manually fetch it from the previous job
         // having last_job_result empty can happen either when the job was suspended and is being restarted, or if it's a
         // flow restart from a specific step
-        if last_job_result.is_some() {
-            last_job_result.unwrap()
+        if let Some(last_job_result) = last_job_result {
+            last_job_result
         } else {
             match get_previous_job_result(db, flow_job.workspace_id.as_str(), &status).await? {
                 None => Arc::new(to_raw_value(&json!("{}"))),
@@ -2260,17 +2251,18 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
             );
             Ok(Marc::new(hm))
         } else if let Some(id) = get_args_from_id {
-            let row = sqlx::query_as::<_, RawArgs>(
-                "SELECT args FROM completed_job WHERE id = $1 AND workspace_id = $2",
-            )
-            .bind(id)
-            .bind(&flow_job.workspace_id)
-            .fetch_optional(db)
-            .await?;
+            let row = fetch_optional_with_fallback!(
+                db,
+                query_as,
+                RowArgs,
+                "SELECT args FROM {} WHERE id = $1 AND workspace_id = $2",
+                "job_args" || "completed_job",
+                id,
+                &flow_job.workspace_id
+            )?;
+
             if let Some(raw_args) = row {
-                Ok(Marc::new(
-                    raw_args.args.map(|x| x.0).unwrap_or_else(HashMap::new),
-                ))
+                Ok(Marc::new(raw_args.args.map(|x| x.0).unwrap_or_default()))
             } else {
                 Ok(Marc::new(HashMap::new()))
             }
@@ -2389,7 +2381,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
 
     let mut tx: QueueTransaction<'_, R> = (rsmq.clone(), db.begin().await?).into();
     let nargs = args.as_ref();
-    for i in (0..len).into_iter() {
+    for i in 0..len {
         if i % 100 == 0 && i != 0 {
             tracing::info!(id = %flow_job.id, root_id = %job_root, "pushed (non-commited yet) first {i} subflows of {len}");
             sqlx::query!(
@@ -3902,13 +3894,15 @@ async fn get_previous_job_result(
             Ok(Some(retrieve_flow_jobs_results(db, w_id, flow_jobs).await?))
         }
         Some(FlowStatusModule::Success { job, .. }) => Ok(Some(
-            sqlx::query_scalar::<_, Json<Box<RawValue>>>(
-                "SELECT result FROM completed_job WHERE id = $1 AND workspace_id = $2",
-            )
-            .bind(job)
-            .bind(w_id)
-            .fetch_one(db)
-            .await?
+            fetch_one_with_fallback!(
+                db,
+                query_scalar,
+                Json<Box<RawValue>>,
+                "SELECT result FROM {} WHERE id = $1 AND workspace_id = $2",
+                "completed_jobs_result" || "completed_job",
+                job,
+                w_id
+            )?
             .0,
         )),
         _ => Ok(None),

frontend/.gitignore

@@ -9,4 +9,5 @@ tests-out/
 storageState.json
 .env.production
 dist/
-static/tsdocs/
+static/tsdocs/
+!build/.gitkeep

frontend/src/lib/components/InputTransformForm.svelte

@@ -6,15 +6,13 @@
 	import ArgInput from './ArgInput.svelte'
 	import FieldHeader from './FieldHeader.svelte'
 	import DynamicInputHelpBox from './flows/content/DynamicInputHelpBox.svelte'
-	import type { PropPickerWrapperContext } from './prop_picker'
+	import type { PropPickerWrapperContext } from './flows/propPicker/PropPickerWrapper.svelte'
 	import { codeToStaticTemplate, getDefaultExpr } from './flows/utils'
 	import SimpleEditor from './SimpleEditor.svelte'
-	import { Button } from '$lib/components/common'
-	import AnimatedButton from '$lib/components/common/button/AnimatedButton.svelte'
+	import { Button } from './common'
 	import ToggleButtonGroup from '$lib/components/common/toggleButton-v2/ToggleButtonGroup.svelte'
 	import ToggleButton from '$lib/components/common/toggleButton-v2/ToggleButton.svelte'
-	import { fade } from 'svelte/transition'
-	import { tick } from 'svelte'
+
 	import type VariableEditor from './VariableEditor.svelte'
 	import type ItemPicker from './ItemPicker.svelte'
 	import type { InputTransform } from '$lib/gen'
@@ -25,8 +23,7 @@
 	import type { FlowCopilotContext } from './copilot/flow'
 	import StepInputGen from './copilot/StepInputGen.svelte'
 	import type { PickableProperties } from './flows/previousResults'
-	import { buildPrefixRegex } from './flows/previousResults'
-	import { twMerge } from 'tailwind-merge'
+
 	export let schema: Schema | { properties?: Record<string, any>; required?: string[] }
 	export let arg: InputTransform | any
 	export let argName: string
@@ -60,8 +57,6 @@
 	const { shouldUpdatePropertyType, exprsToSet } =
 		getContext<FlowCopilotContext | undefined>('FlowCopilotContext') || {}
 
-	const { inputMatches } = getContext<PropPickerWrapperContext>('PropPickerWrapper')
-
 	function setExpr() {
 		const newArg = $exprsToSet?.[argName]
 		if (newArg) {
@@ -136,54 +131,6 @@
 		}
 	}
 
-	let codeInjectionDetected = false
-
-	const dynamicTemplateRegexPairs = buildPrefixRegex([
-		'flow_input',
-		'results',
-		'resource',
-		'variable'
-	])
-
-	function checkCodeInjection(rawValue: string) {
-		if (!arg || !rawValue || rawValue.length < 3 || !dynamicTemplateRegexPairs) {
-			return undefined
-		}
-		const matches = dynamicTemplateRegexPairs.filter(({ regex }) => regex.test(rawValue))
-		if (matches.length > 0) {
-			return matches.map((m) => ({ word: m.word, value: rawValue }))
-		}
-		return undefined
-	}
-
-	async function setJavaScriptExpr(rawValue: string) {
-		arg = {
-			type: 'javascript',
-			expr: rawValue
-		}
-		propertyType = 'javascript'
-		monaco?.setCode('')
-		monaco?.insertAtCursor(rawValue)
-		await tick()
-		monaco?.focus()
-		await tick()
-		monaco?.setCursorToEnd()
-	}
-
-	function handleKeyUp(e: KeyboardEvent) {
-		if (
-			e.key === 'Tab' &&
-			isStaticTemplate(inputCat) &&
-			propertyType == 'static' &&
-			!noDynamicToggle &&
-			codeInjectionDetected
-		) {
-			setJavaScriptExpr(arg.value)
-		} else {
-			stepInputGen?.onKeyUp?.(e)
-		}
-	}
-
 	function isStaticTemplate(inputCat: InputCat) {
 		return inputCat === 'string' || inputCat === 'sql' || inputCat == 'yaml'
 	}
@@ -219,24 +166,7 @@
 
 	const { focusProp, propPickerConfig } = getContext<PropPickerWrapperContext>('PropPickerWrapper')
 
-	$: updateStaticInput(inputCat, propertyType, arg)
-
-	function updateStaticInput(
-		inputCat: InputCat,
-		propertyType: 'static' | 'javascript',
-		arg: InputTransform | any
-	) {
-		if (!isStaticTemplate(inputCat)) {
-			return
-		}
-		if (propertyType == 'static') {
-			setPropertyType(arg?.value)
-			codeInjectionDetected = !!checkCodeInjection(arg?.value)
-		} else if (propertyType == 'javascript' && focused) {
-			setPropertyType(arg?.expr)
-			$inputMatches = checkCodeInjection(arg?.expr)
-		}
-	}
+	$: isStaticTemplate(inputCat) && propertyType == 'static' && setPropertyType(arg?.value)
 
 	function setDefaultCode() {
 		if (!arg?.value) {
@@ -256,22 +186,11 @@
 	let stepInputGen: StepInputGen | undefined = undefined
 
 	loadResourceTypes()
-
-	$: connecting =
-		$propPickerConfig?.propName == argName && $propPickerConfig?.insertionMode == 'connect'
 </script>
 
 {#if arg != undefined}
-	<div
-		class={twMerge(
-			'pl-2 pt-2 pb-2 ml-2 relative hover:bg-surface hover:shadow-md transition-all duration-200',
-			$propPickerConfig?.propName == argName
-				? 'bg-surface border-l-4 border-blue-500 shadow-md rounded-l-md z-2000'
-				: 'hover:rounded-md',
-			$$props.class
-		)}
-	>
-		<div class="flex flex-row justify-between gap-1 pb-1 px-2">
+	<div class={$$props.class}>
+		<div class="flex flex-row justify-between gap-1 pb-1">
 			<div class="flex flex-wrap grow">
 				<FieldHeader
 					label={argName}
@@ -403,181 +322,147 @@
 						</ToggleButtonGroup>
 					</div>
 
-					<AnimatedButton animate={connecting} baseRadius="6px" animationDuration="1s">
-						<Button
-							variant="border"
-							color="light"
-							size="xs2"
-							btnClasses={connecting ? 'text-blue-500' : 'text-primary'}
-							on:click={() => {
-								focusProp(argName, 'connect', (path) => {
-									connectProperty(path)
-									dispatch('change', { argName })
-									return true
-								})
-							}}
-						>
-							<Plug size={16} /> &rightarrow;
-						</Button>
-					</AnimatedButton>
+					<Button
+						title="Connect to another node's output"
+						variant="border"
+						color="light"
+						size="xs2"
+						on:click={() => {
+							focusProp(argName, 'connect', (path) => {
+								connectProperty(path)
+								dispatch('change', { argName })
+								return true
+							})
+						}}
+						id="flow-editor-plug"
+					>
+						<Plug size={16} /> &rightarrow;
+					</Button>
 				</div>
 			{/if}
 		</div>
 
 		<div class="max-w-xs" />
 		<!-- svelte-ignore a11y-no-static-element-interactions -->
-		<div class="relative" on:keyup={handleKeyUp}>
-			<!-- {#if $propPickerConfig?.propName == argName && $propPickerConfig?.insertionMode == 'connect'}
+		<div
+			class="relative {$propPickerConfig?.propName == argName
+				? 'outline outline-offset-1 outline-1 outline-blue-500 rounded-md'
+				: ''}"
+			on:keyup={stepInputGen?.onKeyUp}
+		>
+			{#if $propPickerConfig?.propName == argName && $propPickerConfig?.insertionMode == 'connect'}
 				<span
 					class={'text-white  z-50 px-1 text-2xs py-0.5 font-bold rounded-t-sm w-fit absolute top-0 right-0 bg-blue-500'}
 				>
 					Connect input &rightarrow;
 				</span>
-			{/if} -->
+			{/if}
 			<!-- {inputCat}
 			{propertyType} -->
-			<div class="relative flex flex-row items-top gap-2 justify-between">
-				<div class="min-w-0 grow">
-					{#if isStaticTemplate(inputCat) && propertyType == 'static' && !noDynamicToggle}
-						{#if argName && schema?.properties?.[argName]?.description}
-							<div class="text-xs italic pb-1 text-secondary">
-								<pre class="font-main">{schema.properties[argName].description}</pre>
-							</div>
-						{/if}
-						<div class="mt-2 min-h-[28px]">
-							{#if arg}
-								<TemplateEditor
-									bind:this={monacoTemplate}
-									{extraLib}
-									on:focus={onFocus}
-									on:blur={() => {
-										focused = false
-									}}
-									bind:code={arg.value}
-									fontSize={14}
-									on:change={() => {
-										dispatch('change', { argName })
-									}}
-								/>
-							{/if}
-						</div>
-
-						{#if codeInjectionDetected}
-							<Button
-								size="xs"
-								color="light"
-								btnClasses="font-normal text-xs w-fit bg-green-100 text-green-800 hover:bg-green-100 dark:text-green-400 dark:bg-green-700 dark:hover:bg-green-700"
-								on:click={() => setJavaScriptExpr(arg.value)}
-							>
-								<span class="font-normal"
-									>JavaScript expression detected - press
-									<span class="font-bold">TAB</span> to exit static mode
-								</span>
-							</Button>
-						{/if}
-					{:else if (propertyType === undefined || propertyType == 'static') && schema?.properties?.[argName]}
-						<ArgInput
-							{resourceTypes}
-							noMargin
-							compact
-							bind:this={argInput}
+			{#if isStaticTemplate(inputCat) && propertyType == 'static' && !noDynamicToggle}
+				{#if argName && schema?.properties?.[argName]?.description}
+					<div class="text-xs italic pb-1 text-secondary">
+						<pre class="font-main">{schema.properties[argName].description}</pre>
+					</div>
+				{/if}
+				<div class="mt-2 min-h-[28px]">
+					{#if arg}
+						<TemplateEditor
+							bind:this={monacoTemplate}
+							{extraLib}
 							on:focus={onFocus}
 							on:blur={() => {
 								focused = false
 							}}
-							shouldDispatchChanges
+							bind:code={arg.value}
+							fontSize={14}
 							on:change={() => {
 								dispatch('change', { argName })
 							}}
-							label={argName}
-							bind:editor={monaco}
-							bind:description={schema.properties[argName].description}
-							bind:value={arg.value}
-							type={schema.properties[argName].type}
-							oneOf={schema.properties[argName].oneOf}
-							required={schema.required?.includes(argName)}
-							bind:pattern={schema.properties[argName].pattern}
-							bind:valid={inputCheck}
-							defaultValue={schema.properties[argName].default}
-							bind:enum_={schema.properties[argName].enum}
-							bind:format={schema.properties[argName].format}
-							contentEncoding={schema.properties[argName].contentEncoding}
-							bind:itemsType={schema.properties[argName].items}
-							properties={schema.properties[argName].properties}
-							nestedRequired={schema.properties[argName].required}
-							displayHeader={false}
-							extra={argExtra}
-							{variableEditor}
-							{itemPicker}
-							bind:pickForField
-							showSchemaExplorer
-							nullable={schema.properties[argName].nullable}
-							bind:title={schema.properties[argName].title}
-							bind:placeholder={schema.properties[argName].placeholder}
 						/>
-					{:else if arg.expr != undefined}
-						<div class="border mt-2">
-							<SimpleEditor
-								bind:this={monaco}
-								bind:code={arg.expr}
-								on:change={() => {
-									dispatch('change', { argName })
-								}}
-								{extraLib}
-								lang="javascript"
-								shouldBindKey={false}
-								on:focus={() => {
-									focused = true
-									focusProp(argName, 'insert', (path) => {
-										monaco?.insertAtCursor(path)
-										return false
-									})
-								}}
-								on:change={() => {
-									dispatch('change', { argName })
-								}}
-								on:blur={() => {
-									focused = false
-								}}
-								autoHeight
-							/>
-						</div>
-						<DynamicInputHelpBox />
-						<div class="mb-2" />
-					{:else}
-						Not recognized input type {argName} ({arg.expr}, {propertyType})
-						<div class="flex mt-2">
-							<Button
-								variant="border"
-								size="xs"
-								on:click={() => {
-									arg.expr = ''
-								}}>Set expr to empty string</Button
-							></div
-						>
 					{/if}
 				</div>
-
-				{#if $propPickerConfig?.propName == argName && ($propPickerConfig?.insertionMode == 'insert' || $propPickerConfig?.insertionMode == 'append')}
-					<div class="text-blue-500 mt-2" in:fade={{ duration: 200 }}>
-						<svg
-							xmlns="http://www.w3.org/2000/svg"
-							width="20"
-							height="20"
-							viewBox="0 0 24 24"
-							fill="currentColor"
-							stroke="currentColor"
-							stroke-width="2"
-							stroke-linecap="round"
-							stroke-linejoin="round"
-						>
-							<polyline points="24 24 12 12 24 0" />
-						</svg>
-					</div>
-				{:else}
-					<div class="w-0" />
-				{/if}
-			</div>
+			{:else if (propertyType === undefined || propertyType == 'static') && schema?.properties?.[argName]}
+				<ArgInput
+					{resourceTypes}
+					noMargin
+					compact
+					bind:this={argInput}
+					on:focus={onFocus}
+					on:blur={() => {
+						focused = false
+					}}
+					shouldDispatchChanges
+					on:change={() => {
+						dispatch('change', { argName })
+					}}
+					label={argName}
+					bind:editor={monaco}
+					bind:description={schema.properties[argName].description}
+					bind:value={arg.value}
+					type={schema.properties[argName].type}
+					oneOf={schema.properties[argName].oneOf}
+					required={schema.required?.includes(argName)}
+					bind:pattern={schema.properties[argName].pattern}
+					bind:valid={inputCheck}
+					defaultValue={schema.properties[argName].default}
+					bind:enum_={schema.properties[argName].enum}
+					bind:format={schema.properties[argName].format}
+					contentEncoding={schema.properties[argName].contentEncoding}
+					bind:itemsType={schema.properties[argName].items}
+					properties={schema.properties[argName].properties}
+					nestedRequired={schema.properties[argName].required}
+					displayHeader={false}
+					extra={argExtra}
+					{variableEditor}
+					{itemPicker}
+					bind:pickForField
+					showSchemaExplorer
+					nullable={schema.properties[argName].nullable}
+					bind:title={schema.properties[argName].title}
+					bind:placeholder={schema.properties[argName].placeholder}
+				/>
+			{:else if arg.expr != undefined}
+				<div class="border mt-2">
+					<SimpleEditor
+						bind:this={monaco}
+						bind:code={arg.expr}
+						on:change={() => {
+							dispatch('change', { argName })
+						}}
+						{extraLib}
+						lang="javascript"
+						shouldBindKey={false}
+						on:focus={() => {
+							focused = true
+							focusProp(argName, 'insert', (path) => {
+								monaco?.insertAtCursor(path)
+								return false
+							})
+						}}
+						on:change={() => {
+							dispatch('change', { argName })
+						}}
+						on:blur={() => {
+							focused = false
+						}}
+						autoHeight
+					/>
+				</div>
+				<DynamicInputHelpBox />
+				<div class="mb-2" />
+			{:else}
+				Not recognized input type {argName} ({arg.expr}, {propertyType})
+				<div class="flex mt-2">
+					<Button
+						variant="border"
+						size="xs"
+						on:click={() => {
+							arg.expr = ''
+						}}>Set expr to empty string</Button
+					></div
+				>
+			{/if}
 		</div>
 	</div>
 {/if}

frontend/src/lib/components/InputTransformSchemaForm.svelte

@@ -84,7 +84,7 @@
 	{#if keys.length > 0}
 		{#each keys as argName (argName)}
 			{#if (!filter || filter.includes(argName)) && Object.keys(schema.properties ?? {}).includes(argName)}
-				<div class="z-10 pt-2 relative">
+				<div class="z-10 pt-4">
 					<InputTransformForm
 						{previousModuleId}
 						bind:arg={args[argName]}
@@ -128,7 +128,7 @@
 >
 	<div
 		slot="submission"
-		class="flex flex-row-reverse w-full border-t border-gray-200 rounded-bl-lg rounded-br-lg"
+		class="flex flex-row-reverse w-full bg-surface border-t border-gray-200 rounded-bl-lg rounded-br-lg"
 	>
 		<Button
 			variant="border"

frontend/src/lib/components/SimpleEditor.svelte

@@ -405,15 +405,6 @@
 			editor && editor.dispose()
 		} catch (err) {}
 	})
-
-	export function setCursorToEnd(): void {
-		if (editor) {
-			const lastLine = editor.getModel()?.getLineCount() ?? 1
-			const lastColumn = editor.getModel()?.getLineMaxColumn(lastLine) ?? 1
-			editor.setPosition({ lineNumber: lastLine, column: lastColumn })
-			editor.focus()
-		}
-	}
 </script>
 
 <EditorTheme />

frontend/src/lib/components/common/button/AnimatedButton.svelte

@@ -1,87 +0,0 @@
-<script lang="ts">
-	import { twMerge } from 'tailwind-merge'
-
-	export let marginWidth = '2px'
-	export let animationDuration = '2s'
-	export let baseRadius = '4px'
-	export let animate = true
-	export let wrapperClasses = ''
-	export let ringColor = 'transparent'
-
-	let clientWidth = 0
-	let clientHeight = 0
-
-	$: circleRadius = Math.ceil(
-		Math.sqrt(clientWidth * clientWidth + clientHeight * clientHeight) / 2
-	)
-</script>
-
-<div
-	class={twMerge('gradient-button', wrapperClasses)}
-	style="--margin-width: {marginWidth}; --animation-duration: {animationDuration}; --base-radius: {baseRadius}; --circle-radius: {circleRadius}; --ring-color: {ringColor}"
-	class:animate
-	bind:clientWidth
-	bind:clientHeight
->
-	<slot />
-</div>
-
-<style>
-	.gradient-button {
-		position: relative;
-		padding: var(--margin-width, 2px);
-		font-size: inherit;
-		border: none;
-		border-radius: calc(var(--base-radius) + var(--margin-width, 2px));
-		color: currentColor;
-		background: inherit;
-		z-index: 1;
-		overflow: hidden;
-	}
-
-	/* Circular gradient */
-	.gradient-button::before {
-		content: '';
-		position: absolute;
-		top: 50%;
-		left: 50%;
-		transform: translate(-50%, -50%);
-		width: calc(var(--circle-radius, 300px) * 2px);
-		height: calc(var(--circle-radius, 300px) * 2px);
-		background: var(--ring-color, transparent);
-		border-radius: 50%;
-		z-index: -1;
-		animation: none;
-	}
-
-	.gradient-button.animate::before {
-		background: conic-gradient(from 0deg, #a4c7f2, #0c79fd, #104688, #a4c7f2);
-		animation: rotate var(--animation-duration, 2s) linear infinite;
-	}
-
-	/* inner background */
-	.gradient-button::after {
-		content: '';
-		position: absolute;
-		top: var(--margin-width, 2px);
-		right: var(--margin-width, 2px);
-		bottom: var(--margin-width, 2px);
-		left: var(--margin-width, 2px);
-		background: inherit;
-		border-radius: var(--base-radius);
-		z-index: -1;
-	}
-
-	@keyframes rotate {
-		from {
-			transform: translate(-50%, -50%) rotate(0deg);
-		}
-		to {
-			transform: translate(-50%, -50%) rotate(360deg);
-		}
-	}
-
-	.gradient-button.animate:hover::before {
-		animation-duration: 1s;
-	}
-</style>

frontend/src/lib/components/common/popup/Popup.svelte

@@ -19,9 +19,9 @@
 </script>
 
 <Popover on:close class="leading-none">
-	<PopoverButton let:open>
+	<PopoverButton>
 		<div use:floatingRef>
-			<slot name="button" {open} />
+			<slot name="button" />
 		</div>
 	</PopoverButton>
 	<ConditionalPortal condition={shouldUsePortal} {target}>

frontend/src/lib/components/flows/FlowEditor.svelte

@@ -4,13 +4,11 @@
 	import FlowModuleSchemaMap from './map/FlowModuleSchemaMap.svelte'
 	import WindmillIcon from '../icons/WindmillIcon.svelte'
 	import { Skeleton } from '../common'
-	import { getContext, setContext } from 'svelte'
+	import { getContext } from 'svelte'
 	import type { FlowEditorContext } from './types'
 	import type { FlowCopilotContext } from '../copilot/flow'
 	import { classNames } from '$lib/utils'
-	import { writable } from 'svelte/store'
-	import type { PropPickerWrapperContext, PropPickerConfig } from '$lib/components/prop_picker'
-	import type { PickableProperties } from '$lib/components/flows/previousResults'
+
 	const { flowStore } = getContext<FlowEditorContext>('FlowEditorContext')
 
 	export let loading: boolean
@@ -26,23 +24,6 @@
 
 	const { currentStepStore: copilotCurrentStepStore } =
 		getContext<FlowCopilotContext>('FlowCopilotContext')
-
-	const propPickerConfig = writable<PropPickerConfig | undefined>(undefined)
-	setContext<PropPickerWrapperContext>('PropPickerWrapper', {
-		propPickerConfig,
-		inputMatches: writable(undefined),
-		focusProp: (propName, insertionMode, onSelect) => {
-			propPickerConfig.set({
-				propName,
-				insertionMode,
-				onSelect
-			})
-		},
-		clearFocus: () => {
-			propPickerConfig.set(undefined)
-		},
-		filteredPickableProperties: writable<PickableProperties | undefined>(undefined)
-	})
 </script>
 
 <div

frontend/src/lib/components/flows/content/FlowConstants.svelte

@@ -4,7 +4,7 @@
 	import { Alert, Badge } from '$lib/components/common'
 	import type { FlowModule, FlowModuleValue, InputTransform, PathScript, RawScript } from '$lib/gen'
 	import { getContext, setContext } from 'svelte'
-	import type { PropPickerWrapperContext } from '$lib/components/prop_picker'
+	import type { PropPickerWrapperContext } from '../propPicker/PropPickerWrapper.svelte'
 	import { writable } from 'svelte/store'
 	import Toggle from '../../Toggle.svelte'
 	import InputTransformSchemaForm from '$lib/components/InputTransformSchemaForm.svelte'
@@ -79,9 +79,7 @@
 	setContext<PropPickerWrapperContext>('PropPickerWrapper', {
 		focusProp: () => {},
 		propPickerConfig: writable(undefined),
-		inputMatches: writable(undefined),
-		clearFocus: () => {},
-		filteredPickableProperties: writable(undefined)
+		clearFocus: () => {}
 	})
 </script>
 

frontend/src/lib/components/flows/content/FlowModuleComponent.svelte

@@ -378,11 +378,10 @@
 							class={advancedSelected === 'runtime' ? 'h-[calc(100%-68px)]' : 'h-[calc(100%-34px)]'}
 						>
 							{#if selected === 'inputs' && (flowModule.value.type == 'rawscript' || flowModule.value.type == 'script' || flowModule.value.type == 'flow')}
-								<div class="h-full overflow-auto px-2 bg-surface" id="flow-editor-step-input">
+								<div class="h-full overflow-auto px-2" id="flow-editor-step-input">
 									<PropPickerWrapper
 										pickableProperties={stepPropPicker.pickableProperties}
 										error={failureModule}
-										noPadding
 									>
 										<InputTransformSchemaForm
 											bind:this={inputTransformSchemaForm}

frontend/src/lib/components/flows/map/FlowModuleSchemaItem.svelte

@@ -28,8 +28,7 @@
 	import DrawerContent from '$lib/components/common/drawer/DrawerContent.svelte'
 	import { getDependeeAndDependentComponents } from '../flowExplorer'
 	import { replaceId } from '../flowStore'
-	import FlowPropPicker from '$lib/components/flows/propPicker/FlowPropPicker.svelte'
-	import type { PropPickerWrapperContext } from '$lib/components/prop_picker'
+
 	export let selected: boolean = false
 	export let deletable: boolean = false
 	export let retry: boolean = false
@@ -47,24 +46,17 @@
 	export let concurrency: boolean = false
 	export let retries: number | undefined = undefined
 	export let warningMessage: string | undefined = undefined
-	let pickableIds: Record<string, any> | undefined = undefined
 
 	const { flowInputsStore } = getContext<{ flowInputsStore: Writable<FlowInput | undefined> }>(
 		'FlowGraphContext'
 	)
 
 	const flowEditorContext = getContext<FlowEditorContext>('FlowEditorContext')
-
 	const dispatch = createEventDispatcher()
 
 	const { currentStepStore: copilotCurrentStepStore } =
 		getContext<FlowCopilotContext | undefined>('FlowCopilotContext') || {}
 
-	const { propPickerConfig, filteredPickableProperties } =
-		getContext<PropPickerWrapperContext>('PropPickerWrapper')
-
-	$: filteredPickableProperties && (pickableIds = $filteredPickableProperties?.priorIds)
-
 	let editId = false
 
 	let newId: string = id ?? ''
@@ -274,19 +266,6 @@ hover:border-blue-700 hover:!visible {hover ? '' : '!hidden'}"
 			{/if}
 		</div>
 	</div>
-
-	{#if id && $propPickerConfig && pickableIds && Object.keys(pickableIds).includes(id)}
-		<div class="absolute -bottom-[18px] right-[50%] translate-x-[50%]">
-			<FlowPropPicker
-				json={{
-					[id]: pickableIds[id]
-				}}
-				prefix={'results'}
-				viewOnly={false}
-			/>
-		</div>
-	{/if}
-
 	{#if deletable}
 		<button
 			class="absolute -top-[10px] -right-[10px] rounded-full h-[20px] w-[20px] trash center-center text-secondary
@@ -301,7 +280,7 @@ hover:border-blue-700 hover:!visible {hover ? '' : '!hidden'}"
 
 		{#if id !== 'preprocessor'}
 			<button
-				class="absolute -top-[10px] right-[60px] rounded-full h-[20px] w-[20px] center-center text-secondary
+				class="absolute -top-[10px] right-[60px] rounded-full h-[20px] w-[20px] trash center-center text-secondary
 outline-[1px] outline dark:outline-gray-500 outline-gray-300 bg-surface duration-150 hover:bg-blue-400 hover:text-white
  {hover ? '' : '!hidden'}"
 				on:click|preventDefault|stopPropagation={(event) => dispatch('move')}

frontend/src/lib/components/flows/map/VirtualItem.svelte

@@ -4,8 +4,6 @@
 	import { classNames } from '$lib/utils'
 	import { createEventDispatcher, getContext } from 'svelte'
 	import type { FlowCopilotContext } from '$lib/components/copilot/flow'
-	import type { PropPickerConfig } from '$lib/components/prop_picker'
-	import FlowPropPicker from '$lib/components/flows/propPicker/FlowPropPicker.svelte'
 
 	export let label: string | undefined = undefined
 	export let bgColor: string = ''
@@ -16,9 +14,6 @@
 	export let borderColor: string | undefined = undefined
 	export let hideId: boolean = false
 	export let preLabel: string | undefined = undefined
-	export let propPickerConfig: PropPickerConfig | undefined = undefined
-	export let inputJson = {}
-	export let prefix = ''
 
 	const dispatch = createEventDispatcher<{
 		insert: {
@@ -38,7 +33,7 @@
 <!-- svelte-ignore a11y-no-static-element-interactions -->
 <div
 	class={classNames(
-		'w-full flex relative rounded-sm',
+		'w-full flex relative overflow-hidden rounded-sm',
 		selectable ? 'cursor-pointer' : '',
 		selected ? 'outline outline-offset-1 outline-2  outline-gray-600' : '',
 		label === 'Input' && $copilotCurrentStepStore === 'Input' ? 'z-[901]' : ''
@@ -81,9 +76,4 @@
 			{/if}
 		</div>
 	</div>
-	{#if propPickerConfig && Object.keys(inputJson).length > 0}
-		<div class="absolute -bottom-[18px] right-[50%] translate-x-[50%]">
-			<FlowPropPicker json={inputJson} {prefix} />
-		</div>
-	{/if}
 </div>

frontend/src/lib/components/flows/previousResults.ts

@@ -260,42 +260,3 @@ declare const approvers: string
 }
 `
 }
-
-export function buildPrefixRegex(words: string[]): Array<{ regex: RegExp; word: string }> {
-	return words.map((word) => {
-		const prefixes: string[] = []
-		for (let i = 1; i <= word.length; i++) {
-			prefixes.push(word.slice(0, i) + '$')
-		}
-		prefixes.push(word + '\\.')
-		prefixes.push(word + '\\[')
-
-		return {
-			regex: new RegExp(`^(${prefixes.join('|')}).*`),
-			word
-		}
-	})
-}
-
-export function filterNestedObject(obj: any, nestedKeys: string[]) {
-	if (nestedKeys.length === 0) return obj
-	if (nestedKeys.length === 1) {
-		if (nestedKeys[0] === '') {
-			return obj
-		}
-		const regexes = buildPrefixRegex(Object.keys(obj))
-		const matches = regexes.filter(({ regex }) => regex.test(nestedKeys[0]))
-		const filteredObj = {}
-		matches.forEach(({ word }) => {
-			if (obj.hasOwnProperty(word)) {
-				filteredObj[word] = obj[word]
-			}
-		})
-		return filteredObj
-	}
-	const [key, ...rest] = nestedKeys
-	if (obj && typeof obj === 'object' && key in obj) {
-		return filterNestedObject(obj[key], rest)
-	}
-	return undefined
-}

frontend/src/lib/components/flows/propPicker/FlowPropPicker.svelte

@@ -1,64 +0,0 @@
-<script lang="ts">
-	import ObjectViewer from '$lib/components/propertyPicker/ObjectViewer.svelte'
-	import AnimatedButton from '$lib/components/common/button/AnimatedButton.svelte'
-	import { Popup } from '$lib/components/common'
-	import { Plug } from 'lucide-svelte'
-	import { twMerge } from 'tailwind-merge'
-	import type { PropPickerWrapperContext } from '$lib/components/prop_picker'
-	import { getContext } from 'svelte'
-	import Popover from '$lib/components/Popover.svelte'
-
-	export let json = {}
-	export let prefix = ''
-	export let viewOnly = false
-
-	const { propPickerConfig } = getContext<PropPickerWrapperContext>('PropPickerWrapper')
-</script>
-
-<button
-	on:click|preventDefault|stopPropagation={(e) => {
-		e.preventDefault()
-		e.stopPropagation()
-	}}
-	on:keydown|preventDefault|stopPropagation
-	data-prop-picker
->
-	<AnimatedButton
-		animate={$propPickerConfig?.insertionMode === 'connect' && !viewOnly}
-		wrapperClasses="h-[20px] w-[20px] "
-		baseRadius="9999px"
-		marginWidth="1px"
-	>
-		<Popup floatingConfig={{ strategy: 'fixed', placement: 'bottom-start' }}>
-			<svelte:fragment slot="button" let:open>
-				<Popover disablePopup={open}>
-					<svelte:fragment slot="text">node outputs</svelte:fragment>
-					<button
-						class={twMerge(
-							'rounded-full trash center-center h-[18px] w-[18px]',
-							viewOnly
-								? 'outline-[1px] outline dark:outline-gray-500 outline-gray-300 duration-150 bg-surface hover:bg-surface-hover text-secondary'
-								: $propPickerConfig?.insertionMode == 'connect'
-								? 'bg-surface text-blue-500'
-								: 'outline-[1px] outline dark:outline-gray-500 outline-gray-300 duration-150 bg-blue-500 hover:bg-blue-700 text-white'
-						)}
-					>
-						<Plug size={12} strokeWidth={2} />
-					</button>
-				</Popover>
-			</svelte:fragment>
-			<div data-prop-picker>
-				<ObjectViewer
-					{json}
-					topBrackets={false}
-					pureViewer={viewOnly}
-					{prefix}
-					on:select={(e) => {
-						$propPickerConfig?.onSelect(e.detail)
-						$propPickerConfig = undefined
-					}}
-				/>
-			</div>
-		</Popup>
-	</AnimatedButton>
-</button>

frontend/src/lib/components/flows/propPicker/PropPickerWrapper.svelte

@@ -1,13 +1,31 @@
+<script context="module" lang="ts">
+	type InsertionMode = 'append' | 'connect' | 'insert'
+
+	type SelectCallback = (path: string) => boolean
+
+	type PropPickerConfig = {
+		insertionMode: InsertionMode
+		propName: string
+		onSelect: SelectCallback
+	}
+
+	export type PropPickerWrapperContext = {
+		propPickerConfig: Writable<PropPickerConfig | undefined>
+		focusProp: (propName: string, insertionMode: InsertionMode, onSelect: SelectCallback) => void
+		clearFocus: () => void
+	}
+</script>
+
 <script lang="ts">
 	import PropPicker from '$lib/components/propertyPicker/PropPicker.svelte'
 	import PropPickerResult from '$lib/components/propertyPicker/PropPickerResult.svelte'
-	import { clickOutside } from '$lib/utils'
-	import { createEventDispatcher, getContext } from 'svelte'
+	import { clickOutside, sendUserToast } from '$lib/utils'
+	import { createEventDispatcher, setContext } from 'svelte'
 	import { Pane, Splitpanes } from 'svelte-splitpanes'
+	import { writable, type Writable } from 'svelte/store'
 	import type { PickableProperties } from '../previousResults'
 	import { twMerge } from 'tailwind-merge'
-	import AnimatedButton from '$lib/components/common/button/AnimatedButton.svelte'
-	import type { PropPickerWrapperContext } from '$lib/components/prop_picker'
+
 	export let pickableProperties: PickableProperties | undefined
 	export let result: any = undefined
 	export let extraResults: any = undefined
@@ -17,98 +35,74 @@
 	export let notSelectable = false
 	export let noPadding: boolean = false
 
+	const propPickerConfig = writable<PropPickerConfig | undefined>(undefined)
 	const dispatch = createEventDispatcher()
 
-	const { propPickerConfig } = getContext<PropPickerWrapperContext>('PropPickerWrapper')
-
-	async function getPropPickerElements(): Promise<HTMLElement[]> {
-		return Array.from(
-			document.querySelectorAll('[data-prop-picker], [data-prop-picker] *')
-		) as HTMLElement[]
-	}
+	setContext<PropPickerWrapperContext>('PropPickerWrapper', {
+		propPickerConfig,
+		focusProp: (propName, insertionMode, onSelect) => {
+			propPickerConfig.set({
+				propName,
+				insertionMode,
+				onSelect
+			})
+		},
+		clearFocus: () => {
+			propPickerConfig.set(undefined)
+		}
+	})
 </script>
 
 <div
 	class="h-full w-full"
-	data-prop-picker-root
-	use:clickOutside={{ capture: true, exclude: getPropPickerElements }}
-	on:click_outside={() => {
-		propPickerConfig.set(undefined)
-	}}
+	use:clickOutside
+	on:click_outside={() => propPickerConfig.set(undefined)}
 >
-	<Splitpanes class={$propPickerConfig ? 'splitpanes-remove-splitter' : ''}>
+	<Splitpanes>
 		<Pane
 			minSize={20}
 			size={60}
-			class={twMerge('relative !transition-none ', noPadding ? '' : 'p-2')}
+			class={twMerge('relative !transition-none', noPadding ? '' : 'p-2')}
 		>
 			<slot />
 		</Pane>
 		<Pane
 			minSize={20}
 			size={40}
-			class="!transition-none z-1000 {$propPickerConfig ? 'ml-[-1px]' : ''}"
+			class="pt-2 relative !transition-none {$propPickerConfig ? 'border-2 border-blue-500' : ''}"
 		>
-			<AnimatedButton
-				animate={$propPickerConfig?.insertionMode == 'connect'}
-				baseRadius="4px"
-				wrapperClasses="h-full w-full pt-2"
-				marginWidth="4px"
-				ringColor={$propPickerConfig?.insertionMode == 'insert' ||
-				$propPickerConfig?.insertionMode == 'append'
-					? '#3b82f6'
-					: 'transparent'}
-				animationDuration="1s"
-			>
-				{#if result}
-					<PropPickerResult
-						{result}
-						{extraResults}
-						{flow_input}
-						allowCopy={!notSelectable && !$propPickerConfig}
-						on:select={({ detail }) => {
-							dispatch('select', detail)
-							if ($propPickerConfig?.onSelect(detail)) {
-								propPickerConfig.set(undefined)
-							}
-						}}
-					/>
-				{:else if pickableProperties}
-					<PropPicker
-						{displayContext}
-						{error}
-						{pickableProperties}
-						{notSelectable}
-						allowCopy={!notSelectable && !$propPickerConfig}
-						on:select={({ detail }) => {
-							dispatch('select', detail)
-							if ($propPickerConfig?.onSelect(detail)) {
-								propPickerConfig.set(undefined)
-							}
-						}}
-					/>
-				{/if}
-			</AnimatedButton>
+			{#if result}
+				<PropPickerResult
+					{result}
+					{extraResults}
+					{flow_input}
+					on:select={({ detail }) => {
+						if (!notSelectable && !$propPickerConfig) {
+							sendUserToast('Set cursor within an input or click on the plug first', true)
+						}
+						dispatch('select', detail)
+						if ($propPickerConfig?.onSelect(detail)) {
+							propPickerConfig.set(undefined)
+						}
+					}}
+				/>
+			{:else if pickableProperties}
+				<PropPicker
+					{displayContext}
+					{error}
+					{pickableProperties}
+					{notSelectable}
+					on:select={({ detail }) => {
+						if (!notSelectable && !$propPickerConfig) {
+							sendUserToast('Set cursor within an input or click on the plug first', true)
+						}
+						dispatch('select', detail)
+						if ($propPickerConfig?.onSelect(detail)) {
+							propPickerConfig.set(undefined)
+						}
+					}}
+				/>
+			{/if}
 		</Pane>
 	</Splitpanes>
 </div>
-
-<style>
-	:global(.splitpanes-remove-splitter > .splitpanes__pane) {
-		background-color: inherit !important;
-	}
-	:global(.splitpanes-remove-splitter > .splitpanes__splitter) {
-		background-color: transparent !important;
-		width: 0 !important;
-		border: none !important;
-	}
-
-	:global(.splitpanes__pane) {
-		overflow-y: auto;
-		scrollbar-width: none;
-	}
-
-	:global(.splitpanes__pane:hover) {
-		scrollbar-width: thin;
-	}
-</style>

frontend/src/lib/components/graph/renderers/nodes/ForLoopStartNode.svelte

@@ -5,8 +5,7 @@
 	import { getStateColor } from '../../util'
 	import type { GraphModuleState } from '../../model'
 	import type { GraphEventHandlers } from '../../graphBuilder'
-	import { getContext } from 'svelte'
-	import type { PropPickerWrapperContext } from '$lib/components/prop_picker'
+
 	export let data: {
 		offset: number
 		id: string
@@ -14,21 +13,11 @@
 		flowModuleStates: Record<string, GraphModuleState> | undefined
 		eventHandlers: GraphEventHandlers
 	}
-
-	const { propPickerConfig, filteredPickableProperties } =
-		getContext<PropPickerWrapperContext>('PropPickerWrapper')
-
-	$: filteredInput = filterIterFromInput($filteredPickableProperties?.flow_input)
-
-	function filterIterFromInput(inputJson: Record<string, any> | undefined): Record<string, any> {
-		if (!inputJson || typeof inputJson !== 'object' || !inputJson.iter) return {}
-		return { iter: inputJson.iter }
-	}
 </script>
 
 <NodeWrapper let:darkMode offset={data.offset}>
 	<VirtualItem
-		label={'Do one iterations'}
+		label={'Do one iteration'}
 		selectable={false}
 		selected={false}
 		id={data.id}
@@ -38,8 +27,5 @@
 		on:select={(e) => {
 			data?.eventHandlers?.select(e.detail)
 		}}
-		propPickerConfig={$propPickerConfig}
-		inputJson={filteredInput}
-		prefix="flow_input"
 	/>
 </NodeWrapper>

frontend/src/lib/components/graph/renderers/nodes/InputNode.svelte

@@ -7,7 +7,6 @@
 	import { getContext } from 'svelte'
 	import type { Writable } from 'svelte/store'
 	import InsertModuleButton from '$lib/components/flows/map/InsertModuleButton.svelte'
-	import type { PropPickerWrapperContext } from '$lib/components/prop_picker'
 
 	export let data: {
 		hasPreprocessor: boolean
@@ -24,20 +23,6 @@
 	const { selectedId } = getContext<{
 		selectedId: Writable<string | undefined>
 	}>('FlowGraphContext')
-
-	const { propPickerConfig, filteredPickableProperties } =
-		getContext<PropPickerWrapperContext>('PropPickerWrapper')
-
-	function filterIterFromInput(inputJson: Record<string, any> | undefined): Record<string, any> {
-		if (!inputJson || typeof inputJson !== 'object') return {}
-
-		const newJson = { ...inputJson }
-		delete newJson.iter
-
-		return newJson
-	}
-
-	$: filteredInput = filterIterFromInput($filteredPickableProperties?.flow_input)
 </script>
 
 <NodeWrapper let:darkMode>
@@ -79,8 +64,5 @@
 		on:select={(e) => {
 			data.eventHandlers?.select(e.detail)
 		}}
-		propPickerConfig={$propPickerConfig}
-		inputJson={filteredInput}
-		prefix="flow_input"
 	/>
 </NodeWrapper>

frontend/src/lib/components/prop_picker.ts

@@ -1,20 +0,0 @@
-import type { Writable } from 'svelte/store'
-import type { PickableProperties } from '$lib/components/flows/previousResults'
-
-type InsertionMode = 'append' | 'connect' | 'insert'
-
-type SelectCallback = (path: string) => boolean
-
-export type PropPickerConfig = {
-	insertionMode: InsertionMode
-	propName: string
-	onSelect: SelectCallback
-}
-
-export type PropPickerWrapperContext = {
-	propPickerConfig: Writable<PropPickerConfig | undefined>
-	filteredPickableProperties: Writable<PickableProperties | undefined>
-	inputMatches: Writable<{ word: string; value: string }[] | undefined>
-	focusProp: (propName: string, insertionMode: InsertionMode, onSelect: SelectCallback) => void
-	clearFocus: () => void
-}

frontend/src/lib/components/propertyPicker/ObjectViewer.svelte

@@ -1,12 +1,13 @@
 <script lang="ts">
-	import { copyToClipboard, truncate } from '$lib/utils'
+	import { copyToClipboard, pluralize, truncate } from '$lib/utils'
 
 	import { createEventDispatcher } from 'svelte'
+	import { Badge } from '../common'
 	import { computeKey } from './utils'
+	import WarningMessage from './WarningMessage.svelte'
 	import { NEVER_TESTED_THIS_FAR } from '../flows/models'
 	import Portal from '$lib/components/Portal.svelte'
-	import { Button } from '$lib/components/common'
-	import Popover from '$lib/components/Popover.svelte'
+
 	import { Download, PanelRightOpen } from 'lucide-svelte'
 	import S3FilePicker from '../S3FilePicker.svelte'
 	import { workspaceStore } from '$lib/stores'
@@ -18,9 +19,9 @@
 	export let collapsed = (level != 0 && level % 3 == 0) || Array.isArray(json)
 	export let rawKey = false
 	export let topBrackets = false
+	export let topLevelNode = false
 	export let allowCopy = true
 	export let collapseLevel: number | undefined = undefined
-	export let prefix = ''
 
 	let s3FileViewer: S3FilePicker
 
@@ -49,21 +50,12 @@
 
 	const dispatch = createEventDispatcher()
 
-	function computeFullKey(key: string, rawKey: boolean) {
-		if (rawKey) {
-			return `${prefix}('${key}')`
-		}
-		const keyToSelect = computeKey(key, isArray, currentPath)
-		const separator = !prefix || keyToSelect.startsWith('[') ? '' : '.'
-		return prefix + separator + keyToSelect
-	}
-
 	function selectProp(key: string, value: any | undefined = undefined) {
-		const fullKey = computeFullKey(key, rawKey)
 		if (pureViewer && allowCopy) {
-			copyToClipboard(fullKey)
+			const valueToCopy = value !== undefined ? value : computeKey(key, isArray, currentPath)
+			copyToClipboard(valueToCopy)
 		}
-		dispatch('select', fullKey)
+		dispatch('select', rawKey ? key : computeKey(key, isArray, currentPath))
 	}
 
 	$: keyLimit = isArray ? 1 : 100
@@ -81,40 +73,33 @@
 			{#if level != 0 && keys.length > 1}
 				<!-- svelte-ignore a11y-click-events-have-key-events -->
 				<!-- svelte-ignore a11y-no-static-element-interactions -->
-				<Button
-					color="light"
-					size="xs2"
-					variant="border"
-					on:click={collapse}
-					wrapperClasses="inline-flex w-fit h-5"
-					btnClasses="font-semibold text-primary border-nord-300 rounded-[0.275rem]">-</Button
-				>
+				<span class="cursor-pointer border hover:bg-surface-hover px-1 rounded" on:click={collapse}>
+					-
+				</span>
 			{/if}
 			{#if level == 0 && topBrackets}<span class="h-0">{openBracket}</span>{/if}
 			<ul class={`w-full pl-2 ${level === 0 ? 'border-none' : 'border-l border-dotted'}`}>
 				{#each keys.length > keyLimit ? keys.slice(0, keyLimit) : keys as key, index (key)}
 					<li>
-						<Popover>
-							<svelte:fragment slot="text">{computeFullKey(key, rawKey)}</svelte:fragment>
-							<Button
-								on:click={() => selectProp(key)}
-								size="xs2"
-								color="dark"
-								variant="contained"
-								wrapperClasses="inline-flex p-0 whitespace-nowrap w-fit h-4"
-								btnClasses="font-normal rounded-[0.275rem]"
-							>
-								<span class={pureViewer ? 'cursor-auto' : ''}>
-									{!isArray ? key : index}
-								</span>
-							</Button>
-						</Popover>
-						:
+						<button on:click={() => selectProp(key)} class="whitespace-nowrap">
+							{#if topLevelNode}
+								<Badge baseClass="border border-blue-600" color="indigo">{key}</Badge>
+							{:else}
+								<span
+									class="key {pureViewer
+										? 'cursor-auto'
+										: 'border '} font-semibold rounded px-1 hover:bg-surface-hover text-2xs text-secondary"
+								>
+									{!isArray ? key : index}</span
+								>
+							{/if}:
+						</button>
+
 						{#if getTypeAsString(json[key]) === 'object'}
 							<svelte:self
 								json={json[key]}
 								level={level + 1}
-								currentPath={computeFullKey(key, isArray)}
+								currentPath={computeKey(key, isArray, currentPath)}
 								{pureViewer}
 								{allowCopy}
 								on:select
@@ -122,38 +107,28 @@
 								collapsed={collapseLevel !== undefined ? level + 1 >= collapseLevel : undefined}
 							/>
 						{:else}
-							<Popover disablePopup={!json[key]}>
-								<svelte:fragment slot="text">
-									{JSON.stringify(json[key])}
-								</svelte:fragment>
-								<button
-									class="val text-left {pureViewer
-										? 'cursor-auto'
-										: ''} rounded px-1 {getTypeAsString(json[key])}"
-									on:click={() => {
-										if (json[key]) {
-											copyToClipboard(json[key])
-										}
-									}}
-									disabled={false}
-								>
-									{#if json[key] === NEVER_TESTED_THIS_FAR}
-										<span class="text-2xs text-tertiary font-normal">
-											Test the flow to see a value
-										</span>
-									{:else if json[key] == undefined}
-										<span class="text-2xs">undefined</span>
-									{:else if json[key] == null}
-										<span class="text-2xs">null</span>
-									{:else if typeof json[key] == 'string'}
-										<span class="text-2xs">"{truncate(json[key], 200)}"</span>
-									{:else}
-										<span class="text-2xs">
-											{truncate(JSON.stringify(json[key]), 200)}
-										</span>
-									{/if}
-								</button>
-							</Popover>
+							<button
+								class="val text-left {pureViewer
+									? 'cursor-auto'
+									: ''} rounded px-1 hover:bg-blue-100 dark:hover:bg-blue-100/10 {getTypeAsString(
+									json[key]
+								)}"
+								on:click={() => selectProp(key, json[key])}
+							>
+								{#if json[key] === NEVER_TESTED_THIS_FAR}
+									<WarningMessage />
+								{:else if json[key] == undefined}
+									<span class="text-2xs">undefined</span>
+								{:else if json[key] == null}
+									<span class="text-2xs">null</span>
+								{:else if typeof json[key] == 'string'}
+									<span title={json[key]} class="text-2xs">"{truncate(json[key], 200)}"</span>
+								{:else}
+									<span title={JSON.stringify(json[key])} class="text-2xs">
+										{truncate(JSON.stringify(json[key]), 200)}
+									</span>
+								{/if}
+							</button>
 						{/if}
 					</li>
 				{/each}
@@ -192,18 +167,17 @@
 
 	<!-- svelte-ignore a11y-click-events-have-key-events -->
 	<!-- svelte-ignore a11y-no-static-element-interactions -->
-
+	<span
+		class="border border-blue-600 rounded px-1 cursor-pointer hover:bg-gray-200"
+		class:hidden={!fullyCollapsed}
+		on:click={collapse}
+	>
+		{openBracket}{collapsedSymbol}{closeBracket}
+	</span>
 	{#if fullyCollapsed}
-		<Button
-			color="light"
-			size="xs2"
-			variant="border"
-			on:click={collapse}
-			wrapperClasses="inline-flex w-fit h-5"
-			btnClasses="font-semibold border-nord-300 rounded-[0.275rem] p-1"
-		>
-			{openBracket}{collapsedSymbol}{closeBracket}
-		</Button>
+		<span class="text-tertiary text-xs">
+			{pluralize(Object.keys(json).length, Array.isArray(json) ? 'item' : 'key')}
+		</span>
 	{/if}
 {:else if topBrackets}
 	<span class="text-primary">{openBracket}{closeBracket}</span>

frontend/src/lib/components/propertyPicker/PropPicker.svelte

@@ -3,45 +3,41 @@
 	import { workspaceStore } from '$lib/stores'
 	import { getContext } from 'svelte'
 	import { Badge, Button } from '../common'
-	import type { PropPickerWrapperContext } from '../prop_picker'
+	import type { PropPickerWrapperContext } from '../flows/propPicker/PropPickerWrapper.svelte'
+	import { createEventDispatcher } from 'svelte'
 
 	import ObjectViewer from './ObjectViewer.svelte'
 	import { keepByKey } from './utils'
 	import type { PickableProperties } from '../flows/previousResults'
 	import ClearableInput from '../common/clearableInput/ClearableInput.svelte'
-	import { filterNestedObject } from '../flows/previousResults'
 
 	export let pickableProperties: PickableProperties
 	export let displayContext = true
 	export let notSelectable: boolean
 	export let error: boolean = false
-	export let allowCopy = false
 
 	$: previousId = pickableProperties?.previousId
 	let variables: Record<string, string> = {}
 	let resources: Record<string, any> = {}
 	let displayVariable = false
 	let displayResources = false
-	let allResultsCollapsed = true
-	let flowInputsFiltered: Record<string, any> = {}
-	let resultByIdFiltered: Record<string, any> = {}
-	let collapsableInitialState:
-		| {
-				allResultsCollapsed: boolean
-				displayVariable: boolean
-				displayResources: boolean
-		  }
-		| undefined
+
+	const dispatch = createEventDispatcher()
 
 	const EMPTY_STRING = ''
 	let search = ''
 
-	const { propPickerConfig, filteredPickableProperties, inputMatches } =
-		getContext<PropPickerWrapperContext>('PropPickerWrapper')
+	const { propPickerConfig } = getContext<PropPickerWrapperContext>('PropPickerWrapper')
 
-	$filteredPickableProperties = { ...pickableProperties }
+	$: flowInputsFiltered =
+		search === EMPTY_STRING
+			? pickableProperties.flow_input
+			: keepByKey(pickableProperties.flow_input, search)
 
-	$: filterPickableProperties(), updateCollapsable(), search, $inputMatches
+	$: resultByIdFiltered =
+		search === EMPTY_STRING
+			? pickableProperties.priorIds
+			: keepByKey(pickableProperties.priorIds, search)
 
 	$: suggestedPropsFiltered = $propPickerConfig
 		? keepByKey(pickableProperties.priorIds, $propPickerConfig.propName)
@@ -66,90 +62,23 @@
 			).map((resource) => [resource.path, resource.description ?? ''])
 		)
 	}
-
-	function filterPickableProperties() {
-		flowInputsFiltered = pickableProperties.flow_input
-		resultByIdFiltered = pickableProperties.priorIds
-
-		if ($inputMatches) {
-			if (!$inputMatches.some((match) => match.word === 'flow_input')) {
-				flowInputsFiltered = []
-			}
-			if (!$inputMatches.some((match) => match.word === 'results')) {
-				resultByIdFiltered = []
-			}
-			if ($inputMatches.length == 1) {
-				if ($inputMatches[0].word === 'flow_input') {
-					let [, ...nestedKeys] = $inputMatches[0].value.split('.')
-					flowInputsFiltered = filterNestedObject(flowInputsFiltered, nestedKeys)
-				} else if ($inputMatches[0].word === 'results') {
-					let [, ...nestedKeys] = $inputMatches[0].value.split('.')
-					resultByIdFiltered = filterNestedObject(resultByIdFiltered, nestedKeys)
-				}
-			}
-		}
-
-		if (flowInputsFiltered && search !== EMPTY_STRING) {
-			flowInputsFiltered = keepByKey(flowInputsFiltered, search)
-		}
-		if (resultByIdFiltered && search !== EMPTY_STRING) {
-			resultByIdFiltered = keepByKey(resultByIdFiltered, search)
-		}
-
-		if ($filteredPickableProperties) {
-			resultByIdFiltered && ($filteredPickableProperties.priorIds = resultByIdFiltered)
-			flowInputsFiltered && ($filteredPickableProperties.flow_input = flowInputsFiltered)
-		}
-	}
-
-	async function updateCollapsable() {
-		if (!$inputMatches || $inputMatches.length !== 1) {
-			resetCollapsable()
-			return
-		}
-
-		if (!collapsableInitialState) {
-			collapsableInitialState = { allResultsCollapsed, displayVariable, displayResources }
-		}
-
-		if ($inputMatches[0].word === 'variable') {
-			await loadVariables()
-			displayVariable = true
-			return
-		}
-		if ($inputMatches[0].word === 'resource') {
-			await loadResources()
-			displayResources = true
-			return
-		}
-		if ($inputMatches[0].word === 'results') {
-			allResultsCollapsed = false
-			return
-		}
-	}
-
-	function resetCollapsable() {
-		if (!collapsableInitialState) {
-			return
-		}
-		;({ allResultsCollapsed, displayVariable, displayResources } = collapsableInitialState)
-		collapsableInitialState = undefined
-	}
 </script>
 
-<div class="flex flex-col h-full !bg-surface rounded overflow-hidden">
+<div class="flex flex-col h-full">
 	<div class="px-2">
 		{#if !notSelectable}
-			{#if $propPickerConfig}
-				<!-- <Badge large color="blue">
-					{`Selected: ${$propPickerConfig?.propName}`}
-				</Badge> -->
-				<Badge large color="blue">
-					{`Mode: ${$propPickerConfig?.insertionMode}`}
-				</Badge>
-			{:else}
-				<Badge large color="blue">&leftarrow; Edit or connect an input</Badge>
-			{/if}
+			<div class="flex flex-row space-x-1">
+				{#if $propPickerConfig}
+					<Badge large color="blue">
+						{`Selected: ${$propPickerConfig?.propName}`}
+					</Badge>
+					<Badge large color="blue">
+						{`Mode: ${$propPickerConfig?.insertionMode}`}
+					</Badge>
+				{:else}
+					<Badge large color="blue">&leftarrow; Edit or connect an input</Badge>
+				{/if}
+			</div>
 		{/if}
 		<ClearableInput bind:value={search} placeholder="Search prop..." wrapperClass="py-2" />
 	</div>
@@ -157,26 +86,28 @@
 		class="overflow-y-auto px-2 pt-2 grow"
 		class:bg-surface-secondary={!$propPickerConfig && !notSelectable}
 	>
-		{#if flowInputsFiltered && Object.keys(flowInputsFiltered).length > 0}
-			<div class="flex justify-between items-center space-x-1">
-				<span class="font-normal text-sm text-secondary">Flow Input</span>
-				<div class="flex space-x-2 items-center" />
-			</div>
-			<div class="overflow-y-auto mb-2">
-				<ObjectViewer
-					{allowCopy}
-					pureViewer={!$propPickerConfig}
-					json={flowInputsFiltered}
-					prefix="flow_input"
-					on:select
-				/>
-			</div>
-		{/if}
+		<div class="flex justify-between items-center space-x-1">
+			<span class="font-bold text-sm">Flow Input</span>
+			<div class="flex space-x-2 items-center" />
+		</div>
+		<div class="overflow-y-auto mb-2">
+			<ObjectViewer
+				allowCopy={false}
+				pureViewer={!$propPickerConfig}
+				json={flowInputsFiltered}
+				on:select={(e) => {
+					dispatch(
+						'select',
+						e.detail?.startsWith('[') ? `flow_input${e.detail}` : `flow_input.${e.detail}`
+					)
+				}}
+			/>
+		</div>
 		{#if error}
-			<span class="font-normal text-sm text-secondary">Error</span>
+			<span class="font-bold text-sm">Error</span>
 			<div class="overflow-y-auto mb-2">
 				<ObjectViewer
-					{allowCopy}
+					allowCopy={false}
 					pureViewer={!$propPickerConfig}
 					json={{
 						error: {
@@ -191,186 +122,160 @@
 			</div>
 			{#if Object.keys(pickableProperties.priorIds).length > 0}
 				{#if suggestedPropsFiltered && Object.keys(suggestedPropsFiltered).length > 0}
-					<span class="font-normal text-sm text-secondary">Suggested Results</span>
+					<span class="font-bold text-sm">Suggested Results</span>
 					<div class="overflow-y-auto mb-2">
 						<ObjectViewer
-							{allowCopy}
+							allowCopy={false}
+							topLevelNode
 							pureViewer={!$propPickerConfig}
 							collapsed={false}
 							json={suggestedPropsFiltered}
-							prefix="results"
-							on:select
+							on:select={(e) => {
+								dispatch('select', `results.${e.detail}`)
+							}}
 						/>
 					</div>
 				{/if}
-				<span class="font-normal text-sm text-secondary">All Results</span>
+				<span class="font-bold text-sm">All Results</span>
 				<div class="overflow-y-auto mb-2">
 					<ObjectViewer
-						{allowCopy}
+						allowCopy={false}
+						topLevelNode
 						pureViewer={!$propPickerConfig}
 						collapsed={true}
 						json={resultByIdFiltered}
-						prefix="results"
-						on:select
+						on:select={(e) => {
+							dispatch('select', `results.${e.detail}`)
+						}}
 					/>
 				</div>
 			{/if}
 		{:else}
-			{@const json = Object.fromEntries(
-				Object.entries(resultByIdFiltered).filter(([k, v]) => k == previousId)
-			)}
-			{#if previousId && Object.keys(json).length > 0}
-				<span class="font-normal text-sm text-secondary">Previous Result</span>
+			{#if previousId}
+				<span class="font-bold text-sm">Previous Result</span>
 				<div class="overflow-y-auto mb-2">
 					<ObjectViewer
-						{allowCopy}
+						allowCopy={false}
+						topLevelNode
 						pureViewer={!$propPickerConfig}
-						{json}
-						prefix="results"
-						on:select
+						json={Object.fromEntries(
+							Object.entries(resultByIdFiltered).filter(([k, v]) => k == previousId)
+						)}
+						on:select={(e) => {
+							dispatch('select', `results.${e.detail}`)
+						}}
 					/>
 				</div>
 			{/if}
 			{#if pickableProperties.hasResume}
-				<span class="font-normal text-sm text-secondary">Resume payloads</span>
+				<span class="font-bold text-sm">Resume payloads</span>
 				<div class="overflow-y-auto mb-2">
 					<ObjectViewer
-						{allowCopy}
+						allowCopy={false}
+						topLevelNode
 						pureViewer={!$propPickerConfig}
 						json={{
 							resume: 'The resume payload',
 							resumes: 'All resume payloads from all approvers',
 							approvers: 'The list of approvers'
 						}}
-						on:select
+						on:select={(e) => {
+							dispatch('select', `${e.detail}`)
+						}}
 					/>
 				</div>
 			{/if}
 			{#if Object.keys(pickableProperties.priorIds).length > 0}
-				{#if !$inputMatches && suggestedPropsFiltered && Object.keys(suggestedPropsFiltered).length > 0}
-					<span class="font-normal text-sm text-secondary">Suggested Results</span>
+				{#if suggestedPropsFiltered && Object.keys(suggestedPropsFiltered).length > 0}
+					<span class="font-bold text-sm">Suggested Results</span>
 					<div class="overflow-y-auto mb-2">
 						<ObjectViewer
-							{allowCopy}
+							allowCopy={false}
+							topLevelNode
 							pureViewer={!$propPickerConfig}
 							collapsed={false}
 							json={suggestedPropsFiltered}
-							prefix="results"
-							on:select
-						/>
-					</div>
-				{/if}
-				{#if Object.keys(resultByIdFiltered).length > 0}
-					<div class="overflow-y-auto mb-2">
-						<span class="font-normal text-sm text-tertiary">All Results :</span>
-						{#if !allResultsCollapsed}
-							<Button
-								color="light"
-								size="xs2"
-								variant="border"
-								on:click={() => {
-									allResultsCollapsed = true
-								}}
-								wrapperClasses="inline-flex w-fit h-4"
-								btnClasses="font-normal text-primary border-nord-300 rounded-[0.275rem]">-</Button
-							>
-						{/if}
-
-						<ObjectViewer
-							{allowCopy}
-							pureViewer={!$propPickerConfig}
-							bind:collapsed={allResultsCollapsed}
-							json={resultByIdFiltered}
-							prefix="results"
-							on:select
+							on:select={(e) => {
+								dispatch('select', `results.${e.detail}`)
+							}}
 						/>
 					</div>
 				{/if}
+				<span class="font-bold text-sm">All Results</span>
+				<div class="overflow-y-auto mb-2">
+					<ObjectViewer
+						allowCopy={false}
+						topLevelNode
+						pureViewer={!$propPickerConfig}
+						collapsed={true}
+						json={resultByIdFiltered}
+						on:select={(e) => {
+							dispatch('select', `results.${e.detail}`)
+						}}
+					/>
+				</div>
 			{/if}
 		{/if}
 
 		{#if displayContext}
-			{#if !$inputMatches || $inputMatches.some((match) => match.word === 'variable')}
-				<div class="overflow-y-auto mb-2">
-					<span class="font-normal text-sm text-secondary">Variables :</span>
-
-					{#if displayVariable}
+			<span class="font-bold text-sm">Variables </span>
+			<div class="overflow-y-auto mb-2">
+				{#if displayVariable}
+					<div class="flex">
 						<Button
 							color="light"
-							size="xs2"
+							size="xs"
 							variant="border"
 							on:click={() => {
 								displayVariable = false
-							}}
-							wrapperClasses="inline-flex w-fit h-4"
-							btnClasses="font-normal text-primary border-nord-300 rounded-[0.275rem]">-</Button
+							}}>-</Button
 						>
-
-						<ObjectViewer
-							{allowCopy}
-							pureViewer={!$propPickerConfig}
-							rawKey={true}
-							json={variables}
-							prefix="variable"
-							on:select
-						/>
-					{:else}
-						<Button
-							color="light"
-							size="xs2"
-							variant="border"
-							on:click={async () => {
-								await loadVariables()
-								displayVariable = true
-							}}
-							wrapperClasses="inline-flex w-fit h-5"
-							btnClasses="font-semibold border-nord-300 rounded-[0.275rem] p-1"
-						>
-							{'{...}'}
-						</Button>
-					{/if}
-				</div>
-			{/if}
-			{#if !$inputMatches || $inputMatches.some((match) => match.word === 'resource')}
-				<div class="overflow-y-auto mb-2">
-					<span class="font-normal text-sm text-secondary">Resources :</span>
-
-					{#if displayResources}
-						<Button
-							color="light"
-							size="xs2"
-							variant="border"
-							on:click={() => {
-								displayResources = false
-							}}
-							wrapperClasses="inline-flex w-fit h-5"
-							btnClasses="font-semibold text-primary border-nord-300 rounded-[0.275rem]">-</Button
-						>
-						<ObjectViewer
-							{allowCopy}
-							pureViewer={!$propPickerConfig}
-							rawKey={true}
-							json={resources}
-							prefix="resource"
-							on:select
-						/>
-					{:else}
-						<Button
-							color="light"
-							size="xs2"
-							variant="border"
-							on:click={async () => {
-								await loadResources()
-								displayResources = true
-							}}
-							wrapperClasses="inline-flex w-fit h-5"
-							btnClasses="font-semibold border-nord-300 rounded-[0.275rem] p-1"
-						>
-							{'{...}'}
-						</Button>
-					{/if}
-				</div>
-			{/if}
+					</div>
+					<ObjectViewer
+						allowCopy={false}
+						pureViewer={!$propPickerConfig}
+						rawKey={true}
+						json={variables}
+						on:select={(e) => dispatch('select', `variable('${e.detail}')`)}
+					/>
+				{:else}
+					<button
+						class="border border-blue-600 key font-normal rounded hover:bg-blue-100 px-1"
+						on:click={async () => {
+							await loadVariables()
+							displayVariable = true
+						}}>{'{...}'}</button
+					>
+				{/if}
+			</div>
+			<span class="font-bold text-sm">Resources</span>
+			<div class="overflow-y-auto mb-2">
+				{#if displayResources}
+					<Button
+						color="light"
+						variant="border"
+						size="xs"
+						on:click={() => {
+							displayResources = false
+						}}>-</Button
+					>
+					<ObjectViewer
+						allowCopy={false}
+						pureViewer={!$propPickerConfig}
+						rawKey={true}
+						json={resources}
+						on:select={(e) => dispatch('select', `resource('${e.detail}')`)}
+					/>
+				{:else}
+					<button
+						class="border border-blue-600 px-1 key font-normal rounded hover:bg-blue-100"
+						on:click={async () => {
+							await loadResources()
+							displayResources = true
+						}}>{'{...}'}</button
+					>
+				{/if}
+			</div>
 		{/if}
 	</div>
 </div>

frontend/src/lib/components/propertyPicker/PropPickerResult.svelte

@@ -1,21 +1,31 @@
 <script lang="ts">
+	import { createEventDispatcher } from 'svelte'
 	import ObjectViewer from './ObjectViewer.svelte'
 
-	export let allowCopy = false
 	export let result: any
 	export let extraResults: any = undefined
 	export let flow_input: any = undefined
+
+	const dispatch = createEventDispatcher()
 </script>
 
 <div class="w-full px-2">
-	<span class="font-normal text-sm text-secondary">Result</span>
+	<span class="font-bold text-sm">Result</span>
 	<div class="overflow-y-auto mb-2 w-full">
-		<ObjectViewer {allowCopy} json={{ result, ...(extraResults ? extraResults : {}) }} on:select />
+		<ObjectViewer
+			allowCopy={false}
+			json={{ result, ...(extraResults ? extraResults : {}) }}
+			on:select
+		/>
 	</div>
 	{#if flow_input}
-		<span class="font-normal text-sm text-secondary">Flow Input</span>
+		<span class="font-bold text-sm">Flow Input</span>
 		<div class="overflow-y-auto w-full">
-			<ObjectViewer {allowCopy} json={flow_input} prefix="flow_input" on:select />
+			<ObjectViewer
+				allowCopy={false}
+				json={flow_input}
+				on:select={(e) => dispatch('select', `flow_input.${e.detail}`)}
+			/>
 		</div>
 	{/if}
 </div>

frontend/src/lib/utils.ts

@@ -171,36 +171,13 @@ export function validatePassword(password: string): boolean {
 
 const portalDivs = ['app-editor-select']
 
-interface ClickOutsideOptions {
-	capture?: boolean
-	exclude?: (() => Promise<HTMLElement[]>) | HTMLElement[] | undefined
-}
-
-export function clickOutside(
-	node: Node,
-	options?: ClickOutsideOptions | boolean
-): { destroy(): void; update(newOptions: ClickOutsideOptions | boolean): void } {
-	const handleClick = async (event: MouseEvent) => {
+export function clickOutside(node: Node, capture?: boolean): { destroy(): void } {
+	const handleClick = (event: MouseEvent) => {
 		const target = event.target as HTMLElement
-		const opts = typeof options === 'boolean' ? { capture: options } : options
 
-		let excludedElements: HTMLElement[] = []
-		if (opts?.exclude) {
-			if (Array.isArray(opts.exclude)) {
-				excludedElements = opts.exclude
-			} else {
-				excludedElements = await opts.exclude()
-			}
-		}
-
-		const isExcluded = excludedElements.some((excludedEl) => {
-			const contains = excludedEl?.contains?.(target)
-			const isTarget = target === excludedEl
-			return contains || isTarget
-		})
-
-		if (node && !node.contains(target) && !event.defaultPrevented && !isExcluded) {
+		if (node && !node.contains(target) && !event.defaultPrevented) {
 			const portalDivsSelector = portalDivs.map((id) => `#${id}`).join(', ')
+
 			const parent = target.closest(portalDivsSelector)
 
 			if (!parent) {
@@ -209,15 +186,11 @@ export function clickOutside(
 		}
 	}
 
-	const capture = typeof options === 'boolean' ? options : options?.capture ?? true
-	document.addEventListener('click', handleClick, capture)
+	document.addEventListener('click', handleClick, capture ?? true)
 
 	return {
-		update(newOptions: ClickOutsideOptions | boolean) {
-			options = newOptions
-		},
 		destroy() {
-			document.removeEventListener('click', handleClick, capture)
+			document.removeEventListener('click', handleClick, capture ?? true)
 		}
 	}
 }
@@ -751,7 +724,7 @@ export async function tryEvery({
 		try {
 			await tryCode()
 			break
-		} catch (err) {}
+		} catch (err) { }
 		i++
 	}
 	if (i >= times) {
@@ -975,4 +948,5 @@ export function getSchemaFromProperties(properties: { [name: string]: SchemaProp
 export function validateFileExtension(ext: string) {
 	const validExtensionRegex = /^[a-zA-Z0-9]+([._][a-zA-Z0-9]+)*$/
 	return validExtensionRegex.test(ext)
+
 }