diff --git a/backend/.sqlx/query-4410b2d6e52556569cda9d945f791756a80ff2835d74bfab85f53141b6e32351.json b/backend/.sqlx/query-1301f873a829db137573b8b39449f6160f2adf44f864f26a99b8eab5818fbd50.json similarity index 66% rename from backend/.sqlx/query-4410b2d6e52556569cda9d945f791756a80ff2835d74bfab85f53141b6e32351.json rename to backend/.sqlx/query-1301f873a829db137573b8b39449f6160f2adf44f864f26a99b8eab5818fbd50.json index 63c326f5e2..2999ae4bf2 100644 --- a/backend/.sqlx/query-4410b2d6e52556569cda9d945f791756a80ff2835d74bfab85f53141b6e32351.json +++ b/backend/.sqlx/query-1301f873a829db137573b8b39449f6160f2adf44f864f26a99b8eab5818fbd50.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT \n path, \n script_path, \n is_flow, \n route_path, \n authentication_resource_path,\n workspace_id, \n is_async, \n authentication_method AS \"authentication_method: _\", \n edited_by, \n email, \n static_asset_config AS \"static_asset_config: _\",\n wrap_body,\n raw_string,\n workspaced_route,\n is_static_website,\n error_handler_path,\n error_handler_args as \"error_handler_args: _\",\n retry as \"retry: _\"\n FROM \n http_trigger \n WHERE \n http_method = $1\n ", + "query": "\n SELECT\n path,\n script_path,\n is_flow,\n route_path,\n authentication_resource_path,\n workspace_id,\n request_type AS \"request_type: _\",\n authentication_method AS \"authentication_method: _\",\n edited_by,\n email,\n static_asset_config AS \"static_asset_config: _\",\n wrap_body,\n raw_string,\n workspaced_route,\n is_static_website,\n error_handler_path,\n error_handler_args as \"error_handler_args: _\",\n retry as \"retry: _\"\n FROM\n http_trigger\n WHERE\n http_method = $1\n ", "describe": { "columns": [ { @@ -35,8 +35,19 @@ }, { "ordinal": 6, - "name": "is_async", - "type_info": "Bool" + "name": "request_type: _", + "type_info": { + "Custom": { + "name": "request_type", + "kind": { + "Enum": [ + "sync", + "async", + "sync_sse" + ] + } + } + } }, { "ordinal": 7, @@ -147,5 +158,5 @@ true ] }, - "hash": "4410b2d6e52556569cda9d945f791756a80ff2835d74bfab85f53141b6e32351" + "hash": "1301f873a829db137573b8b39449f6160f2adf44f864f26a99b8eab5818fbd50" } diff --git a/backend/.sqlx/query-1f6b773ce34fe51d03d6f9a2345481629c62453eebbb08f82dd2da23389bc117.json b/backend/.sqlx/query-1f6b773ce34fe51d03d6f9a2345481629c62453eebbb08f82dd2da23389bc117.json new file mode 100644 index 0000000000..f69ec4f2a6 --- /dev/null +++ b/backend/.sqlx/query-1f6b773ce34fe51d03d6f9a2345481629c62453eebbb08f82dd2da23389bc117.json @@ -0,0 +1,73 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO http_trigger (\n workspace_id,\n path,\n route_path,\n route_path_key,\n workspaced_route,\n authentication_resource_path,\n wrap_body,\n raw_string,\n script_path,\n summary,\n description,\n is_flow,\n request_type,\n authentication_method,\n http_method,\n static_asset_config,\n edited_by,\n email,\n edited_at,\n is_static_website,\n error_handler_path,\n error_handler_args,\n retry\n )\n VALUES (\n $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, now(), $19, $20, $21, $22\n )\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Varchar", + "Varchar", + "Bool", + "Varchar", + "Bool", + "Bool", + "Varchar", + "Varchar", + "Text", + "Bool", + { + "Custom": { + "name": "request_type", + "kind": { + "Enum": [ + "sync", + "async", + "sync_sse" + ] + } + } + }, + { + "Custom": { + "name": "authentication_method", + "kind": { + "Enum": [ + "none", + "windmill", + "api_key", + "basic_http", + "custom_script", + "signature" + ] + } + } + }, + { + "Custom": { + "name": "http_method", + "kind": { + "Enum": [ + "get", + "post", + "put", + "delete", + "patch" + ] + } + } + }, + "Jsonb", + "Varchar", + "Varchar", + "Bool", + "Varchar", + "Jsonb", + "Jsonb" + ] + }, + "nullable": [] + }, + "hash": "1f6b773ce34fe51d03d6f9a2345481629c62453eebbb08f82dd2da23389bc117" +} diff --git a/backend/.sqlx/query-465144ea7e2930203618d9814a3e20c77b4363cf9e7c655d395f3fe40c247f61.json b/backend/.sqlx/query-465144ea7e2930203618d9814a3e20c77b4363cf9e7c655d395f3fe40c247f61.json new file mode 100644 index 0000000000..7869fe789e --- /dev/null +++ b/backend/.sqlx/query-465144ea7e2930203618d9814a3e20c77b4363cf9e7c655d395f3fe40c247f61.json @@ -0,0 +1,74 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE\n http_trigger\n SET\n route_path = $1,\n route_path_key = $2,\n workspaced_route = $3,\n wrap_body = $4,\n raw_string = $5,\n authentication_resource_path = $6,\n script_path = $7,\n path = $8,\n is_flow = $9,\n http_method = $10,\n static_asset_config = $11,\n edited_by = $12,\n email = $13,\n request_type = $14,\n authentication_method = $15,\n summary = $16,\n description = $17,\n edited_at = now(),\n is_static_website = $18,\n error_handler_path = $19,\n error_handler_args = $20,\n retry = $21\n WHERE\n workspace_id = $22 AND\n path = $23\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Bool", + "Bool", + "Bool", + "Varchar", + "Varchar", + "Varchar", + "Bool", + { + "Custom": { + "name": "http_method", + "kind": { + "Enum": [ + "get", + "post", + "put", + "delete", + "patch" + ] + } + } + }, + "Jsonb", + "Varchar", + "Varchar", + { + "Custom": { + "name": "request_type", + "kind": { + "Enum": [ + "sync", + "async", + "sync_sse" + ] + } + } + }, + { + "Custom": { + "name": "authentication_method", + "kind": { + "Enum": [ + "none", + "windmill", + "api_key", + "basic_http", + "custom_script", + "signature" + ] + } + } + }, + "Varchar", + "Text", + "Bool", + "Varchar", + "Jsonb", + "Jsonb", + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "465144ea7e2930203618d9814a3e20c77b4363cf9e7c655d395f3fe40c247f61" +} diff --git a/backend/.sqlx/query-4c64bd0e364f536597db83161b5a27ff58b5ec7148bc94807423593bcfa27b90.json b/backend/.sqlx/query-4c64bd0e364f536597db83161b5a27ff58b5ec7148bc94807423593bcfa27b90.json new file mode 100644 index 0000000000..dfe905113e --- /dev/null +++ b/backend/.sqlx/query-4c64bd0e364f536597db83161b5a27ff58b5ec7148bc94807423593bcfa27b90.json @@ -0,0 +1,71 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE\n http_trigger\n SET\n wrap_body = $1,\n raw_string = $2,\n authentication_resource_path = $3,\n script_path = $4,\n path = $5,\n is_flow = $6,\n http_method = $7,\n static_asset_config = $8,\n edited_by = $9,\n email = $10,\n request_type = $11,\n authentication_method = $12,\n summary = $13,\n description = $14,\n edited_at = now(),\n is_static_website = $15,\n error_handler_path = $16,\n error_handler_args = $17,\n retry = $18\n WHERE\n workspace_id = $19 AND\n path = $20\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Bool", + "Bool", + "Varchar", + "Varchar", + "Varchar", + "Bool", + { + "Custom": { + "name": "http_method", + "kind": { + "Enum": [ + "get", + "post", + "put", + "delete", + "patch" + ] + } + } + }, + "Jsonb", + "Varchar", + "Varchar", + { + "Custom": { + "name": "request_type", + "kind": { + "Enum": [ + "sync", + "async", + "sync_sse" + ] + } + } + }, + { + "Custom": { + "name": "authentication_method", + "kind": { + "Enum": [ + "none", + "windmill", + "api_key", + "basic_http", + "custom_script", + "signature" + ] + } + } + }, + "Varchar", + "Text", + "Bool", + "Varchar", + "Jsonb", + "Jsonb", + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "4c64bd0e364f536597db83161b5a27ff58b5ec7148bc94807423593bcfa27b90" +} diff --git a/backend/.sqlx/query-5cc0d9e3dcd9c20e6e6ec1acf38c8f97b5ece60bfe24fc3783a83de47e3aa583.json b/backend/.sqlx/query-5cc0d9e3dcd9c20e6e6ec1acf38c8f97b5ece60bfe24fc3783a83de47e3aa583.json deleted file mode 100644 index fc2540e85d..0000000000 --- a/backend/.sqlx/query-5cc0d9e3dcd9c20e6e6ec1acf38c8f97b5ece60bfe24fc3783a83de47e3aa583.json +++ /dev/null @@ -1,62 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n INSERT INTO http_trigger (\n workspace_id, \n path, \n route_path, \n route_path_key,\n workspaced_route,\n authentication_resource_path,\n wrap_body,\n raw_string,\n script_path, \n summary,\n description,\n is_flow, \n is_async, \n authentication_method, \n http_method, \n static_asset_config, \n edited_by, \n email, \n edited_at, \n is_static_website,\n error_handler_path,\n error_handler_args,\n retry\n ) \n VALUES (\n $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, now(), $19, $20, $21, $22\n )\n ", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Varchar", - "Varchar", - "Bool", - "Varchar", - "Bool", - "Bool", - "Varchar", - "Varchar", - "Text", - "Bool", - "Bool", - { - "Custom": { - "name": "authentication_method", - "kind": { - "Enum": [ - "none", - "windmill", - "api_key", - "basic_http", - "custom_script", - "signature" - ] - } - } - }, - { - "Custom": { - "name": "http_method", - "kind": { - "Enum": [ - "get", - "post", - "put", - "delete", - "patch" - ] - } - } - }, - "Jsonb", - "Varchar", - "Varchar", - "Bool", - "Varchar", - "Jsonb", - "Jsonb" - ] - }, - "nullable": [] - }, - "hash": "5cc0d9e3dcd9c20e6e6ec1acf38c8f97b5ece60bfe24fc3783a83de47e3aa583" -} diff --git a/backend/.sqlx/query-85d945cf5ade707291a161078ff96ddb29140dd72ce5115657418eec503b205d.json b/backend/.sqlx/query-85d945cf5ade707291a161078ff96ddb29140dd72ce5115657418eec503b205d.json deleted file mode 100644 index fae73f363c..0000000000 --- a/backend/.sqlx/query-85d945cf5ade707291a161078ff96ddb29140dd72ce5115657418eec503b205d.json +++ /dev/null @@ -1,63 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n UPDATE \n http_trigger \n SET \n route_path = $1, \n route_path_key = $2,\n workspaced_route = $3,\n wrap_body = $4,\n raw_string = $5,\n authentication_resource_path = $6,\n script_path = $7, \n path = $8, \n is_flow = $9, \n http_method = $10, \n static_asset_config = $11, \n edited_by = $12, \n email = $13, \n is_async = $14, \n authentication_method = $15, \n summary = $16,\n description = $17,\n edited_at = now(), \n is_static_website = $18,\n error_handler_path = $19,\n error_handler_args = $20,\n retry = $21\n WHERE \n workspace_id = $22 AND \n path = $23\n ", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Bool", - "Bool", - "Bool", - "Varchar", - "Varchar", - "Varchar", - "Bool", - { - "Custom": { - "name": "http_method", - "kind": { - "Enum": [ - "get", - "post", - "put", - "delete", - "patch" - ] - } - } - }, - "Jsonb", - "Varchar", - "Varchar", - "Bool", - { - "Custom": { - "name": "authentication_method", - "kind": { - "Enum": [ - "none", - "windmill", - "api_key", - "basic_http", - "custom_script", - "signature" - ] - } - } - }, - "Varchar", - "Text", - "Bool", - "Varchar", - "Jsonb", - "Jsonb", - "Text", - "Text" - ] - }, - "nullable": [] - }, - "hash": "85d945cf5ade707291a161078ff96ddb29140dd72ce5115657418eec503b205d" -} diff --git a/backend/.sqlx/query-714fb0f66ceb536aee8cb9ae0144757b999d25870fda37fe904e09dd5c742015.json b/backend/.sqlx/query-9360d00990822f153ff09c7905ae3180f07d02f38ac12d07a5664d93f160e7ee.json similarity index 68% rename from backend/.sqlx/query-714fb0f66ceb536aee8cb9ae0144757b999d25870fda37fe904e09dd5c742015.json rename to backend/.sqlx/query-9360d00990822f153ff09c7905ae3180f07d02f38ac12d07a5664d93f160e7ee.json index 9f3f4b0c8c..c702405099 100644 --- a/backend/.sqlx/query-714fb0f66ceb536aee8cb9ae0144757b999d25870fda37fe904e09dd5c742015.json +++ b/backend/.sqlx/query-9360d00990822f153ff09c7905ae3180f07d02f38ac12d07a5664d93f160e7ee.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT\n route_path,\n http_method AS \"http_method: _\",\n is_async,\n workspaced_route,\n summary,\n description,\n authentication_method AS \"authentication_method: _\",\n authentication_resource_path\n FROM\n http_trigger\n WHERE\n path ~ ANY($1) AND\n route_path ~ ANY($2) AND\n workspace_id = $3\n ", + "query": "\n SELECT\n route_path,\n http_method AS \"http_method: _\",\n request_type AS \"request_type: _\",\n workspaced_route,\n summary,\n description,\n authentication_method AS \"authentication_method: _\",\n authentication_resource_path\n FROM\n http_trigger\n WHERE\n path ~ ANY($1) AND\n route_path ~ ANY($2) AND\n workspace_id = $3\n ", "describe": { "columns": [ { @@ -28,8 +28,19 @@ }, { "ordinal": 2, - "name": "is_async", - "type_info": "Bool" + "name": "request_type: _", + "type_info": { + "Custom": { + "name": "request_type", + "kind": { + "Enum": [ + "sync", + "async", + "sync_sse" + ] + } + } + } }, { "ordinal": 3, @@ -89,5 +100,5 @@ true ] }, - "hash": "714fb0f66ceb536aee8cb9ae0144757b999d25870fda37fe904e09dd5c742015" + "hash": "9360d00990822f153ff09c7905ae3180f07d02f38ac12d07a5664d93f160e7ee" } diff --git a/backend/.sqlx/query-a7a23229d6915d7fdeea8073d31be6e7d9f7a8581bdbb5914f2c5b49f37dbc36.json b/backend/.sqlx/query-a7a23229d6915d7fdeea8073d31be6e7d9f7a8581bdbb5914f2c5b49f37dbc36.json deleted file mode 100644 index b34c87ca4c..0000000000 --- a/backend/.sqlx/query-a7a23229d6915d7fdeea8073d31be6e7d9f7a8581bdbb5914f2c5b49f37dbc36.json +++ /dev/null @@ -1,60 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n UPDATE \n http_trigger \n SET \n wrap_body = $1,\n raw_string = $2,\n authentication_resource_path = $3,\n script_path = $4, \n path = $5, \n is_flow = $6, \n http_method = $7, \n static_asset_config = $8, \n edited_by = $9, \n email = $10, \n is_async = $11, \n authentication_method = $12, \n summary = $13,\n description = $14,\n edited_at = now(), \n is_static_website = $15,\n error_handler_path = $16,\n error_handler_args = $17,\n retry = $18\n WHERE \n workspace_id = $19 AND \n path = $20\n ", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Bool", - "Bool", - "Varchar", - "Varchar", - "Varchar", - "Bool", - { - "Custom": { - "name": "http_method", - "kind": { - "Enum": [ - "get", - "post", - "put", - "delete", - "patch" - ] - } - } - }, - "Jsonb", - "Varchar", - "Varchar", - "Bool", - { - "Custom": { - "name": "authentication_method", - "kind": { - "Enum": [ - "none", - "windmill", - "api_key", - "basic_http", - "custom_script", - "signature" - ] - } - } - }, - "Varchar", - "Text", - "Bool", - "Varchar", - "Jsonb", - "Jsonb", - "Text", - "Text" - ] - }, - "nullable": [] - }, - "hash": "a7a23229d6915d7fdeea8073d31be6e7d9f7a8581bdbb5914f2c5b49f37dbc36" -} diff --git a/backend/migrations/20251016163921_http_trigger_request_type.down.sql b/backend/migrations/20251016163921_http_trigger_request_type.down.sql new file mode 100644 index 0000000000..a924e23147 --- /dev/null +++ b/backend/migrations/20251016163921_http_trigger_request_type.down.sql @@ -0,0 +1,19 @@ +-- Add down migration script here +-- Add back the is_async column +ALTER TABLE http_trigger ADD COLUMN is_async BOOLEAN; + +-- Migrate request_type values back to is_async +-- 'async' -> TRUE +-- 'sync' or 'sync_sse' -> FALSE +UPDATE http_trigger SET is_async = CASE + WHEN request_type = 'async'::REQUEST_TYPE THEN TRUE + ELSE FALSE +END; + +-- Make is_async NOT NULL with default +ALTER TABLE http_trigger ALTER COLUMN is_async SET NOT NULL; +ALTER TABLE http_trigger ALTER COLUMN is_async SET DEFAULT FALSE; + +-- Drop the request_type column and type +ALTER TABLE http_trigger DROP COLUMN request_type; +DROP TYPE REQUEST_TYPE; diff --git a/backend/migrations/20251016163921_http_trigger_request_type.up.sql b/backend/migrations/20251016163921_http_trigger_request_type.up.sql new file mode 100644 index 0000000000..980f39ee24 --- /dev/null +++ b/backend/migrations/20251016163921_http_trigger_request_type.up.sql @@ -0,0 +1,23 @@ +-- Add up migration script here +-- Create the request_type enum type +CREATE TYPE REQUEST_TYPE AS ENUM ('sync', 'async', 'sync_sse'); + +-- Add the new request_type column with a default value +ALTER TABLE http_trigger ADD COLUMN request_type REQUEST_TYPE; + +-- Migrate existing is_async values to request_type +-- is_async = FALSE -> 'sync' +-- is_async = TRUE -> 'async' +UPDATE http_trigger SET request_type = CASE + WHEN is_async = TRUE THEN 'async'::REQUEST_TYPE + ELSE 'sync'::REQUEST_TYPE +END; + +-- Make request_type NOT NULL now that all values are populated +ALTER TABLE http_trigger ALTER COLUMN request_type SET NOT NULL; + +-- Set default for new rows +ALTER TABLE http_trigger ALTER COLUMN request_type SET DEFAULT 'sync'::REQUEST_TYPE; + +-- Drop the old is_async column +ALTER TABLE http_trigger DROP COLUMN is_async; diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index d1c1a6e5bc..935788245c 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -16894,6 +16894,13 @@ components: - delete - patch + HttpRequestType: + type: string + enum: + - sync + - async + - sync_sse + HttpTrigger: allOf: - $ref: "#/components/schemas/TriggerExtraProperty" @@ -16920,8 +16927,8 @@ components: type: string description: type: string - is_async: - type: boolean + request_type: + $ref: "#/components/schemas/HttpRequestType" authentication_method: $ref: "#/components/schemas/AuthenticationMethod" is_static_website: @@ -16941,7 +16948,7 @@ components: required: - route_path - - is_async + - request_type - authentication_method - http_method - is_static_website @@ -16983,6 +16990,9 @@ components: type: string is_async: type: boolean + description: Deprecated, use request_type instead + request_type: + $ref: "#/components/schemas/HttpRequestType" authentication_method: $ref: "#/components/schemas/AuthenticationMethod" is_static_website: @@ -17003,7 +17013,6 @@ components: - script_path - route_path - is_flow - - is_async - authentication_method - http_method - is_static_website @@ -17042,6 +17051,9 @@ components: $ref: "#/components/schemas/HttpMethod" is_async: type: boolean + description: Deprecated, use request_type instead + request_type: + $ref: "#/components/schemas/HttpRequestType" authentication_method: $ref: "#/components/schemas/AuthenticationMethod" is_static_website: @@ -17061,7 +17073,6 @@ components: - script_path - is_flow - kind - - is_async - authentication_method - http_method - is_static_website diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index e54a61df9f..0ff66f67d4 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -6896,7 +6896,7 @@ async fn get_job_update_sse( #[derive(Serialize)] #[serde(tag = "type", rename_all = "lowercase")] -enum JobUpdateSSEStream { +pub enum JobUpdateSSEStream { Update(JobUpdate), Error { error: String }, NotFound, @@ -6909,8 +6909,7 @@ lazy_static::lazy_static! { std::env::var("TIMEOUT_SSE_STREAM").unwrap_or("60".to_string()).parse::().unwrap_or(60); } - -fn start_job_update_sse_stream( +pub fn start_job_update_sse_stream( opt_authed: Option, opt_tokened: OptTokened, db: DB, diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index c7cfcf070d..587c566522 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -101,7 +101,8 @@ mod inkeep_oss; mod inputs; mod integration; mod live_migrations; -pub mod openapi; +#[cfg(feature = "http_trigger")] +mod openapi; #[cfg(all(feature = "private", feature = "parquet"))] pub mod s3_proxy_ee; mod s3_proxy_oss; @@ -475,7 +476,17 @@ pub async fn run_server( .nest("/variables", variables::workspaced_service()) .nest("/workspaces", workspaces::workspaced_service()) .nest("/oidc", oidc_oss::workspaced_service()) - .nest("/openapi", openapi::openapi_service()) + .nest("/openapi", { + #[cfg(feature = "http_trigger")] + { + openapi::openapi_service() + } + + #[cfg(not(feature = "http_trigger"))] + { + Router::new() + } + }) .merge(triggers_service), ) .nest("/workspaces", workspaces::global_service()) diff --git a/backend/windmill-api/src/openapi.rs b/backend/windmill-api/src/openapi.rs index 6cc950523a..04d17a6127 100644 --- a/backend/windmill-api/src/openapi.rs +++ b/backend/windmill-api/src/openapi.rs @@ -9,6 +9,7 @@ use axum::{ }; use http::{header, HeaderValue, Method, StatusCode}; use indexmap::IndexMap; +use itertools::Itertools; use serde::{Deserialize, Serialize}; use serde_json::{to_value, Map, Value}; use sqlx::PgConnection; @@ -20,18 +21,13 @@ use windmill_common::{ DB, }; -use crate::db::ApiAuthed; - -#[cfg(feature = "http_trigger")] -use { - crate::{ - resources::try_get_resource_from_db_as, - triggers::http::{ - http_trigger_args::HttpMethod, http_trigger_auth::ApiKeyAuthentication, - AuthenticationMethod, - }, +use crate::{ + db::ApiAuthed, + resources::try_get_resource_from_db_as, + triggers::http::{ + http_trigger_args::HttpMethod, http_trigger_auth::ApiKeyAuthentication, + AuthenticationMethod, RequestType, }, - itertools::Itertools, }; lazy_static::lazy_static! { @@ -49,6 +45,7 @@ const BASIC_HTTP_AUTH_SCHEME: &'static str = "BasicHttp"; const DEFAULT_REQUEST_KEY: &'static str = "defaultRequest"; const DEFAULT_ASYNC_RESPONSE_KEY: &'static str = "AsyncResponse"; const DEFAULT_SYNC_RESPONSE_KEY: &'static str = "SyncResponse"; +const DEFAULT_SYNC_SSE_RESPONSE_KEY: &'static str = "SyncSseResponse"; const DEFAULT_PAYLOAD_PARAM_KEY: &'static str = "PayloadParam"; pub fn openapi_service() -> Router { @@ -166,7 +163,7 @@ pub enum Kind { pub struct FuturePath { route_path: String, kind: Kind, - is_async: Option, + request_type: Option, summary: Option, description: Option, security_scheme: Option, @@ -176,12 +173,12 @@ impl FuturePath { pub fn new( route_path: String, kind: Kind, - is_async: Option, + request_type: Option, summary: Option, description: Option, security_scheme: Option, ) -> FuturePath { - FuturePath { route_path, kind, is_async, summary, description, security_scheme } + FuturePath { route_path, kind, request_type, summary, description, security_scheme } } } @@ -241,6 +238,7 @@ fn from_route_path_to_openapi_path( vec![ format!("/run/{}", &normalized_path), format!("/run_wait_result/{}", &normalized_path), + format!("/run_and_stream/{}", &normalized_path), ] }; @@ -282,19 +280,23 @@ fn generate_paths( }) }; - let generate_response = |is_async: bool| { - let responses = if is_async { - serde_json::json!({ + let generate_response = |request_type: RequestType| { + let responses = match request_type { + RequestType::Async => serde_json::json!({ "200": { "$ref": format!("#/components/responses/{DEFAULT_ASYNC_RESPONSE_KEY}") } - }) - } else { - serde_json::json!(serde_json::json!({ + }), + RequestType::Sync => serde_json::json!({ "200": { "$ref": format!("#/components/responses/{DEFAULT_SYNC_RESPONSE_KEY}") } - })) + }), + RequestType::SyncSse => serde_json::json!({ + "200": { + "$ref": format!("#/components/responses/{DEFAULT_SYNC_SSE_RESPONSE_KEY}") + } + }), }; responses @@ -347,12 +349,19 @@ fn generate_paths( path_object }); - let is_async; + let request_type; let (methods, is_webhook) = match &path.kind { Kind::Webhook(_) => { - is_async = route_path.starts_with("/run/"); - let methods = if is_async { + request_type = if route_path.starts_with("/run/") { + RequestType::Async + } else if route_path.starts_with("/run_and_stream/") { + RequestType::SyncSse + } else { + RequestType::Sync + }; + + let methods = if request_type == RequestType::Async { vec![Method::POST] } else { vec![Method::GET, Method::POST] @@ -369,7 +378,7 @@ fn generate_paths( ) .into()); } - is_async = path.is_async.unwrap_or(true); + request_type = path.request_type.unwrap_or(RequestType::Sync); (vec![method.to_owned()], false) } }; @@ -401,7 +410,7 @@ fn generate_paths( ); } - method_map.insert("responses", generate_response(is_async)); + method_map.insert("responses", generate_response(request_type)); path_object.insert(method.to_string().to_lowercase(), to_value(&method_map)?); } @@ -428,19 +437,6 @@ pub fn transform_to_minified_postgres_regex(glob: &str) -> String { regex } -#[derive(Debug, Default)] -pub struct ServerToSet { - pub http_route: bool, - pub webhook_flow: bool, - pub webhook_script: bool, -} - -impl ServerToSet { - pub fn new(http_route: bool, webhook_flow: bool, webhook_script: bool) -> ServerToSet { - ServerToSet { http_route, webhook_flow, webhook_script } - } -} - fn header_to_pascal_case(header: &str) -> String { header .split(|c: char| c == '-' || c == '_' || c == ' ') @@ -582,7 +578,17 @@ fn generate_components(future_paths: &[FuturePath]) -> Map { "application/octet-stream": {} } }, - + DEFAULT_SYNC_SSE_RESPONSE_KEY: { + "description": "Returns an SSE stream.", + "content": { + "text/event-stream": { + "schema": { + "type": "string", + "description": "Stream of SSE" + }, + } + } + } })); components @@ -652,7 +658,6 @@ struct GenerateOpenAPI { openapi_spec_format: Format, } -#[cfg(feature = "http_trigger")] async fn http_routes_to_future_paths( db: &DB, user_db: UserDB, @@ -683,7 +688,7 @@ async fn http_routes_to_future_paths( struct MinifiedHttpTrigger { route_path: String, http_method: HttpMethod, - is_async: bool, + request_type: RequestType, workspaced_route: bool, summary: Option, description: Option, @@ -697,7 +702,7 @@ async fn http_routes_to_future_paths( SELECT route_path, http_method AS "http_method: _", - is_async, + request_type AS "request_type: _", workspaced_route, summary, description, @@ -765,7 +770,7 @@ async fn http_routes_to_future_paths( let future_path = FuturePath::new( route_path, Kind::HttpRoute(HttpRouteConfig::new(method)), - Some(http_route.is_async), + Some(http_route.request_type), http_route.summary, http_route.description, auth_method, @@ -777,18 +782,6 @@ async fn http_routes_to_future_paths( Ok(openapi_future_paths) } -#[cfg(not(feature = "http_trigger"))] -async fn http_routes_to_future_paths( - _db: &DB, - _user_db: UserDB, - _authed: &ApiAuthed, - _pg_pool: &mut PgConnection, - _http_route_filters: Option<&[HttpRouteFilter]>, - _w_id: &str, -) -> Result> { - Ok(Vec::new()) -} - async fn webhook_to_future_paths( pg_pool: &mut PgConnection, webhook_filters: Option<&[WebhookFilter]>, diff --git a/backend/windmill-api/src/triggers/http/handler.rs b/backend/windmill-api/src/triggers/http/handler.rs index 4fa10b1e1f..a887dc32a7 100644 --- a/backend/windmill-api/src/triggers/http/handler.rs +++ b/backend/windmill-api/src/triggers/http/handler.rs @@ -1,10 +1,11 @@ use super::{ - http_trigger_args::RawHttpTriggerArgs, AuthenticationMethod, HttpMethod, TriggerRoute, - HTTP_ACCESS_CACHE, HTTP_AUTH_CACHE, HTTP_ROUTERS_CACHE, + http_trigger_args::RawHttpTriggerArgs, 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, resources::try_get_resource_from_db_as, triggers::{ http::{ @@ -12,7 +13,8 @@ use crate::{ RouteExists, ROUTE_PATH_KEY_RE, VALID_ROUTE_PATH_RE, }, trigger_helpers::{ - get_runnable_format, trigger_runnable, trigger_runnable_and_wait_for_result, RunnableId, + get_runnable_format, trigger_runnable, trigger_runnable_and_wait_for_result, + trigger_runnable_inner, RunnableId, }, Trigger, TriggerCrud, TriggerData, }, @@ -26,6 +28,7 @@ use axum::{ routing::{get, post}, Extension, Json, Router, }; +use futures::StreamExt; use http::{HeaderMap, StatusCode}; use sqlx::PgConnection; use std::{ @@ -174,33 +177,35 @@ pub async fn insert_new_trigger_into_db( ) -> Result<()> { require_admin(authed.is_admin, &authed.username)?; + let request_type = trigger.config.request_type; + sqlx::query!( r#" INSERT INTO http_trigger ( - workspace_id, - path, - route_path, + workspace_id, + path, + route_path, route_path_key, workspaced_route, authentication_resource_path, wrap_body, raw_string, - script_path, + script_path, summary, description, - is_flow, - is_async, - authentication_method, - http_method, - static_asset_config, - edited_by, - email, - edited_at, + is_flow, + request_type, + authentication_method, + http_method, + static_asset_config, + edited_by, + email, + edited_at, is_static_website, error_handler_path, error_handler_args, retry - ) + ) VALUES ( $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, now(), $19, $20, $21, $22 ) @@ -217,7 +222,7 @@ pub async fn insert_new_trigger_into_db( trigger.config.summary, trigger.config.description, trigger.base.is_flow, - trigger.config.is_async, + request_type as _, trigger.config.authentication_method as _, trigger.config.http_method as _, trigger.config.static_asset_config as _, @@ -358,7 +363,7 @@ impl TriggerCrud for HttpTrigger { const ADDITIONAL_SELECT_FIELDS: &[&'static str] = &[ "route_path", "route_path_key", - "is_async", + "request_type", "authentication_method", "http_method", "summary", @@ -459,35 +464,37 @@ impl TriggerCrud for HttpTrigger { let route_path_key = check_if_route_exist(db, &trigger.config, workspace_id, Some(path)).await?; + let request_type = trigger.config.request_type; + sqlx::query!( r#" - UPDATE - http_trigger - SET - route_path = $1, + UPDATE + http_trigger + SET + route_path = $1, route_path_key = $2, workspaced_route = $3, wrap_body = $4, raw_string = $5, authentication_resource_path = $6, - script_path = $7, - path = $8, - is_flow = $9, - http_method = $10, - static_asset_config = $11, - edited_by = $12, - email = $13, - is_async = $14, - authentication_method = $15, + script_path = $7, + path = $8, + is_flow = $9, + http_method = $10, + static_asset_config = $11, + edited_by = $12, + email = $13, + request_type = $14, + authentication_method = $15, summary = $16, description = $17, - edited_at = now(), + edited_at = now(), is_static_website = $18, error_handler_path = $19, error_handler_args = $20, retry = $21 - WHERE - workspace_id = $22 AND + WHERE + workspace_id = $22 AND path = $23 "#, route_path, @@ -503,7 +510,7 @@ impl TriggerCrud for HttpTrigger { trigger.config.static_asset_config as _, &authed.username, &authed.email, - trigger.config.is_async, + request_type as _, trigger.config.authentication_method as _, trigger.config.summary, trigger.config.description, @@ -517,32 +524,34 @@ impl TriggerCrud for HttpTrigger { .execute(&mut *tx) .await?; } else { + let request_type = trigger.config.request_type; + sqlx::query!( r#" - UPDATE - http_trigger - SET + UPDATE + http_trigger + SET wrap_body = $1, raw_string = $2, authentication_resource_path = $3, - script_path = $4, - path = $5, - is_flow = $6, - http_method = $7, - static_asset_config = $8, - edited_by = $9, - email = $10, - is_async = $11, - authentication_method = $12, + script_path = $4, + path = $5, + is_flow = $6, + http_method = $7, + static_asset_config = $8, + edited_by = $9, + email = $10, + request_type = $11, + authentication_method = $12, summary = $13, description = $14, - edited_at = now(), + edited_at = now(), is_static_website = $15, error_handler_path = $16, error_handler_args = $17, retry = $18 - WHERE - workspace_id = $19 AND + WHERE + workspace_id = $19 AND path = $20 "#, trigger.config.wrap_body, @@ -555,7 +564,7 @@ impl TriggerCrud for HttpTrigger { trigger.config.static_asset_config as _, &authed.username, &authed.email, - trigger.config.is_async, + request_type as _, trigger.config.authentication_method as _, trigger.config.summary, trigger.config.description, @@ -793,6 +802,7 @@ async fn route_job( args: RawHttpTriggerArgs, ) -> std::result::Result { let route_path = route_path.to_path().trim_end_matches("/"); + let (trigger, called_path, params, authed) = get_http_route_trigger( route_path, &auth_cache, @@ -1021,8 +1031,69 @@ async fn route_job( ) .map_err(|e| e.into_response())?; - if trigger.is_async { - trigger_runnable( + // Handle execution based on the execution mode + match trigger.request_type { + RequestType::SyncSse => { + // Trigger the job (always async when streaming) + let (uuid, _, _) = trigger_runnable_inner( + &db, + 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, + ) + .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, + ); + + 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, @@ -1037,9 +1108,8 @@ async fn route_job( None, ) .await - .map_err(|e| e.into_response()) - } else { - trigger_runnable_and_wait_for_result( + .map_err(|e| e.into_response()), + RequestType::Sync => trigger_runnable_and_wait_for_result( &db, Some(user_db), authed, @@ -1053,6 +1123,6 @@ async fn route_job( format!("http_trigger/{}", trigger.path), ) .await - .map_err(|e| e.into_response()) + .map_err(|e| e.into_response()), } } diff --git a/backend/windmill-api/src/triggers/http/mod.rs b/backend/windmill-api/src/triggers/http/mod.rs index 2b29f0e644..2c5df24913 100644 --- a/backend/windmill-api/src/triggers/http/mod.rs +++ b/backend/windmill-api/src/triggers/http/mod.rs @@ -35,7 +35,7 @@ pub struct TriggerRoute { is_flow: bool, route_path: String, workspace_id: String, - is_async: bool, + request_type: RequestType, authentication_method: AuthenticationMethod, edited_by: String, email: String, @@ -66,6 +66,15 @@ pub enum HttpMethod { Patch, } +#[derive(Serialize, Deserialize, sqlx::Type, Debug, Clone, Copy, PartialEq)] +#[sqlx(type_name = "REQUEST_TYPE", rename_all = "snake_case")] +#[serde(rename_all = "snake_case")] +pub enum RequestType { + Sync, + Async, + SyncSse, +} + impl TryFrom<&http::Method> for HttpMethod { type Error = Error; fn try_from(method: &http::Method) -> Result { @@ -96,7 +105,7 @@ pub enum AuthenticationMethod { pub struct HttpConfig { pub route_path: String, pub route_path_key: String, - pub is_async: bool, + pub request_type: RequestType, pub authentication_method: AuthenticationMethod, pub http_method: HttpMethod, pub summary: Option, @@ -109,11 +118,11 @@ pub struct HttpConfig { pub raw_string: bool, } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize)] pub struct HttpConfigRequest { #[serde(default)] pub route_path: String, - pub is_async: bool, + pub request_type: RequestType, pub authentication_method: AuthenticationMethod, pub http_method: HttpMethod, pub summary: Option, @@ -126,6 +135,61 @@ pub struct HttpConfigRequest { pub raw_string: Option, } +#[derive(Deserialize)] +struct HttpConfigRequestHelper { + #[serde(default)] + route_path: String, + request_type: Option, + is_async: Option, + authentication_method: AuthenticationMethod, + http_method: HttpMethod, + summary: Option, + description: Option, + static_asset_config: Option>, + is_static_website: bool, + authentication_resource_path: Option, + workspaced_route: Option, + wrap_body: Option, + raw_string: Option, +} + +impl<'de> Deserialize<'de> for HttpConfigRequest { + fn deserialize(deserializer: D) -> std::result::Result + where + D: serde::Deserializer<'de>, + { + let helper = HttpConfigRequestHelper::deserialize(deserializer)?; + + // Determine request_type with backward compatibility + let request_type = if let Some(mode) = helper.request_type { + mode + } else if let Some(is_async) = helper.is_async { + if is_async { + RequestType::Async + } else { + RequestType::Sync + } + } else { + RequestType::Sync + }; + + Ok(HttpConfigRequest { + route_path: helper.route_path, + request_type, + authentication_method: helper.authentication_method, + http_method: helper.http_method, + summary: helper.summary, + description: helper.description, + static_asset_config: helper.static_asset_config, + is_static_website: helper.is_static_website, + authentication_resource_path: helper.authentication_resource_path, + workspaced_route: helper.workspaced_route, + wrap_body: helper.wrap_body, + raw_string: helper.raw_string, + }) + } +} + // Regex patterns for route validation lazy_static::lazy_static! { static ref ROUTE_PATH_KEY_RE: regex::Regex = regex::Regex::new(r"/?:[-\w]+").unwrap(); @@ -174,17 +238,17 @@ pub async fn refresh_routers(db: &DB) -> Result<(bool, RwLockReadGuard<'_, Route let triggers = sqlx::query_as!( TriggerRoute, r#" - SELECT - path, - script_path, - is_flow, - route_path, + SELECT + path, + script_path, + is_flow, + route_path, authentication_resource_path, - workspace_id, - is_async, - authentication_method AS "authentication_method: _", - edited_by, - email, + workspace_id, + request_type AS "request_type: _", + authentication_method AS "authentication_method: _", + edited_by, + email, static_asset_config AS "static_asset_config: _", wrap_body, raw_string, @@ -193,9 +257,9 @@ pub async fn refresh_routers(db: &DB) -> Result<(bool, RwLockReadGuard<'_, Route error_handler_path, error_handler_args as "error_handler_args: _", retry as "retry: _" - FROM - http_trigger - WHERE + FROM + http_trigger + WHERE http_method = $1 "#, &http_method as &HttpMethod @@ -277,3 +341,66 @@ pub async fn refresh_routers_loop( } }); } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_request_type_backward_compatibility() { + // Test with new request_type field + let json_new = r#"{ + "route_path": "/test", + "request_type": "sync_sse", + "authentication_method": "none", + "http_method": "get", + "is_static_website": false + }"#; + let config: HttpConfigRequest = serde_json::from_str(json_new).unwrap(); + assert_eq!(config.request_type, RequestType::SyncSse); + + // Test with legacy is_async = true + let json_legacy_async = r#"{ + "route_path": "/test", + "is_async": true, + "authentication_method": "none", + "http_method": "get", + "is_static_website": false + }"#; + let config: HttpConfigRequest = serde_json::from_str(json_legacy_async).unwrap(); + assert_eq!(config.request_type, RequestType::Async); + + // Test with legacy is_async = false + let json_legacy_sync = r#"{ + "route_path": "/test", + "is_async": false, + "authentication_method": "none", + "http_method": "get", + "is_static_website": false + }"#; + let config: HttpConfigRequest = serde_json::from_str(json_legacy_sync).unwrap(); + assert_eq!(config.request_type, RequestType::Sync); + + // Test with neither field (default to sync) + let json_default = r#"{ + "route_path": "/test", + "authentication_method": "none", + "http_method": "get", + "is_static_website": false + }"#; + let config: HttpConfigRequest = serde_json::from_str(json_default).unwrap(); + assert_eq!(config.request_type, RequestType::Sync); + + // Test that request_type takes precedence over is_async + let json_both = r#"{ + "route_path": "/test", + "request_type": "sync_sse", + "is_async": true, + "authentication_method": "none", + "http_method": "get", + "is_static_website": false + }"#; + let config: HttpConfigRequest = serde_json::from_str(json_both).unwrap(); + assert_eq!(config.request_type, RequestType::SyncSse); + } +} diff --git a/backend/windmill-api/src/triggers/trigger_helpers.rs b/backend/windmill-api/src/triggers/trigger_helpers.rs index 993f458343..a012b1071b 100644 --- a/backend/windmill-api/src/triggers/trigger_helpers.rs +++ b/backend/windmill-api/src/triggers/trigger_helpers.rs @@ -477,7 +477,7 @@ pub trait TriggerJobArgs { } #[allow(dead_code)] -async fn trigger_runnable_inner( +pub async fn trigger_runnable_inner( db: &DB, user_db: Option, authed: ApiAuthed, diff --git a/frontend/src/lib/components/triggers/http/RouteEditorInner.svelte b/frontend/src/lib/components/triggers/http/RouteEditorInner.svelte index 67cd7639e0..c17c3b479e 100644 --- a/frontend/src/lib/components/triggers/http/RouteEditorInner.svelte +++ b/frontend/src/lib/components/triggers/http/RouteEditorInner.svelte @@ -76,7 +76,7 @@ let isValid = $state(false) let dirtyRoutePath = $state(false) let dirtyPath = $state(false) - let is_async = $state(false) + let request_type = $state<'sync' | 'async' | 'sync_sse'>('sync') let authentication_method = $state('none') let route_path = $state('') let http_method = $state<'get' | 'post' | 'put' | 'patch' | 'delete'>('post') @@ -223,7 +223,7 @@ is_flow = defaultValues?.is_flow ?? nis_flow edit = false itemKind = nis_flow ? 'flow' : 'script' - is_async = defaultValues?.is_async ?? false + request_type = defaultValues?.request_type ?? 'sync' authentication_method = defaultValues?.authentication_method ?? 'none' route_path = defaultValues?.route_path ?? '' dirtyRoutePath = false @@ -263,7 +263,7 @@ path = cfg?.path ?? '' route_path = cfg?.route_path ?? '' http_method = cfg?.http_method ?? 'post' - is_async = cfg?.is_async ?? false + request_type = cfg?.request_type ?? 'sync' workspaced_route = cfg?.workspaced_route ?? false wrap_body = cfg?.wrap_body ?? false raw_string = cfg?.raw_string ?? false @@ -342,7 +342,7 @@ path, route_path, http_method, - is_async, + request_type, workspaced_route, wrap_body, raw_string, @@ -464,7 +464,7 @@ initialScriptPath = '' is_flow = false http_method = 'get' - is_async = false + request_type = 'sync' is_static_website = ev.detail === 'static_website' if (is_static_website) { authentication_method = 'none' @@ -617,13 +617,20 @@ {#snippet action()} { - is_async = detail === 'async' + request_type = detail }} disabled={!can_write || !!static_asset_config} > {#snippet children({ item, disabled })} + diff --git a/frontend/src/lib/components/triggers/http/utils.ts b/frontend/src/lib/components/triggers/http/utils.ts index 7591ccc4bd..31c70dc5dd 100644 --- a/frontend/src/lib/components/triggers/http/utils.ts +++ b/frontend/src/lib/components/triggers/http/utils.ts @@ -46,7 +46,7 @@ export async function saveHttpRouteFromCfg( path: routeCfg.path, script_path: routeCfg.script_path, is_flow: routeCfg.is_flow, - is_async: routeCfg.is_async, + request_type: routeCfg.request_type, authentication_method: routeCfg.authentication_method, route_path: routeCfg.route_path, http_method: routeCfg.http_method, @@ -139,7 +139,7 @@ function processOpenApiDocument( http_method: method, authentication_method: 'none', workspaced_route: false, - is_async: true, + request_type: 'async', script_path: '', raw_string: false, is_flow: false, diff --git a/frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte b/frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte index 4271728c1d..d77a76f543 100644 --- a/frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte +++ b/frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte @@ -10,7 +10,7 @@ } from '$lib/consts' import bash from 'svelte-highlight/languages/bash' import { Tabs, Tab, TabContent, Button } from '$lib/components/common' - import { ArrowDownRight, ArrowUpRight, Clipboard, RssIcon } from 'lucide-svelte' + import { ArrowDownRight, ArrowUpRight, Clipboard } from 'lucide-svelte' import { Highlight } from 'svelte-highlight' import { typescript } from 'svelte-highlight/languages' import ClipboardPanel from '../../details/ClipboardPanel.svelte' @@ -49,7 +49,6 @@ hash?: string path: string } - sse: {} } sync: { get: { @@ -59,7 +58,13 @@ hash?: string path: string } - sse: { + } + sync_sse: { + get: { + hash?: string + path: string + } + post: { hash?: string path: string } @@ -67,23 +72,17 @@ } = $derived(isFlow ? computeFlowWebhooks(path) : computeScriptWebhooks(hash, path)) let selectedTab: string = $state('rest') let userSettings: UserSettings | undefined = $state() - let requestType = $state(DEFAULT_WEBHOOK_TYPE) as 'async' | 'sync' - let callMethod = $state('post') as 'get' | 'post' | 'sse' + let requestType = $state(DEFAULT_WEBHOOK_TYPE) as 'async' | 'sync' | 'sync_sse' + let callMethod = $state('post') as 'get' | 'post' let runnableId = $state('path') as 'hash' | 'path' let tokenType = $state('headers') as 'query' | 'headers' $effect(() => { - if (requestType === 'async' && (callMethod === 'get' || callMethod === 'sse')) { + if (requestType === 'async' && callMethod === 'get') { callMethod = 'post' } }) - $effect(() => { - if (callMethod === 'sse' && tokenType === 'headers') { - tokenType = 'query' - } - }) - let cleanedRunnableArgs = $derived.by(() => { readFieldsRecursively(runnableArgs) return isObject(runnableArgs) && 'wm_trigger' in runnableArgs @@ -94,7 +93,7 @@ webhooks[requestType][callMethod][runnableId] + (tokenType === 'query' ? `?token=${token}${ - callMethod === 'get' || callMethod === 'sse' + callMethod === 'get' || requestType === 'sync_sse' ? `&payload=${encodeURIComponent(btoa(JSON.stringify(cleanedRunnableArgs ?? {})))}` : '' }` @@ -113,8 +112,7 @@ post: { hash: `${webhookBase}/run/h/${hash}`, path: `${webhookBase}/run/p/${path}` - }, - sse: {} + } }, sync: { get: { @@ -123,8 +121,14 @@ post: { hash: `${webhookBase}/run_wait_result/h/${hash}`, path: `${webhookBase}/run_wait_result/p/${path}` + } + }, + sync_sse: { + get: { + path: `${webhookBase}/run_and_stream/p/${path}`, + hash: `${webhookBase}/run_and_stream/h/${hash}` }, - sse: { + post: { hash: `${webhookBase}/run_and_stream/h/${hash}`, path: `${webhookBase}/run_and_stream/p/${path}` } @@ -143,8 +147,7 @@ get: {}, post: { path: urlAsync - }, - sse: {} + } }, sync: { get: { @@ -152,8 +155,13 @@ }, post: { path: urlSync + } + }, + sync_sse: { + get: { + path: urlStream }, - sse: { + post: { path: urlStream } } @@ -173,31 +181,57 @@ } function fetchCode() { - if (callMethod === 'sse') { + if (requestType === 'sync_sse') { return ` -import { EventSource } from "eventsource"; - export async function main() { - const endpoint = \`${url}\`; + const response = await fetch(\`${url}\`, { + method: '${callMethod === 'get' ? 'GET' : 'POST'}', + headers: ${JSON.stringify(headers(), null, 2).replaceAll('\n', '\n ')}, + body: ${callMethod === 'get' ? 'undefined' : `JSON.stringify(${JSON.stringify(cleanedRunnableArgs ?? {}, null, 2).replaceAll('\n', '\n ')})`} + }); + + if (!response.ok) { + const text = await response.text() + throw new Error(\`\${response.status} \${text}\`) + } - return new Promise((resolve, reject) => { - const eventSource = new EventSource(endpoint); + if (!response.body) { + throw new Error("Response body is empty"); + } - eventSource.onmessage = (event) => { - const data = JSON.parse(event.data); - console.log(data); - if (data.completed) { - eventSource.close(); - resolve(); - } - }; + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; - eventSource.onerror = (error) => { - console.error('EventSource error:', error); - eventSource.close(); - reject(error); - }; - }); + try { + while (true) { + const { done, value } = await reader.read(); + + if (done) break; + + buffer += decoder.decode(value, { stream: true }); + const lines = buffer.split("\\n"); + + // Keep the last incomplete line in buffer + buffer = lines.pop() || ""; + + for (const line of lines) { + if (line.startsWith("data: ")) { + const jsonData = line.slice(6); // Remove 'data: ' prefix + const data = JSON.parse(jsonData); + console.log(data); + + if (data.completed) { + reader.cancel(); + return; + } + } + } + } + } catch (error) { + console.error("Stream error:", error); + throw error; + } }` } if (requestType === 'sync') { @@ -292,16 +326,17 @@ function waitForJobCompletion(UUID) { return `TOKEN='${token}' ${callMethod !== 'get' ? `BODY='${JSON.stringify(cleanedRunnableArgs ?? {})}'` : ''} URL='${url}' -${requestType === 'sync' ? 'RESULT' : 'UUID'}=$(curl -s ${ +${requestType === 'sync' ? 'RESULT=$(' : requestType === 'async' ? 'UUID=$(' : ''}curl -s ${ callMethod != 'get' ? "-H 'Content-Type: application/json'" : '' } ${tokenType === 'headers' ? `-H "Authorization: Bearer $TOKEN"` : ''} -X ${ callMethod === 'get' ? 'GET' : 'POST' - } ${callMethod !== 'get' ? `-d "$BODY" ` : ''}$URL) + } ${callMethod !== 'get' ? `-d "$BODY" ` : ''}$URL${requestType === 'sync' || requestType === 'async' ? ')' : ''} ${ requestType === 'sync' ? 'echo -E $RESULT | jq' - : ` + : requestType === 'async' + ? ` URL="${location.origin}/api/w/${$workspaceStore}/jobs_u/completed/get_result_maybe/$UUID" while true; do curl -s -H "Authorization: Bearer $TOKEN" $URL -o res.json @@ -313,6 +348,7 @@ while true; do sleep 1 fi done` + : '' }` } @@ -370,6 +406,15 @@ done` tooltip="Triggers the execution, wait for the job to complete and return it as a response." {item} /> + {/snippet} @@ -390,19 +435,7 @@ done` selectedColor="#fb923c" value="get" {item} - disabled={requestType !== 'sync'} - /> - {/snippet} @@ -426,12 +459,7 @@ done` > {#snippet children({ item })} - + {/snippet} @@ -443,12 +471,10 @@ done`
REST - {#if SCRIPT_VIEW_SHOW_EXAMPLE_CURL && callMethod !== 'sse'} + {#if SCRIPT_VIEW_SHOW_EXAMPLE_CURL} Curl {/if} - - {callMethod === 'sse' ? 'Event Source' : 'Fetch'} - + Fetch {#snippet content()} {#key token} @@ -486,7 +512,7 @@ done` }} > - +
{/key} {/key} @@ -508,7 +534,7 @@ done` }} > - + {/key}{/key}{/key}{/key} {/key} diff --git a/frontend/src/lib/utils_deployable.ts b/frontend/src/lib/utils_deployable.ts index 2a81423b47..762206913e 100644 --- a/frontend/src/lib/utils_deployable.ts +++ b/frontend/src/lib/utils_deployable.ts @@ -359,7 +359,7 @@ export async function getTriggerValue(kind: TriggerKind, path: string, workspace http_method, route_path, static_asset_config, - is_async, + request_type, authentication_method, is_static_website, authentication_resource_path @@ -374,7 +374,7 @@ export async function getTriggerValue(kind: TriggerKind, path: string, workspace http_method, route_path, static_asset_config, - is_async, + request_type, authentication_method, is_static_website, authentication_resource_path