starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-07 00:01:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: websocket authentication (#4635)

Browse Source 
* feat: websocket authentication

* npm check, sqlx and preprocessor template

* better run job

* nit
This commit is contained in:
HugoCasa
2024-11-05 11:54:08 +01:00
committed by GitHub
parent 77735d859c
commit 4dda0fb8cd
15 changed files with 948 additions and 544 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-acbf74cf3302bfcf7615285070d3f8958932bb8a2dda715f1b9152ab44442780.json → backend/.sqlx/query-0b94bd4c98a11ca1b7e5e34dd1ee6fcb0b7a54ed4218fa3cf23cc929d009d50f.json
Generated
+4 -2
View File
@@ -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"
}
backend/.sqlx/query-4e9668a46bad9e82baa51422946d373b18b6577198df7545c94bd19be3446775.json
Generated
-108
View File
@@ -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"
}
backend/.sqlx/query-5303cb9dd5903aa4791ef8e5e5881a50a832e65c8c9632e2e12cd9c2747f2fc7.json
Generated
-98
View File
@@ -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"
}
backend/.sqlx/query-cd33a9d63f4706a7e3b1e23cd0a4b2e3ecb30aae6510d1fcd08493b07c8b0952.json
Generated
+16
View File
@@ -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"
}
backend/.sqlx/query-f2baee15e6d1fecd6d2d7b39fda1b50a15ec8bd349f1081af54da5dc2f5e3021.json
Generated
-101
View File
@@ -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"
}
backend/migrations/20241030150214_ws_auth.down.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add down migration script here
ALTER TABLE websocket_trigger DROP COLUMN initial_messages, DROP COLUMN url_runnable_args;
backend/migrations/20241030150214_ws_auth.up.sql
+2
View File
@@ -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 '{}';
backend/windmill-api/openapi.yaml
+50
View File
@@ -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
backend/windmill-api/src/users.rs
+2 -1
View File
@@ -129,7 +129,8 @@ fn username_override_from_label(label: Option<String>) -> Option<String> {
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)
}
backend/windmill-api/src/websocket_triggers.rs
+480 -156
View File
@@ -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<bool>,
filters: Vec<serde_json::Value>,
filters: Vec<Box<RawValue>>,
initial_messages: Vec<Box<RawValue>>,
url_runnable_args: Box<RawValue>,
}
#[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<RawValue>, is_flow: bool },
}
#[derive(FromRow, Serialize, Clone)]
@@ -70,7 +95,9 @@ pub struct WebsocketTrigger {
extra_perms: serde_json::Value,
error: Option<String>,
enabled: bool,
filters: Vec<serde_json::Value>,
filters: Vec<sqlx::types::Json<Box<RawValue>>>,
initial_messages: Vec<sqlx::types::Json<Box<RawValue>>>,
url_runnable_args: sqlx::types::Json<Box<RawValue>>,
}
#[derive(Deserialize)]
@@ -79,7 +106,9 @@ struct EditWebsocketTrigger {
url: String,
script_path: String,
is_flow: bool,
filters: Vec<serde_json::Value>,
filters: Vec<Box<RawValue>>,
initial_messages: Vec<Box<RawValue>>,
url_runnable_args: Box<RawValue>,
}
#[derive(Deserialize)]
@@ -133,14 +162,13 @@ async fn get_websocket_trigger(
) -> error::JsonResult<WebsocketTrigger> {
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<Box<RawValue>>],
initial_messages.as_slice() as &[sqlx::types::Json<Box<RawValue>>],
sqlx::types::Json(ct.url_runnable_args) as sqlx::types::Json<Box<RawValue>>,
&authed.username,
&authed.email,
w_id,
@@ -335,8 +380,7 @@ async fn listen_to_unlistened_websockets(
rsmq: &Option<rsmq_async::MultiplexedRsmq>,
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<RawValue>,
ws_trigger: &WebsocketTrigger,
username_override: String,
db: &DB,
rsmq: Option<rsmq_async::MultiplexedRsmq>,
) -> error::Result<String> {
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::<Option<HashMap<String, Box<RawValue>>>>(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<sqlx::types::Json<Box<RawValue>>>,
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<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>,
db: &DB,
rsmq: Option<rsmq_async::MultiplexedRsmq>,
) -> error::Result<()> {
let initial_messages: Vec<InitialMessage> = 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<rsmq_async::MultiplexedRsmq>,
) -> error::Result<String> {
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<rsmq_async::MultiplexedRsmq>,
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<Filter> = 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<rsmq_async::MultiplexedRsmq>,
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,
)
frontend/src/lib/components/ScriptPicker.svelte
+2 -2
View File
@@ -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}
frontend/src/lib/components/details/EmailTriggerPanel.svelte
+8 -2
View File
@@ -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
</script>
frontend/src/lib/components/triggers/WebsocketTriggerEditorInner.svelte
+328 -32
View File
@@ -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<string, any> = {}
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<string, Schema> = {}
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<void> {
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)
</script>
<Drawer size="700px" bind:this={drawer}>
<Drawer size="800px" bind:this={drawer}>
<DrawerContent
title={edit
? can_write
@@ -180,7 +275,11 @@
{/if}
<Button
startIcon={{ icon: Save }}
disabled={pathError != '' || urlError != '' || emptyString(script_path) || !can_write}
disabled={pathError != '' ||
urlError != '' ||
invalidInitialMessages ||
emptyString(script_path) ||
!can_write}
on:click={updateTrigger}
>
Save
@@ -214,31 +313,90 @@
</div>
<Section label="Websocket">
<div class="flex flex-col w-full gap-4">
<label class="block grow w-full">
<div class="text-secondary text-sm flex items-center gap-1 w-full justify-between">
<div>
URL
<Required required={true} />
<div class="mb-2">
<ToggleButtonGroup
selected={url.startsWith('$') ? 'runnable' : 'static'}
on:selected={(ev) => {
url = ev.detail === 'runnable' ? '$script:' : ''
url_runnable_args = {}
}}
>
<ToggleButton value="static" label="Static URL" />
<ToggleButton value="runnable" label="Runnable result as URL" />
</ToggleButtonGroup>
</div>
{#if url.startsWith('$')}
<div class="flex flex-col w-full gap-4">
<div class="block grow w-full">
<div class="text-secondary text-sm flex items-center gap-1 w-full justify-between">
<div>
Runnable
<Required required={true} />
</div>
</div>
<ScriptPicker
allowFlow={true}
itemKind={url.startsWith('$flow:') ? 'flow' : 'script'}
initialPath={url.split(':')[1] ?? ''}
on:select={(ev) => {
dirtyUrl = true
const { path, itemKind } = ev.detail
url = `$${itemKind}:${path ?? ''}`
}}
/>
<div class="text-red-600 dark:text-red-400 text-2xs mt-1.5">
{dirtyUrl ? urlError : ''}
</div>
</div>
<input
type="text"
autocomplete="off"
bind:value={url}
disabled={!can_write}
on:input={() => {
dirtyUrl = true
}}
class={urlError === ''
? ''
: 'border border-red-700 bg-red-100 border-opacity-30 focus:border-red-700 focus:border-opacity-30 focus-visible:ring-red-700 focus-visible:ring-opacity-25 focus-visible:border-red-700'}
/>
<div class="text-red-600 dark:text-red-400 text-2xs mt-1.5">
{dirtyUrl ? urlError : ''}
</div>
</label>
</div>
</div>
{#if url.split(':')[1]?.length > 0}
{#if urlRunnableSchema}
<p class="font-semibold text-sm mt-4 mb-2">Arguments</p>
{#await import('$lib/components/SchemaForm.svelte')}
<Loader2 class="animate-spin mt-2" />
{:then Module}
<Module.default
schema={urlRunnableSchema}
bind:args={url_runnable_args}
shouldHideNoInputs
class="text-xs"
/>
{/await}
{#if urlRunnableSchema.properties && Object.keys(urlRunnableSchema.properties).length === 0}
<div class="text-xs texg-gray-700">This runnable takes no arguments</div>
{/if}
{:else}
<Loader2 class="animate-spin mt-2" />
{/if}
{/if}
{:else}
<div class="flex flex-col w-full gap-4">
<label class="block grow w-full">
<div class="text-secondary text-sm flex items-center gap-1 w-full justify-between">
<div>
URL
<Required required={true} />
</div>
</div>
<input
type="text"
autocomplete="off"
bind:value={url}
disabled={!can_write}
on:input={() => {
dirtyUrl = true
}}
class={urlError === ''
? ''
: 'border border-red-700 bg-red-100 border-opacity-30 focus:border-red-700 focus:border-opacity-30 focus-visible:ring-red-700 focus-visible:ring-opacity-25 focus-visible:border-red-700'}
/>
<div class="text-red-600 dark:text-red-400 text-2xs mt-1.5">
{dirtyUrl ? urlError : ''}
</div>
</label>
</div>
{/if}
</Section>
<Section label="Runnable">
@@ -258,6 +416,144 @@
</div>
</Section>
<Section label="Initial messages">
<p class="text-xs mb-1 text-tertiary">
Initial messages are sent at the beginning of the connection. They are sent in order.<br
/>
Raw messages and runnable results are supported.
</p>
<div class="flex flex-col gap-4 mt-1">
{#each initial_messages as v, i}
<div class="flex w-full gap-2 items-center">
<div class="w-full flex flex-col gap-2 border p-2 rounded-md">
<div class="flex flex-row gap-2 w-full">
<label class="flex flex-col w-full">
<div class="text-secondary text-sm">Type</div>
<select
class="w-20"
on:change={(e) => {
if (e.target?.['value'] === 'raw_message') {
initial_messages[i] = {
raw_message: '""'
}
} else {
initial_messages[i] = {
runnable_result: {
path: '',
args: {},
is_flow: false
}
}
}
}}
value={'runnable_result' in v ? 'runnable_result' : 'raw_message'}
>
<option value="raw_message">Raw message</option>
<option value="runnable_result">Runnable result</option>
</select>
</label>
</div>
{#if 'raw_message' in v}
<div class="flex flex-col w-full">
<div class="text-secondary text-sm">
Raw JSON message (if a string, wrapping quotes will be discarded)
</div>
<JsonEditor
on:change={(ev) => {
const { code } = ev.detail
v = {
raw_message: code
}
}}
code={v.raw_message}
/>
</div>
{:else if 'runnable_result' in v}
<div class="flex flex-col w-full">
<div class="text-secondary text-sm">Runnable</div>
<ScriptPicker
allowFlow={true}
itemKind={v.runnable_result?.is_flow ? 'flow' : 'script'}
initialPath={v.runnable_result?.path ?? ''}
on:select={(ev) => {
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}
<p class="font-semibold text-sm mt-4 mb-2">Arguments</p>
{#await import('$lib/components/SchemaForm.svelte')}
<Loader2 class="animate-spin mt-2" />
{:then Module}
<Module.default
{schema}
bind:args={v.runnable_result.args}
shouldHideNoInputs
class="text-xs"
/>
{/await}
{#if schema && schema.properties && Object.keys(schema.properties).length === 0}
<div class="text-xs texg-gray-700">This runnable takes no arguments</div
>
{/if}
{:else}
<Loader2 class="animate-spin mt-2" />
{/if}
{/if}
</div>
{:else}
Unknown type
{/if}
</div>
<button
transition:fade|local={{ duration: 100 }}
class="rounded-full p-1 bg-surface-secondary duration-200 hover:bg-surface-hover"
aria-label="Clear"
on:click={() => {
initial_messages = initial_messages.filter((_, index) => index !== i)
}}
>
<X size={14} />
</button>
</div>
{/each}
<div class="flex items-baseline">
<Button
variant="border"
color="light"
size="xs"
btnClasses="mt-1"
on:click={() => {
if (initial_messages == undefined || !Array.isArray(initial_messages)) {
initial_messages = []
}
initial_messages = initial_messages.concat({
raw_message: '""'
})
}}
startIcon={{ icon: Plus }}
>
Add item
</Button>
</div>
</div>
</Section>
<Section label="Filters">
<p class="text-xs mb-1 text-tertiary">
Filters will limit the execution of the trigger to only messages that match all
@@ -266,8 +562,8 @@
</p>
<div class="flex flex-col gap-4 mt-1">
{#each filters as v, i}
<div class="flex w-full gap-4 items-center">
<div class="w-full flex flex-col gap-2">
<div class="flex w-full gap-2 items-center">
<div class="w-full flex flex-col gap-2 border p-2 rounded-md">
<div class="flex flex-row gap-2 w-full">
<label class="flex flex-col w-full">
<div class="text-secondary text-sm">Type</div>
frontend/src/lib/script_helpers.ts
+10
View File
@@ -522,6 +522,9 @@ export async function preprocessor(
params: Record<string, string>
query: Record<string, string>
headers: Record<string, string>
},
websocket?: {
url: string // The websocket url
}
},
/* your other args */
@@ -543,6 +546,9 @@ export async function preprocessor(
params: Record<string, string>
query: Record<string, string>
headers: Record<string, string>
},
websocket?: {
url: string // The websocket url
}
},
/* your other args */
@@ -590,9 +596,13 @@ class Http(TypedDict):
query: dict[str, str]
headers: dict[str, str]
class Websocket(TypedDict):
url: str # The websocket url
class WmTrigger(TypedDict):
kind: Literal["http", "email", "webhook", "websocket"]
http: Http | None
websocket: Websocket | None
def preprocessor(
wm_trigger: WmTrigger,
frontend/src/routes/(root)/(logged)/websocket_triggers/+page.svelte
+44 -42
View File
@@ -45,20 +45,22 @@
loading = false
}
let status: {
[path: string]: { error: string | undefined; last_server_ping: string | undefined }
} = {}
let interval = setInterval(async () => {
try {
status = (
await WebsocketTriggerService.listWebsocketTriggers({
workspace: $workspaceStore!
})
).reduce((acc, x) => {
acc[x.path] = x
return acc
}, {})
const newTriggers = await WebsocketTriggerService.listWebsocketTriggers({
workspace: $workspaceStore!
})
for (let i = 0; i < triggers.length; i++) {
const newTrigger = newTriggers.find((x) => x.path === triggers[i].path)
if (newTrigger) {
triggers[i] = {
...triggers[i],
error: newTrigger.error,
last_server_ping: newTrigger.last_server_ping,
enabled: newTrigger.enabled
}
}
}
} catch (err) {
console.error(err)
}
@@ -258,7 +260,7 @@
<div class="border rounded-md divide-y">
{#each items.slice(0, nbDisplayed) as { path, edited_by, edited_at, script_path, url, is_flow, extra_perms, canWrite, marked, error, last_server_ping, enabled } (path)}
{@const href = `${is_flow ? '/flows/get' : '/scripts/get'}/${script_path}`}
{@const wsStatus = status[path] ?? { error, last_server_ping }}
{@const ping = last_server_ping ? new Date(last_server_ping) : undefined}
<div
class="hover:bg-surface-hover w-full items-center px-4 py-2 gap-4 first-of-type:!border-t-0
@@ -278,7 +280,11 @@
{@html marked}
</span>
{:else}
{url}
{url.startsWith('$script:')
? 'URL: ' + url.replace('$script:', 'result of script ')
: url.startsWith('$flow:')
? 'URL: ' + url.replace('$flow:', 'result of flow ')
: url}
{/if}
</div>
<div class="text-secondary text-xs truncate text-left font-light">
@@ -294,34 +300,30 @@
</div>
<div class="w-10">
{#if enabled}
{@const ping = wsStatus.last_server_ping
? new Date(wsStatus.last_server_ping)
: undefined}
{#if !ping || ping.getTime() < new Date().getTime() - 15 * 1000 || wsStatus.error}
<Popover notClickable>
<span class="flex h-4 w-4">
<Circle
class="text-red-600 animate-ping absolute inline-flex fill-current"
size={12}
/>
<Circle class="text-red-600 relative inline-flex fill-current" size={12} />
</span>
<div slot="text">
Websocket is not connected{wsStatus.error ? ': ' + wsStatus.error : ''}
</div>
</Popover>
{:else}
<Popover notClickable>
<span class="flex h-4 w-4">
<Circle
class="text-green-600 relative inline-flex fill-current"
size={12}
/>
</span>
<div slot="text"> Websocket is connected </div>
</Popover>
{/if}
{#if (enabled && (!ping || ping.getTime() < new Date().getTime() - 15 * 1000 || error)) || (!enabled && error)}
<Popover notClickable>
<span class="flex h-4 w-4">
<Circle
class="text-red-600 animate-ping absolute inline-flex fill-current"
size={12}
/>
<Circle class="text-red-600 relative inline-flex fill-current" size={12} />
</span>
<div slot="text">
{#if enabled}
Websocket is not connected{error ? ': ' + error : ''}
{:else}
Websocket was disabled because of an error: {error}
{/if}
</div>
</Popover>
{:else if enabled}
<Popover notClickable>
<span class="flex h-4 w-4">
<Circle class="text-green-600 relative inline-flex fill-current" size={12} />
</span>
<div slot="text"> Websocket is connected </div>
</Popover>
{/if}
</div>