starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-21 16:02:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: webhook by flow version (#7062)

Browse Source 
* ok

* done

* update sqlx and repo ref

* fix
This commit is contained in:
dieriba
2025-11-14 15:36:24 +01:00
committed by GitHub
parent 3dd293e450
commit 09cdfb4556
25 changed files with 853 additions and 380 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json
Generated
+1 -1
View File
@@ -15,7 +15,7 @@
]
},
"nullable": [
null
true
]
},
"hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55"
backend/.sqlx/query-7cab8475996f0abd7ca7888b460bd3faf3bef6e34200c258e3704981817c241e.json
Generated
+23
View File
@@ -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"
}
backend/.sqlx/query-a7468e9054beed88636786c5495ac3b9d9a6086ae6212ad9237b39a0346d7d26.json
Generated
+66
View File
@@ -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"
}
backend/.sqlx/query-b82f97f2a858e20b1fb6d9b0b3eab5bab2bafa91689eac5073c1678c338afb27.json
Generated
+23
View File
@@ -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"
}
backend/.sqlx/query-e1409c67b93881cf68f6ac9c8bae0856cf426c7e7860c8b5b799972baa8e6945.json
Generated
-66
View File
@@ -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"
}
backend/.sqlx/query-e2474b7855c8b08f927f2b987421e773e537a8eb0a113477764bbab14e1f3a3d.json → backend/.sqlx/query-f0c9670c27d9a3b8e8f316fd28f7d3ede2ecc3225ff2879ff5dfd59c4ef236d5.json
Generated
+3 -3
View File
@@ -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"
}
backend/.sqlx/query-fc9753f501974c4b570d710c7621ff1f6787a852bb27acf49593af963a6aacca.json
Generated
+23
View File
@@ -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"
}
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
a0471e9f11c9cc181cfd9eb2738858f7bebc30ee
31ee9d3449f05cd0328c0fcf43e2b161dee767b9
backend/tests/worker.rs
+1 -1
View File
@@ -2769,7 +2769,7 @@ async fn test_result_format(db: Pool<Postgres>) -> 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",
)
backend/windmill-api/openapi.yaml
+222
View File
@@ -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
backend/windmill-api/src/flows.rs
+3 -1
View File
@@ -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
backend/windmill-api/src/jobs.rs
+356 -142
View File
@@ -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<HashMap<String, Box<RawValue>>> {
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<JobTriggerKind>,
) -> error::Result<(Uuid, Option<String>)> {
#[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<JobTriggerKind>,
) -> error::Result<Response> {
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<JobTriggerKind>,
) -> error::Result<(Uuid, Option<String>)> {
#[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<DB>,
Extension(user_db): Extension<UserDB>,
Path((w_id, version)): Path<(String, i64)>,
Query(run_query): Query<RunJobQuery>,
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<JobTriggerKind>,
) -> error::Result<(Uuid, Option<String>)> {
#[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<String>,
username: &str,
) -> error::Result<(Box<RawValue>, 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<RawValue>, 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<String>,
username: &str,
) -> error::Result<Response> {
@@ -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<DB>,
Extension(user_db): Extension<UserDB>,
Path((w_id, version)): Path<(String, i64)>,
Query(run_query): Query<RunJobQuery>,
method: hyper::http::Method,
args: RawWebhookArgs,
) -> error::Result<Response> {
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<DB>,
@@ -5423,20 +5597,10 @@ pub async fn stream_job(
is_get: bool,
) -> error::Result<Response> {
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<UserDB>,
Extension(db): Extension<DB>,
Path((w_id, version)): Path<(String, i64)>,
Query(run_query): Query<RunJobQuery>,
args: RawWebhookArgs,
) -> error::Result<Response> {
#[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<UserDB>,
Extension(db): Extension<DB>,
Path((w_id, version)): Path<(String, i64)>,
Query(run_query): Query<RunJobQuery>,
args: RawWebhookArgs,
) -> error::Result<Response> {
#[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::<Uuid>()
.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::<Uuid>()
.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;
}
backend/windmill-api/src/triggers/trigger_helpers.rs
+36 -15
View File
@@ -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<i64> {
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!(
backend/windmill-common/src/flows.rs
+1
View File
@@ -66,6 +66,7 @@ pub struct FlowWithStarred {
pub starred: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub lock_error_logs: Option<String>,
pub version_id: i64,
}
fn is_none_or_false(b: &Option<bool>) -> bool {
backend/windmill-common/src/lib.rs
+32 -17
View File
@@ -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>>(
backend/windmill-queue/src/schedule.rs
+4 -7
View File
@@ -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?;
(
frontend/src/lib/components/ScriptBuilder.svelte
+1 -1
View File
@@ -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}
/>
frontend/src/lib/components/triggers/TriggersEditor.svelte
+4 -4
View File
@@ -41,7 +41,7 @@
newItem?: boolean
currentPath: string
fakeInitialPath?: string
hash?: string | undefined
runnableVersion?: string | undefined
args?: Record<string, any>
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
frontend/src/lib/components/triggers/TriggersWrapper.svelte
+4 -4
View File
@@ -24,7 +24,7 @@
initialPath: string
fakeInitialPath: string
currentPath: string
hash?: string
runnableVersion?: string
isDeployed: boolean
small: boolean
args: Record<string, any>
@@ -45,7 +45,7 @@
initialPath,
fakeInitialPath,
currentPath,
hash,
runnableVersion,
small,
args,
newItem,
@@ -68,7 +68,7 @@
<WebhooksPanel
{isFlow}
path={initialPath || fakeInitialPath}
{hash}
{runnableVersion}
token=""
{args}
scopes={isFlow ? [`jobs:run:flows:${currentPath}`] : [`jobs:run:scripts:${currentPath}`]}
@@ -80,7 +80,7 @@
scopes={isFlow ? [`jobs:run:flows:${currentPath}`] : [`jobs:run:scripts:${currentPath}`]}
path={initialPath || fakeInitialPath}
{isFlow}
{hash}
runnableVersion={!isFlow ? runnableVersion : undefined}
{onEmailDomain}
/>
{:else if selectedTrigger.type === 'schedule'}
frontend/src/lib/components/triggers/email/DefaultEmailConfigSection.svelte
+6 -6
View File
@@ -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 @@
<ToggleButtonGroup class="w-auto" bind:selected={requestType}>
{#snippet children({ item })}
<ToggleButton label="By path" value="path" {item} />
<ToggleButton label="By hash" value="hash" {item} />
<ToggleButton label="By hash" value="runnableVersion" {item} />
{/snippet}
</ToggleButtonGroup>
</div>
frontend/src/lib/components/triggers/email/DefaultEmailPanel.svelte
+3 -3
View File
@@ -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 @@
<Skeleton layout={[[18]]} />
{:else}
{#if emailDomain}
<DefaultEmailConfigSection {hash} {token} {path} {isFlow} {userSettings} {emailDomain} />
<DefaultEmailConfigSection {runnableVersion} {token} {path} {isFlow} {userSettings} {emailDomain} />
{:else}
<div>
<Alert title="Email triggers are disabled" size="xs" type="warning">
frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte
+35 -104
View File
@@ -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}
</ToggleButtonGroup>
</Label>
{#if !isFlow}
<Label label="Reference type">
<ToggleButtonGroup bind:selected={runnableId}>
{#snippet children({ item })}
<ToggleButton label="Path" value="path" {item} />
<ToggleButton label="Hash" value="hash" disabled={!hash} {item} />
{/snippet}
</ToggleButtonGroup>
</Label>
{/if}
<Label label="Reference type">
<ToggleButtonGroup bind:selected={runnableId}>
{#snippet children({ item })}
<ToggleButton label="Path" value="path" {item} />
<ToggleButton
label={isFlow ? 'Flow version' : 'Hash'}
value="runnableVersion"
disabled={!runnableVersion}
{item}
/>
{/snippet}
</ToggleButtonGroup>
</Label>
<Label label="Token configuration">
<ToggleButtonGroup bind:selected={tokenType}>
{#snippet children({ item })}
frontend/src/lib/components/triggers/webhook/WebhooksPanel.svelte
+3 -3
View File
@@ -10,7 +10,7 @@
args?: Record<string, any>
scopes?: string[]
isFlow?: boolean
hash?: string | undefined
runnableVersion?: string | undefined
path: string
newItem?: boolean
}
@@ -20,7 +20,7 @@
args = {},
scopes = [],
isFlow = false,
hash = undefined,
runnableVersion = undefined,
path,
newItem = false
}: Props = $props()
@@ -39,5 +39,5 @@
{isFlow ? 'flow' : 'script'}.
</Alert>
{/if}
<WebhooksConfigSection {isFlow} {path} {hash} {token} runnableArgs={args} {scopes} />
<WebhooksConfigSection {isFlow} {path} {runnableVersion} {token} runnableArgs={args} {scopes} />
</Section>
frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte
+1
View File
@@ -675,6 +675,7 @@
{#if flow}
<TriggersEditor
{args}
runnableVersion={flow.version_id?.toString()}
initialPath={flow.path}
currentPath={flow.path}
noEditor={true}
frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte
+1 -1
View File
@@ -791,7 +791,7 @@
{#if script}
<TriggersEditor
{args}
hash={script.hash}
runnableVersion={script.hash}
initialPath={script.path}
currentPath={script.path}
noEditor={true}