diff --git a/Dockerfile b/Dockerfile index 8683aa9738..cff245d3e4 100644 --- a/Dockerfile +++ b/Dockerfile @@ -175,9 +175,9 @@ RUN /usr/local/bin/python3 -m pip install pip-tools COPY --from=builder /frontend/build /static_frontend COPY --from=builder /windmill/target/release/windmill ${APP}/windmill -COPY --from=denoland/deno:2.0.2 --chmod=755 /usr/bin/deno /usr/bin/deno +COPY --from=denoland/deno:2.0.4 --chmod=755 /usr/bin/deno /usr/bin/deno -COPY --from=oven/bun:1.1.32 /usr/local/bin/bun /usr/bin/bun +COPY --from=oven/bun:1.1.34 /usr/local/bin/bun /usr/bin/bun COPY --from=php:8.3.7-cli /usr/local/bin/php /usr/bin/php COPY --from=composer:2.7.6 /usr/bin/composer /usr/bin/composer diff --git a/backend/.sqlx/query-acbf74cf3302bfcf7615285070d3f8958932bb8a2dda715f1b9152ab44442780.json b/backend/.sqlx/query-0b94bd4c98a11ca1b7e5e34dd1ee6fcb0b7a54ed4218fa3cf23cc929d009d50f.json similarity index 50% rename from backend/.sqlx/query-acbf74cf3302bfcf7615285070d3f8958932bb8a2dda715f1b9152ab44442780.json rename to backend/.sqlx/query-0b94bd4c98a11ca1b7e5e34dd1ee6fcb0b7a54ed4218fa3cf23cc929d009d50f.json index eef46ce41d..75288e80ae 100644 --- a/backend/.sqlx/query-acbf74cf3302bfcf7615285070d3f8958932bb8a2dda715f1b9152ab44442780.json +++ b/backend/.sqlx/query-0b94bd4c98a11ca1b7e5e34dd1ee6fcb0b7a54ed4218fa3cf23cc929d009d50f.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "UPDATE websocket_trigger SET url = $1, script_path = $2, path = $3, is_flow = $4, filters = $5, edited_by = $6, email = $7, edited_at = now(), server_id = NULL, last_server_ping = NULL, error = NULL\n WHERE workspace_id = $8 AND path = $9", + "query": "UPDATE websocket_trigger SET url = $1, script_path = $2, path = $3, is_flow = $4, filters = $5, initial_messages = $6, url_runnable_args = $7, edited_by = $8, email = $9, edited_at = now(), server_id = NULL, last_server_ping = NULL, error = NULL\n WHERE workspace_id = $10 AND path = $11", "describe": { "columns": [], "parameters": { @@ -10,6 +10,8 @@ "Varchar", "Bool", "JsonbArray", + "JsonbArray", + "Jsonb", "Varchar", "Varchar", "Text", @@ -18,5 +20,5 @@ }, "nullable": [] }, - "hash": "acbf74cf3302bfcf7615285070d3f8958932bb8a2dda715f1b9152ab44442780" + "hash": "0b94bd4c98a11ca1b7e5e34dd1ee6fcb0b7a54ed4218fa3cf23cc929d009d50f" } diff --git a/backend/.sqlx/query-4e9668a46bad9e82baa51422946d373b18b6577198df7545c94bd19be3446775.json b/backend/.sqlx/query-4e9668a46bad9e82baa51422946d373b18b6577198df7545c94bd19be3446775.json deleted file mode 100644 index 2f96d31ecb..0000000000 --- a/backend/.sqlx/query-4e9668a46bad9e82baa51422946d373b18b6577198df7545c94bd19be3446775.json +++ /dev/null @@ -1,108 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO websocket_trigger (workspace_id, path, url, script_path, is_flow, enabled, filters, edited_by, email, edited_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, now()) RETURNING *", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "path", - "type_info": "Varchar" - }, - { - "ordinal": 1, - "name": "url", - "type_info": "Varchar" - }, - { - "ordinal": 2, - "name": "script_path", - "type_info": "Varchar" - }, - { - "ordinal": 3, - "name": "is_flow", - "type_info": "Bool" - }, - { - "ordinal": 4, - "name": "workspace_id", - "type_info": "Varchar" - }, - { - "ordinal": 5, - "name": "edited_by", - "type_info": "Varchar" - }, - { - "ordinal": 6, - "name": "email", - "type_info": "Varchar" - }, - { - "ordinal": 7, - "name": "edited_at", - "type_info": "Timestamptz" - }, - { - "ordinal": 8, - "name": "extra_perms", - "type_info": "Jsonb" - }, - { - "ordinal": 9, - "name": "server_id", - "type_info": "Varchar" - }, - { - "ordinal": 10, - "name": "last_server_ping", - "type_info": "Timestamptz" - }, - { - "ordinal": 11, - "name": "error", - "type_info": "Text" - }, - { - "ordinal": 12, - "name": "enabled", - "type_info": "Bool" - }, - { - "ordinal": 13, - "name": "filters", - "type_info": "JsonbArray" - } - ], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Varchar", - "Varchar", - "Bool", - "Bool", - "JsonbArray", - "Varchar", - "Varchar" - ] - }, - "nullable": [ - false, - false, - false, - false, - false, - false, - false, - false, - false, - true, - true, - true, - false, - false - ] - }, - "hash": "4e9668a46bad9e82baa51422946d373b18b6577198df7545c94bd19be3446775" -} diff --git a/backend/.sqlx/query-5303cb9dd5903aa4791ef8e5e5881a50a832e65c8c9632e2e12cd9c2747f2fc7.json b/backend/.sqlx/query-5303cb9dd5903aa4791ef8e5e5881a50a832e65c8c9632e2e12cd9c2747f2fc7.json deleted file mode 100644 index a95b00b8c8..0000000000 --- a/backend/.sqlx/query-5303cb9dd5903aa4791ef8e5e5881a50a832e65c8c9632e2e12cd9c2747f2fc7.json +++ /dev/null @@ -1,98 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT *\n FROM websocket_trigger\n WHERE enabled IS TRUE AND (server_id IS NULL OR last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds')", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "path", - "type_info": "Varchar" - }, - { - "ordinal": 1, - "name": "url", - "type_info": "Varchar" - }, - { - "ordinal": 2, - "name": "script_path", - "type_info": "Varchar" - }, - { - "ordinal": 3, - "name": "is_flow", - "type_info": "Bool" - }, - { - "ordinal": 4, - "name": "workspace_id", - "type_info": "Varchar" - }, - { - "ordinal": 5, - "name": "edited_by", - "type_info": "Varchar" - }, - { - "ordinal": 6, - "name": "email", - "type_info": "Varchar" - }, - { - "ordinal": 7, - "name": "edited_at", - "type_info": "Timestamptz" - }, - { - "ordinal": 8, - "name": "extra_perms", - "type_info": "Jsonb" - }, - { - "ordinal": 9, - "name": "server_id", - "type_info": "Varchar" - }, - { - "ordinal": 10, - "name": "last_server_ping", - "type_info": "Timestamptz" - }, - { - "ordinal": 11, - "name": "error", - "type_info": "Text" - }, - { - "ordinal": 12, - "name": "enabled", - "type_info": "Bool" - }, - { - "ordinal": 13, - "name": "filters", - "type_info": "JsonbArray" - } - ], - "parameters": { - "Left": [] - }, - "nullable": [ - false, - false, - false, - false, - false, - false, - false, - false, - false, - true, - true, - true, - false, - false - ] - }, - "hash": "5303cb9dd5903aa4791ef8e5e5881a50a832e65c8c9632e2e12cd9c2747f2fc7" -} diff --git a/backend/.sqlx/query-cd33a9d63f4706a7e3b1e23cd0a4b2e3ecb30aae6510d1fcd08493b07c8b0952.json b/backend/.sqlx/query-cd33a9d63f4706a7e3b1e23cd0a4b2e3ecb30aae6510d1fcd08493b07c8b0952.json new file mode 100644 index 0000000000..36b1323980 --- /dev/null +++ b/backend/.sqlx/query-cd33a9d63f4706a7e3b1e23cd0a4b2e3ecb30aae6510d1fcd08493b07c8b0952.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE websocket_trigger SET enabled = FALSE, error = $1, server_id = NULL, last_server_ping = NULL WHERE workspace_id = $2 AND path = $3", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "cd33a9d63f4706a7e3b1e23cd0a4b2e3ecb30aae6510d1fcd08493b07c8b0952" +} diff --git a/backend/.sqlx/query-f2baee15e6d1fecd6d2d7b39fda1b50a15ec8bd349f1081af54da5dc2f5e3021.json b/backend/.sqlx/query-f2baee15e6d1fecd6d2d7b39fda1b50a15ec8bd349f1081af54da5dc2f5e3021.json deleted file mode 100644 index 2e75b4e74b..0000000000 --- a/backend/.sqlx/query-f2baee15e6d1fecd6d2d7b39fda1b50a15ec8bd349f1081af54da5dc2f5e3021.json +++ /dev/null @@ -1,101 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT *\n FROM websocket_trigger\n WHERE workspace_id = $1 AND path = $2", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "path", - "type_info": "Varchar" - }, - { - "ordinal": 1, - "name": "url", - "type_info": "Varchar" - }, - { - "ordinal": 2, - "name": "script_path", - "type_info": "Varchar" - }, - { - "ordinal": 3, - "name": "is_flow", - "type_info": "Bool" - }, - { - "ordinal": 4, - "name": "workspace_id", - "type_info": "Varchar" - }, - { - "ordinal": 5, - "name": "edited_by", - "type_info": "Varchar" - }, - { - "ordinal": 6, - "name": "email", - "type_info": "Varchar" - }, - { - "ordinal": 7, - "name": "edited_at", - "type_info": "Timestamptz" - }, - { - "ordinal": 8, - "name": "extra_perms", - "type_info": "Jsonb" - }, - { - "ordinal": 9, - "name": "server_id", - "type_info": "Varchar" - }, - { - "ordinal": 10, - "name": "last_server_ping", - "type_info": "Timestamptz" - }, - { - "ordinal": 11, - "name": "error", - "type_info": "Text" - }, - { - "ordinal": 12, - "name": "enabled", - "type_info": "Bool" - }, - { - "ordinal": 13, - "name": "filters", - "type_info": "JsonbArray" - } - ], - "parameters": { - "Left": [ - "Text", - "Text" - ] - }, - "nullable": [ - false, - false, - false, - false, - false, - false, - false, - false, - false, - true, - true, - true, - false, - false - ] - }, - "hash": "f2baee15e6d1fecd6d2d7b39fda1b50a15ec8bd349f1081af54da5dc2f5e3021" -} diff --git a/backend/migrations/20241030150214_ws_auth.down.sql b/backend/migrations/20241030150214_ws_auth.down.sql new file mode 100644 index 0000000000..6fed6a173b --- /dev/null +++ b/backend/migrations/20241030150214_ws_auth.down.sql @@ -0,0 +1,2 @@ +-- Add down migration script here +ALTER TABLE websocket_trigger DROP COLUMN initial_messages, DROP COLUMN url_runnable_args; \ No newline at end of file diff --git a/backend/migrations/20241030150214_ws_auth.up.sql b/backend/migrations/20241030150214_ws_auth.up.sql new file mode 100644 index 0000000000..ddefe851e5 --- /dev/null +++ b/backend/migrations/20241030150214_ws_auth.up.sql @@ -0,0 +1,2 @@ +-- Add up migration script here +ALTER TABLE websocket_trigger ADD COLUMN initial_messages JSONB[] DEFAULT '{}', ADD COLUMN url_runnable_args JSONB DEFAULT '{}'; \ No newline at end of file diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 0949843665..2c5a8cec85 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -11625,6 +11625,12 @@ components: required: - key - value + initial_messages: + type: array + items: + $ref: "#/components/schemas/WebsocketTriggerInitialMessage" + url_runnable_args: + $ref: "#/components/schemas/ScriptArgs" required: - path @@ -11638,6 +11644,8 @@ components: - workspace_id - enabled - filters + - initial_messages + - url_runnable_args NewWebsocketTrigger: type: object @@ -11663,6 +11671,12 @@ components: required: - key - value + initial_messages: + type: array + items: + $ref: "#/components/schemas/WebsocketTriggerInitialMessage" + url_runnable_args: + $ref: "#/components/schemas/ScriptArgs" required: - path @@ -11670,6 +11684,8 @@ components: - url - is_flow - filters + - initial_messages + - url_runnable_args EditWebsocketTrigger: type: object @@ -11693,6 +11709,12 @@ components: required: - key - value + initial_messages: + type: array + items: + $ref: "#/components/schemas/WebsocketTriggerInitialMessage" + url_runnable_args: + $ref: "#/components/schemas/ScriptArgs" required: - path @@ -11700,6 +11722,34 @@ components: - url - is_flow - filters + - initial_messages + - url_runnable_args + + WebsocketTriggerInitialMessage: + anyOf: + - type: object + properties: + raw_message: + type: string + required: + - raw_message + - type: object + properties: + runnable_result: + type: object + properties: + path: + type: string + args: + $ref: "#/components/schemas/ScriptArgs" + is_flow: + type: boolean + required: + - path + - args + - is_flow + required: + - runnable_result Group: type: object diff --git a/backend/windmill-api/src/users.rs b/backend/windmill-api/src/users.rs index c9a812c4a0..005dfad342 100644 --- a/backend/windmill-api/src/users.rs +++ b/backend/windmill-api/src/users.rs @@ -129,7 +129,8 @@ fn username_override_from_label(label: Option) -> Option { Some(label) if label.starts_with("webhook-") || label.starts_with("http-") - || label.starts_with("email-") => + || label.starts_with("email-") + || label.starts_with("ws-") => { Some(label) } diff --git a/backend/windmill-api/src/websocket_triggers.rs b/backend/windmill-api/src/websocket_triggers.rs index b94ed85461..e10cd94c78 100644 --- a/backend/windmill-api/src/websocket_triggers.rs +++ b/backend/windmill-api/src/websocket_triggers.rs @@ -1,9 +1,10 @@ +use anyhow::Context; use axum::{ extract::{Path, Query}, routing::{delete, get, post}, Extension, Json, Router, }; -use futures::StreamExt; +use futures::{stream::SplitSink, SinkExt, StreamExt}; use http::StatusCode; use itertools::Itertools; use rand::seq::SliceRandom; @@ -11,16 +12,20 @@ use serde::{ de::{self, MapAccess, Visitor}, Deserialize, Deserializer, Serialize, }; -use serde_json::Value; +use serde_json::{value::RawValue, Value}; use sql_builder::{bind::Bind, SqlBuilder}; use sqlx::prelude::FromRow; use std::{collections::HashMap, fmt}; -use tokio_tungstenite::connect_async; +use tokio::net::TcpStream; +use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream}; +use uuid::Uuid; use windmill_audit::{audit_ee::audit_log, ActionKind}; use windmill_common::{ db::UserDB, - error::{self, JsonResult}, - utils::{not_found_if_none, paginate, require_admin, Pagination, StripPath}, + error::{self, to_anyhow, JsonResult}, + utils::{ + not_found_if_none, paginate, report_critical_error, require_admin, Pagination, StripPath, + }, worker::{to_raw_value, CLOUD_HOSTED}, INSTANCE_NAME, }; @@ -28,9 +33,7 @@ use windmill_queue::PushArgsOwned; use crate::{ db::{ApiAuthed, DB}, - jobs::{ - run_wait_result_flow_by_path_internal, run_wait_result_script_by_path_internal, RunJobQuery, - }, + jobs::{run_flow_by_path_inner, run_script_by_path_inner, RunJobQuery}, users::fetch_api_authed, }; @@ -52,7 +55,29 @@ struct NewWebsocketTrigger { script_path: String, is_flow: bool, enabled: Option, - filters: Vec, + filters: Vec>, + initial_messages: Vec>, + url_runnable_args: Box, +} + +#[derive(Deserialize)] +struct JsonFilter { + key: String, + value: serde_json::Value, +} + +#[derive(Deserialize)] +#[serde(untagged)] +enum Filter { + JsonFilter(JsonFilter), +} + +#[derive(Deserialize)] +enum InitialMessage { + #[serde(rename = "raw_message")] + RawMessage(String), + #[serde(rename = "runnable_result")] + RunnableResult { path: String, args: Box, is_flow: bool }, } #[derive(FromRow, Serialize, Clone)] @@ -70,7 +95,9 @@ pub struct WebsocketTrigger { extra_perms: serde_json::Value, error: Option, enabled: bool, - filters: Vec, + filters: Vec>>, + initial_messages: Vec>>, + url_runnable_args: sqlx::types::Json>, } #[derive(Deserialize)] @@ -79,7 +106,9 @@ struct EditWebsocketTrigger { url: String, script_path: String, is_flow: bool, - filters: Vec, + filters: Vec>, + initial_messages: Vec>, + url_runnable_args: Box, } #[derive(Deserialize)] @@ -133,14 +162,13 @@ async fn get_websocket_trigger( ) -> error::JsonResult { let mut tx = user_db.begin(&authed).await?; let path = path.to_path(); - let trigger = sqlx::query_as!( - WebsocketTrigger, + let trigger = sqlx::query_as::<_, WebsocketTrigger>( r#"SELECT * FROM websocket_trigger WHERE workspace_id = $1 AND path = $2"#, - w_id, - path, ) + .bind(w_id) + .bind(path) .fetch_optional(&mut *tx) .await?; tx.commit().await?; @@ -164,19 +192,27 @@ async fn create_websocket_trigger( require_admin(authed.is_admin, &authed.username)?; let mut tx = user_db.begin(&authed).await?; - sqlx::query_as!( - WebsocketTrigger, - "INSERT INTO websocket_trigger (workspace_id, path, url, script_path, is_flow, enabled, filters, edited_by, email, edited_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, now()) RETURNING *", - w_id, - ct.path, - ct.url, - ct.script_path, - ct.is_flow, - ct.enabled.unwrap_or(true), - &ct.filters, - &authed.username, - &authed.email + + let filters = ct.filters.into_iter().map(sqlx::types::Json).collect_vec(); + let initial_messages = ct + .initial_messages + .into_iter() + .map(sqlx::types::Json) + .collect_vec(); + sqlx::query_as::<_, WebsocketTrigger>( + "INSERT INTO websocket_trigger (workspace_id, path, url, script_path, is_flow, enabled, filters, initial_messages, url_runnable_args, edited_by, email, edited_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, now()) RETURNING *", ) + .bind(&w_id) + .bind(&ct.path) + .bind(ct.url) + .bind(ct.script_path) + .bind(ct.is_flow) + .bind(ct.enabled.unwrap_or(true)) + .bind(filters.as_slice()) + .bind(initial_messages.as_slice()) + .bind(sqlx::types::Json(ct.url_runnable_args)) + .bind(&authed.username) + .bind(&authed.email) .fetch_one(&mut *tx).await?; audit_log( @@ -204,15 +240,24 @@ async fn update_websocket_trigger( let path = path.to_path(); let mut tx = user_db.begin(&authed).await?; + let filters = ct.filters.into_iter().map(sqlx::types::Json).collect_vec(); + let initial_messages = ct + .initial_messages + .into_iter() + .map(sqlx::types::Json) + .collect_vec(); + // important to update server_id, last_server_ping and error to NULL to stop current websocket listener sqlx::query!( - "UPDATE websocket_trigger SET url = $1, script_path = $2, path = $3, is_flow = $4, filters = $5, edited_by = $6, email = $7, edited_at = now(), server_id = NULL, last_server_ping = NULL, error = NULL - WHERE workspace_id = $8 AND path = $9", + "UPDATE websocket_trigger SET url = $1, script_path = $2, path = $3, is_flow = $4, filters = $5, initial_messages = $6, url_runnable_args = $7, edited_by = $8, email = $9, edited_at = now(), server_id = NULL, last_server_ping = NULL, error = NULL + WHERE workspace_id = $10 AND path = $11", ct.url, ct.script_path, ct.path, ct.is_flow, - &ct.filters, + filters.as_slice() as &[sqlx::types::Json>], + initial_messages.as_slice() as &[sqlx::types::Json>], + sqlx::types::Json(ct.url_runnable_args) as sqlx::types::Json>, &authed.username, &authed.email, w_id, @@ -335,8 +380,7 @@ async fn listen_to_unlistened_websockets( rsmq: &Option, killpill_rx: &tokio::sync::broadcast::Receiver<()>, ) -> () { - match sqlx::query_as!( - WebsocketTrigger, + match sqlx::query_as::<_, WebsocketTrigger>( r#"SELECT * FROM websocket_trigger WHERE enabled IS TRUE AND (server_id IS NULL OR last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds')"# @@ -422,8 +466,6 @@ impl<'de, 'a> Visitor<'de> for SupersetVisitor<'a> { if key == self.key { // Deserialize the value for the key and check if it's a superset let json_value: Value = map.next_value()?; - tracing::info!("json_value: {:?}", json_value); - tracing::info!("value_to_check: {:?}", self.value_to_check); return Ok(is_superset(&json_value, self.value_to_check)); } else { // Skip the value if it's not the one we're interested in @@ -470,149 +512,429 @@ where deserializer.deserialize_map(SupersetVisitor { key, value_to_check }) } +async fn wait_runnable_result( + path: String, + is_flow: bool, + args: &Box, + ws_trigger: &WebsocketTrigger, + username_override: String, + db: &DB, + rsmq: Option, +) -> error::Result { + let user_db = UserDB::new(db.clone()); + let authed = fetch_api_authed( + ws_trigger.edited_by.clone(), + ws_trigger.email.clone(), + &ws_trigger.workspace_id, + &db, + username_override, + ) + .await?; + + let args = serde_json::from_str::>>>(args.get()) + .map_err(|e| error::Error::BadRequest(format!("invalid json: {}", e)))? + .unwrap_or_else(HashMap::new); + + let label_prefix = Some(format!("ws-{}-", ws_trigger.path)); + let (_, job_id) = if is_flow { + run_flow_by_path_inner( + authed, + db.clone(), + user_db, + rsmq.clone(), + ws_trigger.workspace_id.clone(), + StripPath(path.clone()), + RunJobQuery::default(), + PushArgsOwned { args, extra: None }, + label_prefix, + ) + .await? + } else { + run_script_by_path_inner( + authed, + db.clone(), + user_db, + rsmq.clone(), + ws_trigger.workspace_id.clone(), + StripPath(path.clone()), + RunJobQuery::default(), + PushArgsOwned { args, extra: None }, + label_prefix, + ) + .await? + }; + + let start_time = tokio::time::Instant::now(); + + loop { + if start_time.elapsed() > tokio::time::Duration::from_secs(300) { + return Err(anyhow::anyhow!( + "Timed out after 5m waiting for runnable {path} (is_flow: {is_flow}) to complete", + ) + .into()); + } + + #[derive(sqlx::FromRow)] + struct RawResult { + result: Option>>, + success: bool, + } + + let result = sqlx::query_as::<_, RawResult>( + "SELECT result, success FROM completed_job WHERE id = $1 AND workspace_id = $2", + ) + .bind(Uuid::parse_str(&job_id).unwrap()) + .bind(&ws_trigger.workspace_id) + .fetch_optional(db) + .await; + + match result { + Ok(Some(r)) => { + if !r.success { + return Err(anyhow::anyhow!( + "Runnable {path} (is_flow: {is_flow}) failed: {:?}", + r.result + ) + .into()); + } else { + return Ok(r.result.map(|r| r.get().to_owned()).unwrap_or_default()); + } + } + Ok(None) => { + // not yet done, wait for 5s and check again + tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; + } + Err(err) => { + return Err(anyhow::anyhow!( + "Error fetching job result for runnable {path} (is_flow: {is_flow}): {err}", + ) + .into()); + } + } + } +} + +async fn send_initial_messages( + ws_trigger: &WebsocketTrigger, + mut writer: SplitSink>, Message>, + db: &DB, + rsmq: Option, +) -> error::Result<()> { + let initial_messages: Vec = ws_trigger + .initial_messages + .iter() + .filter_map(|m| serde_json::from_str(m.get()).ok()) + .collect_vec(); + + for start_message in initial_messages { + match start_message { + InitialMessage::RawMessage(msg) => { + let msg = if msg.starts_with("\"") && msg.ends_with("\"") { + msg[1..msg.len() - 1].to_string() + } else { + msg + }; + tracing::info!( + "Sending raw message initial message to websocket {}: {}", + ws_trigger.url, + msg + ); + writer + .send(tokio_tungstenite::tungstenite::Message::Text(msg)) + .await + .map_err(to_anyhow) + .with_context(|| "failed to send raw message")?; + } + InitialMessage::RunnableResult { path, is_flow, args } => { + tracing::info!( + "Running runnable {path} (is_flow: {is_flow}) for initial message to websocket {}", + ws_trigger.url, + ); + + let result = wait_runnable_result( + path.clone(), + is_flow, + &args, + ws_trigger, + "init".to_string(), + db, + rsmq.clone(), + ) + .await?; + + tracing::info!( + "Sending runnable {path} (is_flow: {is_flow}) result to websocket {}", + ws_trigger.url + ); + + let result = if result.starts_with("\"") && result.ends_with("\"") { + result[1..result.len() - 1].to_string() + } else { + result + }; + + writer + .send(tokio_tungstenite::tungstenite::Message::Text(result)) + .await + .map_err(to_anyhow) + .with_context(|| { + format!("Failed to send runnable {path} (is_flow: {is_flow}) result") + })?; + } + } + } + + Ok(()) +} + +async fn get_url_from_runnable( + path: &str, + is_flow: bool, + ws_trigger: &WebsocketTrigger, + db: &DB, + rsmq: Option, +) -> error::Result { + tracing::info!("Running runnable {path} (is_flow: {is_flow}) to get websocket URL",); + + let result = wait_runnable_result( + path.to_string(), + is_flow, + &ws_trigger.url_runnable_args.0, + ws_trigger, + "url".to_string(), + db, + rsmq, + ) + .await?; + + if result.starts_with("\"") && result.ends_with("\"") { + Ok(result[1..result.len() - 1].to_string()) + } else { + Err(anyhow::anyhow!("Runnable {path} (is_flow: {is_flow}) did not return a string").into()) + } +} + +async fn update_ping(db: &DB, ws_trigger: &WebsocketTrigger, error: Option<&str>) -> Option<()> { + match sqlx::query_scalar!( + "UPDATE websocket_trigger SET last_server_ping = now(), error = $1 WHERE workspace_id = $2 AND path = $3 AND server_id = $4 AND enabled IS TRUE RETURNING 1", + error, + ws_trigger.workspace_id, + ws_trigger.path, + *INSTANCE_NAME + ).fetch_optional(db).await { + Ok(updated) => { + if updated.flatten().is_none() { + tracing::info!("Websocket {} changed, disabled, or deleted, stopping...", ws_trigger.url); + return None; + } + }, + Err(err) => { + tracing::warn!("Error updating ping of websocket {}: {:?}", ws_trigger.url, err); + } + }; + + Some(()) +} + +async fn loop_ping(db: &DB, ws_trigger: &WebsocketTrigger, error: Option<&str>) -> () { + loop { + if let None = update_ping(db, ws_trigger, error).await { + return; + } + tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; + } +} + +async fn disable_with_error(db: &DB, ws_trigger: &WebsocketTrigger, error: String) { + match sqlx::query!( + "UPDATE websocket_trigger SET enabled = FALSE, error = $1, server_id = NULL, last_server_ping = NULL WHERE workspace_id = $2 AND path = $3", + error, + ws_trigger.workspace_id, + ws_trigger.path, + ) + .execute(db).await { + Ok(_) => { + report_critical_error(format!("Disabling websocket {} because of error: {}", ws_trigger.url, error), db.clone()).await; + }, + Err(disable_err) => { + report_critical_error( + format!("Could not disable websocket {} with err {}, disabling because of error {}", ws_trigger.path, disable_err, error), + db.clone() + ).await; + } + } +} + async fn listen_to_websocket( ws_trigger: WebsocketTrigger, db: DB, rsmq: Option, mut killpill_rx: tokio::sync::broadcast::Receiver<()>, ) -> () { - async fn update_ping(db: DB, ws_trigger: &WebsocketTrigger, error: Option<&str>) -> Option<()> { - match sqlx::query_scalar!( - "UPDATE websocket_trigger SET last_server_ping = now(), error = $1 WHERE workspace_id = $2 AND path = $3 AND server_id = $4 AND enabled IS TRUE RETURNING 1", - error, - ws_trigger.workspace_id, - ws_trigger.path, - *INSTANCE_NAME - ).fetch_optional(&db).await { - Ok(updated) => { - if updated.flatten().is_none() { - tracing::info!("Websocket {} changed, disabled, or deleted, stopping...", ws_trigger.url); - return None; - } - }, - Err(err) => { - tracing::warn!("Error updating ping of websocket {}: {:?}", ws_trigger.url, err); - } - }; - - Some(()) - } + update_ping(&db, &ws_trigger, Some("Connecting...")).await; let url = ws_trigger.url.as_str(); - #[derive(Deserialize)] - struct JsonFilter { - key: String, - value: serde_json::Value, - } - - #[derive(Deserialize)] - #[serde(untagged)] - enum Filter { - JsonFilter(JsonFilter), - } let filters: Vec = ws_trigger .filters .iter() - .filter_map(|m| serde_json::from_value(m.clone()).ok()) + .filter_map(|m| serde_json::from_str(m.get()).ok()) .collect_vec(); loop { + let connect_url = if url.starts_with("$") { + if url.starts_with("$flow:") || url.starts_with("$script:") { + let path = url.splitn(2, ':').nth(1).unwrap(); + tokio::select! { + biased; + _ = killpill_rx.recv() => { + return; + }, + _ = loop_ping(&db, &ws_trigger, Some( + "Waiting on runnable to return websocket URL..." + )) => { + return; + }, + url_result = get_url_from_runnable(path, url.starts_with("$flow:"), &ws_trigger, &db, rsmq.clone()) => match url_result { + Ok(url) => url, + Err(err) => { + disable_with_error( + &db, + &ws_trigger, + format!( + "Error getting websocket URL from runnable after 5 tries: {:?}", + err + ), + ) + .await; + return; + } + }, + } + } else { + disable_with_error( + &db, + &ws_trigger, + format!("Invalid websocket runnable path: {}", url), + ) + .await; + return; + } + } else { + url.to_string() + }; + tokio::select! { biased; _ = killpill_rx.recv() => { return; }, - connection = connect_async(url) => { + _ = loop_ping(&db, &ws_trigger, Some("Connecting...")) => { + return; + }, + connection = connect_async(connect_url) => { match connection { Ok((ws_stream, _)) => { tracing::info!("Listening to websocket {}", url); - if let None = update_ping(db.clone(), &ws_trigger, None).await { + if let None = update_ping(&db, &ws_trigger, None).await { return; } + let (writer, mut reader) = ws_stream.split(); let mut last_ping = tokio::time::Instant::now(); - let (_, mut read) = ws_stream.split(); - loop { - tokio::select! { - biased; - _ = killpill_rx.recv() => { - return; + + tokio::select! { + biased; + _ = killpill_rx.recv() => { + return; + } + _ = async { + if let Err(err) = send_initial_messages(&ws_trigger, writer, &db, rsmq.clone()).await { + disable_with_error(&db, &ws_trigger, format!("Error sending initial messages: {:?}", err)).await; + } else { + // if initial messages sent successfully, wait forever + futures::future::pending::<()>().await; } - msg = read.next() => { - if let Some(msg) = msg { - if last_ping.elapsed() > tokio::time::Duration::from_secs(5) { - if let None = update_ping(db.clone(), &ws_trigger, None).await { + } => { + // was disabled => exit + return; + }, + _ = async { + loop { + tokio::select! { + biased; + msg = reader.next() => { + if let Some(msg) = msg { + if last_ping.elapsed() > tokio::time::Duration::from_secs(5) { + if let None = update_ping(&db, &ws_trigger, None).await { + return; + } + last_ping = tokio::time::Instant::now(); + } + match msg { + Ok(msg) => { + match msg { + tokio_tungstenite::tungstenite::Message::Text(text) => { + let mut should_handle = true; + for filter in &filters { + match filter { + Filter::JsonFilter(JsonFilter { key, value }) => { + let mut deserializer = serde_json::Deserializer::from_str(text.as_str()); + should_handle = match is_value_superset(&mut deserializer, key, &value) { + Ok(filter_match) => { + filter_match + }, + Err(err) => { + tracing::warn!("Error deserializing filter for websocket {}: {:?}", url, err); + false + } + }; + } + } + if !should_handle { + break; + } + } + if should_handle { + if let Err(err) = run_job(&db, rsmq.clone(), &ws_trigger, text).await { + report_critical_error(format!("Failed to trigger job from websocket {}: {:?}", ws_trigger.url, err), db.clone()).await; + }; + } + }, + _ => {} + } + }, + Err(err) => { + tracing::error!("Error reading from websocket {}: {:?}", url, err); + } + } + } else { + tracing::error!("Websocket {} closed", url); + if let None = + update_ping(&db, &ws_trigger, Some("Websocket closed")).await + { + return; + } + tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; + break; + } + }, + _ = tokio::time::sleep(tokio::time::Duration::from_secs(5)) => { + if let None = update_ping(&db, &ws_trigger, None).await { return; } last_ping = tokio::time::Instant::now(); - } - match msg { - Ok(msg) => { - match msg { - tokio_tungstenite::tungstenite::Message::Text(text) => { - let mut should_handle = true; - for filter in &filters { - match filter { - Filter::JsonFilter(JsonFilter { key, value }) => { - let mut deserializer = serde_json::Deserializer::from_str(text.as_str()); - should_handle = match is_value_superset(&mut deserializer, key, &value) { - Ok(filter_match) => { - filter_match - }, - Err(err) => { - tracing::warn!("Error deserializing filter for websocket {}: {:?}", url, err); - false - } - }; - } - } - if !should_handle { - break; - } - } - if should_handle { - let db_ = db.clone(); - let rsmq_ = rsmq.clone(); - let ws_trigger_ = ws_trigger.clone(); - tokio::spawn(async move { - let url = ws_trigger_.url.clone(); - if let Err(err) = run_job(db_, rsmq_, ws_trigger_, text).await { - tracing::error!("Error running job on websocket {}: {:?}", url, err); - }; - }); - } - }, - _ => {} - } - }, - Err(err) => { - tracing::error!("Error reading from websocket {}: {:?}", url, err); - } - } - } else { - tracing::error!("Websocket {} closed", url); - if let None = - update_ping(db.clone(), &ws_trigger, Some("Websocket closed")).await - { - return; - } - tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; - break; + }, } - }, - _ = tokio::time::sleep(tokio::time::Duration::from_secs(5)) => { - if let None = update_ping(db.clone(), &ws_trigger, None).await { - return; - } - last_ping = tokio::time::Instant::now(); - }, + } + } => { + return; } - } + }; } Err(err) => { tracing::error!("Error connecting to websocket {}: {:?}", url, err); if let None = - update_ping(db.clone(), &ws_trigger, Some(err.to_string().as_str())).await + update_ping(&db, &ws_trigger, Some(err.to_string().as_str())).await { return; } @@ -625,16 +947,18 @@ async fn listen_to_websocket( } async fn run_job( - db: DB, + db: &DB, rsmq: Option, - trigger: WebsocketTrigger, + trigger: &WebsocketTrigger, msg: String, ) -> anyhow::Result<()> { let args = PushArgsOwned { args: HashMap::from([("msg".to_string(), to_raw_value(&msg))]), extra: Some(HashMap::from([( "wm_trigger".to_string(), - to_raw_value(&serde_json::json!({"kind": "websocket"})), + to_raw_value( + &serde_json::json!({"kind": "websocket", "websocket": { "url": trigger.url }}), + ), )])), }; let label_prefix = Some(format!("ws-{}-", trigger.path)); @@ -643,7 +967,7 @@ async fn run_job( trigger.edited_by.clone(), trigger.email.clone(), &trigger.workspace_id, - &db, + db, "anonymous".to_string(), ) .await?; @@ -653,27 +977,27 @@ async fn run_job( let run_query = RunJobQuery::default(); if trigger.is_flow { - run_wait_result_flow_by_path_internal( - db, - run_query, - StripPath(trigger.script_path.to_owned()), + run_flow_by_path_inner( authed, - rsmq, + db.clone(), user_db, - args, + rsmq, trigger.workspace_id.clone(), + StripPath(trigger.script_path.to_owned()), + run_query, + args, label_prefix, ) .await?; } else { - run_wait_result_script_by_path_internal( - db, - run_query, - StripPath(trigger.script_path.to_owned()), + run_script_by_path_inner( authed, - rsmq, + db.clone(), user_db, + rsmq, trigger.workspace_id.clone(), + StripPath(trigger.script_path.to_owned()), + run_query, args, label_prefix, ) diff --git a/docker/DockerfileSlim b/docker/DockerfileSlim index 16c2361a0d..a1b5daca43 100644 --- a/docker/DockerfileSlim +++ b/docker/DockerfileSlim @@ -20,7 +20,7 @@ RUN /usr/local/bin/python3 -m pip install pip-tools # Install UV RUN curl --proto '=https' --tlsv1.2 -LsSf https://github.com/astral-sh/uv/releases/download/0.4.18/uv-installer.sh | sh && mv /root/.cargo/bin/uv /usr/local/bin/uv -COPY --from=oven/bun:1.1.32 /usr/local/bin/bun /usr/bin/bun +COPY --from=oven/bun:1.1.34 /usr/local/bin/bun /usr/bin/bun # add the docker client to call docker from a worker if enabled COPY --from=docker:dind /usr/local/bin/docker /usr/local/bin/ diff --git a/docker/DockerfileSlimEe b/docker/DockerfileSlimEe index 65c6b55a19..94cc160cae 100644 --- a/docker/DockerfileSlimEe +++ b/docker/DockerfileSlimEe @@ -19,7 +19,7 @@ RUN /usr/local/bin/python3 -m pip install pip-tools # Install UV RUN curl --proto '=https' --tlsv1.2 -LsSf https://github.com/astral-sh/uv/releases/download/0.4.18/uv-installer.sh | sh && mv /root/.cargo/bin/uv /usr/local/bin/uv -COPY --from=oven/bun:1.1.32 /usr/local/bin/bun /usr/bin/bun +COPY --from=oven/bun:1.1.34 /usr/local/bin/bun /usr/bin/bun # add the docker client to call docker from a worker if enabled COPY --from=docker:dind /usr/local/bin/docker /usr/local/bin/ diff --git a/frontend/src/lib/components/ArgInput.svelte b/frontend/src/lib/components/ArgInput.svelte index 80537e4d44..bbaf166e0d 100644 --- a/frontend/src/lib/components/ArgInput.svelte +++ b/frontend/src/lib/components/ArgInput.svelte @@ -128,8 +128,10 @@ let s3FilePicker: S3FilePicker let s3FileUploadRawMode: false let isListJson = false + let hasIsListJsonChanged = false let el: HTMLTextAreaElement | undefined = undefined + let inputCat = computeInputCat(type, format, itemsType?.type, enum_, contentEncoding) $: inputCat = computeInputCat(type, format, itemsType?.type, enum_, contentEncoding) @@ -170,6 +172,43 @@ $: computeDefaultValue(value, inputCat, defaultValue, nullable) + let lastValue: any = undefined + + // By setting isListJson to true, we can render inputs even if the value is not an array of the correct type + // This avoids the issue of the input being rendered as a string with value: [object Object], or as a number with value: NaN + function checkArrayValueType() { + try { + if (Array.isArray(value) && value.length > 0) { + const firstItem = value?.[0] + const type = itemsType?.type + + switch (type) { + case 'string': + if (typeof firstItem !== 'string') { + isListJson = true + } + break + case 'number': + if (typeof firstItem !== 'number') { + isListJson = true + } + break + } + } + } catch (e) { + console.error(e) + } + + lastValue = value + } + + $: !isListJson && + inputCat === 'list' && + value != lastValue && + itemsType?.type && + !hasIsListJsonChanged && + checkArrayValueType() + $: defaultValue != undefined && handleDefaultValueChange() let oldDefaultValue = defaultValue @@ -518,6 +557,11 @@
{ + // Once the user has changed the input type, we should not change it back automatically + if (!hasIsListJsonChanged) { + hasIsListJsonChanged = true + } + evalValueToRaw() isListJson = !isListJson }} diff --git a/frontend/src/lib/components/InputTransformForm.svelte b/frontend/src/lib/components/InputTransformForm.svelte index 64530dd3ba..706d00b7ae 100644 --- a/frontend/src/lib/components/InputTransformForm.svelte +++ b/frontend/src/lib/components/InputTransformForm.svelte @@ -9,9 +9,11 @@ import type { PropPickerWrapperContext } from './flows/propPicker/PropPickerWrapper.svelte' import { codeToStaticTemplate, getDefaultExpr } from './flows/utils' import SimpleEditor from './SimpleEditor.svelte' - import { Button } from './common' + import { Button } from '$lib/components/common' + import AnimatedButton from '$lib/components/common/button/AnimatedButton.svelte' import ToggleButtonGroup from '$lib/components/common/toggleButton-v2/ToggleButtonGroup.svelte' import ToggleButton from '$lib/components/common/toggleButton-v2/ToggleButton.svelte' + import { fade } from 'svelte/transition' import type VariableEditor from './VariableEditor.svelte' import type ItemPicker from './ItemPicker.svelte' @@ -23,7 +25,7 @@ import type { FlowCopilotContext } from './copilot/flow' import StepInputGen from './copilot/StepInputGen.svelte' import type { PickableProperties } from './flows/previousResults' - + import { twMerge } from 'tailwind-merge' export let schema: Schema | { properties?: Record; required?: string[] } export let arg: InputTransform | any export let argName: string @@ -186,11 +188,22 @@ let stepInputGen: StepInputGen | undefined = undefined loadResourceTypes() + + $: connecting = + $propPickerConfig?.propName == argName && $propPickerConfig?.insertionMode == 'connect' {#if arg != undefined} -
-
+
+
- + +
{/if}
-
- {#if $propPickerConfig?.propName == argName && $propPickerConfig?.insertionMode == 'connect'} +
+ - {#if isStaticTemplate(inputCat) && propertyType == 'static' && !noDynamicToggle} - {#if argName && schema?.properties?.[argName]?.description} -
- 
{schema.properties[argName].description}
-
- {/if} -
- {#if arg} - +
+ {#if isStaticTemplate(inputCat) && propertyType == 'static' && !noDynamicToggle} + {#if argName && schema?.properties?.[argName]?.description} +
+ 
{schema.properties[argName].description}
+
+ {/if} +
+ {#if arg} + { + focused = false + }} + bind:code={arg.value} + fontSize={14} + on:change={() => { + dispatch('change', { argName }) + }} + /> + {/if} +
+ {:else if (propertyType === undefined || propertyType == 'static') && schema?.properties?.[argName]} + { focused = false }} - bind:code={arg.value} - fontSize={14} + shouldDispatchChanges on:change={() => { dispatch('change', { argName }) }} + label={argName} + bind:editor={monaco} + bind:description={schema.properties[argName].description} + bind:value={arg.value} + type={schema.properties[argName].type} + oneOf={schema.properties[argName].oneOf} + required={schema.required?.includes(argName)} + bind:pattern={schema.properties[argName].pattern} + bind:valid={inputCheck} + defaultValue={schema.properties[argName].default} + bind:enum_={schema.properties[argName].enum} + bind:format={schema.properties[argName].format} + contentEncoding={schema.properties[argName].contentEncoding} + bind:itemsType={schema.properties[argName].items} + properties={schema.properties[argName].properties} + nestedRequired={schema.properties[argName].required} + displayHeader={false} + extra={argExtra} + {variableEditor} + {itemPicker} + bind:pickForField + showSchemaExplorer + nullable={schema.properties[argName].nullable} + bind:title={schema.properties[argName].title} + bind:placeholder={schema.properties[argName].placeholder} /> + {:else if arg.expr != undefined} +
+ { + dispatch('change', { argName }) + }} + {extraLib} + lang="javascript" + shouldBindKey={false} + on:focus={() => { + focused = true + focusProp(argName, 'insert', (path) => { + monaco?.insertAtCursor(path) + return false + }) + }} + on:change={() => { + dispatch('change', { argName }) + }} + on:blur={() => { + focused = false + }} + autoHeight + /> +
+ +
+ {:else} + Not recognized input type {argName} ({arg.expr}, {propertyType}) +
+
{/if}
- {:else if (propertyType === undefined || propertyType == 'static') && schema?.properties?.[argName]} - { - focused = false - }} - shouldDispatchChanges - on:change={() => { - dispatch('change', { argName }) - }} - label={argName} - bind:editor={monaco} - bind:description={schema.properties[argName].description} - bind:value={arg.value} - type={schema.properties[argName].type} - oneOf={schema.properties[argName].oneOf} - required={schema.required?.includes(argName)} - bind:pattern={schema.properties[argName].pattern} - bind:valid={inputCheck} - defaultValue={schema.properties[argName].default} - bind:enum_={schema.properties[argName].enum} - bind:format={schema.properties[argName].format} - contentEncoding={schema.properties[argName].contentEncoding} - bind:itemsType={schema.properties[argName].items} - properties={schema.properties[argName].properties} - nestedRequired={schema.properties[argName].required} - displayHeader={false} - extra={argExtra} - {variableEditor} - {itemPicker} - bind:pickForField - showSchemaExplorer - nullable={schema.properties[argName].nullable} - bind:title={schema.properties[argName].title} - bind:placeholder={schema.properties[argName].placeholder} - /> - {:else if arg.expr != undefined} -
- { - dispatch('change', { argName }) - }} - {extraLib} - lang="javascript" - shouldBindKey={false} - on:focus={() => { - focused = true - focusProp(argName, 'insert', (path) => { - monaco?.insertAtCursor(path) - return false - }) - }} - on:change={() => { - dispatch('change', { argName }) - }} - on:blur={() => { - focused = false - }} - autoHeight - /> -
- -
- {:else} - Not recognized input type {argName} ({arg.expr}, {propertyType}) -
-
- {/if} + + {#if $propPickerConfig?.propName == argName} +
+ + + +
+ {:else} +
+ {/if} +
{/if} diff --git a/frontend/src/lib/components/InputTransformSchemaForm.svelte b/frontend/src/lib/components/InputTransformSchemaForm.svelte index 488797c6db..e9a0f4899d 100644 --- a/frontend/src/lib/components/InputTransformSchemaForm.svelte +++ b/frontend/src/lib/components/InputTransformSchemaForm.svelte @@ -67,24 +67,26 @@
{#if enableAi} - - Object.keys(schema.properties ?? {}).includes(argName) && - Object.keys(args ?? {}).includes(argName) && - ((args[argName].type === 'static' && !args[argName].value) || - (args[argName].type === 'javascript' && !args[argName].expr)) - ) - : []} - {schema} - /> +
+ + Object.keys(schema.properties ?? {}).includes(argName) && + Object.keys(args ?? {}).includes(argName) && + ((args[argName].type === 'static' && !args[argName].value) || + (args[argName].type === 'javascript' && !args[argName].expr)) + ) + : []} + {schema} + /> +
{/if} {#if keys.length > 0} {#each keys as argName (argName)} {#if (!filter || filter.includes(argName)) && Object.keys(schema.properties ?? {}).includes(argName)} -
+
+ {/if} + { - dispatch('select', `results.${e.detail}`) - }} + prefix="results" + on:select />
{/if} {/if} {#if displayContext} - Variables
+ Variables: + {#if displayVariable} -
- -
+ + dispatch('select', `variable('${e.detail}')`)} + prefix="variable" + on:select /> {:else} - + {'{...}'} + {/if}
- Resources +
+ Resources: + {#if displayResources} - dispatch('select', `resource('${e.detail}')`)} + prefix="resource" + on:select /> {:else} - + {'{...}'} + {/if}
{/if} diff --git a/frontend/src/lib/components/propertyPicker/PropPickerResult.svelte b/frontend/src/lib/components/propertyPicker/PropPickerResult.svelte index adf9d76aec..cbb2f9c33c 100644 --- a/frontend/src/lib/components/propertyPicker/PropPickerResult.svelte +++ b/frontend/src/lib/components/propertyPicker/PropPickerResult.svelte @@ -1,31 +1,21 @@
- Result + Result
- +
{#if flow_input} - Flow Input + Flow Input
- dispatch('select', `flow_input.${e.detail}`)} - /> +
{/if}
diff --git a/frontend/src/lib/components/propertyPicker/utils.ts b/frontend/src/lib/components/propertyPicker/utils.ts index 129bfdf671..27515b4190 100644 --- a/frontend/src/lib/components/propertyPicker/utils.ts +++ b/frontend/src/lib/components/propertyPicker/utils.ts @@ -1,44 +1,35 @@ -function filterByKey(obj: Object, key: string): Object { - if (Object(obj) !== obj) { - return obj +function filterByKey(obj: Object, key: string): any { + if (typeof obj !== 'object' || obj === null) { + return undefined } else if (Array.isArray(obj)) { - return obj.map((o) => filterByKey(o, key)) - } else { - return Object.fromEntries( - Object.entries(obj) - .filter(([k, v]) => !k.includes(key)) - .map(([k, v]) => [k, filterByKey(v, key)]) - ) - } -} - -function diff(target: Object, source: Object): Object { - if (Array.isArray(target)) { - return target - } - - const result = {} - - Object.keys(target ?? {}).forEach((key: string) => { - if (typeof source[key] === 'object') { - const difference = diff(target[key], source[key]) - - if (Object.keys(difference).length > 0) { - result[key] = difference - } - } else if (source[key] !== target[key]) { - result[key] = target[key] + let a = obj + .map((k, o) => + typeof o == 'object' ? filterByKey(o, key) : String(k - 1).includes(key) ? o : undefined + ) + .filter((v) => v !== undefined) + if (a.length === 0) { + return undefined + } else { + return a } - }) - - return result + } else { + let o = Object.fromEntries( + Object.entries(obj) + .map(([k, v]) => (k.includes(key) ? [k, v] : [k, filterByKey(v, key)])) + .filter(([k, v]) => v !== undefined) + ) + if (Object.keys(o).length === 0) { + return undefined + } + return o + } } export function keepByKey(json: Object | undefined, key: string): Object { if (!json) { return {} } - return diff(json, filterByKey(json, key)) + return filterByKey(json, key) ?? {} } // https://stackoverflow.com/questions/23377217/way-to-test-if-a-string-is-valid-identifier-name-in-javascript diff --git a/frontend/src/lib/components/triggers/WebsocketTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/WebsocketTriggerEditorInner.svelte index 39f9ce3cae..0ee30613fc 100644 --- a/frontend/src/lib/components/triggers/WebsocketTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/WebsocketTriggerEditorInner.svelte @@ -5,9 +5,17 @@ import Path from '$lib/components/Path.svelte' import Required from '$lib/components/Required.svelte' import ScriptPicker from '$lib/components/ScriptPicker.svelte' - import { WebsocketTriggerService } from '$lib/gen' + import { + FlowService, + ScriptService, + WebsocketTriggerService, + type Flow, + type Script, + type ScriptArgs, + type WebsocketTriggerInitialMessage + } from '$lib/gen' import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores' - import { canWrite, emptyString, sendUserToast } from '$lib/utils' + import { canWrite, emptySchema, emptyString, sendUserToast } from '$lib/utils' import { createEventDispatcher } from 'svelte' import Section from '$lib/components/Section.svelte' import { Loader2, Save, X, Plus } from 'lucide-svelte' @@ -15,6 +23,9 @@ import Toggle from '../Toggle.svelte' import { fade } from 'svelte/transition' import JsonEditor from '../apps/editor/settingsPanel/inputEditor/JsonEditor.svelte' + import type { Schema } from '$lib/common' + import ToggleButtonGroup from '../common/toggleButton-v2/ToggleButtonGroup.svelte' + import ToggleButton from '../common/toggleButton-v2/ToggleButton.svelte' let drawer: Drawer let is_flow: boolean = false @@ -34,6 +45,8 @@ key: string value: any }[] = [] + let initial_messages: WebsocketTriggerInitialMessage[] = [] + let url_runnable_args: Record = {} let dirtyPath = false let can_write = true let drawerLoading = true @@ -74,6 +87,8 @@ path = '' initialPath = '' filters = [] + initial_messages = [] + url_runnable_args = {} dirtyPath = false } finally { drawerLoading = false @@ -93,10 +108,81 @@ url = s.url enabled = s.enabled filters = s.filters + initial_messages = s.initial_messages + url_runnable_args = s.url_runnable_args can_write = canWrite(s.path, s.extra_perms, $userStore) } + let initialMessageRunnableSchemas: Record = {} + async function loadInitialMessageRunnableSchemas( + initialMessageRunnables: { + path: string + is_flow: boolean + }[] + ) { + for (const { path, is_flow } of initialMessageRunnables) { + if (!path) { + continue + } + try { + let schema: Schema | undefined = emptySchema() + let scriptOrFlow: Script | Flow = is_flow + ? await FlowService.getFlowByPath({ workspace: $workspaceStore!, path }) + : await ScriptService.getScriptByPath({ workspace: $workspaceStore!, path }) + schema = scriptOrFlow.schema as Schema + if (schema && schema.properties) { + initialMessageRunnableSchemas[(is_flow ? 'flow/' : '') + path] = schema + } + } catch (err) { + sendUserToast( + `Could not query runnable schema for ${is_flow ? 'flow' : 'script'} ${path}: ${err}`, + true + ) + } + } + } + $: initialMessageRunnables = initial_messages + .map((v) => ('runnable_result' in v ? v.runnable_result : undefined)) + .filter((v): v is { path: string; is_flow: boolean; args: ScriptArgs } => !!v) + $: loadInitialMessageRunnableSchemas(initialMessageRunnables) + + let urlRunnableSchema: Schema | undefined = emptySchema() + async function loadUrlRunnableSchema(url: string) { + if (url.startsWith('$')) { + const path = url.split(':')[1] + if (path && path.length > 0) { + try { + let scriptOrFlow: Script | Flow = url.startsWith('$flow:') + ? await FlowService.getFlowByPath({ + workspace: $workspaceStore!, + path: url.split(':')[1] + }) + : await ScriptService.getScriptByPath({ + workspace: $workspaceStore!, + path: url.split(':')[1] + }) + urlRunnableSchema = scriptOrFlow.schema as Schema + } catch (err) { + sendUserToast( + `Could not query runnable schema for ${url.startsWith('$flow:') ? 'flow' : 'script'} ${ + url.split(':')[1] + }: ${err}`, + true + ) + } + } + } + } + $: loadUrlRunnableSchema(url) + + $: invalidInitialMessages = initial_messages.some((v) => { + if ('runnable_result' in v) { + return !v.runnable_result.path + } + return false + }) + async function updateTrigger(): Promise { if (edit) { await WebsocketTriggerService.updateWebsocketTrigger({ @@ -107,7 +193,9 @@ script_path, is_flow, url, - filters + filters, + initial_messages, + url_runnable_args } }) sendUserToast(`Route ${path} updated`) @@ -120,7 +208,9 @@ is_flow, url, enabled: true, - filters + filters, + initial_messages, + url_runnable_args } }) sendUserToast(`Route ${path} created`) @@ -139,7 +229,12 @@ clearTimeout(validateTimeout) } validateTimeout = setTimeout(() => { - if (/^(ws:|wss:)\/\/[^\s]+$/.test(url) === false) { + console.log('validating ' + url) + if (url.startsWith('$')) { + if (/^(\$script|\$flow):[^\s]+$/.test(url) === false) { + urlError = 'Invalid runnable path' + } + } else if (/^(ws:|wss:)\/\/[^\s]+$/.test(url) === false) { urlError = 'Invalid websocket URL' } validateTimeout = undefined @@ -148,7 +243,7 @@ $: validateUrl(url) - + Save @@ -214,31 +313,90 @@
-
-
@@ -258,6 +416,144 @@
+
+

+ Initial messages are sent at the beginning of the connection. They are sent in order.
+ Raw messages and runnable results are supported. +

+
+ {#each initial_messages as v, i} +
+
+
+ +
+ {#if 'raw_message' in v} +
+
+ Raw JSON message (if a string, wrapping quotes will be discarded) +
+ { + const { code } = ev.detail + v = { + raw_message: code + } + }} + code={v.raw_message} + /> +
+ {:else if 'runnable_result' in v} +
+
Runnable
+ { + const { path, itemKind } = ev.detail + v = { + runnable_result: { + path: path ?? '', + args: {}, + is_flow: itemKind === 'flow' + } + } + }} + /> + + {#if v.runnable_result?.path} + {@const schema = + initialMessageRunnableSchemas[ + v.runnable_result.is_flow + ? 'flow/' + v.runnable_result.path + : v.runnable_result.path + ]} + {#if schema} +

Arguments

+ {#await import('$lib/components/SchemaForm.svelte')} + + {:then Module} + + {/await} + {#if schema && schema.properties && Object.keys(schema.properties).length === 0} +
This runnable takes no arguments
+ {/if} + {:else} + + {/if} + {/if} +
+ {:else} + Unknown type + {/if} +
+ +
+ {/each} + +
+ +
+
+
+

Filters will limit the execution of the trigger to only messages that match all @@ -266,8 +562,8 @@

{#each filters as v, i} -
-
+
+