feat: accept direct file upload for webhook/http (s3) (#4903)

* feat: accept direct file upload for webhook/http (s3) * nit * nit
2026-10-09 16:02:38 +00:00 · 2024-12-12 00:33:19 +01:00
parent c580b7cbae
commit a8d95434cc
6 changed files with 525 additions and 363 deletions
@@ -0,0 +1,469 @@
+use std::collections::HashMap;
+
+#[cfg(feature = "parquet")]
+use crate::job_helpers_ee::get_workspace_s3_resource;
+use axum::{
+    extract::{FromRequest, FromRequestParts, Query, Request},
+    http::{HeaderMap, Uri},
+    response::{IntoResponse, Response},
+};
+use bytes::Bytes;
+use http::{header::CONTENT_TYPE, request::Parts};
+#[cfg(feature = "parquet")]
+use object_store::{Attribute, Attributes};
+use serde::Deserialize;
+use serde_json::value::RawValue;
+use sqlx::types::JsonRawValue;
+#[cfg(feature = "parquet")]
+use windmill_common::s3_helpers::build_object_store_client;
+use windmill_common::{error::Error, worker::to_raw_value, DB};
+use windmill_queue::PushArgsOwned;
+
+use crate::db::ApiAuthed;
+#[cfg(feature = "parquet")]
+use crate::job_helpers_ee::{get_random_file_name, upload_file_internal};
+
+pub struct WebhookArgs {
+    pub args: PushArgsOwned,
+    pub file_req: Option<Request>,
+}
+
+impl WebhookArgs {
+    #[cfg(not(feature = "parquet"))]
+    pub async fn to_push_args_owned(
+        self,
+        _authed: &ApiAuthed,
+        _db: &DB,
+        _w_id: &str,
+    ) -> Result<PushArgsOwned, Error> {
+        return Err(Error::BadRequest(format!(
+            "Uploading files requires the parquet feature"
+        )));
+    }
+
+    #[cfg(feature = "parquet")]
+    pub async fn to_push_args_owned(
+        mut self,
+        authed: &ApiAuthed,
+        db: &DB,
+        w_id: &str,
+    ) -> Result<PushArgsOwned, Error> {
+        if let Some(req) = self.file_req {
+            {
+                let (_, s3_resource) =
+                    get_workspace_s3_resource(authed, db, None, "", w_id, None).await?;
+
+                if let Some(s3_resource) = s3_resource {
+                    let s3_client = build_object_store_client(&s3_resource).await?;
+
+                    let content_type = req
+                        .headers()
+                        .get(CONTENT_TYPE)
+                        .map(|x| x.to_str().ok().map(|x| x.to_string()))
+                        .flatten();
+
+                    let file_extension = content_type
+                        .as_ref()
+                        .map(|mime_str| {
+                            mime_guess::get_mime_extensions_str(mime_str)
+                                .map(|x| x.first().map(|x| x.to_string()))
+                        })
+                        .flatten()
+                        .flatten();
+
+                    let file_key = get_random_file_name(file_extension);
+
+                    let options = Attributes::from_iter(vec![
+                        (
+                            Attribute::ContentType,
+                            content_type.unwrap_or("application/octet-stream".to_string()),
+                        ),
+                        (Attribute::ContentDisposition, "inline".to_string()),
+                    ])
+                    .into();
+
+                    upload_file_internal(s3_client, &file_key, req, options).await?;
+
+                    self.args.args.insert(
+                        "body".to_string(),
+                        to_raw_value(&serde_json::json!({
+                            "s3": &file_key
+                        })),
+                    );
+
+                    return Ok(self.args);
+                }
+            }
+
+            return Err(Error::BadRequest(format!(
+                "You need to connect your workspace to an S3 bucket to upload files"
+            )));
+        }
+
+        Ok(self.args)
+    }
+}
+
+#[derive(Deserialize)]
+pub struct RequestQuery {
+    pub raw: Option<bool>,
+    pub wrap_body: Option<bool>,
+    pub include_header: Option<String>,
+}
+
+#[axum::async_trait]
+impl<S> FromRequest<S, axum::body::Body> for WebhookArgs
+where
+    S: Send + Sync,
+{
+    type Rejection = Response;
+
+    async fn from_request(
+        req: Request<axum::body::Body>,
+        _state: &S,
+    ) -> Result<Self, Self::Rejection> {
+        let (content_type, mut extra, use_raw, wrap_body) = {
+            let headers_map = req.headers();
+            let content_type_header = headers_map.get(CONTENT_TYPE);
+            let content_type = content_type_header.and_then(|value| value.to_str().ok());
+            let uri = req.uri();
+            let query = Query::<RequestQuery>::try_from_uri(uri).unwrap().0;
+            let mut extra = build_extra(&headers_map, query.include_header);
+            let query_decode = DecodeQueries::from_uri(uri);
+            if let Some(DecodeQueries(queries)) = query_decode {
+                extra.extend(queries);
+            }
+            let raw = query.raw.as_ref().is_some_and(|x| *x);
+            let wrap_body = query.wrap_body.as_ref().is_some_and(|x| *x);
+            (content_type, extra, raw, wrap_body)
+        };
+
+        let no_content_type = content_type.is_none();
+        if no_content_type || content_type.unwrap().starts_with("application/json") {
+            let bytes = Bytes::from_request(req, _state)
+                .await
+                .map_err(IntoResponse::into_response)?;
+            if no_content_type && bytes.is_empty() {
+                if use_raw {
+                    extra.insert("raw_string".to_string(), to_raw_value(&"".to_string()));
+                }
+                let mut args = HashMap::new();
+                if wrap_body {
+                    args.insert("body".to_string(), to_raw_value(&serde_json::json!({})));
+                }
+                return Ok(Self {
+                    args: PushArgsOwned { extra: Some(extra), args: args },
+                    file_req: None,
+                });
+            }
+            let str = String::from_utf8(bytes.to_vec())
+                .map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)).into_response())?;
+
+            PushArgsOwned::from_json(extra, use_raw, wrap_body, str)
+                .await
+                .map(|args| Self { args, file_req: None })
+        } else if content_type
+            .unwrap()
+            .starts_with("application/cloudevents+json")
+        {
+            let bytes = Bytes::from_request(req, _state)
+                .await
+                .map_err(IntoResponse::into_response)?;
+            let str = String::from_utf8(bytes.to_vec())
+                .map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)).into_response())?;
+
+            PushArgsOwned::from_ce_json(extra, use_raw, str)
+                .await
+                .map(|args| Self { args, file_req: None })
+        } else if content_type
+            .unwrap()
+            .starts_with("application/cloudevents-batch+json")
+        {
+            Err(
+                Error::BadRequest(format!("Cloud events batching is not supported yet"))
+                    .into_response(),
+            )
+        } else if content_type.unwrap().starts_with("text/plain") {
+            let bytes = Bytes::from_request(req, _state)
+                .await
+                .map_err(IntoResponse::into_response)?;
+            let str = String::from_utf8(bytes.to_vec())
+                .map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)).into_response())?;
+            extra.insert("raw_string".to_string(), to_raw_value(&str));
+            Ok(Self {
+                args: PushArgsOwned { extra: Some(extra), args: HashMap::new() },
+                file_req: None,
+            })
+        } else if content_type
+            .unwrap()
+            .starts_with("application/x-www-form-urlencoded")
+        {
+            let bytes = Bytes::from_request(req, _state)
+                .await
+                .map_err(IntoResponse::into_response)?;
+
+            if use_raw {
+                let raw_string = String::from_utf8(bytes.to_vec()).map_err(|e| {
+                    Error::BadRequest(format!("invalid utf8: {}", e)).into_response()
+                })?;
+                extra.insert("raw_string".to_string(), to_raw_value(&raw_string));
+            }
+
+            let payload: HashMap<String, Option<String>> = serde_urlencoded::from_bytes(&bytes)
+                .map_err(|e| {
+                    Error::BadRequest(format!("invalid urlencoded data: {}", e)).into_response()
+                })?;
+            let payload = payload
+                .into_iter()
+                .map(|(k, v)| (k, to_raw_value(&v)))
+                .collect::<HashMap<_, _>>();
+
+            return Ok(Self {
+                args: PushArgsOwned { extra: Some(extra), args: payload },
+                file_req: None,
+            });
+        } else {
+            return Ok(Self {
+                args: PushArgsOwned { extra: None, args: HashMap::new() },
+                file_req: Some(req),
+            });
+
+            // Err(StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response())
+        }
+    }
+}
+
+lazy_static::lazy_static! {
+    static ref INCLUDE_HEADERS: Vec<String> = std::env::var("INCLUDE_HEADERS")
+        .ok().map(|x| x
+        .split(',')
+        .map(|s| s.to_string())
+        .collect()).unwrap_or_default();
+}
+
+pub fn build_extra(
+    headers: &HeaderMap,
+    include_header: Option<String>,
+) -> HashMap<String, Box<RawValue>> {
+    let mut args = HashMap::new();
+    let whitelist = include_header
+        .map(|s| s.split(",").map(|s| s.to_string()).collect::<Vec<_>>())
+        .unwrap_or_default();
+
+    whitelist
+        .iter()
+        .chain(INCLUDE_HEADERS.iter())
+        .for_each(|h| {
+            if let Some(v) = headers.get(h) {
+                args.insert(
+                    h.to_string().to_lowercase().replace('-', "_"),
+                    to_raw_value(&v.to_str().unwrap().to_string()),
+                );
+            }
+        });
+    args
+}
+
+#[derive(Deserialize)]
+pub struct IncludeQuery {
+    pub include_query: Option<String>,
+}
+
+pub struct DecodeQueries(pub HashMap<String, Box<RawValue>>);
+
+#[axum::async_trait]
+impl<S> FromRequestParts<S> for DecodeQueries
+where
+    S: Send + Sync,
+{
+    type Rejection = Response;
+
+    async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
+        Ok(DecodeQueries::from_uri(&parts.uri).unwrap_or_else(|| DecodeQueries(HashMap::new())))
+    }
+}
+
+impl DecodeQueries {
+    pub fn from_uri(uri: &Uri) -> Option<Self> {
+        let query = uri.query();
+        if query.is_none() {
+            return None;
+        }
+        let query = query.unwrap();
+        let include_query = serde_urlencoded::from_str::<IncludeQuery>(query)
+            .map(|x| x.include_query)
+            .ok()
+            .flatten()
+            .unwrap_or_default();
+        let parse_query_args = include_query
+            .split(",")
+            .map(|s| s.to_string())
+            .collect::<Vec<_>>();
+        let mut args = HashMap::new();
+        if !parse_query_args.is_empty() {
+            let queries =
+                serde_urlencoded::from_str::<HashMap<String, String>>(query).unwrap_or_default();
+            parse_query_args.iter().for_each(|h| {
+                if let Some(v) = queries.get(h) {
+                    args.insert(h.to_string(), to_raw_value(v));
+                }
+            });
+        }
+        Some(DecodeQueries(args))
+    }
+}
+
+// impl<'c> PushArgs<'c> {
+//     pub fn insert<K: Into<String>, V: Into<Box<RawValue>>>(&mut self, k: K, v: V) {
+//         self.extra.insert(k.into(), v.into());
+//     }
+// }
+
+fn restructure_cloudevents_metadata(
+    mut p: HashMap<String, Box<RawValue>>,
+) -> Result<HashMap<String, Box<RawValue>>, Error> {
+    let data = p
+        .remove("data")
+        .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() != '{';
+
+    if wrap_body {
+        let args = serde_json::from_str::<Option<Box<RawValue>>>(&str)
+            .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)))?
+            .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null));
+        let mut hm = HashMap::new();
+        hm.insert("body".to_string(), args);
+        hm.insert("WEBHOOK__METADATA__".to_string(), to_raw_value(&p));
+        Ok(hm)
+    } else {
+        let mut hm = serde_json::from_str::<Option<HashMap<String, Box<JsonRawValue>>>>(&str)
+            .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)))?
+            .unwrap_or_else(HashMap::new);
+        hm.insert("WEBHOOK__METADATA__".to_string(), to_raw_value(&p));
+        Ok(hm)
+    }
+}
+
+trait PushArgsOwnedExt: Sized {
+    async fn from_json(
+        extra: HashMap<String, Box<RawValue>>,
+        use_raw: bool,
+        force_wrap_body: bool,
+        str: String,
+    ) -> Result<Self, Response>;
+
+    async fn from_ce_json(
+        extra: HashMap<String, Box<RawValue>>,
+        use_raw: bool,
+        str: String,
+    ) -> Result<Self, Response>;
+}
+
+impl PushArgsOwnedExt for PushArgsOwned {
+    async fn from_json(
+        mut extra: HashMap<String, Box<RawValue>>,
+        use_raw: bool,
+        force_wrap_body: bool,
+        str: String,
+    ) -> Result<Self, Response> {
+        if use_raw {
+            extra.insert("raw_string".to_string(), to_raw_value(&str));
+        }
+
+        let wrap_body = force_wrap_body || str.len() > 0 && str.chars().next().unwrap() != '{';
+
+        if wrap_body {
+            let args = serde_json::from_str::<Option<Box<RawValue>>>(&str)
+                .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)).into_response())?
+                .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null));
+            let mut hm = HashMap::new();
+            hm.insert("body".to_string(), args);
+            Ok(PushArgsOwned { extra: Some(extra), args: hm })
+        } else {
+            let hm = serde_json::from_str::<Option<HashMap<String, Box<JsonRawValue>>>>(&str)
+                .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)).into_response())?
+                .unwrap_or_else(HashMap::new);
+            Ok(PushArgsOwned { extra: Some(extra), args: hm })
+        }
+    }
+
+    async fn from_ce_json(
+        mut extra: HashMap<String, Box<RawValue>>,
+        use_raw: bool,
+        str: String,
+    ) -> Result<Self, Response> {
+        if use_raw {
+            extra.insert("raw_string".to_string(), to_raw_value(&str));
+        }
+
+        let hm = serde_json::from_str::<HashMap<String, Box<RawValue>>>(&str).map_err(|e| {
+            Error::BadRequest(format!("invalid cloudevents+json: {}", e)).into_response()
+        })?;
+        let hm = restructure_cloudevents_metadata(hm).map_err(|e| e.into_response())?;
+        Ok(PushArgsOwned { extra: Some(extra), args: hm })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+
+    use super::*;
+
+    #[tokio::test]
+    async fn test_cloudevents_json_payload() {
+        let r1 = r#"
+        {
+            "specversion" : "1.0",
+            "type" : "com.example.someevent",
+            "source" : "/mycontext",
+            "subject": null,
+            "id" : "C234-1234-1234",
+            "time" : "2018-04-05T17:31:00Z",
+            "comexampleextension1" : "value",
+            "comexampleothervalue" : 5,
+            "datacontenttype" : "application/json",
+            "data" : {
+                "appinfoA" : "abc",
+                "appinfoB" : 123,
+                "appinfoC" : true
+            }
+        }
+        "#;
+        let r2 = r#"
+        {
+            "specversion" : "1.0",
+            "type" : "com.example.someevent",
+            "source" : "/mycontext",
+            "subject": null,
+            "id" : "C234-1234-1234",
+            "time" : "2018-04-05T17:31:00Z",
+            "comexampleextension1" : "value",
+            "comexampleothervalue" : 5,
+            "datacontenttype" : "application/json",
+            "data" : 1.5
+        }
+        "#;
+        let extra = HashMap::new();
+
+        let a1 = PushArgsOwned::from_ce_json(extra.clone(), false, r1.to_string())
+            .await
+            .expect("Failed to parse the cloudevent");
+        let a2 = PushArgsOwned::from_ce_json(extra.clone(), false, r2.to_string())
+            .await
+            .expect("Failed to parse the cloudevent");
+
+        a1.args.get("WEBHOOK__METADATA__").expect(
+            "CloudEvents should generate a neighboring `webhook-metadata` field in PushArgs",
+        );
+        assert_eq!(
+            a2.args
+                .get("body")
+                .expect("Cloud events with a data field with no wrapping curly brackets should be inside of a `body` field in PushArgs")
+                .to_string(),
+            "1.5"
+        );
+    }
+}
@@ -18,9 +18,12 @@ use windmill_common::{
    error::{JsonResult, Result},
    utils::{not_found_if_none, StripPath},
 };
