windmill/backend/windmill-api/src/triggers/http/handler.rs

use super::{
    effective_allowed_origins, http_trigger_args::RawHttpTriggerArgs, match_origin,
    refresh_routers, AuthenticationMethod, HttpMethod, RequestType, TriggerRoute,
    HTTP_ACCESS_CACHE, HTTP_AUTH_CACHE, HTTP_ROUTERS_CACHE,
};
use crate::{
    auth::{AuthCache, OptTokened},
    db::{ApiAuthed, DB},
    jobs::start_job_update_sse_stream,
    triggers::trigger_helpers::{
        get_runnable_format, trigger_runnable, trigger_runnable_and_wait_for_result,
        trigger_runnable_inner, RunnableId,
    },
    utils::{check_scopes, ExpiringCacheEntry},
};
use axum::{
    extract::Path,
    response::{IntoResponse, Response},
    routing::get,
    Extension, Router,
};
use futures::StreamExt;
use http::{HeaderMap, StatusCode};
use std::{collections::HashMap, sync::Arc};
use windmill_common::{
    db::UserDB,
    error::{Error, Result},
    global_settings::HTTP_ROUTE_DEFAULT_ALLOWED_ORIGINS,
    jobs::JobTriggerKind,
    triggers::{TriggerKind, TriggerMetadata},
    utils::{not_found_if_none, StripPath},
};
use windmill_store::resources::try_get_resource_from_db_as;
use windmill_trigger::TriggerMode;

#[cfg(feature = "parquet")]
use {
    crate::job_helpers_oss::get_workspace_s3_resource,
    windmill_object_store::build_object_store_client,
};

/// Which router a request's CORS decision must be looked up in.
///
/// A preflight names the method it is asking about in
/// `Access-Control-Request-Method`; the routers are keyed by method, so without
/// that header there is nothing to look up.
fn cors_lookup_method(req: &axum::extract::Request) -> Option<HttpMethod> {
    let method = req.method();
    if method == http::Method::OPTIONS {
        req.headers()
            .get(http::header::ACCESS_CONTROL_REQUEST_METHOD)
            .and_then(|method| method.to_str().ok())
            .and_then(|method| http::Method::try_from(method).ok())
            .as_ref()
            .and_then(routable_method)
    } else {
        routable_method(method)
    }
}

/// The router key a request method maps to. `HEAD` resolves the `GET` route it
/// mirrors, and does so for a preflight naming it too: browsers send
/// `Access-Control-Request-Method: HEAD` when the HEAD carries a non-safelisted
/// header, and answering that preflight from a different route than the request
/// itself resolves is how the two come to disagree.
fn routable_method(method: &http::Method) -> Option<HttpMethod> {
    if method == http::Method::HEAD {
        Some(HttpMethod::Get)
    } else {
        HttpMethod::try_from(method).ok()
    }
}

/// The key to look a request up by, matching what `route_job` resolves it to.
///
/// `Path<StripPath>` percent-decodes before `get_http_route_trigger` builds its
/// lookup key, so decoding here is what keeps the two agreeing: on the raw path,
/// `/us%65rs` misses the trigger registered at `/users` that goes on to serve the
/// request, and the response would carry the permissive default instead of that
/// trigger's allowlist.
fn cors_lookup_path(raw_path: &str) -> Option<String> {
    let decoded = urlencoding::decode(raw_path).ok()?;
    // `StripPath::to_path` strips one leading slash and the handler trims
    // trailing ones, before a single `/` is prefixed back on.
    let stripped = decoded.strip_prefix('/').unwrap_or(&decoded);
    Some(format!("/{}", stripped.trim_end_matches('/')))
}

/// What the middleware should stamp, decided while the routers guard is held.
///
/// Deliberately small and owned: the allowlist itself never leaves the guard,
/// so a large one is scanned in place instead of being copied per request onto
/// a path an unauthenticated preflight can reach.
#[derive(Clone)]
enum CorsDecision {
    /// No allowlist applies, so the permissive default stands.
    Unrestricted,
    /// An allowlist applies. `allow_origin` is the value to echo, present only
    /// when the request's own `Origin` is on the list.
    Restricted { route_method: Option<HttpMethod>, allow_origin: Option<http::HeaderValue> },
    /// The routers could not be read, so nothing is known about this path.
    Unavailable,
}

