diff --git a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json index 713ccb9dd3..36ddb8ab9f 100644 --- a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json +++ b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json @@ -15,7 +15,7 @@ ] }, "nullable": [ - null + true ] }, "hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55" diff --git a/backend/.sqlx/query-7cab8475996f0abd7ca7888b460bd3faf3bef6e34200c258e3704981817c241e.json b/backend/.sqlx/query-7cab8475996f0abd7ca7888b460bd3faf3bef6e34200c258e3704981817c241e.json new file mode 100644 index 0000000000..4311abf5ff --- /dev/null +++ b/backend/.sqlx/query-7cab8475996f0abd7ca7888b460bd3faf3bef6e34200c258e3704981817c241e.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT \n path \n FROM \n flow_version \n WHERE \n id = $1 AND \n workspace_id = $2\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Int8", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "7cab8475996f0abd7ca7888b460bd3faf3bef6e34200c258e3704981817c241e" +} diff --git a/backend/.sqlx/query-a7468e9054beed88636786c5495ac3b9d9a6086ae6212ad9237b39a0346d7d26.json b/backend/.sqlx/query-a7468e9054beed88636786c5495ac3b9d9a6086ae6212ad9237b39a0346d7d26.json new file mode 100644 index 0000000000..cead5d7019 --- /dev/null +++ b/backend/.sqlx/query-a7468e9054beed88636786c5495ac3b9d9a6086ae6212ad9237b39a0346d7d26.json @@ -0,0 +1,66 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT\n flow_version.id AS version,\n flow_version.value->>'early_return' as early_return, \n flow_version.value->>'preprocessor_module' IS NOT NULL as has_preprocessor, \n (flow_version.value->>'chat_input_enabled')::boolean as chat_input_enabled, \n flow.tag, \n flow.dedicated_worker, \n flow.on_behalf_of_email, \n flow.edited_by\n FROM \n flow_version\n INNER JOIN flow\n ON flow.path = flow_version.path AND\n flow.workspace_id = flow_version.workspace_id\n WHERE \n flow_version.workspace_id = $1 AND\n flow_version.path = $2 AND\n flow_version.id = $3\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "version", + "type_info": "Int8" + }, + { + "ordinal": 1, + "name": "early_return", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "has_preprocessor", + "type_info": "Bool" + }, + { + "ordinal": 3, + "name": "chat_input_enabled", + "type_info": "Bool" + }, + { + "ordinal": 4, + "name": "tag", + "type_info": "Varchar" + }, + { + "ordinal": 5, + "name": "dedicated_worker", + "type_info": "Bool" + }, + { + "ordinal": 6, + "name": "on_behalf_of_email", + "type_info": "Text" + }, + { + "ordinal": 7, + "name": "edited_by", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Int8" + ] + }, + "nullable": [ + false, + null, + null, + null, + true, + true, + true, + false + ] + }, + "hash": "a7468e9054beed88636786c5495ac3b9d9a6086ae6212ad9237b39a0346d7d26" +} diff --git a/backend/.sqlx/query-b82f97f2a858e20b1fb6d9b0b3eab5bab2bafa91689eac5073c1678c338afb27.json b/backend/.sqlx/query-b82f97f2a858e20b1fb6d9b0b3eab5bab2bafa91689eac5073c1678c338afb27.json new file mode 100644 index 0000000000..64b774cbd9 --- /dev/null +++ b/backend/.sqlx/query-b82f97f2a858e20b1fb6d9b0b3eab5bab2bafa91689eac5073c1678c338afb27.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT \n path \n FROM \n flow_version \n WHERE \n id = $1 AND \n workspace_id = $2\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Int8", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "b82f97f2a858e20b1fb6d9b0b3eab5bab2bafa91689eac5073c1678c338afb27" +} diff --git a/backend/.sqlx/query-e1409c67b93881cf68f6ac9c8bae0856cf426c7e7860c8b5b799972baa8e6945.json b/backend/.sqlx/query-e1409c67b93881cf68f6ac9c8bae0856cf426c7e7860c8b5b799972baa8e6945.json deleted file mode 100644 index da6ca296b1..0000000000 --- a/backend/.sqlx/query-e1409c67b93881cf68f6ac9c8bae0856cf426c7e7860c8b5b799972baa8e6945.json +++ /dev/null @@ -1,66 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT tag, dedicated_worker, flow_version.value->>'early_return' as early_return, flow_version.value->>'preprocessor_module' IS NOT NULL as has_preprocessor, (flow_version.value->>'chat_input_enabled')::boolean as chat_input_enabled, on_behalf_of_email, edited_by, flow_version.id AS version\n FROM flow\n INNER JOIN flow_version\n ON flow_version.id = $3\n WHERE flow.path = $1 and flow.workspace_id = $2", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "tag", - "type_info": "Varchar" - }, - { - "ordinal": 1, - "name": "dedicated_worker", - "type_info": "Bool" - }, - { - "ordinal": 2, - "name": "early_return", - "type_info": "Text" - }, - { - "ordinal": 3, - "name": "has_preprocessor", - "type_info": "Bool" - }, - { - "ordinal": 4, - "name": "chat_input_enabled", - "type_info": "Bool" - }, - { - "ordinal": 5, - "name": "on_behalf_of_email", - "type_info": "Text" - }, - { - "ordinal": 6, - "name": "edited_by", - "type_info": "Varchar" - }, - { - "ordinal": 7, - "name": "version", - "type_info": "Int8" - } - ], - "parameters": { - "Left": [ - "Text", - "Text", - "Int8" - ] - }, - "nullable": [ - true, - true, - null, - null, - null, - true, - false, - false - ] - }, - "hash": "e1409c67b93881cf68f6ac9c8bae0856cf426c7e7860c8b5b799972baa8e6945" -} diff --git a/backend/.sqlx/query-e2474b7855c8b08f927f2b987421e773e537a8eb0a113477764bbab14e1f3a3d.json b/backend/.sqlx/query-f0c9670c27d9a3b8e8f316fd28f7d3ede2ecc3225ff2879ff5dfd59c4ef236d5.json similarity index 69% rename from backend/.sqlx/query-e2474b7855c8b08f927f2b987421e773e537a8eb0a113477764bbab14e1f3a3d.json rename to backend/.sqlx/query-f0c9670c27d9a3b8e8f316fd28f7d3ede2ecc3225ff2879ff5dfd59c4ef236d5.json index c7c7214a87..19842dda7d 100644 --- a/backend/.sqlx/query-e2474b7855c8b08f927f2b987421e773e537a8eb0a113477764bbab14e1f3a3d.json +++ b/backend/.sqlx/query-f0c9670c27d9a3b8e8f316fd28f7d3ede2ecc3225ff2879ff5dfd59c4ef236d5.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT\n value->'preprocessor_module'->'value' as \"preprocessor_module: _\",\n schema as \"schema: _\"\n FROM flow_version\n WHERE\n path = $1\n AND workspace_id = $2\n ORDER BY created_at DESC\n LIMIT 1", + "query": "SELECT\n value->'preprocessor_module'->'value' as \"preprocessor_module: _\",\n schema as \"schema: _\"\n FROM flow_version\n WHERE\n id = $1\n AND workspace_id = $2", "describe": { "columns": [ { @@ -16,7 +16,7 @@ ], "parameters": { "Left": [ - "Text", + "Int8", "Text" ] }, @@ -25,5 +25,5 @@ true ] }, - "hash": "e2474b7855c8b08f927f2b987421e773e537a8eb0a113477764bbab14e1f3a3d" + "hash": "f0c9670c27d9a3b8e8f316fd28f7d3ede2ecc3225ff2879ff5dfd59c4ef236d5" } diff --git a/backend/.sqlx/query-fc9753f501974c4b570d710c7621ff1f6787a852bb27acf49593af963a6aacca.json b/backend/.sqlx/query-fc9753f501974c4b570d710c7621ff1f6787a852bb27acf49593af963a6aacca.json new file mode 100644 index 0000000000..b699cc582a --- /dev/null +++ b/backend/.sqlx/query-fc9753f501974c4b570d710c7621ff1f6787a852bb27acf49593af963a6aacca.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT path FROM flow_version WHERE id = $1 AND workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Int8", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "fc9753f501974c4b570d710c7621ff1f6787a852bb27acf49593af963a6aacca" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 4633dc7d63..f6823b4692 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -a0471e9f11c9cc181cfd9eb2738858f7bebc30ee \ No newline at end of file +31ee9d3449f05cd0328c0fcf43e2b161dee767b9 \ No newline at end of file diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 72b6aeea52..57d7833c9d 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -2769,7 +2769,7 @@ async fn test_result_format(db: Pool) -> anyhow::Result<()> { let response = windmill_api::jobs::run_wait_result( &db, Uuid::parse_str(ordered_result_job_id).unwrap(), - "test-workspace".to_string(), + "test-workspace", None, "test-user", ) diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 8fa61418ce..e57812e2e2 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -5956,6 +5956,80 @@ paths: application/json: schema: {} + /w/{workspace}/jobs/run_wait_result/fv/{version}: + post: + summary: run flow by version and wait until completion + operationId: runWaitResultFlowByVersion + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - name: version + description: flow version ID + in: path + required: true + schema: + type: integer + format: int64 + - $ref: "#/components/parameters/IncludeHeader" + - $ref: "#/components/parameters/QueueLimit" + - $ref: "#/components/parameters/NewJobId" + - $ref: "#/components/parameters/SkipPreprocessor" + - name: memory_id + description: memory ID for chat-enabled flows + in: query + schema: + type: string + format: uuid + + requestBody: + description: script args + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/ScriptArgs" + + responses: + "200": + description: job result + content: + application/json: + schema: {} + + get: + summary: run flow by version with GET and wait until completion + operationId: runWaitResultFlowByVersionGet + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - name: version + description: flow version ID + in: path + required: true + schema: + type: integer + format: int64 + - $ref: "#/components/parameters/IncludeHeader" + - $ref: "#/components/parameters/QueueLimit" + - $ref: "#/components/parameters/Payload" + - $ref: "#/components/parameters/NewJobId" + - $ref: "#/components/parameters/SkipPreprocessor" + - name: memory_id + description: memory ID for chat-enabled flows + in: query + schema: + type: string + format: uuid + + responses: + "200": + description: job result + content: + application/json: + schema: {} + /w/{workspace}/jobs/run_and_stream/f/{path}: post: summary: run flow by path and stream updates via SSE @@ -6032,6 +6106,94 @@ paths: schema: type: string + /w/{workspace}/jobs/run_and_stream/fv/{version}: + post: + summary: run flow by version and stream updates via SSE + operationId: runAndStreamFlowByVersion + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - name: version + description: flow version ID + in: path + required: true + schema: + type: integer + format: int64 + - $ref: "#/components/parameters/IncludeHeader" + - $ref: "#/components/parameters/QueueLimit" + - $ref: "#/components/parameters/NewJobId" + - $ref: "#/components/parameters/SkipPreprocessor" + - name: memory_id + description: memory ID for chat-enabled flows + in: query + schema: + type: string + format: uuid + - name: poll_delay_ms + description: delay between polling for job updates in milliseconds + in: query + schema: + type: integer + format: int64 + + requestBody: + description: flow args + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/ScriptArgs" + + responses: + "200": + description: server-sent events stream of job updates + content: + text/event-stream: + schema: + type: string + + get: + summary: run flow by version with GET and stream updates via SSE + operationId: runAndStreamFlowByVersionGet + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - name: version + description: flow version ID + in: path + required: true + schema: + type: integer + format: int64 + - $ref: "#/components/parameters/IncludeHeader" + - $ref: "#/components/parameters/QueueLimit" + - $ref: "#/components/parameters/Payload" + - $ref: "#/components/parameters/NewJobId" + - $ref: "#/components/parameters/SkipPreprocessor" + - name: memory_id + description: memory ID for chat-enabled flows + in: query + schema: + type: string + format: uuid + - name: poll_delay_ms + description: delay between polling for job updates in milliseconds + in: query + schema: + type: integer + format: int64 + + responses: + "200": + description: server-sent events stream of job updates + content: + text/event-stream: + schema: + type: string + /w/{workspace}/jobs/run_and_stream/p/{path}: post: summary: run script by path and stream updates via SSE @@ -7680,6 +7842,64 @@ paths: type: string format: uuid + /w/{workspace}/jobs/run/fv/{version}: + post: + summary: run flow by version + operationId: runFlowByVersion + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - name: version + description: flow version ID + in: path + required: true + schema: + type: integer + format: int64 + - name: scheduled_for + description: when to schedule this job (leave empty for immediate run) + in: query + schema: + type: string + format: date-time + - name: scheduled_in_secs + description: schedule the script to execute in the number of seconds starting now + in: query + schema: + type: integer + - $ref: "#/components/parameters/SkipPreprocessor" + - $ref: "#/components/parameters/ParentJob" + - $ref: "#/components/parameters/WorkerTag" + - $ref: "#/components/parameters/NewJobId" + - $ref: "#/components/parameters/IncludeHeader" + - name: invisible_to_owner + description: make the run invisible to the the flow owner (default false) + in: query + schema: + type: boolean + - name: memory_id + description: memory ID for chat-enabled flows + in: query + schema: + type: string + format: uuid + requestBody: + description: flow args + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/ScriptArgs" + responses: + "201": + description: job created + content: + text/plain: + schema: + type: string + format: uuid + /w/{workspace}/jobs/run/batch_rerun_jobs: post: summary: re-run multiple jobs @@ -18721,6 +18941,8 @@ components: properties: lock_error_logs: type: string + version_id: + type: number ExtraPerms: type: object diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 0b4bdbd0fb..dd141d46f6 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -1135,6 +1135,7 @@ async fn get_flow_by_path( flow.timeout, flow.visible_to_runner_only, flow.on_behalf_of_email, + flow_version.id AS version_id, flow_version.schema, flow_version.value, flow_version.created_at AS edited_at, @@ -1174,8 +1175,9 @@ async fn get_flow_by_path( flow.timeout, flow.visible_to_runner_only, flow.on_behalf_of_email, + flow_version.id AS version_id, flow_version.schema, - flow_version.value, + flow_version.value, flow_version.created_at AS edited_at, flow_version.created_by AS edited_by, NULL AS starred diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 90cdb4e747..dcb2b6cd08 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -45,7 +45,7 @@ use windmill_common::{email_oss::send_email_html, server::load_smtp_config}; use windmill_common::variables::get_workspace_key; -use crate::triggers::trigger_helpers::ScriptId; +use crate::triggers::trigger_helpers::{FlowId, ScriptId}; use crate::{ add_webhook_allowed_origin, args::{self, RawWebhookArgs}, @@ -95,8 +95,9 @@ use windmill_common::{ }; use windmill_common::{ - get_latest_deployed_hash_for_path, get_latest_flow_version_info_for_path, - get_script_info_for_hash, utils::empty_as_none, FlowVersionInfo, ScriptHashInfo, BASE_URL, + get_flow_version_info_from_version, get_latest_deployed_hash_for_path, + get_latest_flow_version_info_for_path, get_script_info_for_hash, utils::empty_as_none, + FlowVersionInfo, ScriptHashInfo, BASE_URL, }; use windmill_queue::{ cancel_job, get_result_and_success_by_id_from_flow, job_is_complete, push, PushArgs, @@ -124,6 +125,13 @@ pub fn workspaced_service() -> Router { .layer(cors.clone()) .layer(ce_headers.clone()), ) + .route( + "/run/fv/:version", + post(run_flow_by_version) + .head(|| async { "" }) + .layer(cors.clone()) + .layer(ce_headers.clone()), + ) .route( "/run/batch_rerun_jobs", post(batch_rerun_jobs) @@ -176,6 +184,14 @@ pub fn workspaced_service() -> Router { .layer(cors.clone()) .layer(ce_headers.clone()), ) + .route( + "/run_wait_result/fv/:version", + post(run_wait_result_flow_by_version) + .get(run_wait_result_flow_by_version_get) + .head(|| async { "" }) + .layer(cors.clone()) + .layer(ce_headers.clone()), + ) .route( "/run_and_stream/f/*script_path", get(stream_flow_by_path) @@ -184,6 +200,14 @@ pub fn workspaced_service() -> Router { .layer(cors.clone()) .layer(ce_headers.clone()), ) + .route( + "/run_and_stream/fv/:version", + get(stream_flow_by_version) + .post(stream_flow_by_version) + .head(|| async { "" }) + .layer(cors.clone()) + .layer(ce_headers.clone()), + ) .route( "/run_and_stream/p/*script_path", get(stream_script_by_path) @@ -1750,6 +1774,22 @@ impl RunJobQuery { Ok(None) } } + + fn payload_as_args(&self) -> error::Result>> { + let payload_r = self.payload.clone().map(decode_payload).map(|x| { + x.map_err(|e| { + error::Error::internal_err(format!("Impossible to decode query payload: {e:#?}")) + }) + }); + + let payload_as_args = if let Some(payload) = payload_r { + payload? + } else { + HashMap::new() + }; + + Ok(payload_as_args) + } } #[derive(Deserialize, Clone)] @@ -4000,24 +4040,17 @@ pub async fn run_flow_by_path( Ok((StatusCode::CREATED, uuid.to_string())) } -pub async fn run_flow_by_path_inner( - authed: ApiAuthed, - db: DB, +pub async fn run_flow( + authed: &ApiAuthed, + db: &DB, user_db: UserDB, - w_id: String, - flow_path: StripPath, + w_id: &str, + flow_path: &str, + flow_version_info: FlowVersionInfo, run_query: RunJobQuery, args: PushArgsOwned, trigger_kind: Option, ) -> error::Result<(Uuid, Option)> { - #[cfg(feature = "enterprise")] - check_license_key_valid().await?; - - let flow_path = flow_path.to_path(); - check_scopes(&authed, || format!("jobs:run:flows:{flow_path}"))?; - - let userdb_authed = UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() }; - let FlowVersionInfo { version, tag, @@ -4028,8 +4061,7 @@ pub async fn run_flow_by_path_inner( edited_by, early_return, .. - } = get_latest_flow_version_info_for_path(Some(userdb_authed), &db, &w_id, &flow_path, true) - .await?; + } = flow_version_info; let tag = run_query.tag.clone().or(tag); @@ -4110,9 +4142,146 @@ pub async fn run_flow_by_path_inner( } tx.commit().await?; + Ok((uuid, early_return)) } +pub async fn run_flow_and_wait_result( + authed: &ApiAuthed, + db: &DB, + user_db: UserDB, + w_id: &str, + flow_path: &str, + flow_version_info: FlowVersionInfo, + run_query: RunJobQuery, + args: PushArgsOwned, + trigger_kind: Option, +) -> error::Result { + let (uuid, early_return) = run_flow( + authed, + db, + user_db, + w_id, + flow_path, + flow_version_info, + run_query, + args, + trigger_kind, + ) + .await?; + + run_wait_result(&db, uuid, w_id, early_return, &authed.username).await +} + +pub async fn run_flow_by_path_inner( + authed: ApiAuthed, + db: DB, + user_db: UserDB, + w_id: String, + flow_path: StripPath, + run_query: RunJobQuery, + args: PushArgsOwned, + trigger_kind: Option, +) -> error::Result<(Uuid, Option)> { + #[cfg(feature = "enterprise")] + check_license_key_valid().await?; + + let flow_path = flow_path.to_path(); + check_scopes(&authed, || format!("jobs:run:flows:{flow_path}"))?; + + let userdb_authed = UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() }; + + let flow_version_info = + get_latest_flow_version_info_for_path(Some(userdb_authed), &db, &w_id, &flow_path, true) + .await?; + + run_flow( + &authed, + &db, + user_db, + &w_id, + flow_path, + flow_version_info, + run_query, + args, + trigger_kind, + ) + .await +} + +pub async fn run_flow_by_version( + authed: ApiAuthed, + Extension(db): Extension, + Extension(user_db): Extension, + Path((w_id, version)): Path<(String, i64)>, + Query(run_query): Query, + args: RawWebhookArgs, +) -> error::Result<(StatusCode, String)> { + let args = args + .to_args_from_runnable( + &authed, + &db, + &w_id, + RunnableId::from_flow_version(version), + run_query.skip_preprocessor, + ) + .await?; + + let (uuid, _) = + run_flow_by_version_inner(authed, db, user_db, w_id, version, run_query, args, None) + .await?; + + Ok((StatusCode::CREATED, uuid.to_string())) +} + +pub async fn run_flow_by_version_inner( + authed: ApiAuthed, + db: DB, + user_db: UserDB, + w_id: String, + version: i64, + run_query: RunJobQuery, + args: PushArgsOwned, + trigger_kind: Option, +) -> error::Result<(Uuid, Option)> { + #[cfg(feature = "enterprise")] + check_license_key_valid().await?; + + let flow_path = sqlx::query_scalar!( + r#" + SELECT + path + FROM + flow_version + WHERE + id = $1 AND + workspace_id = $2 + "#, + version, + &w_id + ) + .fetch_one(&db) + .await?; + + check_scopes(&authed, || format!("jobs:run:flows:{flow_path}"))?; + + let flow_version_info = + get_flow_version_info_from_version(&db, version, &w_id, &flow_path).await?; + + run_flow( + &authed, + &db, + user_db, + &w_id, + &flow_path, + flow_version_info, + run_query, + args, + trigger_kind, + ) + .await +} + #[cfg(not(feature = "enterprise"))] pub async fn restart_flow( _authed: ApiAuthed, @@ -4600,7 +4769,7 @@ pub struct WindmillCompositeResult { pub async fn run_wait_result_internal( db: &DB, uuid: Uuid, - w_id: String, + w_id: &str, node_id_for_empty_return: Option, username: &str, ) -> error::Result<(Box, bool)> { @@ -4616,7 +4785,7 @@ pub async fn run_wait_result_internal( let mut g = Guard { done: false, id: uuid, - w_id: w_id.clone(), + w_id: w_id.to_string(), db: db.clone(), username: username.to_string(), }; @@ -4628,7 +4797,7 @@ pub async fn run_wait_result_internal( if let Some(node_id_for_empty_return) = node_id_for_empty_return.as_ref() { let result_and_success = get_result_and_success_by_id_from_flow( &db, - &w_id, + w_id, &uuid, node_id_for_empty_return, None, @@ -4785,7 +4954,7 @@ pub fn result_to_response(result: Box, success: bool) -> error::Result pub async fn run_wait_result( db: &DB, uuid: Uuid, - w_id: String, + w_id: &str, node_id_for_empty_return: Option, username: &str, ) -> error::Result { @@ -4945,18 +5114,11 @@ pub async fn run_wait_result_job_by_path_get( if method == http::Method::HEAD { return Ok(Json(serde_json::json!("")).into_response()); } - let payload_r = run_query.payload.map(decode_payload).map(|x| { - x.map_err(|e| Error::internal_err(format!("Impossible to decode query payload: {e:#?}"))) - }); - let payload_args = if let Some(payload) = payload_r { - payload? - } else { - HashMap::new() - }; + let payload_as_args = run_query.payload_as_args()?; let mut args = args.process_args(&authed, &db, &w_id, None).await?; - args.body = args::Body::HashMap(payload_args); + args.body = args::Body::HashMap(payload_as_args); let args = args .to_args_from_runnable( @@ -5033,7 +5195,7 @@ pub async fn run_wait_result_job_by_path_get( .await?; tx.commit().await?; - let wait_result = run_wait_result(&db, uuid, w_id, None, &authed.username).await; + let wait_result = run_wait_result(&db, uuid, &w_id, None, &authed.username).await; if delete_after_use.unwrap_or(false) { delete_job_metadata_after_use(&db, uuid).await?; } @@ -5058,20 +5220,10 @@ pub async fn run_wait_result_flow_by_path_get( if method == http::Method::HEAD { return Ok(Json(serde_json::json!("")).into_response()); } - let payload_r = run_query.payload.clone().map(decode_payload).map(|x| { - x.map_err(|e| { - error::Error::internal_err(format!("Impossible to decode query payload: {e:#?}")) - }) - }); - - let payload_args = if let Some(payload) = payload_r { - payload? - } else { - HashMap::new() - }; + let payload_as_args = run_query.payload_as_args()?; let mut args = args.process_args(&authed, &db, &w_id, None).await?; - args.body = args::Body::HashMap(payload_args); + args.body = args::Body::HashMap(payload_as_args); let args = args .to_args_from_runnable( @@ -5188,7 +5340,7 @@ pub async fn run_wait_result_script_by_path_internal( .await?; tx.commit().await?; - let wait_result = run_wait_result(&db, uuid, w_id, None, &authed.username).await; + let wait_result = run_wait_result(&db, uuid, &w_id, None, &authed.username).await; if delete_after_use.unwrap_or(false) { delete_job_metadata_after_use(&db, uuid).await?; } @@ -5311,7 +5463,7 @@ pub async fn run_wait_result_script_by_hash( .await?; tx.commit().await?; - let wait_result = run_wait_result(&db, uuid, w_id, None, &authed.username).await; + let wait_result = run_wait_result(&db, uuid, &w_id, None, &authed.username).await; if delete_after_use.unwrap_or(false) { delete_job_metadata_after_use(&db, uuid).await?; } @@ -5368,6 +5520,28 @@ pub async fn stream_flow_by_path( .await } +pub async fn stream_flow_by_version( + authed: ApiAuthed, + Extension(db): Extension, + Extension(user_db): Extension, + Path((w_id, version)): Path<(String, i64)>, + Query(run_query): Query, + method: hyper::http::Method, + args: RawWebhookArgs, +) -> error::Result { + stream_job( + authed, + db, + user_db, + w_id, + RunnableId::from_flow_version(version), + args, + run_query, + method == http::Method::GET, + ) + .await +} + pub async fn stream_script_by_path( authed: ApiAuthed, Extension(db): Extension, @@ -5423,20 +5597,10 @@ pub async fn stream_job( is_get: bool, ) -> error::Result { let args = if is_get { - let payload_r = run_query.payload.clone().map(decode_payload).map(|x| { - x.map_err(|e| { - Error::internal_err(format!("Impossible to decode query payload: {e:#?}")) - }) - }); - - let payload_args = if let Some(payload) = payload_r { - payload? - } else { - HashMap::new() - }; + let payload_as_args = run_query.payload_as_args()?; let mut args = args.process_args(&authed, &db, &w_id, None).await?; - args.body = args::Body::HashMap(payload_args); + args.body = args::Body::HashMap(payload_as_args); let args = args .to_args_from_runnable(&db, &w_id, runnable_id.clone(), run_query.skip_preprocessor) @@ -5483,7 +5647,7 @@ pub async fn stream_job( .await? .0 } - RunnableId::FlowPath(flow_path) => { + RunnableId::FlowId(FlowId::FlowPath(flow_path)) => { run_flow_by_path_inner( authed.clone(), db.clone(), @@ -5497,6 +5661,20 @@ pub async fn stream_job( .await? .0 } + RunnableId::FlowId(FlowId::FlowVersion(version)) => { + run_flow_by_version_inner( + authed.clone(), + db.clone(), + user_db, + w_id.clone(), + version, + run_query, + args, + None + ) + .await? + .0 + } }; let opt_authed = Some(authed.clone()); @@ -5552,101 +5730,137 @@ pub async fn run_wait_result_flow_by_path_internal( let flow_path = flow_path.to_path(); - let scheduled_for = run_query.get_scheduled_for(&db).await?; - let userdb_authed = UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() }; - let FlowVersionInfo { - tag, - dedicated_worker, - early_return, - has_preprocessor, - chat_input_enabled, - on_behalf_of_email, - edited_by, - version, - } = get_latest_flow_version_info_for_path(Some(userdb_authed), &db, &w_id, &flow_path, true) - .await?; + let flow_version_info = + get_latest_flow_version_info_for_path(Some(userdb_authed), &db, &w_id, &flow_path, true) + .await?; - let tag = run_query.tag.clone().or(tag); - check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?; - - let (email, permissioned_as, push_authed, tx) = - if let Some(on_behalf_of_email) = on_behalf_of_email.as_ref() { - ( - on_behalf_of_email, - username_to_permissioned_as(&edited_by), - None, - PushIsolationLevel::IsolatedRoot(db.clone()), - ) - } else { - ( - &authed.email, - username_to_permissioned_as(&authed.username), - Some(authed.clone().into()), - PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into()), - ) - }; - - let (uuid, mut tx) = push( + run_flow_and_wait_result( + &authed, &db, - tx, + user_db, &w_id, - JobPayload::Flow { - path: flow_path.to_string(), - dedicated_worker, - version, - apply_preprocessor: !run_query.skip_preprocessor.unwrap_or(false) - && has_preprocessor.unwrap_or(false), - }, - PushArgs { args: &args.args, extra: args.extra }, - authed.display_username(), - email, - permissioned_as, - authed.token_prefix.as_deref(), - scheduled_for, - None, - run_query.parent_job, - None, - run_query.root_job, - run_query.job_id, - false, - false, - None, - !run_query.invisible_to_owner.unwrap_or(false), - tag, - None, - None, - None, - push_authed.as_ref(), - false, - None, - None, + flow_path, + flow_version_info, + run_query, + args, None, ) + .await +} + +pub async fn run_wait_result_flow_by_version_get( + method: hyper::http::Method, + authed: ApiAuthed, + Extension(user_db): Extension, + Extension(db): Extension, + Path((w_id, version)): Path<(String, i64)>, + Query(run_query): Query, + args: RawWebhookArgs, +) -> error::Result { + #[cfg(feature = "enterprise")] + check_license_key_valid().await?; + + let flow_path = sqlx::query_scalar!( + "SELECT path FROM flow_version WHERE id = $1 AND workspace_id = $2", + version, + &w_id + ) + .fetch_one(&db) .await?; - // Set conversation_id if provided (for agent memory) - if let Some(memory_id) = run_query.memory_id { - set_flow_memory_id(&mut tx, uuid, memory_id).await?; + check_scopes(&authed, || format!("jobs:run:flows:{flow_path}"))?; + + if method == http::Method::HEAD { + return Ok(Json(serde_json::json!("")).into_response()); } - // Handle conversation messages for chat-enabled flows - if chat_input_enabled.unwrap_or(false) { - handle_chat_conversation_messages( - &mut tx, - &authed, + let payload_as_args = run_query.payload_as_args()?; + + let mut args = args.process_args(&authed, &db, &w_id, None).await?; + args.body = args::Body::HashMap(payload_as_args); + + let args = args + .to_args_from_runnable( + &db, &w_id, - &flow_path.to_string(), - &run_query, - args.args.get("user_message"), + RunnableId::from_flow_version(version), + run_query.skip_preprocessor, ) .await?; - } - tx.commit().await?; + let flow_version_info = + get_flow_version_info_from_version(&db, version, &w_id, &flow_path).await?; - run_wait_result(&db, uuid, w_id, early_return, &authed.username).await + run_flow_and_wait_result( + &authed, + &db, + user_db, + &w_id, + &flow_path, + flow_version_info, + run_query, + args, + None, + ) + .await +} + +pub async fn run_wait_result_flow_by_version( + authed: ApiAuthed, + Extension(user_db): Extension, + Extension(db): Extension, + Path((w_id, version)): Path<(String, i64)>, + Query(run_query): Query, + args: RawWebhookArgs, +) -> error::Result { + #[cfg(feature = "enterprise")] + check_license_key_valid().await?; + + let flow_path = sqlx::query_scalar!( + r#" + SELECT + path + FROM + flow_version + WHERE + id = $1 AND + workspace_id = $2 + "#, + version, + &w_id + ) + .fetch_one(&db) + .await?; + + check_scopes(&authed, || format!("jobs:run:flows:{flow_path}"))?; + + let args = args + .to_args_from_runnable( + &authed, + &db, + &w_id, + RunnableId::from_flow_version(version), + run_query.skip_preprocessor, + ) + .await?; + + let flow_version_info = + get_flow_version_info_from_version(&db, version, &w_id, &flow_path).await?; + + run_flow_and_wait_result( + &authed, + &db, + user_db, + &w_id, + &flow_path, + flow_version_info, + run_query, + args, + None, + ) + .await } async fn run_preview_script( @@ -5745,7 +5959,7 @@ async fn run_wait_result_preview_script( let uuid = uuid .parse::() .map_err(|_| Error::BadRequest("Invalid UUID".to_string()))?; - let result = run_wait_result(&db, uuid, w_id, None, &authed.username).await; + let result = run_wait_result(&db, uuid, &w_id, None, &authed.username).await; return result; } @@ -5983,7 +6197,7 @@ async fn run_dependencies_job( .await?; tx.commit().await?; - let wait_result = run_wait_result(&db, uuid, w_id, None, &authed.username).await; + let wait_result = run_wait_result(&db, uuid, &w_id, None, &authed.username).await; wait_result } @@ -6053,7 +6267,7 @@ async fn run_flow_dependencies_job( .await?; tx.commit().await?; - let wait_result = run_wait_result(&db, uuid, w_id, None, &authed.username).await; + let wait_result = run_wait_result(&db, uuid, &w_id, None, &authed.username).await; wait_result } @@ -6449,7 +6663,7 @@ async fn run_wait_result_preview_flow( let uuid = uuid .parse::() .map_err(|_| Error::BadRequest("Invalid UUID".to_string()))?; - let result = run_wait_result(&db, uuid, w_id, None, &authed.username).await; + let result = run_wait_result(&db, uuid, &w_id, None, &authed.username).await; return result; } diff --git a/backend/windmill-api/src/triggers/trigger_helpers.rs b/backend/windmill-api/src/triggers/trigger_helpers.rs index d72a88e773..e592c43bb3 100644 --- a/backend/windmill-api/src/triggers/trigger_helpers.rs +++ b/backend/windmill-api/src/triggers/trigger_helpers.rs @@ -24,7 +24,6 @@ use windmill_common::{ users::username_to_permissioned_as, utils::StripPath, worker::to_raw_value, - FlowVersionInfo, }; use windmill_queue::{push, PushArgs, PushArgsOwned, PushIsolationLevel}; @@ -59,7 +58,7 @@ struct PartialSchema { #[derive(Debug, Clone, Eq, PartialEq, Hash)] pub enum RunnableId { - FlowPath(String), + FlowId(FlowId), ScriptId(ScriptId), HubScript(String), } @@ -78,7 +77,33 @@ impl RunnableId { } pub fn from_flow_path(path: &str) -> Self { - Self::FlowPath(path.to_string()) + Self::FlowId(FlowId::FlowPath(path.to_string())) + } + + pub fn from_flow_version(version: i64) -> Self { + Self::FlowId(FlowId::FlowVersion(version)) + } +} + +#[derive(Debug, Clone, Eq, PartialEq, Hash)] +pub enum FlowId { + FlowPath(String), + FlowVersion(i64), +} + +impl FlowId { + async fn get_flow_version_id(self, workspace_id: &str, db: &DB) -> Result { + let version_id = match self { + FlowId::FlowPath(path) => { + let info = + get_latest_flow_version_info_for_path(None, db, workspace_id, &path, true) + .await?; + info.version + } + FlowId::FlowVersion(version) => version, + }; + + Ok(version_id) } } @@ -258,9 +283,8 @@ pub async fn get_runnable_format( }, ) } - RunnableId::FlowPath(path) => { - let FlowVersionInfo { version, .. } = - get_latest_flow_version_info_for_path(None, &db, workspace_id, &path, true).await?; + RunnableId::FlowId(flow_id) => { + let version = flow_id.get_flow_version_id(workspace_id, db).await?; let key = ( HubOrWorkspaceId::WorkspaceId(workspace_id.to_string()), @@ -271,7 +295,7 @@ pub async fn get_runnable_format( let runnable_format = RUNNABLE_FORMAT_VERSION_CACHE.get(&key); if let Some(runnable_format) = runnable_format { - tracing::debug!("Using cached runnable format for flow {path}"); + tracing::debug!("Using cached runnable format for flow version {version}"); return Ok(runnable_format); } @@ -282,11 +306,9 @@ pub async fn get_runnable_format( schema as \"schema: _\" FROM flow_version WHERE - path = $1 - AND workspace_id = $2 - ORDER BY created_at DESC - LIMIT 1", - path, + id = $1 + AND workspace_id = $2", + version, workspace_id, ) .fetch_one(db) @@ -614,8 +636,7 @@ pub async fn trigger_runnable_and_wait_for_result( ) .await?; let (result, success) = - run_wait_result_internal(db, uuid, workspace_id.to_string(), early_return, &username) - .await?; + run_wait_result_internal(db, uuid, &workspace_id, early_return, &username).await?; if delete_after_use.unwrap_or(false) { delete_job_metadata_after_use(&db, uuid).await?; @@ -658,7 +679,7 @@ pub async fn trigger_runnable_and_wait_for_raw_result( .await?; let (result, success) = - run_wait_result_internal(db, uuid, workspace_id.to_string(), early_return, &username) + run_wait_result_internal(db, uuid, &workspace_id, early_return, &username) .await .with_context(|| { format!( diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 929736eff5..5b81d35f75 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -66,6 +66,7 @@ pub struct FlowWithStarred { pub starred: Option, #[serde(skip_serializing_if = "Option::is_none")] pub lock_error_logs: Option, + pub version_id: i64, } fn is_none_or_false(b: &Option) -> bool { diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index b0a3f5e4ea..b91dcfaf95 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -769,7 +769,7 @@ where } } -pub fn get_latest_flow_version_info_for_path_from_version< +pub fn get_flow_version_info_from_version< 'a, 'e, A: sqlx::Acquire<'e, Database = Postgres> + Send + 'a, @@ -782,7 +782,6 @@ pub fn get_latest_flow_version_info_for_path_from_version< async move { // as instructed in the docstring of sqlx::Acquire let key = (w_id.to_string(), version); - match FLOW_INFO_CACHE.get(&key) { Some(info) => { tracing::debug!("Using cached flow version info for {version} ({path})"); @@ -791,21 +790,37 @@ pub fn get_latest_flow_version_info_for_path_from_version< _ => { tracing::debug!("Fetching flow version info for {version} ({path})"); let mut conn = db.acquire().await?; - let info = sqlx::query_as!( - FlowVersionInfo, - "SELECT tag, dedicated_worker, flow_version.value->>'early_return' as early_return, flow_version.value->>'preprocessor_module' IS NOT NULL as has_preprocessor, (flow_version.value->>'chat_input_enabled')::boolean as chat_input_enabled, on_behalf_of_email, edited_by, flow_version.id AS version - FROM flow - INNER JOIN flow_version - ON flow_version.id = $3 - WHERE flow.path = $1 and flow.workspace_id = $2", - path, - w_id, - version - ) - .fetch_optional(&mut *conn) - .await?; + let flow_info = + sqlx::query_as!( + FlowVersionInfo, + r#" + SELECT + flow_version.id AS version, + flow_version.value->>'early_return' as early_return, + flow_version.value->>'preprocessor_module' IS NOT NULL as has_preprocessor, + (flow_version.value->>'chat_input_enabled')::boolean as chat_input_enabled, + flow.tag, + flow.dedicated_worker, + flow.on_behalf_of_email, + flow.edited_by + FROM + flow_version + INNER JOIN flow + ON flow.path = flow_version.path AND + flow.workspace_id = flow_version.workspace_id + WHERE + flow_version.workspace_id = $1 AND + flow_version.path = $2 AND + flow_version.id = $3 + "#, + w_id, + path, + version, + ) + .fetch_optional(&mut *conn) + .await?; - let info = utils::not_found_if_none(info, "flow", path)?; + let info = utils::not_found_if_none(flow_info, "flow", path)?; FLOW_INFO_CACHE.insert(key, info.clone()); @@ -825,7 +840,7 @@ pub async fn get_latest_flow_version_info_for_path<'e>( // as instructed in the docstring of sqlx::Acquire let version = get_latest_flow_version_id_for_path(db_authed, &db.clone(), w_id, path, use_cache).await?; - get_latest_flow_version_info_for_path_from_version(db, version, w_id, path).await + get_flow_version_info_from_version(db, version, w_id, path).await } async fn get_latest_flow_version_for_path<'e, E: sqlx::PgExecutor<'e>>( diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index b9bfb0fe58..52a397ce33 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -17,8 +17,8 @@ use std::str::FromStr; use windmill_common::db::Authed; use windmill_common::ee_oss::LICENSE_KEY_VALID; use windmill_common::flows::Retry; +use windmill_common::get_flow_version_info_from_version; use windmill_common::get_latest_flow_version_id_for_path; -use windmill_common::get_latest_flow_version_info_for_path_from_version; use windmill_common::jobs::check_tag_available_for_workspace_internal; use windmill_common::jobs::JobPayload; use windmill_common::schedule::schedule_to_user; @@ -63,7 +63,7 @@ async fn get_schedule_metadata<'c>( .await?; let FlowVersionInfo { tag, on_behalf_of_email, edited_by, .. } = - get_latest_flow_version_info_for_path_from_version( + get_flow_version_info_from_version( &mut **tx, version, &schedule.workspace_id, @@ -291,16 +291,13 @@ pub async fn push_scheduled_job<'c>( let FlowVersionInfo { version, tag, dedicated_worker, on_behalf_of_email, edited_by, .. - } = get_latest_flow_version_info_for_path_from_version( + } = get_flow_version_info_from_version( &mut *tx, version, &schedule.workspace_id, &schedule.script_path, ) - .warn_after_seconds_with_sql( - 1, - "get_latest_flow_version_info_for_path_from_version".to_string(), - ) + .warn_after_seconds_with_sql(1, "get_flow_version_info_from_version".to_string()) .await?; ( diff --git a/frontend/src/lib/components/ScriptBuilder.svelte b/frontend/src/lib/components/ScriptBuilder.svelte index 03bbe541c1..df82c244ab 100644 --- a/frontend/src/lib/components/ScriptBuilder.svelte +++ b/frontend/src/lib/components/ScriptBuilder.svelte @@ -1668,7 +1668,7 @@ args={hasPreprocessor && selectedInputTab !== 'preprocessor' ? {} : args} isDeployed={savedScript && !savedScript?.draft_only} schema={script.schema} - hash={script.parent_hash} + runnableVersion={script.parent_hash} onDeployTrigger={handleDeployTrigger} /> diff --git a/frontend/src/lib/components/triggers/TriggersEditor.svelte b/frontend/src/lib/components/triggers/TriggersEditor.svelte index c04362f347..2eae503d3e 100644 --- a/frontend/src/lib/components/triggers/TriggersEditor.svelte +++ b/frontend/src/lib/components/triggers/TriggersEditor.svelte @@ -41,7 +41,7 @@ newItem?: boolean currentPath: string fakeInitialPath?: string - hash?: string | undefined + runnableVersion?: string | undefined args?: Record initialPath: string isFlow: boolean @@ -59,7 +59,7 @@ newItem = false, currentPath, fakeInitialPath = '', - hash = undefined, + runnableVersion = undefined, args = {}, initialPath, isFlow, @@ -363,7 +363,7 @@ {initialPath} {fakeInitialPath} {currentPath} - {hash} + {runnableVersion} {isDeployed} small={useVerticalTriggerBar} {args} @@ -416,7 +416,7 @@ {hasPreprocessor} {canHavePreprocessor} args={config} - data={{ args, hash, emailDomain }} + data={{ args, hash: !isFlow ? runnableVersion : undefined, emailDomain }} {isValid} triggerDeployed={!triggersState.selectedTrigger.isDraft} on:applyArgs diff --git a/frontend/src/lib/components/triggers/TriggersWrapper.svelte b/frontend/src/lib/components/triggers/TriggersWrapper.svelte index a6be555a42..5030069f10 100644 --- a/frontend/src/lib/components/triggers/TriggersWrapper.svelte +++ b/frontend/src/lib/components/triggers/TriggersWrapper.svelte @@ -24,7 +24,7 @@ initialPath: string fakeInitialPath: string currentPath: string - hash?: string + runnableVersion?: string isDeployed: boolean small: boolean args: Record @@ -45,7 +45,7 @@ initialPath, fakeInitialPath, currentPath, - hash, + runnableVersion, small, args, newItem, @@ -68,7 +68,7 @@ {:else if selectedTrigger.type === 'schedule'} diff --git a/frontend/src/lib/components/triggers/email/DefaultEmailConfigSection.svelte b/frontend/src/lib/components/triggers/email/DefaultEmailConfigSection.svelte index 0ed7d5b4ae..0aebee6680 100644 --- a/frontend/src/lib/components/triggers/email/DefaultEmailConfigSection.svelte +++ b/frontend/src/lib/components/triggers/email/DefaultEmailConfigSection.svelte @@ -14,12 +14,12 @@ import UserSettings from '$lib/components/UserSettings.svelte' import TextInput from '$lib/components/text_input/TextInput.svelte' - let requestType: 'hash' | 'path' = $state('path') + let requestType: 'runnableVersion' | 'path' = $state('path') function emailAddress() { - const pathOrHash = requestType === 'hash' ? hash : path.replaceAll('/', '.') + const pathOrHash = requestType === 'runnableVersion' ? runnableVersion : path.replaceAll('/', '.') const plainPrefix = `${$workspaceStore}+${ - (requestType === 'hash' ? 'hash.' : isFlow ? 'flow.' : '') + pathOrHash + (requestType === 'runnableVersion' ? 'hash.' : isFlow ? 'flow.' : '') + pathOrHash }+${token}` const encodedPrefix = base32 .stringify(new TextEncoder().encode(plainPrefix), { @@ -32,7 +32,7 @@ interface Props { token?: string isFlow?: boolean - hash?: string | undefined + runnableVersion?: string | undefined path: string userSettings: UserSettings emailDomain?: string | null @@ -42,7 +42,7 @@ let { token = $bindable(''), isFlow = false, - hash = undefined, + runnableVersion = undefined, path, userSettings, emailDomain = null, @@ -90,7 +90,7 @@ {#snippet children({ item })} - + {/snippet} diff --git a/frontend/src/lib/components/triggers/email/DefaultEmailPanel.svelte b/frontend/src/lib/components/triggers/email/DefaultEmailPanel.svelte index 3859237cdd..e8c201ff63 100644 --- a/frontend/src/lib/components/triggers/email/DefaultEmailPanel.svelte +++ b/frontend/src/lib/components/triggers/email/DefaultEmailPanel.svelte @@ -16,7 +16,7 @@ token: string scopes?: string[] isFlow?: boolean - hash?: string | undefined + runnableVersion?: string | undefined path: string onEmailDomain: (domain: string) => void } @@ -25,7 +25,7 @@ token = $bindable(), scopes = [], isFlow = false, - hash = undefined, + runnableVersion = undefined, path, onEmailDomain }: Props = $props() @@ -70,7 +70,7 @@ {:else} {#if emailDomain} - + {:else}
diff --git a/frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte b/frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte index 15027bb15e..bbd7ae9601 100644 --- a/frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte +++ b/frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte @@ -26,7 +26,7 @@ interface Props { isFlow?: boolean path?: string - hash?: string | undefined + runnableVersion?: string | undefined token?: string runnableArgs: any triggerTokens?: TriggerTokens | undefined @@ -36,46 +36,38 @@ let { isFlow = false, path = '', - hash = undefined, + runnableVersion = undefined, token = $bindable(''), runnableArgs, triggerTokens = $bindable(undefined), scopes = [] }: Props = $props() - let webhooks: { - async: { - get: {} - post: { - hash?: string - path: string - } + const WEBHOOK_BASE_URL = `${location.origin}${base}/api/w/${$workspaceStore}/jobs` + + let baseWebhookUrl = $derived.by(() => { + let webhookUrlPath: string + + if (isFlow) { + webhookUrlPath = runnableId == 'path' ? `f/${path}` : `fv/${runnableVersion}` + } else { + webhookUrlPath = runnableId == 'path' ? `p/${path}` : `h/${runnableVersion}` } - sync: { - get: { - path: string - } - post: { - hash?: string - path: string - } + + if (requestType == 'async') { + return `${WEBHOOK_BASE_URL}/run/${webhookUrlPath}` + } else if (requestType == 'sync') { + return `${WEBHOOK_BASE_URL}/run_wait_result/${webhookUrlPath}` + } else { + return `${WEBHOOK_BASE_URL}/run_and_stream/${webhookUrlPath}` } - sync_sse: { - get: { - hash?: string - path: string - } - post: { - hash?: string - path: string - } - } - } = $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' | 'sync_sse' let callMethod = $state('post') as 'get' | 'post' - let runnableId = $state('path') as 'hash' | 'path' + let runnableId = $state('path') as 'runnableVersion' | 'path' let tokenType = $state('headers') as 'query' | 'headers' $effect(() => { @@ -91,7 +83,7 @@ : runnableArgs }) let url: string = $derived( - webhooks[requestType][callMethod][runnableId] + + baseWebhookUrl + (tokenType === 'query' ? `?token=${token}${ callMethod === 'get' || requestType === 'sync_sse' @@ -105,70 +97,6 @@ }`) ) - function computeScriptWebhooks(hash: string | undefined, path: string) { - let webhookBase = `${location.origin}${base}/api/w/${$workspaceStore}/jobs` - return { - async: { - get: {}, - post: { - hash: `${webhookBase}/run/h/${hash}`, - path: `${webhookBase}/run/p/${path}` - } - }, - sync: { - get: { - path: `${webhookBase}/run_wait_result/p/${path}` - }, - 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}` - }, - post: { - hash: `${webhookBase}/run_and_stream/h/${hash}`, - path: `${webhookBase}/run_and_stream/p/${path}` - } - } - } - } - - function computeFlowWebhooks(path: string) { - let webhooksBase = `${location.origin}${base}/api/w/${$workspaceStore}/jobs` - - let urlAsync = `${webhooksBase}/run/f/${path}` - let urlSync = `${webhooksBase}/run_wait_result/f/${path}` - let urlStream = `${webhooksBase}/run_and_stream/f/${path}` - return { - async: { - get: {}, - post: { - path: urlAsync - } - }, - sync: { - get: { - path: urlSync - }, - post: { - path: urlSync - } - }, - sync_sse: { - get: { - path: urlStream - }, - post: { - path: urlStream - } - } - } - } - function headers() { const headers = {} if (callMethod === 'post') { @@ -433,16 +361,19 @@ done` {/snippet} - {#if !isFlow} - - {/if} + {/if} - + diff --git a/frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte b/frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte index 4df466ce49..43eeab456b 100644 --- a/frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte @@ -675,6 +675,7 @@ {#if flow}