feat(backend): implement flow scripts (#4748)

* feat(backend): implement flow scripts * conditionally resolve reduced flow value * conditionally use flow scripts based on the min worker version * grant all to `flow_{version_lite,node}` * do not silently discard errors * env: add flow script feature flag * flow script on cloud automatically --------- Co-authored-by: Ruben Fiszel <ruben@rubenfiszel.com>
2026-08-18 16:02:10 +00:00 · 2024-11-25 23:20:51 +01:00
parent a182b91750
commit e4784e89da
29 changed files with 911 additions and 226 deletions
@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT value as \"value: sqlx::types::Json<Box<RawValue>>\" FROM flow_version WHERE id = $1 LIMIT 1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "value: sqlx::types::Json<Box<RawValue>>",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "00a16dbe1729d530f136279471fa3d89ca7f1a88a48fcf0b1ac07144924c1707"
+}
@@ -42,7 +42,8 @@
                "noop",
                "appdependencies",
                "deploymentcallback",
-                "singlescriptflow"
+                "singlescriptflow",
+                "flowscript"
              ]
            }
          }
@@ -0,0 +1,28 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT lock, code AS \"code!: String\" FROM flow_node WHERE id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "lock",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 1,
+        "name": "code!: String",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Int8"
+      ]
+    },
+    "nullable": [
+      true,
+      true
+    ]
+  },
+  "hash": "15a5db0202c61ed187d9594311cb32cbde4240e143df1c980843dcdfeef3499e"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO flow_version_lite (id, value) VALUES ($1, $2)\n             ON CONFLICT (id) DO UPDATE SET value = EXCLUDED.value",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Int8",
+        "Jsonb"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "2de4ac337d25cc483829818393fa3254556c0307bb6bb596613b1ce2ee7ec4cb"
+}
@@ -32,7 +32,8 @@
                "noop",
                "appdependencies",
                "deploymentcallback",
-                "singlescriptflow"
+                "singlescriptflow",
+                "flowscript"
              ]
            }
          }
@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT coalesce(flow_version_lite.value, flow_version.value) as \"value!: sqlx::types::Json<Box<RawValue>>\" FROM flow \n                    LEFT JOIN flow_version\n                        ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n                    LEFT JOIN flow_version_lite \n                        ON flow_version_lite.id = flow_version.id\n                    WHERE flow.path = $1 AND flow.workspace_id = $2 LIMIT 1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "value!: sqlx::types::Json<Box<RawValue>>",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "625c400bd81369a8142b8c7fa67b2be02f0f945e587c90d7e0f23706bf6e9a31"
+}
@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT coalesce(flow_version_lite.value, flow_version.value) AS \"value!: sqlx::types::Json<Box<sqlx::types::JsonRawValue>>\" \n                FROM flow \n                LEFT JOIN flow_version \n                    ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n                LEFT JOIN flow_version_lite \n                    ON flow_version_lite.id = flow_version.id\n                WHERE flow.path = $1 AND flow.workspace_id = $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "value!: sqlx::types::Json<Box<sqlx::types::JsonRawValue>>",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "721ab4b4ce441a86d43e1b89ab295fbe9c6b59e27c368eaf78b33ecdfa95140f"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE flow_node SET workspace_id = $1 WHERE workspace_id = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "79992fd5650a99a4391e7c9e590d9ee7969f98975406fa13118d2b7e808958cf"
+}
@@ -0,0 +1,28 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT lock, code AS \"code!: String\" FROM flow_node WHERE id = $1 LIMIT 1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "lock",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 1,
+        "name": "code!: String",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Int8"
+      ]
+    },
+    "nullable": [
+      true,
+      true
+    ]
+  },
+  "hash": "8653373c2c8301c902e21e8bf5a11480ff040bb987a48b6f66522bc9b7e2c1d8"
+}
@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT coalesce(flow_version_lite.value, flow_version.value) as \"value!: sqlx::types::Json<Box<RawValue>>\" FROM flow \n                        LEFT JOIN flow_version\n                            ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n                        LEFT JOIN flow_version_lite \n                            ON flow_version_lite.id = flow_version.id\n                        WHERE flow.path = $1 AND flow.workspace_id = $2 LIMIT 1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "value!: sqlx::types::Json<Box<RawValue>>",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "94a38b83d59f8c010bd96852bf57163476791448201bdd22a9e463a4116b9582"
+}
@@ -0,0 +1,27 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        WITH existing AS (\n            SELECT id FROM flow_node\n            WHERE hash = $1 AND path = $2 AND workspace_id = $3 AND code = $4 AND lock = $5 AND flow = $6\n            LIMIT 1\n        ),\n        inserted AS (\n            INSERT INTO flow_node (hash, path, workspace_id, code, lock, flow)\n            VALUES ($1, $2, $3, $4, $5, $6)\n            ON CONFLICT DO NOTHING\n            RETURNING id\n        )\n        SELECT id FROM existing\n        UNION ALL\n        SELECT id FROM inserted\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Int8",
+        "Text",
+        "Text",
+        "Text",
+        "Text",
+        "Jsonb"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "aa98ab0e4b9a0eb41a804ab047a120444af65b1dc43a234821cde5bf6bf8b74f"
+}
@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT flow_version.value as \"value!: sqlx::types::Json<Box<RawValue>>\" FROM flow \n                        LEFT JOIN flow_version\n                            ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n                        WHERE flow.path = $1 AND flow.workspace_id = $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "value!: sqlx::types::Json<Box<RawValue>>",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "dc44783a555d41c9df4cd303af686e43bbf89add915bfb09d94e25e8517d65ff"
+}
@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE flow_node SET path = REGEXP_REPLACE(path,'u/' || $2 || '/(.*)','u/' || $1 || '/\\1') WHERE path LIKE ('u/' || $2 || '/%') AND workspace_id = $3",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "ed12d54e9cdc18672bbb34d2500adc18b72323e06a1f32e7a1a7589649888927"
+}
@@ -46,7 +46,8 @@
                "noop",
                "appdependencies",
                "deploymentcallback",
-                "singlescriptflow"
+                "singlescriptflow",
+                "flowscript"
              ]
            }
          }
@@ -10485,6 +10485,7 @@ dependencies = [
 "candle-transformers",
 "chrono",
 "chrono-tz 0.10.0",
+ "const_format",
 "cookie 0.17.0",
 "cron",
 "datafusion",
@@ -0,0 +1,4 @@
+-- Add down migration script here
+DROP TABLE IF EXISTS flow_version_lite;
+DROP TABLE IF EXISTS flow_node;
+DROP INDEX IF EXISTS flow_node_hash;
@@ -0,0 +1,29 @@
+-- Add up migration script here
+ALTER TYPE JOB_KIND ADD VALUE IF NOT EXISTS 'flowscript';
+
+-- Same as `flow_version` but with a "lite" value (e.g. `RawScript` replaced by `FlowScript`).
+CREATE TABLE flow_version_lite (
+    id BIGSERIAL PRIMARY KEY,
+    value JSONB,
+    FOREIGN KEY (id) REFERENCES flow_version (id) ON DELETE CASCADE
+);
+
+GRANT ALL ON flow_version_lite TO windmill_user;
+GRANT ALL ON flow_version_lite TO windmill_admin;
+
+-- Either a script or a flow value.
+CREATE TABLE flow_node (
+    id BIGSERIAL PRIMARY KEY,
+    workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id),
+    hash BIGINT NOT NULL,
+    path VARCHAR(255) NOT NULL, -- flow path.
+    lock TEXT,
+    code TEXT,
+    flow JSONB,
+    FOREIGN KEY (path, workspace_id) REFERENCES flow (path, workspace_id) ON DELETE CASCADE
+);
+
+GRANT ALL ON flow_node TO windmill_user;
+GRANT ALL ON flow_node TO windmill_admin;
+
+CREATE INDEX flow_node_hash ON flow_node (hash);
@@ -97,6 +97,7 @@ jsonwebtoken = { workspace = true }
 matchit.workspace = true
 tokio-tungstenite.workspace = true
 rdkafka = { workspace = true, optional = true }