/// Whether a route actually serves a static website, rather than merely saying
/// it does.
///
/// `is_static_website` is a caller-set flag that validation ties to nothing: a
/// route can carry it while having no assets configured and a `script_path`
/// that `route_job` runs regardless. Keying the exemption off the flag alone
/// would let one boolean disable a route's allowlist and hand its runnable back
/// the `wm_headers` escape hatch, so the assets have to be there too.
fn serves_a_static_website(trigger: &TriggerRoute) -> bool {
    trigger.is_static_website && trigger.static_asset_config.is_some()
}

/// A static website is never subject to an allowlist, its own included. It has
/// no authentication of its own — the editor does not offer any — so it hands
/// out public files that any non-browser client can already fetch, and
/// restricting which browsers may read them protects nothing while breaking the
/// cross-origin uses that do consult CORS: a webfont, a `crossorigin` asset, a
/// `fetch`.
///
/// A single-file static asset is not exempt. That one can carry an
/// `authentication_method`, so its content need not be public, and an allowlist
/// is what keeps another origin from reading a response its own credentials
/// would not have obtained.
///
/// The CORS verdict for a request, published by whoever resolved its trigger.
///
/// The middleware stamps headers after the handler returns, but only the
/// handler knows which trigger it actually served. Re-deriving that from the
/// routers cache is a second lookup which can disagree with the first when a
/// route is edited, deleted or widened mid-request, and every ordering of the
/// two is wrong in some case. So the verdict travels with the request instead
/// of being worked out twice.
#[derive(Clone, Default)]
struct ResolvedCorsPolicy(std::sync::Arc<std::sync::OnceLock<CorsDecision>>);

impl ResolvedCorsPolicy {
    /// Record what the trigger being served allows. Called once, where the
    /// route is resolved, so the answer cannot drift from the response.
    fn publish(&self, trigger: &TriggerRoute, method: Option<HttpMethod>, headers: &HeaderMap) {
        let decision = if serves_a_static_website(trigger) {
            CorsDecision::Unrestricted
        } else {
            let instance_default = HTTP_ROUTE_DEFAULT_ALLOWED_ORIGINS.load();
            match effective_allowed_origins(
                trigger.allowed_origins.as_deref(),
                instance_default.as_slice(),
            ) {
                None => CorsDecision::Unrestricted,
                Some(allowed_origins) => CorsDecision::Restricted {
                    route_method: method,
                    allow_origin: match_origin(allowed_origins, headers.get(http::header::ORIGIN)),
                },
            }
        };
        let _ = self.0.set(decision);
    }

    fn published(&self) -> Option<CorsDecision> {
        self.0.get().cloned()
    }
}

/// Decide the CORS answer from the routers cache, for a request no handler
/// published a verdict for: a preflight, an unknown path, or any failure ahead
/// of the publish — authentication included, which runs after the route itself
/// resolves.
///
/// Loads the routers when the cache is cold, the way `get_http_route_trigger`
/// does, so a preflight is answered from the same view of the routes as the
/// request that follows it.
async fn resolve_cors_decision(
    db: &DB,
    http_method: HttpMethod,
    requested_path: &str,
    origin: Option<&http::HeaderValue>,
) -> CorsDecision {
    let routers_cache = HTTP_ROUTERS_CACHE.read().await;

    let routers_cache = if routers_cache.routers.is_empty() {
        drop(routers_cache);
        match refresh_routers(db, false).await {
            Ok((_, routers_cache)) => routers_cache,
            Err(err) => {
                tracing::error!("Could not load HTTP routers to resolve CORS: {err:#}");
                return CorsDecision::Unavailable;
            }
        }
    } else {
        routers_cache
    };

    let Some(router) = routers_cache.routers.get(&http_method) else {
        return CorsDecision::Unavailable;
    };

    let route = router.at(requested_path).ok();
    if route
        .as_ref()
        .is_some_and(|trigger| serves_a_static_website(trigger.value))
    {
        return CorsDecision::Unrestricted;
    }
    let route_allowed_origins = route
        .as_ref()
        .and_then(|trigger| trigger.value.allowed_origins.as_deref());

    let instance_default = HTTP_ROUTE_DEFAULT_ALLOWED_ORIGINS.load();
    match effective_allowed_origins(route_allowed_origins, instance_default.as_slice()) {
        None => CorsDecision::Unrestricted,
        Some(allowed_origins) => CorsDecision::Restricted {
            route_method: route.map(|_| http_method),
            allow_origin: match_origin(allowed_origins, origin),
        },
    }
}

