From 4dda0fb8cd8262ad3a2ab2b9d27e7043ac3bb891 Mon Sep 17 00:00:00 2001 From: HugoCasa Date: Tue, 5 Nov 2024 11:54:08 +0100 Subject: [PATCH 1/6] feat: websocket authentication (#4635) * feat: websocket authentication * npm check, sqlx and preprocessor template * better run job * nit --- ...6fcb0b7a54ed4218fa3cf23cc929d009d50f.json} | 6 +- ...d373b18b6577198df7545c94bd19be3446775.json | 108 --- ...81a50a832e65c8c9632e2e12cd9c2747f2fc7.json | 98 --- ...4b2e3ecb30aae6510d1fcd08493b07c8b0952.json | 16 + ...1b50a15ec8bd349f1081af54da5dc2f5e3021.json | 101 --- .../20241030150214_ws_auth.down.sql | 2 + .../migrations/20241030150214_ws_auth.up.sql | 2 + backend/windmill-api/openapi.yaml | 50 ++ backend/windmill-api/src/users.rs | 3 +- .../windmill-api/src/websocket_triggers.rs | 636 +++++++++++++----- .../src/lib/components/ScriptPicker.svelte | 4 +- .../details/EmailTriggerPanel.svelte | 10 +- .../WebsocketTriggerEditorInner.svelte | 360 +++++++++- frontend/src/lib/script_helpers.ts | 10 + .../(logged)/websocket_triggers/+page.svelte | 86 +-- 15 files changed, 948 insertions(+), 544 deletions(-) rename backend/.sqlx/{query-acbf74cf3302bfcf7615285070d3f8958932bb8a2dda715f1b9152ab44442780.json => query-0b94bd4c98a11ca1b7e5e34dd1ee6fcb0b7a54ed4218fa3cf23cc929d009d50f.json} (50%) delete mode 100644 backend/.sqlx/query-4e9668a46bad9e82baa51422946d373b18b6577198df7545c94bd19be3446775.json delete mode 100644 backend/.sqlx/query-5303cb9dd5903aa4791ef8e5e5881a50a832e65c8c9632e2e12cd9c2747f2fc7.json create mode 100644 backend/.sqlx/query-cd33a9d63f4706a7e3b1e23cd0a4b2e3ecb30aae6510d1fcd08493b07c8b0952.json delete mode 100644 backend/.sqlx/query-f2baee15e6d1fecd6d2d7b39fda1b50a15ec8bd349f1081af54da5dc2f5e3021.json create mode 100644 backend/migrations/20241030150214_ws_auth.down.sql create mode 100644 backend/migrations/20241030150214_ws_auth.up.sql 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/frontend/src/lib/components/ScriptPicker.svelte b/frontend/src/lib/components/ScriptPicker.svelte index 53a856c047..f9c117bab3 100644 --- a/frontend/src/lib/components/ScriptPicker.svelte +++ b/frontend/src/lib/components/ScriptPicker.svelte @@ -95,11 +95,11 @@ value={items?.find((x) => x.value == initialPath)} class="grow shrink max-w-full" on:change={() => { - dispatch('select', { path: scriptPath }) + dispatch('select', { path: scriptPath, itemKind }) }} on:input={(ev) => { if (!ev.detail) { - dispatch('select', { path: undefined }) + dispatch('select', { path: undefined, itemKind }) } }} bind:justValue={scriptPath} diff --git a/frontend/src/lib/components/details/EmailTriggerPanel.svelte b/frontend/src/lib/components/details/EmailTriggerPanel.svelte index 578ef02e4c..fca824946f 100644 --- a/frontend/src/lib/components/details/EmailTriggerPanel.svelte +++ b/frontend/src/lib/components/details/EmailTriggerPanel.svelte @@ -39,7 +39,13 @@ let requestType: 'hash' | 'path' = 'path' - function emailAddress() { + function emailAddress( + requestType: 'hash' | 'path', + path: string, + hash: string | undefined, + isFlow: boolean, + token: string + ) { const pathOrHash = requestType === 'hash' ? hash : path.replaceAll('/', '.') const plainPrefix = `${$workspaceStore}+${ (requestType === 'hash' ? 'hash.' : isFlow ? 'flow.' : '') + pathOrHash @@ -54,7 +60,7 @@ export let email: string = '' - $: email = emailAddress() + $: email = emailAddress(requestType, path, hash, isFlow, token) let triggerTokens: TriggerTokens | undefined = undefined 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} -
-
+
+