+const_format.workspace = true

 pin-project.workspace = true
 http.workspace = true
@@ -63,7 +63,7 @@ use windmill_common::{
    db::UserDB,
    error::{self, to_anyhow, Error},
    flow_status::{Approval, FlowStatus, FlowStatusModule},
-    flows::{add_virtual_items_if_necessary, FlowValue},
+    flows::{add_virtual_items_if_necessary, resolve_maybe_value, FlowValue},
    jobs::{script_path_to_payload, CompletedJob, JobKind, JobPayload, QueuedJob, RawCode},
    oauth2::HmacSha256,
    scripts::{ScriptHash, ScriptLang},
@@ -593,7 +593,7 @@ async fn get_flow_job_debug_info(
    Extension(db): Extension<DB>,
    Path((w_id, id)): Path<(String, Uuid)>,
 ) -> error::Result<Response> {
-    let job = get_queued_job_ex(&db, &w_id, id, false, None).await?;
+    let job = GetQuery::new().fetch_queued(&db, id, &w_id).await?;
    if let Some(job) = job {
        let is_flow = job.is_flow();
        if job.is_flow_step || !is_flow {
@@ -630,7 +630,7 @@ async fn get_flow_job_debug_info(
            }
        }
        for job_id in job_ids {
-            let job = get_job_internal(&db, w_id.as_str(), job_id, false, Some(&opt_authed)).await;
+            let job = GetQuery::new().with_auth(&opt_authed).fetch(&db, job_id, &w_id).await;
            if let Ok(job) = job {
                jobs.insert(job.id().to_string(), job);
            }
@@ -658,14 +658,11 @@ async fn get_job(
    Path((w_id, id)): Path<(String, Uuid)>,
    Query(GetJobQuery { no_logs }): Query<GetJobQuery>,
 ) -> error::Result<Response> {
-    let mut job = get_job_internal(
-        &db,
-        w_id.as_str(),
-        id,
-        no_logs.unwrap_or(false),
-        Some(&opt_authed),
-    )
-    .await?;
+    let mut get = GetQuery::new().with_auth(&opt_authed);
+    if no_logs.unwrap_or(false) {
+        get = get.without_logs();
+    }
+    let mut job = get.fetch(&db, id, &w_id).await?;
    job.fetch_outstanding_wait_time(&db).await?;

    log_job_view(&db, opt_authed.as_ref(), &w_id, &id).await?;
@@ -673,136 +670,154 @@ async fn get_job(
    Ok(Json(job).into_response())
 }

-lazy_static::lazy_static! {
-    static ref GET_COMPLETED_JOB_QUERY_NO_LOGS: String = generate_get_job_query(true, "completed_job_view");
-    static ref GET_COMPLETED_JOB_QUERY: String = generate_get_job_query(false, "completed_job_view");
-    static ref GET_QUEUED_JOB_QUERY_NO_LOGS: String = generate_get_job_query(true, "queue_view");
-    static ref GET_QUEUED_JOB_QUERY: String = generate_get_job_query(false, "queue_view");
-}
-fn generate_get_job_query(no_logs: bool, table: &str) -> String {
-    let log_expr = if no_logs {
-        "null".to_string()
-    } else {
-        format!("right({table}.logs, 20000)")
+macro_rules! get_job_query {
+    ("completed_job_view", $($opts:tt)*) => {
+        get_job_query!(
+            @impl "completed_job_view", ($($opts)*),
+            "duration_ms, success, result, deleted, is_skipped, result->'wm_labels' as labels, \
+            CASE WHEN result is null or pg_column_size(result) < 90000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result",
+        )
    };
-    let additional_fields = if table == "completed_job_view" {
-        "duration_ms,      
-        success,        
-        result,    
-        deleted,    
-        is_skipped,
-        result->'wm_labels' as labels,
-        CASE WHEN result is null or pg_column_size(result) < 90000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result"
-    } else {
-        "scheduled_for,  
-        running,       
-        last_ping,        
-        suspend,     
-        suspend_until,
-        same_worker,
-        pre_run_error,           
-        visible_to_owner,          
-        root_job,         
-        leaf_jobs,        
-        tag,       
-        concurrent_limit,      
-        concurrency_time_window_s, 
-        timeout,
-        flow_step_id,
-        cache_ttl
-        "
+    ("queue_view", $($opts:tt)*) => {
+        get_job_query!(
+            @impl "queue_view", ($($opts)*),
+            "scheduled_for, running, last_ping, suspend, suspend_until, same_worker, pre_run_error, visible_to_owner, \
+            root_job, leaf_jobs, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, cache_ttl",
+        )
    };
-    return format!("SELECT 
-    id, {table}.workspace_id, parent_job, created_by, {table}.created_at, started_at, script_hash, script_path, 
-    CASE WHEN args is null or pg_column_size(args) < 90000 THEN args ELSE '{{\"reason\": \"WINDMILL_TOO_BIG\"}}'::jsonb END as args, 
-    {log_expr} as logs, raw_code, canceled, canceled_by, canceled_reason, job_kind,
-    schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, language,
-    raw_lock, email, visible_to_owner, mem_peak, tag, priority, {additional_fields}
-    FROM {table}
-    WHERE id = $1 AND {table}.workspace_id = $2");
-}
-pub async fn get_queued_job_ex(
-    db: &DB,
-    workspace_id: &str,
-    job_id: Uuid,
-    no_logs: bool,
-    // first optional is if authed need to be checked, second is the opt_authed itself
-    opt_authed: Option<&Option<ApiAuthed>>,
-) -> error::Result<Option<JobExtended<QueuedJob>>> {
-    let query = if no_logs {
-        &*GET_QUEUED_JOB_QUERY_NO_LOGS
-    } else {
-        &*GET_QUEUED_JOB_QUERY
-    };
-    let job = sqlx::query_as::<_, JobExtended<QueuedJob>>(query)
-        .bind(job_id)
-        .bind(workspace_id)
-        .fetch_optional(db)
-        .await?;
-
-    if let Some(job) = job.as_ref() {
-        if opt_authed.is_some_and(|x| x.is_none()) && job.created_by != "anonymous" {
-            return Err(Error::BadRequest(
-                "As a non logged in user, you can only see jobs ran by anonymous users".to_string(),
-            ));
+    (@impl $table:literal, (with_logs: $with_logs:expr, $($rest:tt)*), $additional_fields:literal, $($args:tt)*) => {
+        if $with_logs {
+            get_job_query!(@impl $table, ($($rest)*), $additional_fields, logs = const_format::formatcp!("right({}.logs, 20000)", $table), $($args)*)
+        } else {
+            get_job_query!(@impl $table, ($($rest)*), $additional_fields, logs = "null", $($args)*)
        }
-    }
-
-    Ok(job)
-}
-pub async fn get_completed_job_ex(
-    db: &DB,
-    workspace_id: &str,
-    job_id: Uuid,
-    no_logs: bool,
-    // first optional is if authed need to be checked, second is the opt_authed itself
-    opt_authed: Option<&Option<ApiAuthed>>,
-) -> error::Result<Option<JobExtended<CompletedJob>>> {
-    let query = if no_logs {
-        &*GET_COMPLETED_JOB_QUERY_NO_LOGS
-    } else {
-        &*GET_COMPLETED_JOB_QUERY
    };
-    let cjob = sqlx::query_as::<_, JobExtended<CompletedJob>>(query)
-        .bind(job_id)
-        .bind(workspace_id)
-        .fetch_optional(db)
-        .await?;
-
-    if let Some(job) = cjob.as_ref() {
-        if opt_authed.is_some_and(|x| x.is_none()) && job.created_by != "anonymous" {
-            return Err(Error::BadRequest(
-                "As a non logged in user, you can only see jobs ran by anonymous users".to_string(),
-            ));
+    (@impl $table:literal, (with_code: $with_code:expr, $($rest:tt)*), $additional_fields:literal, $($args:tt)*) => {
+        if $with_code {
+            get_job_query!(@impl $table, ($($rest)*), $additional_fields, lock = "raw_lock", code = "raw_code", $($args)*)
+        } else {
+            get_job_query!(@impl $table, ($($rest)*), $additional_fields, lock = "null", code = "null", $($args)*)
        }
+    };
+    (@impl $table:literal, (with_flow: $with_flow:expr, $($rest:tt)*), $additional_fields:literal, $($args:tt)*) => {
+        if $with_flow {
+            get_job_query!(@impl $table, ($($rest)*), $additional_fields, flow = "raw_flow", $($args)*)
+        } else {
+            get_job_query!(@impl $table, ($($rest)*), $additional_fields, flow = "null", $($args)*)
+        }
+    };
+    (@impl $table:literal, (), $additional_fields:literal, $($args:tt)*) => {
+        const_format::formatcp!(
+            "SELECT \
+            id, {table}.workspace_id, parent_job, created_by, {table}.created_at, started_at, script_hash, script_path, \
+            CASE WHEN args is null or pg_column_size(args) < 90000 THEN args ELSE '{{\"reason\": \"WINDMILL_TOO_BIG\"}}'::jsonb END as args, \
+            {logs} as logs, {code} as raw_code, canceled, canceled_by, canceled_reason, job_kind, \
+            schedule_path, permissioned_as, flow_status, {flow} as raw_flow, is_flow_step, language, \
+            {lock} as raw_lock, email, visible_to_owner, mem_peak, tag, priority, {additional_fields} \
+            FROM {table} \
+            WHERE id = $1 AND {table}.workspace_id = $2 LIMIT 1",
+            table = $table,
+            additional_fields = $additional_fields,
+            $($args)*
+        )
    }
-
-    if let Some(mut cjob) = cjob {
-        cjob.inner = format_completed_job_result(cjob.inner);
-        return Ok(Some(cjob));
-    }
-
-    Ok(cjob)
 }
-pub async fn get_job_internal(
-    db: &DB,
-    workspace_id: &str,
-    job_id: Uuid,
-    no_logs: bool,
-    // first optional is if authed need to be checked, second is the opt_authed itself
-    opt_authed: Option<&Option<ApiAuthed>>,
-) -> error::Result<Job> {
-    let cjob = get_completed_job_ex(db, workspace_id, job_id, no_logs, opt_authed.clone())
-        .await?
-        .map(Job::CompletedJob);

-    match cjob {
-        Some(cjob) => Ok(cjob),
-        None => {
-            let job_maybe = get_queued_job_ex(db, workspace_id, job_id, no_logs, opt_authed)
-                .await?
-                .map(Job::QueuedJob);
-            not_found_if_none(job_maybe, "Job", job_id.to_string())
+#[derive(Copy, Clone)]
+struct GetQuery<'a> {
+    with_logs: bool,
+    with_code: bool,
+    with_flow: bool,
+    with_auth: Option<&'a Option<ApiAuthed>>,
+}
+
+impl<'a> GetQuery<'a> {
+    fn new() -> Self {
+        Self { with_logs: true, with_code: true, with_flow: true, with_auth: None }
+    }
+
+    fn without_logs(self) -> Self {
+        Self { with_logs: false, ..self }
+    }
+
+    fn without_code(self) -> Self {
+        Self { with_code: false, ..self }
+    }
+
+    fn without_flow(self) -> Self {
+        Self { with_flow: false, ..self }
+    }
+
+    fn with_auth(self, auth: &'a Option<ApiAuthed>) -> Self {
+        Self { with_auth: Some(auth), ..self }
+    }
+
+    fn check_auth(self, email: Option<&str>) -> error::Result<()> {
+        if let Some(email) = email {
+            if self.with_auth.is_some_and(|x| x.is_none()) && email != "anonymous" {
+                return Err(Error::BadRequest(
+                    "As a non logged in user, you can only see jobs ran by anonymous users".to_string(),
+                ));
+            }
+        }
+        Ok(())
+    }
+
+    async fn fetch_queued(self, db: &DB, job_id: Uuid, workspace_id: &str) -> error::Result<Option<JobExtended<QueuedJob>>> {
+        let query = get_job_query!("queue_view",
+            with_logs: self.with_logs,
+            with_code: self.with_code,
+            with_flow: self.with_flow,
+        );
+        let mut job = sqlx::query_as::<_, JobExtended<QueuedJob>>(query)
+            .bind(job_id)
+            .bind(workspace_id)
+            .fetch_optional(db)
+            .await?;
+
+        self.check_auth(job.as_ref().map(|job| job.created_by.as_str()))?;
+        if self.with_flow {
+            job = resolve_maybe_value(db, workspace_id, self.with_code, job, |job| job.raw_flow.as_mut()).await?;
+        }
+        Ok(job)
+    }
+
+    async fn fetch_completed(self, db: &DB, job_id: Uuid, workspace_id: &str) -> error::Result<Option<JobExtended<CompletedJob>>> {
+        let query = get_job_query!("completed_job_view",
+            with_logs: self.with_logs,
+            with_code: self.with_code,
+            with_flow: self.with_flow,
+        );
+        let mut cjob = sqlx::query_as::<_, JobExtended<CompletedJob>>(query)
+            .bind(job_id)
+            .bind(workspace_id)
+            .fetch_optional(db)
+            .await?;
+
+        self.check_auth(cjob.as_ref().map(|job| job.created_by.as_str()))?;
+        if self.with_flow {
+            cjob = resolve_maybe_value(db, workspace_id, self.with_code, cjob, |job| job.raw_flow.as_mut()).await?;
+        }
+        if let Some(mut cjob) = cjob {
+            cjob.inner = format_completed_job_result(cjob.inner);
+            return Ok(Some(cjob));
+        }
+        Ok(cjob)
+    }
+
+    async fn fetch(self, db: &DB, job_id: Uuid, workspace_id: &str) -> error::Result<Job> {
+        let cjob = self.fetch_completed(db, job_id, workspace_id)
+            .await?
+            .map(Job::CompletedJob);
+
+        match cjob {
+            Some(cjob) => Ok(cjob),
+            None => {
+                let job_maybe = self.fetch_queued(db, job_id, workspace_id)
+                    .await?
+                    .map(Job::QueuedJob);
+                not_found_if_none(job_maybe, "Job", job_id.to_string())
+            }
        }
    }
 }
@@ -1787,8 +1802,12 @@ async fn resume_suspended_job_internal(
    verify_suspended_secret(&w_id, &db, job_id, resume_id, &approver, secret).await?;

    let parent_flow_info = get_suspended_parent_flow_info(job_id, &db).await?;
-    let parent_flow =
-        get_job_internal(&db, w_id.as_str(), parent_flow_info.id, false, None).await?;
+    let parent_flow = GetQuery::new()
+        .without_logs()
+        .without_code()
+        .without_flow()
+        .fetch(&db, parent_flow_info.id, &w_id)
+        .await?;
    let flow_status = parent_flow
        .flow_status()
        .ok_or_else(|| anyhow::anyhow!("unable to find the flow status in the flow job"))?;
@@ -2056,7 +2075,11 @@ pub async fn get_suspended_job_flow(
    .flatten()
    .ok_or_else(|| anyhow::anyhow!("parent flow job not found"))?;

-    let flow = get_job_internal(&db, w_id.as_str(), flow_id, true, None).await?;
+    let flow = GetQuery::new()
+        .without_logs()
+        .without_code()
+        .fetch(&db, flow_id, &w_id)
+        .await?;

    let flow_status = flow
        .flow_status()
@@ -3130,7 +3153,7 @@ pub async fn run_workflow_as_code(
        i += 1;
    }

-    let job = get_queued_job_ex(&db, &w_id, job_id, true, None).await?;
+    let job = GetQuery::new().without_logs().fetch_queued(&db, job_id, &w_id).await?;

    if *CLOUD_HOSTED {
        tracing::info!("workflow_as_code_tracing id {i} ");
@@ -4415,10 +4438,12 @@ async fn add_batch_jobs(
                (value, JobKind::FlowPreview, None)
            } else if let Some(path) = batch_info.path {
                let value_json = sqlx::query!(
-                    "SELECT flow_version.value AS \"value: sqlx::types::Json<Box<RawValue>>\" FROM flow 
+                    "SELECT coalesce(flow_version_lite.value, flow_version.value) as \"value!: sqlx::types::Json<Box<RawValue>>\" FROM flow 
                    LEFT JOIN flow_version
                        ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]
-                    WHERE flow.path = $1 AND flow.workspace_id = $2",
+                    LEFT JOIN flow_version_lite 
+                        ON flow_version_lite.id = flow_version.id
+                    WHERE flow.path = $1 AND flow.workspace_id = $2 LIMIT 1",
                    &path, &w_id
                )
                .fetch_optional(&db)
@@ -5103,7 +5128,7 @@ async fn get_completed_job<'a>(
    Extension(db): Extension<DB>,
    Path((w_id, id)): Path<(String, Uuid)>,
 ) -> error::Result<Response> {
-    let job_o = get_completed_job_ex(&db, &w_id, id, false, Some(&opt_authed)).await?;
+    let job_o = GetQuery::new().with_auth(&opt_authed).fetch_completed(&db, id, &w_id).await?;

    let cj = not_found_if_none(job_o, "Completed Job", id.to_string())?;
    let response = Json(cj).into_response();
@@ -3015,6 +3015,15 @@ async fn update_username_in_workpsace<'c>(
    .execute(&mut **tx)
    .await?;

+    sqlx::query!(
+        r#"UPDATE flow_node SET path = REGEXP_REPLACE(path,'u/' || $2 || '/(.*)','u/' || $1 || '/\1') WHERE path LIKE ('u/' || $2 || '/%') AND workspace_id = $3"#,
+        new_username,
+        old_username,
+        w_id
+    )
+    .execute(&mut **tx)
+    .await?;
+
    sqlx::query!(
        "DELETE FROM flow WHERE path LIKE ('u/' || $1 || '/%') AND workspace_id = $2",
        old_username,
@@ -2882,6 +2882,14 @@ async fn change_workspace_id(
    .execute(&mut *tx)
    .await?;

+    sqlx::query!(
+        "UPDATE flow_node SET workspace_id = $1 WHERE workspace_id = $2",
+        &rw.new_id,
+        &old_id
+    )
+    .execute(&mut *tx)
+    .await?;
+
    sqlx::query!("DELETE FROM flow WHERE workspace_id = $1", &old_id)
        .execute(&mut *tx)
        .await?;
@@ -14,10 +14,14 @@ use std::{

 use rand::Rng;
 use serde::{Deserialize, Serialize, Serializer};
+use sqlx::types::Json;
+use sqlx::types::JsonRawValue;

 use crate::{
+    error::Error,
    more_serde::{default_empty_string, default_id, default_null, default_true, is_default},
    scripts::{Schema, ScriptHash, ScriptLang},
+    worker::to_raw_value,
 };

 #[derive(Serialize, Deserialize, sqlx::FromRow)]
@@ -26,7 +30,7 @@ pub struct Flow {
    pub path: String,
    pub summary: String,
    pub description: String,
-    pub value: sqlx::types::Json<Box<serde_json::value::RawValue>>,
+    pub value: Json<Box<JsonRawValue>>,
    pub edited_by: String,
    pub edited_at: chrono::DateTime<chrono::Utc>,
    pub archived: bool,
@@ -398,6 +402,11 @@ pub enum InputTransform {
    },
 }

+/// Id in the `flow_node` table.
+#[derive(Serialize, Deserialize, Debug, Copy, Clone, Hash)]
+#[serde(transparent)]
+pub struct FlowNodeId(pub i64);
+
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct Branch {
    #[serde(skip_serializing_if = "Option::is_none")]
@@ -480,6 +489,24 @@ pub enum FlowModuleValue {
        is_trigger: Option<bool>,
    },
    Identity,
+    // Internal only, never exposed to the frontend.
+    FlowScript {
+        #[serde(default)]
+        #[serde(alias = "input_transform", serialize_with = "ordered_map")]
+        input_transforms: HashMap<String, InputTransform>,
+        id: FlowNodeId,
+        #[serde(skip_serializing_if = "is_none_or_empty")]
+        tag: Option<String>,
+        language: ScriptLang,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        custom_concurrency_key: Option<String>,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        concurrent_limit: Option<i32>,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        concurrency_time_window_s: Option<i32>,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        is_trigger: Option<bool>,
+    },
 }

 fn is_none_or_empty(expr: &Option<String>) -> bool {
@@ -510,6 +537,7 @@ struct UntaggedFlowModuleValue {
    concurrent_limit: Option<i32>,
    concurrency_time_window_s: Option<i32>,
    is_trigger: Option<bool>,
+    id: Option<FlowNodeId>,
 }

 impl<'de> Deserialize<'de> for FlowModuleValue {
@@ -582,6 +610,20 @@ impl<'de> Deserialize<'de> for FlowModuleValue {
                concurrency_time_window_s: untagged.concurrency_time_window_s,
                is_trigger: untagged.is_trigger,
            }),
+            "flowscript" => Ok(FlowModuleValue::FlowScript {
+                input_transforms: untagged.input_transforms.unwrap_or_default(),
+                id: untagged
+                    .id
+                    .ok_or_else(|| serde::de::Error::missing_field("id"))?,
+                tag: untagged.tag,
+                language: untagged
+                    .language
+                    .ok_or_else(|| serde::de::Error::missing_field("language"))?,
+                custom_concurrency_key: untagged.custom_concurrency_key,
+                concurrent_limit: untagged.concurrent_limit,
+                concurrency_time_window_s: untagged.concurrency_time_window_s,
+                is_trigger: untagged.is_trigger,
+            }),
            "identity" => Ok(FlowModuleValue::Identity),
            other => Err(serde::de::Error::unknown_variant(
                other,
@@ -651,3 +693,85 @@ pub fn add_virtual_items_if_necessary(modules: &mut Vec<FlowModule>) {
        });
    }
 }
+
+/// Resolve the value of a flow if any.
+pub async fn resolve_maybe_value<T>(
+    e: &sqlx::PgPool,
+    workspace_id: &str,
+    with_code: bool,
+    maybe: Option<T>,
+    value_mut: impl FnOnce(&mut T) -> Option<&mut Json<Box<JsonRawValue>>>
+) -> Result<Option<T>, Error> {
+    let Some(mut container) = maybe else { return Ok(None); };
+    let Some(value) = value_mut(&mut container) else { return Ok(Some(container)); };
+    resolve_value(e, workspace_id, &mut value.0, with_code).await?;
+    Ok(Some(container))
+}
+
+/// Resolve modules recursively.
+pub async fn resolve_value(
+    e: &sqlx::PgPool,
+    workspace_id: &str,
+    value: &mut Box<JsonRawValue>,
+    with_code: bool,
+) -> Result<(), Error> {
+    let mut val = serde_json::from_str::<FlowValue>(value.get())
+        .map_err(|err| Error::InternalErr(format!("resolve: Failed to parse flow value: {}", err)))?;
+    for module in &mut val.modules {
+        resolve_module(e, workspace_id, &mut module.value, with_code).await?;
+    }
+    *value = to_raw_value(&val);
+    Ok(())
+}
+
+/// Resolve module value recursively.
+pub async fn resolve_module(
+    e: &sqlx::PgPool,
+    workspace_id: &str,
+    value: &mut Box<JsonRawValue>,
+    with_code: bool,
+) -> Result<(), Error> {
+    use FlowModuleValue::*;
+
+    let mut val = serde_json::from_str::<FlowModuleValue>(value.get())
+        .map_err(|err| Error::InternalErr(format!("resolve: Failed to parse flow module value: {}", err)))?;
+    match &mut val {
+        FlowScript { .. } => {
+            // In order to avoid an unnecessary `.clone()` of `val`, take ownership of it's content
+            // using `std::mem::replace`.
+            let FlowScript {
+                input_transforms, id, tag, language,
+                custom_concurrency_key, concurrent_limit, concurrency_time_window_s, is_trigger
+            } = std::mem::replace(&mut val, Identity) else { unreachable!() };
+            // Load script lock file and code content.
+            let (lock, content) = if !with_code {
+                (Some("...".to_string()), "...".to_string())
+            } else {
+                sqlx::query!("SELECT lock, code AS \"code!: String\" FROM flow_node WHERE id = $1", id.0)
+                    .fetch_one(e)
+                    .await
+                    .map_err(Error::SqlErr)
+                    .map(|record| (record.lock, record.code))?
+            };
+            val = RawScript {
+                input_transforms, content, lock, path: None, tag, language, custom_concurrency_key,
+                concurrent_limit, concurrency_time_window_s, is_trigger
+            };
+        },
+        ForloopFlow { modules, .. } | WhileloopFlow { modules, .. }  => {
+            for module in modules {
+                Box::pin(resolve_module(e, workspace_id, &mut module.value, with_code)).await?;
+            }
+        },
+        BranchOne { branches, .. } | BranchAll { branches, .. } => {
+            for branch in branches {
+                for module in &mut branch.modules {
+                    Box::pin(resolve_module(e, workspace_id, &mut module.value, with_code)).await?;
+                }
+            }
+        }
+        _ => {}
+    }
+    *value = to_raw_value(&val);
+    Ok(())
+}
@@ -16,13 +16,13 @@ pub const PREPROCESSOR_FAKE_ENTRYPOINT: &str = "__WM_PREPROCESSOR";
 use crate::{
    error::{self, to_anyhow, Error},
    flow_status::{FlowStatus, RestartedFrom},
-    flows::{FlowValue, Retry},
+    flows::{FlowNodeId, FlowValue, Retry},
    get_latest_deployed_hash_for_path,
    scripts::{ScriptHash, ScriptLang},
    worker::{to_raw_value, TMP_DIR},
 };

-#[derive(sqlx::Type, Serialize, Deserialize, Debug, PartialEq, Clone)]
+#[derive(sqlx::Type, Serialize, Deserialize, Debug, PartialEq, Copy, Clone)]
 #[sqlx(type_name = "JOB_KIND", rename_all = "lowercase")]
 #[serde(rename_all(serialize = "lowercase"))]
 pub enum JobKind {
@@ -39,6 +39,7 @@ pub enum JobKind {
    AppDependencies,
    Noop,
    DeploymentCallback,
+    FlowScript,
 }

 #[derive(sqlx::FromRow, Debug, Serialize, Clone)]
@@ -263,6 +264,15 @@ pub enum JobPayload {
        priority: Option<i16>,
        apply_preprocessor: bool,
    },
+    FlowScript {
+        id: FlowNodeId, // flow_node(id).
+        language: ScriptLang,
+        custom_concurrency_key: Option<String>,
+        concurrent_limit: Option<i32>,
+        concurrency_time_window_s: Option<i32>,
+        cache_ttl: Option<i32>,
+        dedicated_worker: Option<bool>,
+    },
    Code(RawCode),
    Dependencies {
        path: String,
@@ -93,6 +93,10 @@ lazy_static::lazy_static! {

    pub static ref MIN_VERSION: Arc<RwLock<Version>> = Arc::new(RwLock::new(Version::new(0, 0, 0)));
    pub static ref MIN_VERSION_IS_AT_LEAST_1_427: Arc<RwLock<bool>> = Arc::new(RwLock::new(false));
+    pub static ref MIN_VERSION_IS_AT_LEAST_1_432: Arc<RwLock<bool>> = Arc::new(RwLock::new(false));
+
+    // Features flags:
+    pub static ref DISABLE_FLOW_SCRIPT: bool = std::env::var("DISABLE_FLOW_SCRIPT").ok().is_some_and(|x| x == "1" || x == "true");
 }

 pub async fn make_suspended_pull_query(wc: &WorkerConfig) {
@@ -593,6 +597,7 @@ pub async fn update_min_version<'c, E: sqlx::Executor<'c, Database = sqlx::Postg
    }

    *MIN_VERSION_IS_AT_LEAST_1_427.write().await = min_version >= Version::new(1, 427, 0);
+    *MIN_VERSION_IS_AT_LEAST_1_432.write().await = min_version >= Version::new(1, 432, 0);

    *MIN_VERSION.write().await = min_version.clone();
    min_version >= cur_version
@@ -58,8 +58,9 @@ use windmill_common::{
    users::{SUPERADMIN_NOTIFICATION_EMAIL, SUPERADMIN_SECRET_EMAIL},
    utils::{not_found_if_none, report_critical_error, StripPath},
    worker::{
-        to_raw_value, DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES,
-        MIN_VERSION_IS_AT_LEAST_1_427, NO_LOGS, WORKER_PULL_QUERIES, WORKER_SUSPENDED_PULL_QUERY,
+        to_raw_value, CLOUD_HOSTED, DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES,
+        DISABLE_FLOW_SCRIPT, MIN_VERSION_IS_AT_LEAST_1_427, MIN_VERSION_IS_AT_LEAST_1_432, NO_LOGS,
+        WORKER_PULL_QUERIES, WORKER_SUSPENDED_PULL_QUERY,
    },
    DB, METRICS_ENABLED,
 };
@@ -2930,11 +2931,6 @@ lazy_static::lazy_static! {
    pub static ref RE_ARG_TAG: Regex = Regex::new(r#"\$args\[(\w+)\]"#).unwrap();
 }

-#[derive(sqlx::FromRow)]
-struct FlowRawValue {
-    pub value: sqlx::types::Json<Box<RawValue>>,
-}
-
 // #[instrument(level = "trace", skip_all)]
 pub async fn push<'c, 'd>(
    _db: &Pool<Postgres>,
@@ -3192,6 +3188,29 @@ pub async fn push<'c, 'd>(
                priority,
            )
        }
+        JobPayload::FlowScript {
+            id, // flow_node(id).
+            language,
+            custom_concurrency_key,
+            concurrent_limit,
+            concurrency_time_window_s,
+            cache_ttl,
+            dedicated_worker,
+        } => (
+            Some(id.0),
+            None,
+            None,
+            JobKind::FlowScript,
+            None,
+            None,
+            Some(language),
+            custom_concurrency_key,
+            concurrent_limit,
+            concurrency_time_window_s,
+            cache_ttl,
+            dedicated_worker,
+            None,
+        ),
        JobPayload::ScriptHub { path } => {
            if path == "hub/7771/slack" || path == "hub/7836/slack" {
                permissioned_as = SUPERADMIN_NOTIFICATION_EMAIL.to_string();
@@ -3292,17 +3311,15 @@ pub async fn push<'c, 'd>(
        ),
        JobPayload::FlowDependencies { path, dedicated_worker, version } => {
            let value_json = fetch_scalar_isolated!(
-                sqlx::query_as::<_, FlowRawValue>("SELECT value FROM flow_version WHERE id = $1",)
-                    .bind(&version),
+                sqlx::query_scalar!("SELECT value as \"value: sqlx::types::Json<Box<RawValue>>\" FROM flow_version WHERE id = $1 LIMIT 1", &version),
                tx
            )?
            .ok_or_else(|| Error::InternalErr(format!("not found flow at path {:?}", path)))?;
-            let value =
-                serde_json::from_str::<FlowValue>(value_json.value.get()).map_err(|err| {
-                    Error::InternalErr(format!(
-                        "could not convert json to flow for {path}: {err:?}"
-                    ))
-                })?;
+            let value = serde_json::from_str::<FlowValue>(value_json.get()).map_err(|err| {
+                Error::InternalErr(format!(
+                    "could not convert json to flow for {path}: {err:?}"
+                ))
+            })?;
            (
                Some(version),
                Some(path),
@@ -3464,24 +3481,38 @@ pub async fn push<'c, 'd>(
            )
        }
        JobPayload::Flow { path, dedicated_worker, apply_preprocessor } => {
-            let value_json = fetch_scalar_isolated!(
-                sqlx::query_as::<_, FlowRawValue>(
-                    "SELECT flow_version.value FROM flow 
-                    LEFT JOIN flow_version
-                        ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]
-                    WHERE flow.path = $1 AND flow.workspace_id = $2",
+            // Do not use the lite version unless all workers are updated.
+            let value_json = if *DISABLE_FLOW_SCRIPT || (!*MIN_VERSION_IS_AT_LEAST_1_432.read().await  && !*CLOUD_HOSTED) {
+                fetch_scalar_isolated!(
+                    sqlx::query_scalar!(
+                        "SELECT flow_version.value as \"value!: sqlx::types::Json<Box<RawValue>>\" FROM flow 
+                        LEFT JOIN flow_version
+                            ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]
+                        WHERE flow.path = $1 AND flow.workspace_id = $2",
+                        &path, &workspace_id
+                    ),
+                    tx
                )
-                .bind(&path)
-                .bind(&workspace_id),
-                tx
-            )?
+            } else {
+                fetch_scalar_isolated!(
+                    sqlx::query_scalar!(
+                        "SELECT coalesce(flow_version_lite.value, flow_version.value) as \"value!: sqlx::types::Json<Box<RawValue>>\" FROM flow 
+                        LEFT JOIN flow_version
+                            ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]
+                        LEFT JOIN flow_version_lite 
+                            ON flow_version_lite.id = flow_version.id
+                        WHERE flow.path = $1 AND flow.workspace_id = $2 LIMIT 1",
+                        &path, &workspace_id
+                    ),
+                    tx
+                )
+            }?
            .ok_or_else(|| Error::InternalErr(format!("not found flow at path {:?}", path)))?;
-            let mut value =
-                serde_json::from_str::<FlowValue>(value_json.value.get()).map_err(|err| {
-                    Error::InternalErr(format!(
-                        "could not convert json to flow for {path}: {err:?}"
-                    ))
-                })?;
+            let mut value = serde_json::from_str::<FlowValue>(value_json.get()).map_err(|err| {
+                Error::InternalErr(format!(
+                    "could not convert json to flow for {path}: {err:?}"
+                ))
+            })?;
            let priority = value.priority;
            add_virtual_items_if_necessary(&mut value.modules);
            if same_worker {
@@ -3936,6 +3967,7 @@ pub async fn push<'c, 'd>(
            JobKind::FlowDependencies => "jobs.run.flow_dependencies",
            JobKind::AppDependencies => "jobs.run.app_dependencies",
            JobKind::DeploymentCallback => "jobs.run.deployment_callback",
+            JobKind::FlowScript => "jobs.run.flow_script",
        };

        let audit_author = if format!("u/{user}") != permissioned_as && user != permissioned_as {
@@ -392,6 +392,44 @@ async fn spawn_dedicated_workers_for_flow(
                        workers.push(dedi_w);
                    }
                }
+                FlowModuleValue::FlowScript { id, language, .. } => {
+                    let spawn = sqlx::query!(
+                        "SELECT lock, code AS \"code!: String\" FROM flow_node WHERE id = $1 LIMIT 1",
+                        id.0
+                    )
+                    .fetch_one(db)
+                    .await
+                    .map(|record| SpawnWorker::RawScript {
+                        path: "".to_string(),
+                        content: record.code,
+                        lock: record.lock,
+                        lang: language.clone(),
+                    });
+                    match spawn {
+                        Ok(spawn) => {
+                            if let Some(dedi_w) = spawn_dedicated_worker(
+                                spawn,
+                                w_id,
+                                killpill_tx.clone(),
+                                killpill_rx,
+                                db,
+                                worker_dir,
+                                base_internal_url,
+                                worker_name,
+                                job_completed_tx,
+                                Some(module.id.clone()),
+                            )
+                              .await
+                            {
+                                workers.push(dedi_w);
+                            }
+                        },
+                        Err(err) => tracing::error!(
+                            "failed to get script for module: {:?}, err: {:?}",
+                            module, err
+                        )
+                    }
+                },
                FlowModuleValue::Flow { .. } => (),
                FlowModuleValue::Identity => (),
            }
@@ -422,10 +460,12 @@ pub async fn create_dedicated_worker_map(
        if let Some(flow_path) = _wp.path.strip_prefix("flow/") {
            is_flow_worker = true;
            let value = sqlx::query_scalar!(
-                "SELECT flow_version.value 
+                "SELECT coalesce(flow_version_lite.value, flow_version.value) AS \"value!: sqlx::types::Json<Box<sqlx::types::JsonRawValue>>\" 
                FROM flow 
                LEFT JOIN flow_version 
                    ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]
+                LEFT JOIN flow_version_lite 
+                    ON flow_version_lite.id = flow_version.id
                WHERE flow.path = $1 AND flow.workspace_id = $2",
                flow_path,
                _wp.workspace_id
@@ -434,7 +474,7 @@ pub async fn create_dedicated_worker_map(
            .await;
            if let Ok(v) = value {
                if let Some(v) = v {
-                    let value = serde_json::from_value::<FlowValue>(v).map_err(|err| {
+                    let value = serde_json::from_str::<FlowValue>(v.get()).map_err(|err| {
                        Error::InternalErr(format!(
                            "could not convert json to flow for {flow_path}: {err:?}"
                        ))
@@ -1862,7 +1862,11 @@ async fn handle_queued_job(

    let cached_res_path = if job.cache_ttl.is_some() {
        let version_hash = if let Some(h) = job.script_hash {
-            format!("script_{}", h.to_string())
+            if matches!(job.job_kind, JobKind::FlowScript) {
+                format!("flowscript_{}", h.to_string())
+            } else {
+                format!("script_{}", h.to_string())
+            }
        } else if let Some(rc) = raw_code.as_ref() {
            use std::hash::Hasher;
            let mut s = DefaultHasher::new();
@@ -2273,6 +2277,22 @@ async fn handle_code_execution_job(
            )
            .await?
        }
+        JobKind::FlowScript => {
+            let (lockfile, content) = sqlx::query!(
+                "SELECT lock, code AS \"code!: String\" FROM flow_node WHERE id = $1 LIMIT 1",
+                job.script_hash.unwrap_or(ScriptHash(0)).0
+            )
+            .fetch_one(db)
+            .await
+            .map(|record| (record.lock, record.code))?;
+            ContentReqLangEnvs {
+                content,
+                lockfile,
+                language: job.language.to_owned(),
+                envs: None,
+                codebase: None,
+            }
+        },
        JobKind::DeploymentCallback => {
            get_script_content_by_path(job.script_path.clone(), &job.workspace_id, db).await?
        }
@@ -2253,6 +2253,7 @@ async fn push_next_flow_job(
                Ok(
                    FlowModuleValue::Script { input_transforms, .. }
                    | FlowModuleValue::RawScript { input_transforms, .. }
+                    | FlowModuleValue::FlowScript { input_transforms, .. }
                    | FlowModuleValue::Flow { input_transforms, .. },
                ) => {
                    let ctx = get_transform_context(&flow_job, &previous_id, &status).await?;
@@ -2994,10 +2995,10 @@ async fn compute_next_flow_transform(
    if is_skipped {
        return trivial_next_job(JobPayload::Identity);
    }
-    match &module.get_value()? {
+    match module.get_value()? {
        FlowModuleValue::Identity => trivial_next_job(JobPayload::Identity),
        FlowModuleValue::Flow { path, .. } => {
-            let payload = flow_to_payload(path, &delete_after_use);
+            let payload = flow_to_payload(&path, &delete_after_use);
            Ok(NextFlowTransform::Continue(
                ContinuePayload::SingleJob(payload),
                NextStatus::NextStep,
@@ -3005,7 +3006,7 @@ async fn compute_next_flow_transform(
        }
        FlowModuleValue::Script { path: script_path, hash: script_hash, tag_override, .. } => {
            let payload =
-                script_to_payload(script_hash, script_path, db, flow_job, module, tag_override)
+                script_to_payload(&script_hash, &script_path, db, flow_job, module, &tag_override)
                    .await?;
            Ok(NextFlowTransform::Continue(
                ContinuePayload::SingleJob(payload),
@@ -3036,14 +3037,14 @@ async fn compute_next_flow_transform(
            });
            let payload = raw_script_to_payload(
                path,
-                content,
-                language,
-                lock,
-                custom_concurrency_key,
-                concurrent_limit,
-                concurrency_time_window_s,
+                &content,
+                &language,
+                &lock,
+                &custom_concurrency_key,
+                &concurrent_limit,
+                &concurrency_time_window_s,
                module,
-                tag,
+                &tag,
                &delete_after_use,
            );
            Ok(NextFlowTransform::Continue(
@@ -3051,9 +3052,37 @@ async fn compute_next_flow_transform(
                NextStatus::NextStep,
            ))
        }
+        FlowModuleValue::FlowScript {
+            id, // flow_node(id).
+            tag,
+            language,
+            custom_concurrency_key,
+            concurrent_limit,
+            concurrency_time_window_s,
+            ..
+        } => {
+            let payload = JobPayloadWithTag {
+                payload: JobPayload::FlowScript {
+                    id,
+                    language,
+                    custom_concurrency_key: custom_concurrency_key.clone(),
+                    concurrent_limit,
+                    concurrency_time_window_s,
+                    cache_ttl: module.cache_ttl.map(|x| x as i32),
+                    dedicated_worker: None,
+                },
+                tag: tag.clone(),
+                delete_after_use,
+                timeout: module.timeout,
+            };
+            Ok(NextFlowTransform::Continue(
+                ContinuePayload::SingleJob(payload),
+                NextStatus::NextStep,
+            ))
+        },
        FlowModuleValue::WhileloopFlow { modules, .. } => {
            // if it's a simple single step flow, we will collapse it as an optimization and need to pass flow_input as an arg
-            let is_simple = is_simple_modules(modules, flow);
+            let is_simple = is_simple_modules(&modules, flow);
            let (flow_jobs, flow_jobs_success) = match status_module {
                FlowStatusModule::InProgress {
                    flow_jobs: Some(flow_jobs),
@@ -3077,7 +3106,7 @@ async fn compute_next_flow_transform(
                    },
                    while_loop: true,
                },
-                modules,
+                &modules,
                flow_job,
                is_simple,
                db,
@@ -3089,12 +3118,13 @@ async fn compute_next_flow_transform(
        /* forloop modules are expected set `iter: { value: Value, index: usize }` as job arguments */
        FlowModuleValue::ForloopFlow { modules, iterator, parallel, .. } => {
            // if it's a simple single step flow, we will collapse it as an optimization and need to pass flow_input as an arg
-            let is_simple = !parallel && is_simple_modules(modules, flow);
+            let is_simple = !parallel && is_simple_modules(&modules, flow);

            // if is_simple {
            //     match value {
            //         FlowModuleValue::Script { input_transforms, .. }
            //         | FlowModuleValue::RawScript { input_transforms, .. }
+            //         | FlowModuleValue::FlowScript { input_transforms, .. }
            //         | FlowModuleValue::Flow { input_transforms, .. } => {
            //             Some(input_transforms.clone())
            //         }
@@ -3106,14 +3136,14 @@ async fn compute_next_flow_transform(
                flow_job,
                previous_id,
                status,
-                iterator,
+                &iterator,
                arc_last_job_result,
                resumes,
                resume,
                approvers,
                arc_flow_job_args,
                client,
-                parallel,
+                &parallel,
            )
            .await?;

@@ -3124,7 +3154,7 @@ async fn compute_next_flow_transform(
                        flow,
                        status,
                        ns,
-                        modules,
+                        &modules,
                        flow_job,
                        is_simple,
                        db,
@@ -3150,7 +3180,7 @@ async fn compute_next_flow_transform(

                        let continue_payload = {
                            let flow_value = FlowValue {
-                                modules: (*modules).clone(),
+                                modules,
                                failure_module: flow.failure_module.clone(),
                                same_worker: flow.same_worker,
                                concurrent_limit: None,
@@ -3181,6 +3211,7 @@ async fn compute_next_flow_transform(
                                //     match value {
                                //         FlowModuleValue::Script { input_transforms, .. }
                                //         | FlowModuleValue::RawScript { input_transforms, .. }
+                                //         | FlowModuleValue::FlowScript { input_transforms, .. }
                                //         | FlowModuleValue::Flow { input_transforms, .. } => {
                                //             Some(input_transforms.clone())
                                //         }
@@ -3281,7 +3312,7 @@ async fn compute_next_flow_transform(
                | FlowStatusModule::WaitingForExecutor { .. } => {
                    if branches.is_empty() {
                        return Ok(NextFlowTransform::EmptyInnerFlows);
-                    } else if *parallel {
+                    } else if parallel {
                        return Ok(NextFlowTransform::Continue(
                            ContinuePayload::BranchAllJobs(
                                branches
@@ -3346,7 +3377,7 @@ async fn compute_next_flow_transform(
                    flow_jobs: Some(flow_jobs),
                    flow_jobs_success,
                    ..
-                } if !*parallel => (
+                } if !parallel => (
                    BranchAllStatus { branch: branch + 1, len: len.clone() },
                    flow_jobs.clone(),
                    flow_jobs_success.clone(),
@@ -3439,6 +3470,7 @@ async fn next_loop_iteration(
                    match value {
                        FlowModuleValue::Script { input_transforms, .. }
                        | FlowModuleValue::RawScript { input_transforms, .. }
+                        | FlowModuleValue::FlowScript { input_transforms, .. }
                        | FlowModuleValue::Flow { input_transforms, .. } => {
                            Some(input_transforms.clone())
                        }
@@ -8,7 +8,7 @@ use sqlx::types::Json;
 use uuid::Uuid;
 use windmill_common::error::Error;
 use windmill_common::error::Result;
-use windmill_common::flows::{FlowModule, FlowModuleValue};
+use windmill_common::flows::{FlowModule, FlowModuleValue, FlowNodeId};
 use windmill_common::get_latest_deployed_hash_for_path;
 use windmill_common::jobs::JobPayload;
 use windmill_common::scripts::ScriptHash;
@@ -606,7 +606,9 @@ pub async fn handle_flow_dependency_job(
        occupancy_metrics,
    )
    .await?;
-    let new_flow_value = serde_json::to_value(flow).map_err(to_anyhow)?;
+    let new_flow_value = sqlx::types::Json(
+        serde_json::value::to_raw_value(&flow).map_err(to_anyhow)?
+    );

    // Re-check cancelation to ensure we don't accidentially override a flow.
    if sqlx::query_scalar!("SELECT canceled FROM queue WHERE id = $1", job.id)
@@ -630,7 +632,7 @@ pub async fn handle_flow_dependency_job(

        sqlx::query!(
            "UPDATE flow SET value = $1 WHERE path = $2 AND workspace_id = $3",
-            new_flow_value,
+            &new_flow_value as &sqlx::types::Json<Box<RawValue>>,
            job_path,
            job.workspace_id
        )
@@ -638,12 +640,23 @@ pub async fn handle_flow_dependency_job(
        .await?;
        sqlx::query!(
            "UPDATE flow_version SET value = $1 WHERE id = $2",
-            new_flow_value,
+            &new_flow_value as &sqlx::types::Json<Box<RawValue>>,
            version
        )
        .execute(db)
        .await?;

+        // Compute a lite version of the flow value (`RawScript` => `FlowScript`).
+        let mut value_lite = flow.clone();
+        tx = reduce(tx, &mut value_lite.modules, &job_path, &job.workspace_id).await?;
+        sqlx::query!(
+            "INSERT INTO flow_version_lite (id, value) VALUES ($1, $2)
+             ON CONFLICT (id) DO UPDATE SET value = EXCLUDED.value",
+            version, sqlx::types::Json(to_raw_value(&value_lite)) as sqlx::types::Json<Box<RawValue>>,
+        )
+        .execute(db)
+        .await?;
+
        tx.commit().await?;

        if let Err(e) = handle_deployment_metadata(
@@ -913,7 +926,7 @@ async fn lock_modules<'c>(
        )
        .await;
        //
-        match new_lock {
+        let lock = match new_lock {
            Ok(new_lock) => {
                let dep_path = path.clone().unwrap_or_else(|| job_path.to_string());
                tx = clear_dependency_map_for_item(
@@ -954,20 +967,7 @@ async fn lock_modules<'c>(
                        language = ScriptLang::Bun;
                    };
                }
-                e.value = windmill_common::worker::to_raw_value(&FlowModuleValue::RawScript {
-                    lock: Some(new_lock),
-                    path,
-                    input_transforms,
-                    content,
-                    language,
-                    tag,
-                    custom_concurrency_key,
-                    concurrent_limit,
-                    concurrency_time_window_s,
-                    is_trigger,
-                });
-                new_flow_modules.push(e);
-                continue;
+                Some(new_lock)
            }
            Err(error) => {
                // TODO: Record flow raw script error lock logs
@@ -977,24 +977,123 @@ async fn lock_modules<'c>(
                    error = ?error,
                    "Failed to generate flow lock for raw script"
                );
-                e.value = windmill_common::worker::to_raw_value(&FlowModuleValue::RawScript {
-                    lock: None,
-                    path,
-                    input_transforms,
+                None
+            }
+        };
+        e.value = windmill_common::worker::to_raw_value(&FlowModuleValue::RawScript {
+            lock,
+            path,
+            input_transforms,
+            content,
+            language,
+            tag,
+            custom_concurrency_key,
+            concurrent_limit,
+            concurrency_time_window_s,
+            is_trigger,
+        });
+        new_flow_modules.push(e);
+        continue;
+    }
+    Ok((new_flow_modules, tx, modified_ids))
+}
+
+async fn insert_flow_node<'c>(
+    mut tx: sqlx::Transaction<'c, sqlx::Postgres>,
+    path: &str,
+    workspace_id: &str,
+    code: Option<&String>,
+    lock: Option<&String>,
+    flow: Option<&Json<Box<RawValue>>>,
+) -> Result<(sqlx::Transaction<'c, sqlx::Postgres>, FlowNodeId)> {
+    let hash = {
+        use std::hash::{DefaultHasher, Hasher, Hash};
+    
+        let mut hasher = DefaultHasher::new();
+        code.hash(&mut hasher);
+        lock.hash(&mut hasher);
+        flow.inspect(|flow| flow.get().hash(&mut hasher));
+        hasher.finish() as i64
+    };
+
+    // Insert the flow node if it doesn't exist.
+    let id = sqlx::query_scalar!(
+        r#"
+        WITH existing AS (
+            SELECT id FROM flow_node
+            WHERE hash = $1 AND path = $2 AND workspace_id = $3 AND code = $4 AND lock = $5 AND flow = $6
+            LIMIT 1
+        ),
+        inserted AS (
+            INSERT INTO flow_node (hash, path, workspace_id, code, lock, flow)
+            VALUES ($1, $2, $3, $4, $5, $6)
+            ON CONFLICT DO NOTHING
+            RETURNING id
+        )
+        SELECT id FROM existing
+        UNION ALL
+        SELECT id FROM inserted
+        "#,
+        hash, path, workspace_id, code, lock, flow as Option<&Json<Box<RawValue>>>
+    )
+    .fetch_one(&mut *tx)
+    .await?
+    .ok_or(error::Error::InternalErr("Failed to cache".to_string()))?;
+    Ok((tx, FlowNodeId(id)))
+}
+
+async fn reduce<'c>(
+    mut tx: sqlx::Transaction<'c, sqlx::Postgres>,
+    modules: &mut Vec<FlowModule>,
+    path: &str,
+    workspace_id: &str,
+) -> Result<sqlx::Transaction<'c, sqlx::Postgres>> {
+    use FlowModuleValue::*;
+    for module in &mut *modules {
+        let mut val = serde_json::from_str::<FlowModuleValue>(module.value.get())
+            .map_err(|err| Error::InternalErr(format!("reduce: Failed to parse flow module value: {}", err)))?;
+        match &mut val {
+            RawScript { .. } => {
+                // In order to avoid an unnecessary `.clone()` of `val`, take ownership of it's content
+                // using `std::mem::replace`.
+                let RawScript {
+                    lock,
                    content,
                    language,
+                    input_transforms,
                    tag,
                    custom_concurrency_key,
                    concurrent_limit,
                    concurrency_time_window_s,
                    is_trigger,
-                });
-                new_flow_modules.push(e);
-                continue;
+                    ..
+                } = std::mem::replace(&mut val, Identity) else { unreachable!() };
+                let id;
+                (tx, id) = insert_flow_node(tx, path, workspace_id, Some(&content), lock.as_ref(), None).await?;
+                val = FlowScript {
+                    input_transforms,
+                    id,
+                    tag,
+                    language,
+                    custom_concurrency_key,
+                    concurrent_limit,
+                    concurrency_time_window_s,
+                    is_trigger,
+                };
+            },
+            ForloopFlow { modules, .. } | WhileloopFlow { modules, .. }  => {
+                tx = Box::pin(reduce(tx, &mut *modules, path, workspace_id)).await?;
            }
+            BranchOne { branches, .. } | BranchAll { branches, .. } => {
+                for branch in &mut *branches {
+                    tx = Box::pin(reduce(tx, &mut branch.modules, path, workspace_id)).await?;
+                }
+            }
+            _ => {}
        }
+        module.value = to_raw_value(&val);
    }
-    Ok((new_flow_modules, tx, modified_ids))
+    Ok(tx)
 }

 fn skip_creating_new_lock(language: &ScriptLang, content: &str) -> bool {