async fn conditional_cors_middleware(
    Extension(db): Extension<DB>,
    mut req: axum::extract::Request,
    next: axum::middleware::Next,
) -> Response {
    let origin = req.headers().get(http::header::ORIGIN).cloned();
    // Owned before `next.run` consumes the request. `&Request` is not `Send`
    // (`Body` is not `Sync`), so nothing borrowed from it can cross the await.
    // The URI is carried rather than the decoded path: cloning it is a refcount
    // bump, while decoding allocates, and only the fallback below ever needs it.
    let lookup_method = cors_lookup_method(&req);
    let uri = req.uri().clone();

    let resolved = ResolvedCorsPolicy::default();
    req.extensions_mut().insert(resolved.clone());

    let mut response = next.run(req).await;

    let decision = match resolved.published() {
        // The handler resolved a trigger and said what it served under. That is
        // the policy this response was produced with, so nothing else can be
        // more authoritative.
        Some(decision) => decision,
        // No verdict was published: a preflight, an unknown path, or a request
        // that failed before reaching the publish, authentication included. No
        // runnable produced this body, so reading the cache cannot contradict
        // anything.
        None => match lookup_method.zip(cors_lookup_path(uri.path())) {
            Some((method, path)) => {
                resolve_cors_decision(&db, method, &path, origin.as_ref()).await
            }
            // Not a preflight, not a routable method, or a path that does not
            // decode.
            None => CorsDecision::Unrestricted,
        },
    };

    let headers = response.headers_mut();

    // Check existing headers first to determine what not to insert
    let mut not_insert_origin = false;
    let mut not_insert_methods = false;
    let mut not_insert_headers = false;

    for key in headers.keys() {
        if !not_insert_origin && key == http::header::ACCESS_CONTROL_ALLOW_ORIGIN {
            not_insert_origin = true;
        }
        if !not_insert_methods && key == http::header::ACCESS_CONTROL_ALLOW_METHODS {
            not_insert_methods = true;
        }
        if !not_insert_headers && key == http::header::ACCESS_CONTROL_ALLOW_HEADERS {
            not_insert_headers = true;
        }

        // Early exit if all headers are already present
        if not_insert_origin && not_insert_methods && not_insert_headers {
            break;
        }
    }

    match &decision {
        CorsDecision::Restricted { allow_origin, .. } => {
            // A configured allowlist decides, overriding any `wm_headers` value
            // the runnable set. The preflight is answered before any code runs,
            // so config is the only thing it can consult; letting the response
            // widen what the preflight advertised would make the two disagree
            // and leave the allowlist bounding nothing. A route escapes a
            // stricter instance default — `wm_headers` included — by setting
            // its own list to `*`.
            match allow_origin {
                Some(value) => {
                    headers.insert(http::header::ACCESS_CONTROL_ALLOW_ORIGIN, value.clone())
                }
                // No match: omit the header entirely so the browser blocks the
                // read, and drop any value the runnable set.
                None => headers.remove(http::header::ACCESS_CONTROL_ALLOW_ORIGIN),
            };
            // Appended, not inserted: the answer now depends on the request's
            // Origin, and a shared cache that ignores it would hand one
            // origin's response to another.
            headers.append(http::header::VARY, http::HeaderValue::from_static("origin"));
        }
        // The routers could not be read, so nothing is known about this path;
        // only a preflight or an unresolved request reaches here. Answering a
        // preflight permissively would let a disallowed origin go on to invoke
        // a runnable whose purpose may be a side effect.
        CorsDecision::Unavailable => {
            headers.remove(http::header::ACCESS_CONTROL_ALLOW_ORIGIN);
        }
        CorsDecision::Unrestricted => {
            if !not_insert_origin {
                headers.insert(
                    http::header::ACCESS_CONTROL_ALLOW_ORIGIN,
                    http::HeaderValue::from_static("*"),
                );
            }
        }
    }

    if !not_insert_methods {
        // A route accepts exactly one method, so advertising all seven
        // overstates it. Only a route under an allowlist gets the narrower
        // answer; an unrestricted one advertises the full supported set, since
        // narrowing it would say something about a route the response is not
        // otherwise willing to disclose.
        let restricted_method = match &decision {
            CorsDecision::Restricted { route_method, .. } => *route_method,
            _ => None,
        };
        headers.insert(
            http::header::ACCESS_CONTROL_ALLOW_METHODS,
            http::HeaderValue::from_static(match restricted_method {
                Some(HttpMethod::Get) => "GET, OPTIONS",
                Some(HttpMethod::Post) => "POST, OPTIONS",
                Some(HttpMethod::Put) => "PUT, OPTIONS",
                Some(HttpMethod::Delete) => "DELETE, OPTIONS",
                Some(HttpMethod::Patch) => "PATCH, OPTIONS",
                None => "GET, POST, PUT, DELETE, PATCH, HEAD, OPTIONS",
            }),
        );
    }

    if !not_insert_headers {
        headers.insert(
            http::header::ACCESS_CONTROL_ALLOW_HEADERS,
            http::HeaderValue::from_static("content-type, authorization"),
        );
    }

    response
}