-use windmill_queue::{PushArgs, PushArgsOwned};
+use windmill_queue::PushArgs;

-use crate::db::{ApiAuthed, DB};
+use crate::{
+    args::WebhookArgs,
+    db::{ApiAuthed, DB},
+};

 const KEEP_LAST: i64 = 8;

@@ -86,8 +89,10 @@ pub async fn new_payload(
 pub async fn update_payload(
    Extension(db): Extension<DB>,
    Path((w_id, path)): Path<(String, StripPath)>,
-    args: PushArgsOwned,
+    args: WebhookArgs,
 ) -> Result<StatusCode> {
+    let args = args.args;
+
    let mut tx = db.begin().await?;

    sqlx::query!(
@@ -1,6 +1,7 @@
 #[cfg(feature = "parquet")]
 use crate::job_helpers_ee::get_workspace_s3_resource;
 use crate::{
+    args::WebhookArgs,
    db::{ApiAuthed, DB},
    jobs::{
        run_flow_by_path_inner, run_script_by_path_inner, run_wait_result_flow_by_path_internal,
@@ -32,7 +33,6 @@ use windmill_common::{
    utils::{not_found_if_none, paginate, require_admin, Pagination, StripPath},
    worker::{to_raw_value, CLOUD_HOSTED},
 };
-use windmill_queue::PushArgsOwned;

 lazy_static::lazy_static! {
    static ref ROUTE_PATH_KEY_RE: regex::Regex = regex::Regex::new(r"/:\w+").unwrap();
@@ -531,7 +531,7 @@ async fn route_job(
    Query(query): Query<HashMap<String, String>>,
    method: http::Method,
    headers: HeaderMap,
-    mut args: PushArgsOwned,
+    args: WebhookArgs,
 ) -> impl IntoResponse {
    let route_path = route_path.to_path();
    let (trigger, called_path, params, authed) = match get_http_route_trigger(
@@ -547,6 +547,14 @@ async fn route_job(
        Err(e) => return e.into_response(),
    };

+    let mut args = match args
+        .to_push_args_owned(&authed, &db, &trigger.workspace_id)
+        .await
+    {
+        Ok(args) => args,
+        Err(e) => return e.into_response(),
+    };
+
    #[cfg(not(feature = "parquet"))]
    if trigger.static_asset_config.is_some() {
        return error::Error::InternalErr(
@@ -36,6 +36,7 @@ use crate::db::ApiAuthed;
 use crate::users::get_scope_tags;
 use crate::utils::content_plain;
 use crate::{
+    args::{DecodeQueries, WebhookArgs},
    db::DB,
    users::{check_scopes, require_owner_of_path, OptAuthed},
    utils::require_super_admin,
@@ -83,7 +84,7 @@ use windmill_common::{METRICS_DEBUG_ENABLED, METRICS_ENABLED};
 use windmill_common::{get_latest_deployed_hash_for_path, BASE_URL};
 use windmill_queue::{
    cancel_job, get_queued_job, get_result_by_id_from_running_flow, job_is_complete, push,
-    DecodeQueries, PushArgs, PushArgsOwned, PushIsolationLevel,
+    PushArgs, PushArgsOwned, PushIsolationLevel,
 };

 #[cfg(feature = "prometheus")]
@@ -631,7 +632,10 @@ async fn get_flow_job_debug_info(
            }
        }
        for job_id in job_ids {
-            let job = GetQuery::new().with_auth(&opt_authed).fetch(&db, job_id, &w_id).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);
            }
@@ -832,7 +836,10 @@ impl<'a> GetQuery<'a> {

        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?;
+            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);
@@ -842,14 +849,16 @@ impl<'a> GetQuery<'a> {
    }

    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)
+        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)
+                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())
@@ -2940,8 +2949,10 @@ pub async fn run_flow_by_path(
    Extension(user_db): Extension<UserDB>,
    Path((w_id, flow_path)): Path<(String, StripPath)>,
    Query(run_query): Query<RunJobQuery>,
-    args: PushArgsOwned,
+    args: WebhookArgs,
 ) -> error::Result<(StatusCode, String)> {
+    let args = args.to_push_args_owned(&authed, &db, &w_id).await?;
+
    run_flow_by_path_inner(authed, db, user_db, w_id, flow_path, run_query, args, None).await
 }

@@ -3116,8 +3127,9 @@ pub async fn run_script_by_path(
    Extension(user_db): Extension<UserDB>,
    Path((w_id, script_path)): Path<(String, StripPath)>,
    Query(run_query): Query<RunJobQuery>,
-    args: PushArgsOwned,
+    args: WebhookArgs,
 ) -> error::Result<(StatusCode, String)> {
+    let args = args.to_push_args_owned(&authed, &db, &w_id).await?;
    run_script_by_path_inner(
        authed,
        db,
@@ -3218,7 +3230,10 @@ pub async fn run_workflow_as_code(
        i += 1;
    }

-    let job = GetQuery::new().without_logs().fetch_queued(&db, job_id, &w_id).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} ");
@@ -3779,11 +3794,13 @@ pub async fn run_wait_result_script_by_path(
    Extension(db): Extension<DB>,
    Path((w_id, script_path)): Path<(String, StripPath)>,
    Query(run_query): Query<RunJobQuery>,
-    args: PushArgsOwned,
+    args: WebhookArgs,
 ) -> error::Result<Response> {
    #[cfg(feature = "enterprise")]
    check_license_key_valid().await?;

+    let args = args.to_push_args_owned(&authed, &db, &w_id).await?;
+
    run_wait_result_script_by_path_internal(
        db,
        run_query,
@@ -3861,11 +3878,13 @@ pub async fn run_wait_result_script_by_hash(
    Extension(db): Extension<DB>,
    Path((w_id, script_hash)): Path<(String, ScriptHash)>,
    Query(run_query): Query<RunJobQuery>,
-    args: PushArgsOwned,
+    args: WebhookArgs,
 ) -> error::Result<Response> {
    #[cfg(feature = "enterprise")]
    check_license_key_valid().await?;

+    let args = args.to_push_args_owned(&authed, &db, &w_id).await?;
+
    check_queue_too_long(&db, run_query.queue_limit).await?;

    let hash = script_hash.0;
@@ -3945,11 +3964,13 @@ pub async fn run_wait_result_flow_by_path(
    Extension(db): Extension<DB>,
    Path((w_id, flow_path)): Path<(String, StripPath)>,
    Query(run_query): Query<RunJobQuery>,
-    args: PushArgsOwned,
+    args: WebhookArgs,
 ) -> error::Result<Response> {
    #[cfg(feature = "enterprise")]
    check_license_key_valid().await?;

+    let args = args.to_push_args_owned(&authed, &db, &w_id).await?;
+
    run_wait_result_flow_by_path_internal(
        db, run_query, flow_path, authed, user_db, args, w_id, None,
    )
@@ -4692,12 +4713,12 @@ async fn run_preview_flow_job(
 pub async fn run_job_by_hash(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
-
    Extension(user_db): Extension<UserDB>,
    Path((w_id, script_hash)): Path<(String, ScriptHash)>,
    Query(run_query): Query<RunJobQuery>,
-    args: PushArgsOwned,
+    args: WebhookArgs,
 ) -> error::Result<(StatusCode, String)> {
+    let args = args.to_push_args_owned(&authed, &db, &w_id).await?;
    run_job_by_hash_inner(
        authed,
        db,
@@ -4747,6 +4768,7 @@ pub async fn run_job_by_hash_inner(
    let tag = run_query.tag.clone().or(tag);

    check_tag_available_for_workspace(&w_id, &tag, &authed).await?;
+
    let tx = PushIsolationLevel::Isolated(user_db, authed.clone().into());

    let (uuid, tx) = push(
@@ -46,6 +46,7 @@ use windmill_common::error::AppError;

 mod ai;
 mod apps;
+mod args;
 mod audit;
 mod capture;
 mod concurrency_groups;
@@ -10,23 +10,14 @@ use std::{borrow::Borrow, collections::HashMap, sync::Arc, vec};

 use anyhow::Context;
 use async_recursion::async_recursion;
-use axum::{
-    body::Bytes,
-    extract::{FromRequest, FromRequestParts, Query},
-    http::{request::Parts, Request, Uri},
-    response::{IntoResponse, Response},
-};
 use chrono::{DateTime, Duration, Utc};
 use futures::future::TryFutureExt;
 use itertools::Itertools;
 #[cfg(feature = "prometheus")]
 use prometheus::IntCounter;
 use regex::Regex;
-use reqwest::{
-    header::{HeaderMap, CONTENT_TYPE},
-    Client, StatusCode,
-};
-use serde::{ser::SerializeMap, Deserialize, Serialize};
+use reqwest::Client;
+use serde::{ser::SerializeMap, Serialize};
 use serde_json::{json, value::RawValue};
 use sqlx::{types::Json, FromRow, Pool, Postgres, Transaction};
 #[cfg(feature = "benchmark")]
@@ -2569,340 +2560,6 @@ impl<'c> Serialize for PushArgs<'c> {
    }
 }

-#[derive(Deserialize)]
-pub struct DecodeQuery {
-    pub include_query: Option<String>,
-}
-
-#[derive(Deserialize)]
-pub struct IncludeQuery {
-    pub include_query: Option<String>,
-}
-
-pub struct DecodeQueries(pub HashMap<String, Box<RawValue>>);
-
-#[axum::async_trait]
-impl<S> FromRequestParts<S> for DecodeQueries
-where
-    S: Send + Sync,
-{
-    type Rejection = Response;
-
-    async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
-        Ok(DecodeQueries::from_uri(&parts.uri).unwrap_or_else(|| DecodeQueries(HashMap::new())))
-    }
-}
-
-impl DecodeQueries {
-    fn from_uri(uri: &Uri) -> Option<Self> {
-        let query = uri.query();
-        if query.is_none() {
-            return None;
-        }
-        let query = query.unwrap();
-        let include_query = serde_urlencoded::from_str::<IncludeQuery>(query)
-            .map(|x| x.include_query)
-            .ok()
-            .flatten()
-            .unwrap_or_default();
-        let parse_query_args = include_query
-            .split(",")
-            .map(|s| s.to_string())
-            .collect::<Vec<_>>();
-        let mut args = HashMap::new();
-        if !parse_query_args.is_empty() {
-            let queries =
-                serde_urlencoded::from_str::<HashMap<String, String>>(query).unwrap_or_default();
-            parse_query_args.iter().for_each(|h| {
-                if let Some(v) = queries.get(h) {
-                    args.insert(h.to_string(), to_raw_value(v));
-                }
-            });
-        }
-        Some(DecodeQueries(args))
-    }
-}
-
-// impl<'c> PushArgs<'c> {
-//     pub fn insert<K: Into<String>, V: Into<Box<RawValue>>>(&mut self, k: K, v: V) {
-//         self.extra.insert(k.into(), v.into());
-//     }
-// }
-
-#[derive(Deserialize)]
-pub struct RequestQuery {
-    pub raw: Option<bool>,
-    pub wrap_body: Option<bool>,
-    pub include_header: Option<String>,
-}
-
-fn restructure_cloudevents_metadata(
-    mut p: HashMap<String, Box<RawValue>>,
-) -> Result<HashMap<String, Box<RawValue>>, Error> {
-    let data = p
-        .remove("data")
-        .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() != '{';
-
-    if wrap_body {
-        let args = serde_json::from_str::<Option<Box<RawValue>>>(&str)
-            .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)))?
-            .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null));
-        let mut hm = HashMap::new();
-        hm.insert("body".to_string(), args);
-        hm.insert("WEBHOOK__METADATA__".to_string(), to_raw_value(&p));
-        Ok(hm)
-    } else {
-        let mut hm = serde_json::from_str::<Option<HashMap<String, Box<JsonRawValue>>>>(&str)
-            .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)))?
-            .unwrap_or_else(HashMap::new);
-        hm.insert("WEBHOOK__METADATA__".to_string(), to_raw_value(&p));
-        Ok(hm)
-    }
-}
-
-impl PushArgsOwned {
-    async fn from_json(
-        mut extra: HashMap<String, Box<RawValue>>,
-        use_raw: bool,
-        force_wrap_body: bool,
-        str: String,
-    ) -> Result<Self, Response> {
-        if use_raw {
-            extra.insert("raw_string".to_string(), to_raw_value(&str));
-        }
-
-        let wrap_body = force_wrap_body || str.len() > 0 && str.chars().next().unwrap() != '{';
-
-        if wrap_body {
-            let args = serde_json::from_str::<Option<Box<RawValue>>>(&str)
-                .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)).into_response())?
-                .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null));
-            let mut hm = HashMap::new();
-            hm.insert("body".to_string(), args);
-            Ok(PushArgsOwned { extra: Some(extra), args: hm })
-        } else {
-            let hm = serde_json::from_str::<Option<HashMap<String, Box<JsonRawValue>>>>(&str)
-                .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)).into_response())?
-                .unwrap_or_else(HashMap::new);
-            Ok(PushArgsOwned { extra: Some(extra), args: hm })
-        }
-    }
-
-    async fn from_ce_json(
-        mut extra: HashMap<String, Box<RawValue>>,
-        use_raw: bool,
-        str: String,
-    ) -> Result<Self, Response> {
-        if use_raw {
-            extra.insert("raw_string".to_string(), to_raw_value(&str));
-        }
-
-        let hm = serde_json::from_str::<HashMap<String, Box<RawValue>>>(&str).map_err(|e| {
-            Error::BadRequest(format!("invalid cloudevents+json: {}", e)).into_response()
-        })?;
-        let hm = restructure_cloudevents_metadata(hm).map_err(|e| e.into_response())?;
-        Ok(PushArgsOwned { extra: Some(extra), args: hm })
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[tokio::test]
-    async fn test_cloudevents_json_payload() {
-        let r1 = r#"
-        {
-            "specversion" : "1.0",
-            "type" : "com.example.someevent",
-            "source" : "/mycontext",
-            "subject": null,
-            "id" : "C234-1234-1234",
-            "time" : "2018-04-05T17:31:00Z",
-            "comexampleextension1" : "value",
-            "comexampleothervalue" : 5,
-            "datacontenttype" : "application/json",
-            "data" : {
-                "appinfoA" : "abc",
-                "appinfoB" : 123,
-                "appinfoC" : true
-            }
-        }
-        "#;
-        let r2 = r#"
-        {
-            "specversion" : "1.0",
-            "type" : "com.example.someevent",
-            "source" : "/mycontext",
-            "subject": null,
-            "id" : "C234-1234-1234",
-            "time" : "2018-04-05T17:31:00Z",
-            "comexampleextension1" : "value",
-            "comexampleothervalue" : 5,
-            "datacontenttype" : "application/json",
-            "data" : 1.5
-        }
-        "#;
-        let extra = HashMap::new();
-
-        let a1 = PushArgsOwned::from_ce_json(extra.clone(), false, r1.to_string())
-            .await
-            .expect("Failed to parse the cloudevent");
-        let a2 = PushArgsOwned::from_ce_json(extra.clone(), false, r2.to_string())
-            .await
-            .expect("Failed to parse the cloudevent");
-
-        a1.args.get("WEBHOOK__METADATA__").expect(
-            "CloudEvents should generate a neighboring `webhook-metadata` field in PushArgs",
-        );
-        assert_eq!(
-            a2.args
-                .get("body")
-                .expect("Cloud events with a data field with no wrapping curly brackets should be inside of a `body` field in PushArgs")
-                .to_string(),
-            "1.5"
-        );
-    }
-}
-
-#[axum::async_trait]
-impl<S> FromRequest<S, axum::body::Body> for PushArgsOwned
-where
-    S: Send + Sync,
-{
-    type Rejection = Response;
-
-    async fn from_request(
-        req: Request<axum::body::Body>,
-        _state: &S,
-    ) -> Result<Self, Self::Rejection> {
-        let (content_type, mut extra, use_raw, wrap_body) = {
-            let headers_map = req.headers();
-            let content_type_header = headers_map.get(CONTENT_TYPE);
-            let content_type = content_type_header.and_then(|value| value.to_str().ok());
-            let uri = req.uri();
-            let query = Query::<RequestQuery>::try_from_uri(uri).unwrap().0;
-            let mut extra = build_extra(&headers_map, query.include_header);
-            let query_decode = DecodeQueries::from_uri(uri);
-            if let Some(DecodeQueries(queries)) = query_decode {
-                extra.extend(queries);
-            }
-            let raw = query.raw.as_ref().is_some_and(|x| *x);
-            let wrap_body = query.wrap_body.as_ref().is_some_and(|x| *x);
-            (content_type, extra, raw, wrap_body)
-        };
-
-        let no_content_type = content_type.is_none();
-        if no_content_type || content_type.unwrap().starts_with("application/json") {
-            let bytes = Bytes::from_request(req, _state)
-                .await
-                .map_err(IntoResponse::into_response)?;
-            if no_content_type && bytes.is_empty() {
-                if use_raw {
-                    extra.insert("raw_string".to_string(), to_raw_value(&"".to_string()));
-                }
-                let mut args = HashMap::new();
-                if wrap_body {
-                    args.insert("body".to_string(), to_raw_value(&serde_json::json!({})));
-                }
-                return Ok(PushArgsOwned { extra: Some(extra), args: args });
-            }
-            let str = String::from_utf8(bytes.to_vec())
-                .map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)).into_response())?;
-
-            PushArgsOwned::from_json(extra, use_raw, wrap_body, str).await
-        } else if content_type
-            .unwrap()
-            .starts_with("application/cloudevents+json")
-        {
-            let bytes = Bytes::from_request(req, _state)
-                .await
-                .map_err(IntoResponse::into_response)?;
-            let str = String::from_utf8(bytes.to_vec())
-                .map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)).into_response())?;
-
-            PushArgsOwned::from_ce_json(extra, use_raw, str).await
-        } else if content_type
-            .unwrap()
-            .starts_with("application/cloudevents-batch+json")
-        {
-            Err(
-                Error::BadRequest(format!("Cloud events batching is not supported yet"))
-                    .into_response(),
-            )
-        } else if content_type.unwrap().starts_with("text/plain") {
-            let bytes = Bytes::from_request(req, _state)
-                .await
-                .map_err(IntoResponse::into_response)?;
-            let str = String::from_utf8(bytes.to_vec())
-                .map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)).into_response())?;
-            extra.insert("raw_string".to_string(), to_raw_value(&str));
-            Ok(PushArgsOwned { extra: Some(extra), args: HashMap::new() })
-        } else if content_type
-            .unwrap()
-            .starts_with("application/x-www-form-urlencoded")
-        {
-            let bytes = Bytes::from_request(req, _state)
-                .await
-                .map_err(IntoResponse::into_response)?;
-
-            if use_raw {
-                let raw_string = String::from_utf8(bytes.to_vec()).map_err(|e| {
-                    Error::BadRequest(format!("invalid utf8: {}", e)).into_response()
-                })?;
-                extra.insert("raw_string".to_string(), to_raw_value(&raw_string));
-            }
-
-            let payload: HashMap<String, Option<String>> = serde_urlencoded::from_bytes(&bytes)
-                .map_err(|e| {
-                    Error::BadRequest(format!("invalid urlencoded data: {}", e)).into_response()
-                })?;
-            let payload = payload
-                .into_iter()
-                .map(|(k, v)| (k, to_raw_value(&v)))
-                .collect::<HashMap<_, _>>();
-
-            return Ok(PushArgsOwned { extra: Some(extra), args: payload });
-        } else {
-            Err(StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response())
-        }
-    }
-}
-
-lazy_static::lazy_static! {
-    static ref INCLUDE_HEADERS: Vec<String> = std::env::var("INCLUDE_HEADERS")
-        .ok().map(|x| x
-        .split(',')
-        .map(|s| s.to_string())
-        .collect()).unwrap_or_default();
-}
-
-pub fn build_extra(
-    headers: &HeaderMap,
-    include_header: Option<String>,
-) -> HashMap<String, Box<RawValue>> {
-    let mut args = HashMap::new();
-    let whitelist = include_header
-        .map(|s| s.split(",").map(|s| s.to_string()).collect::<Vec<_>>())
-        .unwrap_or_default();
-
-    whitelist
-        .iter()
-        .chain(INCLUDE_HEADERS.iter())
-        .for_each(|h| {
-            if let Some(v) = headers.get(h) {
-                args.insert(
-                    h.to_string().to_lowercase().replace('-', "_"),
-                    to_raw_value(&v.to_str().unwrap().to_string()),
-                );
-            }
-        });
-    args
-}
-
 impl PushArgsOwned {
    pub fn empty() -> Self {
        PushArgsOwned { extra: None, args: HashMap::new() }