pub fn http_route_trigger_handler() -> Router {
    Router::new()
        .route(
            "/{*path}",
            get(route_job)
                .post(route_job)
                .delete(route_job)
                .put(route_job)
                .patch(route_job)
                .head(|| async { "" })
                .options(|| async { "" }),
        )
        .layer(axum::middleware::from_fn(conditional_cors_middleware))
}

async fn get_http_route_trigger(
    route_path: &str,
    auth_cache: &Arc<AuthCache>,
    token: Option<&String>,
    db: &DB,
    user_db: UserDB,
    method: &http::Method,
) -> Result<(TriggerRoute, String, HashMap<String, String>, ApiAuthed)> {
    let http_method: HttpMethod = method.try_into()?;

    let requested_path = format!("/{}", route_path);

    let routers_cache = HTTP_ROUTERS_CACHE.read().await;

    let routers_cache = if routers_cache.routers.is_empty() {
        tracing::info!("HTTP routers are not loaded, loading from db");
        // refresh_routers takes the write lock, so holding this read guard across it deadlocks.
        drop(routers_cache);
        let (_, routers_cache) = refresh_routers(db, false).await?;
        routers_cache
    } else {
        routers_cache
    };

    let router = routers_cache
        .routers
        .get(&http_method)
        .ok_or(Error::internal_err(
            "HTTP routers could not be loaded".to_string(),
        ))?;

    let trigger_match = router.at(requested_path.as_str()).ok();

    let matchit::Match { value: trigger, params } =
        not_found_if_none(trigger_match, "Trigger", requested_path.as_str())?;

    let params: HashMap<String, String> = params
        .iter()
        .map(|(k, v)| (k.to_string(), v.to_string()))
        .collect();

    let username_override = if let AuthenticationMethod::Windmill = trigger.authentication_method {
        let opt_authed = if let Some(token) = token {
            auth_cache
                .get_authed(Some(trigger.workspace_id.clone()), token)
                .await
        } else {
            None
        };
        if let Some(authed) = opt_authed {
            check_scopes(&authed, || format!("http_triggers:read:{}", &trigger.path))?;

            // check that the user has access to the trigger
            let cache_key = (
                trigger.workspace_id.clone(),
                trigger.path.clone(),
                authed.clone(),
            );
            let exists = match HTTP_ACCESS_CACHE.get(&cache_key) {
                Some(cache_entry) if cache_entry.expiry > std::time::Instant::now() => {
                    tracing::debug!("HTTP access cache hit for route {}", trigger.path);
                    true
                }
                _ => {
                    tracing::debug!("HTTP access cache miss for route {}", trigger.path);
                    let mut tx = user_db.begin(&authed).await?;
                    let exists = sqlx::query_scalar!(
                        r#"
                        SELECT EXISTS(
                            SELECT 1
                            FROM
                                http_trigger
                            WHERE
                                workspace_id = $1 AND
                                path = $2
                        )
                        "#,
                        trigger.workspace_id,
                        trigger.path
                    )
                    .fetch_one(&mut *tx)
                    .await?
                    .unwrap_or(false);
                    if exists {
                        HTTP_ACCESS_CACHE.insert(
                            cache_key,
                            ExpiringCacheEntry {
                                value: (),
                                expiry: std::time::Instant::now()
                                    + std::time::Duration::from_secs(10),
                            },
                        );
                    }
                    exists
                }
            };
            if exists {
                Some(authed.display_username().to_owned())
            } else {
                return Err(Error::NotAuthorized("Unauthorized".to_string()));
            }
        } else {
            return Err(Error::NotAuthorized("Requires authentication".to_string()));
        }
    } else {
        None
    };

    let email = windmill_common::users::get_email_from_permissioned_as(
        &trigger.permissioned_as,
        &trigger.workspace_id,
        db,
    )
    .await?;
    let authed = windmill_api_auth::fetch_api_authed_from_permissioned_as(
        trigger.permissioned_as.clone(),
        email,
        &trigger.workspace_id,
        &db,
        Some(username_override.unwrap_or(format!("HTTP-{}", trigger.path))),
    )
    .await?;

    Ok((trigger.clone(), route_path.to_string(), params, authed))
}

async fn route_job(
    Extension(db): Extension<DB>,
    Extension(user_db): Extension<UserDB>,
    Extension(auth_cache): Extension<Arc<AuthCache>>,
    Extension(cors_policy): Extension<ResolvedCorsPolicy>,
    OptTokened { token }: OptTokened,
    Path(route_path): Path<StripPath>,
    headers: HeaderMap,
    args: RawHttpTriggerArgs,
) -> std::result::Result<impl IntoResponse, Response> {
    let route_path = route_path.to_path().trim_end_matches("/");

    let (trigger, called_path, params, authed) = get_http_route_trigger(
        route_path,
        &auth_cache,
        token.as_ref(),
        &db,
        user_db.clone(),
        &args.0.metadata.method,
    )
    .await
    .map_err(|e| e.into_response())?;

    // Publish before anything else can fail: the CORS middleware stamps this
    // response either way, and it must reflect the trigger actually served.
    cors_policy.publish(&trigger, routable_method(&args.0.metadata.method), &headers);

    if trigger.script_path.is_empty() && trigger.static_asset_config.is_none() {
        return Err(Error::NotFound(format!(
            "Runnable path of HTTP route at path: {}",
            trigger.path
        ))
        .into_response());
    }

    let args = args
        .process_args(
            &authed,
            &db,
            &trigger.workspace_id,
            match trigger.authentication_method {
                AuthenticationMethod::CustomScript | AuthenticationMethod::Signature => true,
                _ => trigger.raw_string,
            },
        )
        .await
        .map_err(|e| e.into_response())?;

    match trigger.authentication_method {
        AuthenticationMethod::None
        | AuthenticationMethod::Windmill
        | AuthenticationMethod::CustomScript => {}
        _ => {
            let resource_path = match trigger.authentication_resource_path {
                Some(resource_path) => resource_path,
                None => {
                    return Err(Error::BadRequest(
                        "Missing authentication resource path".to_string(),
                    )
                    .into_response())
                }
            };

            let cache_key = (
                trigger.workspace_id.clone(),
                resource_path.clone(),
                authed.clone(),
            );

            let authentication_method = match HTTP_AUTH_CACHE.get(&cache_key) {
                Some(cache_entry) if cache_entry.expiry > std::time::Instant::now() => {
                    tracing::debug!("HTTP auth method cache hit for route {}", trigger.path);
                    cache_entry.value
                }
                _ => {
                    tracing::debug!("HTTP auth method cache miss for route {}", trigger.path);
                    let auth_method = try_get_resource_from_db_as::<
                        super::http_trigger_auth::AuthenticationMethod,
                    >(
                        &authed,
                        Some(user_db.clone()),
                        &db,
                        &resource_path,
                        &trigger.workspace_id,
                    )
                    .await
                    .map_err(|e| e.into_response())?;
                    HTTP_AUTH_CACHE.insert(
                        cache_key,
                        ExpiringCacheEntry {
                            value: auth_method.clone(),
                            expiry: std::time::Instant::now() + std::time::Duration::from_secs(60),
                        },
                    );
                    auth_method
                }
            };

            let raw_payload = args.0.metadata.raw_string.as_ref();

            let response = authentication_method
                .authenticate_http_request(&headers, raw_payload)
                .map_err(|e| e.into_response())?;

            if let Some(response) = response {
                return Ok(response);
            }
        }
    }

    #[cfg(not(feature = "parquet"))]
    if trigger.static_asset_config.is_some() {
        return Err(Error::internal_err(
            "Static asset configuration is not supported in this build".to_string(),
        )
        .into_response());
    }

    #[cfg(feature = "parquet")]
    if let Some(sqlx::types::Json(config)) = trigger.static_asset_config {
        let build_static_response_f = async {
            use windmill_object_store::object_store_reexports::ObjectStoreExt;

            let (_, s3_resource_opt) = get_workspace_s3_resource(
                &authed,
                &db,
                None,
                &trigger.workspace_id,
                config.storage.clone(),
            )
            .await?;
            let s3_resource = s3_resource_opt.ok_or_else(|| {
                windmill_object_store::workspace_storage_misconfigured(config.storage.as_deref())
            })?;
            let s3_client = build_object_store_client(&s3_resource).await?;

            let path = if trigger.is_static_website {
                let subpath = params
                    .get("wm_subpath")
                    .cloned()
                    .unwrap_or("index.html".to_string());
                tracing::info!("subpath: {}", subpath);
                format!("{}/{}", config.s3.trim_end_matches('/'), subpath)
            } else {
                config.s3.clone()
            };
            let path = windmill_object_store::object_store_reexports::Path::from(path);
            let s3_object = s3_client.get(&path).await;

            let s3_object = match s3_object {
                Err(
                    windmill_object_store::object_store_reexports::ObjectStoreError::NotFound {
                        ..
                    },
                ) if trigger.is_static_website => {
                    // fallback to index.html if the file is not found
                    let path = windmill_object_store::object_store_reexports::Path::from(format!(
                        "{}/index.html",
                        config.s3.trim_end_matches('/')
                    ));
                    s3_client.get(&path).await
                }
                r => r,
            };

            let s3_object = s3_object.map_err(|err| {
                tracing::warn!("Error retrieving file from S3: {:?}", err);
                let mut msg = format!("Error retrieving file: {err}");
                let mut source = std::error::Error::source(&err);
                while let Some(e) = source {
                    msg.push_str(&format!("\n  caused by: {e}"));
                    source = e.source();
                }
                Error::internal_err(msg)
            })?;

            let mut response_headers = http::HeaderMap::new();
            if let Some(ref e_tag) = s3_object.meta.e_tag {
                if let Some(if_none_match) = headers.get(http::header::IF_NONE_MATCH) {
                    if if_none_match == e_tag {
                        return Ok::<_, Error>((
                            StatusCode::NOT_MODIFIED,
                            response_headers,
                            axum::body::Body::empty(),
                        ));
                    }
                }
                if let Ok(e_tag) = e_tag.parse() {
                    response_headers.insert("etag", e_tag);
                }
            }
            response_headers.insert(
                "content-type",
                s3_object
                    .attributes
                    .get(&windmill_object_store::object_store_reexports::Attribute::ContentType)
                    .map(|s| s.parse().ok())
                    .flatten()
                    .unwrap_or("application/octet-stream".parse().unwrap()),
            );
            response_headers.insert("x-content-type-options", "nosniff".parse().unwrap());
            if !trigger.is_static_website {
                response_headers.insert(
                    "content-disposition",
                    config.filename.as_ref().map_or_else(
                        || {
                            s3_object
                                .attributes
                                .get(&windmill_object_store::object_store_reexports::Attribute::ContentDisposition)
                                .map(|s| s.parse().ok())
                                .flatten()
                                .unwrap_or("inline".parse().unwrap())
                        },
                        |filename| {
                            format!("inline; filename=\"{}\"", filename)
                                .parse()
                                .unwrap_or("inline".parse().unwrap())
                        },
                    ),
                );
                // For single-file triggers, sandbox any HTML/SVG so it can't
                // reach the viewer's session cookie. Allow-scripts/forms/etc.
                // keep the opaque origin (cookies still blocked) while
                // preserving JS for legitimate HTML payloads. Static-website
                // triggers intentionally serve a live web app and cannot be
                // sandboxed; restrict write access to those buckets at the
                // workspace level.
                response_headers.insert(
                    "content-security-policy",
                    "sandbox allow-scripts allow-forms allow-popups allow-modals allow-downloads"
                        .parse()
                        .unwrap(),
                );
            }

            let body_stream = axum::body::Body::from_stream(s3_object.into_stream());
            Ok::<_, Error>((StatusCode::OK, response_headers, body_stream))
        };
        match build_static_response_f.await {
            Ok((status, headers, body_stream)) => {
                return Ok((status, headers, body_stream).into_response())
            }
            Err(e) => return Err(e.into_response()),
        }
    }

    let runnable_format = get_runnable_format(
        if trigger.is_flow {
            RunnableId::from_flow_path(&trigger.script_path)
        } else {
            RunnableId::from_script_path(&trigger.script_path)
        },
        &trigger.workspace_id,
        &db,
        &TriggerKind::Http,
    )
    .await
    .map_err(|e| e.into_response())?;

    let args = args
        .to_args_from_format(
            &trigger.route_path,
            &called_path,
            &trigger.path,
            &params,
            runnable_format,
            trigger.wrap_body,
        )
        .map_err(|e| e.into_response())?;

    let trigger_info = TriggerMetadata::new(Some(trigger.path.clone()), JobTriggerKind::Http);
    if trigger.mode == TriggerMode::Suspended {
        let _ = trigger_runnable(
            &db,
            Some(user_db),
            authed,
            &trigger.workspace_id,
            &trigger.script_path,
            trigger.is_flow,
            args,
            trigger.retry.as_ref(),
            trigger.error_handler_path.as_deref(),
            trigger.error_handler_args.as_ref(),
            format!("http_trigger/{}", trigger.path),
            None,
            true,
            trigger_info,
        )
        .await
        .map_err(|e| e.into_response())?;

        return Ok((
            StatusCode::OK,
            format!(
                "Trigger {} is in suspended mode, jobs are added to the queue but suspended",
                &trigger.path
            ),
        )
            .into_response());
    }

    // Handle execution based on the execution mode
    match trigger.request_type {
        RequestType::SyncSse => {
            // Trigger the job (always async when streaming)
            let (uuid, _, early_return, has_failure_module, _) = trigger_runnable_inner(
                &db,
                None,
                Some(user_db.clone()),
                authed.clone(),
                &trigger.workspace_id,
                &trigger.script_path,
                trigger.is_flow,
                args,
                trigger.retry.as_ref(),
                trigger.error_handler_path.as_deref(),
                trigger.error_handler_args.as_ref(),
                format!("http_trigger/{}", trigger.path),
                None,
                trigger_info,
                None,
            )
            .await
            .map_err(|e| e.into_response())?;

            // Set up SSE stream
            let opt_authed = Some(authed.clone());
            let opt_tokened = OptTokened { token: None };
            let (tx, rx) = tokio::sync::mpsc::channel(32);

            let stream = tokio_stream::wrappers::ReceiverStream::new(rx).map(|x| {
                format!(
                    "data: {}\n\n",
                    serde_json::to_string(&x).unwrap_or_default()
                )
            });

            start_job_update_sse_stream(
                opt_authed,
                opt_tokened,
                db.clone(),
                trigger.workspace_id.clone(),
                uuid,
                None,
                None,
                None,
                None,
                Some(true),
                Some(true),
                None,
                None,
                tx,
                None,
                early_return,
                has_failure_module,
                false,
            );

            let body = axum::body::Body::from_stream(
                stream.map(std::result::Result::<_, std::convert::Infallible>::Ok),
            );

            Ok(Response::builder()
                .status(200)
                .header("Content-Type", "text/event-stream")
                .header("Cache-Control", "no-cache")
                .body(body)
                .map_err(|e| Error::internal_err(e.to_string()).into_response())?)
        }
        RequestType::Async => trigger_runnable(
            &db,
            Some(user_db),
            authed,
            &trigger.workspace_id,
            &trigger.script_path,
            trigger.is_flow,
            args,
            trigger.retry.as_ref(),
            trigger.error_handler_path.as_deref(),
            trigger.error_handler_args.as_ref(),
            format!("http_trigger/{}", trigger.path),
            None,
            false,
            trigger_info,
        )
        .await
        .map_err(|e| e.into_response()),
        RequestType::Sync => trigger_runnable_and_wait_for_result(
            &db,
            Some(user_db),
            authed,
            &trigger.workspace_id,
            &trigger.script_path,
            trigger.is_flow,
            args,
            trigger.retry.as_ref(),
            trigger.error_handler_path.as_deref(),
            trigger.error_handler_args.as_ref(),
            format!("http_trigger/{}", trigger.path),
            trigger_info,
        )
        .await
        .map_err(|e| e.into_response()),
    }
}