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

feat: captures (#4807)

Browse Source 
* feat: captures

* flow UI and improvements

* fix: build

* fix sqlx

* Move Capture WIP

* Add capture to webhook and websocket

* Move connection status viewer in the head

* change trigger section label

* Add popover capture picker using melt ui

* Add shortcut to triggers capture from input form

* remove capture tab in input

* Allways show capture

* remove useless logs

* Add email capture

* Add kafka capture into triggers

* Add edit option in capture table

* use light header for arg input

* Add prefilled group id

* Change button label

* fix logic

* Open resource drawer if prototype has fields

* fix default completion

* Change name Prototype

* Dissociate Editor Mode for triggers

* Fix bug for script

* Fix apply args

* fix apply schema

* Add capture table to script

* fix apply args

* Add capture button for script

* Delete capture tab

* Set capture on when opening triggers capture

* fix connection indicator

* fix minor issues

* Add preprocessor logic

* Use slot in log Panel for captures

* handle capture refresh in script

* Delete capture tab from script editor

* reset kafka resource on toggle static

* fix minor issue

* Allow resource in kafka capture

* use simple capture button in flow

* Remove capture panel

* Polish route trigger editor

* Fix resource saving

* Remove excessive padding

* merge nits

* add workflow_dispatch to build

* add workflow_dispatch to build

* better capture UI

* fix sqlx

* fix build

* fix build

* build

* make initial_messages optional in line with db

* fix npm check

* better handle args for capture webhook and http

* improve migration + http capture fixes

* fix sqlx

* update ee ref

---------

Co-authored-by: Guilhem <guilhemlemouel@gmail.com>
Co-authored-by: Ruben Fiszel <ruben@rubenfiszel.com>
Co-authored-by: Ruben Fiszel <ruben@windmill.dev>
This commit is contained in:
HugoCasa
2025-01-06 12:07:09 +01:00
committed by GitHub
co-authored by Guilhem Ruben Fiszel Ruben Fiszel
parent 3f0a818fa3
commit fcb6ced8df
104 changed files with 5457 additions and 1991 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api/src/capture.rs
+482 -83
View File
@@ -7,138 +7,537 @@
*/
use axum::{
extract::{Extension, Path},
routing::{get, post, put},
Router,
extract::{Extension, Path, Query},
routing::{delete, get, head, post},
Json, Router,
};
#[cfg(feature = "http_trigger")]
use http::HeaderMap;
use hyper::StatusCode;
use sqlx::types::Json;
#[cfg(feature = "http_trigger")]
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use serde_json::value::RawValue;
use sqlx::types::Json as SqlxJson;
#[cfg(feature = "http_trigger")]
use std::collections::HashMap;
use std::fmt;
#[cfg(feature = "http_trigger")]
use windmill_common::error::Error;
use windmill_common::{
db::UserDB,
error::{JsonResult, Result},
utils::{not_found_if_none, StripPath},
worker::{to_raw_value, CLOUD_HOSTED},
};
use windmill_queue::PushArgs;
use windmill_queue::{PushArgs, PushArgsOwned};
#[cfg(feature = "http_trigger")]
use crate::http_triggers::{build_http_trigger_extra, HttpMethod};
#[cfg(all(feature = "enterprise", feature = "kafka"))]
use crate::kafka_triggers_ee::KafkaResourceSecurity;
use crate::{
args::WebhookArgs,
db::{ApiAuthed, DB},
users::fetch_api_authed,
};
const KEEP_LAST: i64 = 8;
const KEEP_LAST: i64 = 20;
pub fn workspaced_service() -> Router {
Router::new()
.route("/*path", put(new_payload))
.route("/*path", get(get_payload))
.route("/set_config", post(set_config))
.route(
"/ping_config/:trigger_kind/:runnable_kind/*path",
post(ping_config),
)
.route("/get_configs/:runnable_kind/*path", get(get_configs))
.route("/list/:runnable_kind/*path", get(list_captures))
.route("/:id", delete(delete_capture))
}
pub fn global_service() -> Router {
Router::new().route("/*path", post(update_payload))
pub fn workspaced_unauthed_service() -> Router {
let router = Router::new().route(
"/webhook/:runnable_kind/*path",
head(|| async {}).post(webhook_payload),
);
#[cfg(feature = "http_trigger")]
{
router.route("/http/:runnable_kind/:path/*route_path", {
head(|| async {}).fallback(http_payload)
})
}
#[cfg(not(feature = "http_trigger"))]
{
router
}
}
pub async fn new_payload(
#[derive(sqlx::Type, Serialize, Deserialize)]
#[sqlx(type_name = "TRIGGER_KIND", rename_all = "lowercase")]
#[serde(rename_all = "lowercase")]
pub enum TriggerKind {
Webhook,
Http,
Websocket,
Kafka,
Email,
}
impl fmt::Display for TriggerKind {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let s = match self {
TriggerKind::Webhook => "webhook",
TriggerKind::Http => "http",
TriggerKind::Websocket => "websocket",
TriggerKind::Kafka => "kafka",
TriggerKind::Email => "email",
};
write!(f, "{}", s)
}
}
#[cfg(feature = "http_trigger")]
#[derive(Serialize, Deserialize)]
struct HttpTriggerConfig {
route_path: String,
http_method: HttpMethod,
}
#[cfg(all(feature = "enterprise", feature = "kafka"))]
#[derive(Serialize, Deserialize)]
#[serde(untagged)]
pub enum KafkaTriggerConfigConnection {
Resource { kafka_resource_path: String },
Static { brokers: Vec<String>, security: KafkaResourceSecurity },
}
#[cfg(all(feature = "enterprise", feature = "kafka"))]
#[derive(Serialize, Deserialize)]
pub struct KafkaTriggerConfig {
#[serde(flatten)]
pub connection: KafkaTriggerConfigConnection,
pub topics: Vec<String>,
pub group_id: String,
}
#[derive(Serialize, Deserialize, Debug)]
pub struct WebsocketTriggerConfig {
pub url: String,
// have to use Value because RawValue is not supported inside untagged
pub url_runnable_args: Option<serde_json::Value>,
}
#[derive(Serialize, Deserialize)]
#[serde(untagged)]
enum TriggerConfig {
#[cfg(feature = "http_trigger")]
Http(HttpTriggerConfig),
Websocket(WebsocketTriggerConfig),
#[cfg(all(feature = "enterprise", feature = "kafka"))]
Kafka(KafkaTriggerConfig),
}
#[derive(Serialize, Deserialize)]
struct NewCaptureConfig {
trigger_kind: TriggerKind,
path: String,
is_flow: bool,
trigger_config: Option<TriggerConfig>,
}
#[derive(Serialize, Deserialize)]
struct CaptureConfig {
trigger_config: Option<SqlxJson<Box<RawValue>>>,
trigger_kind: TriggerKind,
error: Option<String>,
last_server_ping: Option<chrono::DateTime<chrono::Utc>>,
}
async fn get_configs(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Path((w_id, path)): Path<(String, StripPath)>,
) -> Result<StatusCode> {
Path((w_id, runnable_kind, path)): Path<(String, RunnableKind, StripPath)>,
) -> JsonResult<Vec<CaptureConfig>> {
let mut tx = user_db.begin(&authed).await?;
let configs = sqlx::query_as!(
CaptureConfig,
r#"SELECT trigger_config as "trigger_config: _", trigger_kind as "trigger_kind: _", error, last_server_ping
FROM capture_config
WHERE workspace_id = $1 AND path = $2 AND is_flow = $3"#,
&w_id,
&path.to_path(),
matches!(runnable_kind, RunnableKind::Flow),
)
.fetch_all(&mut *tx)
.await?;
tx.commit().await?;
Ok(Json(configs))
}
async fn set_config(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Path(w_id): Path<String>,
Json(nc): Json<NewCaptureConfig>,
) -> Result<()> {
let mut tx = user_db.begin(&authed).await?;
sqlx::query!(
"
INSERT INTO capture
(workspace_id, path, created_by)
VALUES ($1, $2, $3)
ON CONFLICT (workspace_id, path)
DO UPDATE SET created_at = now()
",
"INSERT INTO capture_config
(workspace_id, path, is_flow, trigger_kind, trigger_config, owner, email)
VALUES ($1, $2, $3, $4, $5, $6, $7)
ON CONFLICT (workspace_id, path, is_flow, trigger_kind)
DO UPDATE SET trigger_config = $5, owner = $6, email = $7, server_id = NULL, last_server_ping = NULL, error = NULL",
&w_id,
&path.to_path(),
&nc.path,
nc.is_flow,
nc.trigger_kind as TriggerKind,
nc.trigger_config.map(|x| SqlxJson(to_raw_value(&x))) as Option<SqlxJson<Box<RawValue>>>,
&authed.username,
)
.execute(&mut *tx)
.await?;
/* Retain only KEEP_LAST most recent captures by this user in this workspace. */
sqlx::query!(
"
DELETE FROM capture
WHERE workspace_id = $1
AND created_by = $2
AND created_at <=
( SELECT created_at
FROM capture
WHERE workspace_id = $1
AND created_by = $2
ORDER BY created_at DESC
OFFSET $3
LIMIT 1 )
",
&w_id,
&authed.username,
KEEP_LAST,
&authed.email,
)
.execute(&mut *tx)
.await?;
tx.commit().await?;
Ok(StatusCode::CREATED)
Ok(())
}
pub async fn update_payload(
Extension(db): Extension<DB>,
Path((w_id, path)): Path<(String, StripPath)>,
args: WebhookArgs,
) -> Result<StatusCode> {
let args = args.args;
let mut tx = db.begin().await?;
async fn ping_config(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Path((w_id, trigger_kind, runnable_kind, path)): Path<(
String,
TriggerKind,
RunnableKind,
StripPath,
)>,
) -> Result<()> {
let mut tx = user_db.begin(&authed).await?;
sqlx::query!(
"
UPDATE capture
SET payload = $3
WHERE workspace_id = $1
AND path = $2
",
"UPDATE capture_config SET last_client_ping = now() WHERE workspace_id = $1 AND path = $2 AND is_flow = $3 AND trigger_kind = $4",
&w_id,
&path.to_path(),
Json(PushArgs { args: &args.args, extra: args.extra }) as Json<PushArgs>,
matches!(runnable_kind, RunnableKind::Flow),
trigger_kind as TriggerKind,
)
.execute(&mut *tx)
.await?;
tx.commit().await?;
Ok(())
}
#[derive(Serialize, Deserialize)]
struct Capture {
id: i64,
created_at: chrono::DateTime<chrono::Utc>,
trigger_kind: TriggerKind,
payload: SqlxJson<Box<serde_json::value::RawValue>>,
trigger_extra: Option<SqlxJson<Box<serde_json::value::RawValue>>>,
}
#[derive(Deserialize)]
#[serde(rename_all = "lowercase")]
enum RunnableKind {
Script,
Flow,
}
#[derive(Deserialize)]
struct ListCapturesQuery {
trigger_kind: Option<TriggerKind>,
}
async fn list_captures(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Path((w_id, runnable_kind, path)): Path<(String, RunnableKind, StripPath)>,
Query(query): Query<ListCapturesQuery>,
) -> JsonResult<Vec<Capture>> {
let mut tx = user_db.begin(&authed).await?;
let captures = sqlx::query_as!(
Capture,
r#"SELECT id, created_at, trigger_kind as "trigger_kind: _", payload as "payload: _", trigger_extra as "trigger_extra: _"
FROM capture
WHERE workspace_id = $1
AND path = $2 AND is_flow = $3
AND ($4::trigger_kind IS NULL OR trigger_kind = $4)
ORDER BY created_at DESC"#,
&w_id,
&path.to_path(),
matches!(runnable_kind, RunnableKind::Flow),
query.trigger_kind as Option<TriggerKind>,
)
.fetch_all(&mut *tx)
.await?;
tx.commit().await?;
Ok(Json(captures))
}
async fn delete_capture(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Path((_, id)): Path<(String, i64)>,
) -> Result<()> {
let mut tx = user_db.begin(&authed).await?;
sqlx::query!("DELETE FROM capture WHERE id = $1", id)
.execute(&mut *tx)
.await?;
tx.commit().await?;
Ok(())
}
#[derive(Serialize, Deserialize)]
struct ActiveCaptureOwner {
owner: String,
email: String,
}
pub async fn get_active_capture_owner_and_email(
db: &DB,
w_id: &str,
path: &str,
is_flow: bool,
kind: &TriggerKind,
) -> Result<(String, String)> {
let capture_config = sqlx::query_as!(
ActiveCaptureOwner,
"SELECT owner, email
FROM capture_config
WHERE workspace_id = $1 AND path = $2 AND is_flow = $3 AND trigger_kind = $4 AND last_client_ping > NOW() - INTERVAL '10 seconds'",
&w_id,
&path,
is_flow,
kind as &TriggerKind,
)
.fetch_optional(db)
.await?;
let capture_config = not_found_if_none(
capture_config,
&format!("capture config for {} trigger", kind),
path,
)?;
Ok((capture_config.owner, capture_config.email))
}
#[cfg(feature = "http_trigger")]
async fn get_capture_trigger_config_and_owner<T: DeserializeOwned>(
db: &DB,
w_id: &str,
path: &str,
is_flow: bool,
kind: &TriggerKind,
) -> Result<(T, String, String)> {
#[derive(Deserialize)]
struct CaptureTriggerConfigAndOwner {
trigger_config: Option<SqlxJson<Box<RawValue>>>,
owner: String,
email: String,
}
let capture_config = sqlx::query_as!(
CaptureTriggerConfigAndOwner,
r#"SELECT trigger_config as "trigger_config: _", owner, email
FROM capture_config
WHERE workspace_id = $1 AND path = $2 AND is_flow = $3 AND trigger_kind = $4 AND last_client_ping > NOW() - INTERVAL '10 seconds'"#,
&w_id,
&path,
is_flow,
kind as &TriggerKind,
)
.fetch_optional(db)
.await?;
let capture_config = not_found_if_none(
capture_config,
&format!("capture config for {} trigger", kind),
path,
)?;
let trigger_config = not_found_if_none(
capture_config.trigger_config,
&format!("capture {} trigger config", kind),
path,
)?;
Ok((
serde_json::from_str(trigger_config.get()).map_err(|e| {
Error::InternalErr(format!(
"error parsing capture config for {} trigger: {}",
kind, e
))
})?,
capture_config.owner,
capture_config.email,
))
}
async fn clear_captures_history(db: &DB, w_id: &str) -> Result<()> {
if *CLOUD_HOSTED {
/* Retain only KEEP_LAST most recent captures in this workspace. */
sqlx::query!(
"DELETE FROM capture
WHERE workspace_id = $1
AND created_at <=
(
SELECT created_at
FROM capture
WHERE workspace_id = $1
ORDER BY created_at DESC
OFFSET $2
LIMIT 1
)",
&w_id,
KEEP_LAST,
)
.execute(db)
.await?;
}
Ok(())
}
pub async fn insert_capture_payload(
db: &DB,
w_id: &str,
path: &str,
is_flow: bool,
trigger_kind: &TriggerKind,
payload: PushArgsOwned,
trigger_extra: Option<Box<RawValue>>,
owner: &str,
) -> Result<()> {
sqlx::query!(
"INSERT INTO capture (workspace_id, path, is_flow, trigger_kind, payload, trigger_extra, created_by)
VALUES ($1, $2, $3, $4, $5, $6, $7)",
&w_id,
path,
is_flow,
trigger_kind as &TriggerKind,
SqlxJson(to_raw_value(&PushArgs {
args: &payload.args,
extra: payload.extra
})) as SqlxJson<Box<RawValue>>,
trigger_extra.map(SqlxJson) as Option<SqlxJson<Box<RawValue>>>,
owner,
)
.execute(db)
.await?;
clear_captures_history(db, &w_id).await?;
Ok(())
}
async fn webhook_payload(
Extension(db): Extension<DB>,
Path((w_id, runnable_kind, path)): Path<(String, RunnableKind, StripPath)>,
args: WebhookArgs,
) -> Result<StatusCode> {
let (owner, email) = get_active_capture_owner_and_email(
&db,
&w_id,
&path.to_path(),
matches!(runnable_kind, RunnableKind::Flow),
&TriggerKind::Webhook,
)
.await?;
let authed = fetch_api_authed(owner.clone(), email, &w_id, &db, None).await?;
let args = args.to_push_args_owned(&authed, &db, &w_id).await?;
insert_capture_payload(
&db,
&w_id,
&path.to_path(),
matches!(runnable_kind, RunnableKind::Flow),
&TriggerKind::Webhook,
args,
Some(to_raw_value(&serde_json::json!({
"wm_trigger": {
"kind": "webhook",
}
}))),
&owner,
)
.await?;
Ok(StatusCode::NO_CONTENT)
}
#[derive(sqlx::FromRow)]
struct Payload {
payload: sqlx::types::Json<Box<serde_json::value::RawValue>>,
}
pub async fn get_payload(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Path((w_id, path)): Path<(String, StripPath)>,
) -> JsonResult<Box<serde_json::value::RawValue>> {
let mut tx = user_db.begin(&authed).await?;
#[cfg(feature = "http_trigger")]
async fn http_payload(
Extension(db): Extension<DB>,
Path((w_id, kind, path, route_path)): Path<(String, RunnableKind, String, StripPath)>,
Query(query): Query<HashMap<String, String>>,
method: http::Method,
headers: HeaderMap,
args: WebhookArgs,
) -> Result<StatusCode> {
let route_path = route_path.to_path();
let path = path.replace(".", "/");
let payload = sqlx::query_as::<_, Payload>(
"
SELECT payload
FROM capture
WHERE workspace_id = $1
AND path = $2
",
let (http_trigger_config, owner, email): (HttpTriggerConfig, _, _) =
get_capture_trigger_config_and_owner(
&db,
&w_id,
&path,
matches!(kind, RunnableKind::Flow),
&TriggerKind::Http,
)
.await?;
let authed = fetch_api_authed(owner.clone(), email, &w_id, &db, None).await?;
let args = args.to_push_args_owned(&authed, &db, &w_id).await?;
let mut router = matchit::Router::new();
router.insert(&http_trigger_config.route_path, ()).ok();
let match_ = router.at(route_path).ok();
let match_ = not_found_if_none(match_, "capture http trigger", &route_path)?;
let matchit::Match { params, .. } = match_;
let params: HashMap<String, String> = params
.iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect();
let extra: HashMap<String, Box<RawValue>> = HashMap::from_iter(vec![(
"wm_trigger".to_string(),
build_http_trigger_extra(
&http_trigger_config.route_path,
route_path,
&method,
&params,
&query,
&headers,
)
.await,
)]);
insert_capture_payload(
&db,
&w_id,
&path,
matches!(kind, RunnableKind::Flow),
&TriggerKind::Http,
args,
Some(to_raw_value(&extra)),
&owner,
)
.bind(&w_id)
.bind(&path.to_path())
.fetch_optional(&mut *tx)
.await?;
tx.commit().await?;
not_found_if_none(payload.map(|x| x.payload.0), "capture", path.to_path()).map(axum::Json)
Ok(StatusCode::NO_CONTENT)
}
backend/windmill-api/src/flows.rs
+34
View File
@@ -740,6 +740,24 @@ async fn update_flow(
"Error updating flow due to deleting old flow: {e:#}"
))
})?;
sqlx::query!(
"UPDATE capture_config SET path = $1 WHERE path = $2 AND workspace_id = $3 AND is_flow IS TRUE",
nf.path,
flow_path,
w_id
)
.execute(&mut *tx)
.await?;
sqlx::query!(
"UPDATE capture SET path = $1 WHERE path = $2 AND workspace_id = $3 AND is_flow IS TRUE",
nf.path,
flow_path,
w_id
)
.execute(&mut *tx)
.await?;
}
let version = sqlx::query_scalar!(
@@ -1123,6 +1141,22 @@ async fn delete_flow_by_path(
.execute(&mut *tx)
.await?;
sqlx::query!(
"DELETE FROM capture_config WHERE path = $1 AND workspace_id = $2 AND is_flow IS TRUE",
path,
&w_id
)
.execute(&mut *tx)
.await?;
sqlx::query!(
"DELETE FROM capture WHERE path = $1 AND workspace_id = $2 AND is_flow IS TRUE",
path,
&w_id
)
.execute(&mut *tx)
.await?;
audit_log(
&mut *tx,
&authed,
backend/windmill-api/src/http_triggers.rs
+36 -16
View File
@@ -77,7 +77,7 @@ pub fn workspaced_service() -> Router {
#[derive(Serialize, Deserialize, sqlx::Type)]
#[sqlx(type_name = "HTTP_METHOD", rename_all = "lowercase")]
#[serde(rename_all = "lowercase")]
enum HttpMethod {
pub enum HttpMethod {
Get,
Post,
Put,
@@ -523,6 +523,32 @@ async fn get_http_route_trigger(
Ok((trigger, route_path.0, params, authed))
}
pub async fn build_http_trigger_extra(
route_path: &str,
called_path: &str,
method: &http::Method,
params: &HashMap<String, String>,
query: &HashMap<String, String>,
headers: &HeaderMap,
) -> Box<serde_json::value::RawValue> {
let headers = headers
.iter()
.map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string()))
.collect::<HashMap<String, String>>();
to_raw_value(&serde_json::json!({
"kind": "http",
"http": {
"route": route_path,
"path": called_path,
"method": method.to_string().to_lowercase(),
"params": params,
"query": query,
"headers": headers
},
}))
}
async fn route_job(
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
@@ -639,24 +665,18 @@ async fn route_job(
}
}
let headers = headers
.iter()
.map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string()))
.collect::<HashMap<String, String>>();
let extra = args.extra.get_or_insert_with(HashMap::new);
extra.insert(
"wm_trigger".to_string(),
to_raw_value(&serde_json::json!({
"kind": "http",
"http": {
"route": trigger.route_path,
"path": called_path,
"method": method.to_string().to_lowercase(),
"params": params,
"query": query,
"headers": headers
},
})),
build_http_trigger_extra(
&trigger.route_path,
&called_path,
&method,
&params,
&query,
&headers,
)
.await,
);
let http_method = http::Method::from(trigger.http_method);
backend/windmill-api/src/kafka_triggers_ee.rs
+4
View File
@@ -1,5 +1,9 @@
use crate::db::DB;
use axum::Router;
use serde::{Deserialize, Serialize};
#[derive(Serialize, Deserialize)]
pub struct KafkaResourceSecurity {}
pub fn workspaced_service() -> Router {
Router::new()
backend/windmill-api/src/lib.rs
+1 -4
View File
@@ -250,9 +250,6 @@ pub async fn run_server(
}
}
// #[cfg(feature = "kafka")]
// start_listening().await;
let job_helpers_service = {
#[cfg(feature = "parquet")]
{
@@ -424,7 +421,7 @@ pub async fn run_server(
)
.nest(
"/w/:workspace_id/capture_u",
capture::global_service().layer(cors.clone()),
capture::workspaced_unauthed_service().layer(cors.clone()),
)
.nest(
"/auth",
backend/windmill-api/src/scripts.rs
+34
View File
@@ -662,6 +662,24 @@ async fn create_script_internal<'c>(
.execute(&mut *tx)
.await?;
sqlx::query!(
"UPDATE capture_config SET path = $1 WHERE path = $2 AND workspace_id = $3 AND is_flow IS FALSE",
ns.path,
p_path,
w_id
)
.execute(&mut *tx)
.await?;
sqlx::query!(
"UPDATE capture SET path = $1 WHERE path = $2 AND workspace_id = $3 AND is_flow IS FALSE",
ns.path,
p_path,
w_id
)
.execute(&mut *tx)
.await?;
let mut schedulables = sqlx::query_as::<_, Schedule>(
"UPDATE schedule SET script_path = $1 WHERE script_path = $2 AND path != $2 AND workspace_id = $3 AND is_flow IS false RETURNING *")
.bind(&ns.path)
@@ -1471,6 +1489,22 @@ async fn delete_script_by_path(
.execute(&db)
.await?;
sqlx::query!(
"DELETE FROM capture_config WHERE path = $1 AND workspace_id = $2 AND is_flow IS FALSE",
path,
w_id
)
.execute(&db)
.await?;
sqlx::query!(
"DELETE FROM capture WHERE path = $1 AND workspace_id = $2 AND is_flow IS FALSE",
path,
w_id
)
.execute(&db)
.await?;
audit_log(
&mut *tx,
&authed,
backend/windmill-api/src/websocket_triggers.rs
+476 -219
View File
@@ -15,6 +15,7 @@ use serde::{
use serde_json::{value::RawValue, Value};
use sql_builder::{bind::Bind, SqlBuilder};
use sqlx::prelude::FromRow;
use sqlx::types::Json as SqlxJson;
use std::{collections::HashMap, fmt};
use tokio::net::TcpStream;
use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream};
@@ -30,11 +31,14 @@ use windmill_common::{
use windmill_queue::PushArgsOwned;
use crate::{
capture::{insert_capture_payload, TriggerKind, WebsocketTriggerConfig},
db::{ApiAuthed, DB},
jobs::{run_flow_by_path_inner, run_script_by_path_inner, RunJobQuery},
users::fetch_api_authed,
};
use std::borrow::Cow;
pub fn workspaced_service() -> Router {
Router::new()
.route("/create", post(create_websocket_trigger))
@@ -54,19 +58,19 @@ struct NewWebsocketTrigger {
is_flow: bool,
enabled: Option<bool>,
filters: Vec<Box<RawValue>>,
initial_messages: Vec<Box<RawValue>>,
url_runnable_args: Box<RawValue>,
initial_messages: Option<Vec<Box<RawValue>>>,
url_runnable_args: Option<Box<RawValue>>,
}
#[derive(Deserialize)]
struct JsonFilter {
pub struct JsonFilter {
key: String,
value: serde_json::Value,
}
#[derive(Deserialize)]
#[serde(untagged)]
enum Filter {
pub enum Filter {
JsonFilter(JsonFilter),
}
@@ -93,9 +97,9 @@ pub struct WebsocketTrigger {
extra_perms: serde_json::Value,
error: Option<String>,
enabled: bool,
filters: Vec<sqlx::types::Json<Box<RawValue>>>,
initial_messages: Vec<sqlx::types::Json<Box<RawValue>>>,
url_runnable_args: sqlx::types::Json<Box<RawValue>>,
filters: Vec<SqlxJson<Box<RawValue>>>,
initial_messages: Option<Vec<SqlxJson<Box<RawValue>>>>,
url_runnable_args: Option<SqlxJson<Box<RawValue>>>,
}
#[derive(Deserialize)]
@@ -105,8 +109,8 @@ struct EditWebsocketTrigger {
script_path: String,
is_flow: bool,
filters: Vec<Box<RawValue>>,
initial_messages: Vec<Box<RawValue>>,
url_runnable_args: Box<RawValue>,
initial_messages: Option<Vec<Box<RawValue>>>,
url_runnable_args: Option<Box<RawValue>>,
}
#[derive(Deserialize)]
@@ -190,11 +194,12 @@ async fn create_websocket_trigger(
let mut tx = user_db.begin(&authed).await?;
let filters = ct.filters.into_iter().map(sqlx::types::Json).collect_vec();
let filters = ct.filters.into_iter().map(SqlxJson).collect_vec();
let initial_messages = ct
.initial_messages
.unwrap_or_default()
.into_iter()
.map(sqlx::types::Json)
.map(SqlxJson)
.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 *",
@@ -207,7 +212,7 @@ async fn create_websocket_trigger(
.bind(ct.enabled.unwrap_or(true))
.bind(filters.as_slice())
.bind(initial_messages.as_slice())
.bind(sqlx::types::Json(ct.url_runnable_args))
.bind(ct.url_runnable_args.map(SqlxJson))
.bind(&authed.username)
.bind(&authed.email)
.fetch_one(&mut *tx).await?;
@@ -237,11 +242,12 @@ 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 filters = ct.filters.into_iter().map(SqlxJson).collect_vec();
let initial_messages = ct
.initial_messages
.unwrap_or_default()
.into_iter()
.map(sqlx::types::Json)
.map(SqlxJson)
.collect_vec();
// important to update server_id, last_server_ping and error to NULL to stop current websocket listener
@@ -252,9 +258,9 @@ async fn update_websocket_trigger(
ct.script_path,
ct.path,
ct.is_flow,
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>>,
filters.as_slice() as &[SqlxJson<Box<RawValue>>],
initial_messages.as_slice() as &[SqlxJson<Box<RawValue>>],
ct.url_runnable_args.map(SqlxJson) as Option<SqlxJson<Box<RawValue>>>,
&authed.username,
&authed.email,
w_id,
@@ -386,18 +392,36 @@ async fn listen_to_unlistened_websockets(
Ok(mut triggers) => {
triggers.shuffle(&mut rand::thread_rng());
for trigger in triggers {
maybe_listen_to_websocket(trigger, db.clone(), killpill_rx.resubscribe()).await;
trigger.maybe_listen_to_websocket(db.clone(), killpill_rx.resubscribe()).await;
}
}
Err(err) => {
tracing::error!("Error fetching websocket triggers: {:?}", err);
}
};
match sqlx::query_as!(
CaptureConfigForWebsocket,
r#"SELECT path, is_flow, workspace_id, trigger_config as "trigger_config!: _", owner, email FROM capture_config WHERE trigger_kind = 'websocket' AND last_client_ping > NOW() - INTERVAL '10 seconds' AND trigger_config IS NOT NULL AND (server_id IS NULL OR last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds')"#
)
.fetch_all(db)
.await
{
Ok(mut captures) => {
captures.shuffle(&mut rand::thread_rng());
for capture in captures {
capture.maybe_listen_to_websocket(db.clone(), killpill_rx.resubscribe()).await;
}
}
Err(err) => {
tracing::error!("Error fetching capture websocket triggers: {:?}", err);
}
}
}
pub async fn start_websockets(db: DB, mut killpill_rx: tokio::sync::broadcast::Receiver<()>) -> () {
tokio::spawn(async move {
listen_to_unlistened_websockets(&db, &&killpill_rx).await;
listen_to_unlistened_websockets(&db, &killpill_rx).await;
loop {
tokio::select! {
biased;
@@ -405,37 +429,13 @@ pub async fn start_websockets(db: DB, mut killpill_rx: tokio::sync::broadcast::R
return;
}
_ = tokio::time::sleep(tokio::time::Duration::from_secs(15)) => {
listen_to_unlistened_websockets(&db, &&killpill_rx).await;
listen_to_unlistened_websockets(&db, &killpill_rx).await;
}
}
}
});
}
async fn maybe_listen_to_websocket(
ws_trigger: WebsocketTrigger,
db: DB,
killpill_rx: tokio::sync::broadcast::Receiver<()>,
) -> () {
match sqlx::query_scalar!(
"UPDATE websocket_trigger SET server_id = $1, last_server_ping = now() WHERE enabled IS TRUE AND workspace_id = $2 AND path = $3 AND (server_id IS NULL OR last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') RETURNING true",
*INSTANCE_NAME,
ws_trigger.workspace_id,
ws_trigger.path,
).fetch_optional(&db).await {
Ok(has_lock) => {
if has_lock.flatten().unwrap_or(false) {
tokio::spawn(listen_to_websocket(ws_trigger, db, killpill_rx));
} else {
tracing::info!("Websocket {} already being listened to", ws_trigger.url);
}
},
Err(err) => {
tracing::error!("Error acquiring lock for websocket {}: {:?}", ws_trigger.path, err);
}
};
}
struct SupersetVisitor<'a> {
key: &'a str,
value_to_check: &'a Value,
@@ -505,32 +505,29 @@ where
async fn wait_runnable_result(
path: String,
is_flow: bool,
args: &Box<RawValue>,
ws_trigger: &WebsocketTrigger,
username_override: String,
args: Option<&Box<RawValue>>,
authed: ApiAuthed,
db: &DB,
workspace_id: &str,
trigger_path: &str,
) -> 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,
Some(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 args = if let Some(args) = 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)
} else {
HashMap::new()
};
let label_prefix = Some(format!("ws-{}-", ws_trigger.path));
let label_prefix = Some(format!("ws-{}-", trigger_path));
let (_, job_id) = if is_flow {
run_flow_by_path_inner(
authed,
db.clone(),
user_db,
ws_trigger.workspace_id.clone(),
workspace_id.to_string(),
StripPath(path.clone()),
RunJobQuery::default(),
PushArgsOwned { args, extra: None },
@@ -542,7 +539,7 @@ async fn wait_runnable_result(
authed,
db.clone(),
user_db,
ws_trigger.workspace_id.clone(),
workspace_id.to_string(),
StripPath(path.clone()),
RunJobQuery::default(),
PushArgsOwned { args, extra: None },
@@ -563,7 +560,7 @@ async fn wait_runnable_result(
#[derive(sqlx::FromRow)]
struct RawResult {
result: Option<sqlx::types::Json<Box<RawValue>>>,
result: Option<SqlxJson<Box<RawValue>>>,
success: bool,
}
@@ -571,7 +568,7 @@ async fn wait_runnable_result(
"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)
.bind(workspace_id)
.fetch_optional(db)
.await;
@@ -601,92 +598,34 @@ async fn wait_runnable_result(
}
}
async fn send_initial_messages(
ws_trigger: &WebsocketTrigger,
mut writer: SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>,
db: &DB,
) -> 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,
)
.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")
})?;
}
async fn loop_ping(db: &DB, ws: &WebsocketEnum, error: Option<&str>) -> () {
loop {
if let None = ws.update_ping(db, error).await {
return;
}
tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
}
Ok(())
}
async fn get_url_from_runnable(
path: &str,
is_flow: bool,
ws_trigger: &WebsocketTrigger,
db: &DB,
authed: ApiAuthed,
args: Option<&Box<RawValue>>,
workspace_id: &str,
trigger_path: &str,
) -> 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(),
args,
authed,
db,
workspace_id,
trigger_path,
)
.await?;
@@ -697,78 +636,394 @@ async fn get_url_from_runnable(
}
}
async fn update_ping(db: &DB, ws_trigger: &WebsocketTrigger, error: Option<&str>) -> Option<()> {
match sqlx::query_scalar!(
impl WebsocketTrigger {
async fn maybe_listen_to_websocket(
self,
db: DB,
killpill_rx: tokio::sync::broadcast::Receiver<()>,
) -> () {
match sqlx::query_scalar!(
"UPDATE websocket_trigger SET server_id = $1, last_server_ping = now() WHERE enabled IS TRUE AND workspace_id = $2 AND path = $3 AND (server_id IS NULL OR last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') RETURNING true",
*INSTANCE_NAME,
self.workspace_id,
self.path,
).fetch_optional(&db).await {
Ok(has_lock) => {
if has_lock.flatten().unwrap_or(false) {
tokio::spawn(listen_to_websocket(WebsocketEnum::Trigger(self), db, killpill_rx));
} else {
tracing::info!("Websocket {} already being listened to", self.url);
}
},
Err(err) => {
tracing::error!("Error acquiring lock for websocket {}: {:?}", self.path, err);
}
};
}
async fn update_ping(&self, db: &DB, 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,
self.workspace_id,
self.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);
tracing::info!("Websocket {} changed, disabled, or deleted, stopping...", self.url);
return None;
}
},
Err(err) => {
tracing::warn!("Error updating ping of websocket {}: {:?}", ws_trigger.url, err);
}
};
Err(err) => {
tracing::warn!("Error updating ping of websocket {}: {:?}", self.url, err);
}
};
Some(())
}
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;
async fn disable_with_error(&self, db: &DB, 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,
self.workspace_id,
self.path,
)
.execute(db).await {
Ok(_) => {
report_critical_error(format!("Disabling websocket {} because of error: {}", self.url, error), db.clone(), Some(&self.workspace_id), None).await;
},
Err(disable_err) => {
report_critical_error(
format!("Could not disable websocket {} with err {}, disabling because of error {}", self.path, disable_err, error),
db.clone(),
Some(&self.workspace_id),
None,
).await;
}
}
tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
}
async fn get_url_from_runnable(
&self,
path: &str,
is_flow: bool,
db: &DB,
) -> error::Result<String> {
get_url_from_runnable(
&path,
is_flow,
db,
self.fetch_authed(db, Some("url".to_string())).await?,
self.url_runnable_args.as_ref().map(|r| &r.0),
&self.workspace_id,
&self.path,
)
.await
}
async fn send_initial_messages(
&self,
mut writer: SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>,
db: &DB,
) -> error::Result<()> {
let initial_messages: Vec<InitialMessage> = self
.initial_messages
.as_deref()
.unwrap_or_default()
.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 {}: {}",
self.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 {}",
self.url,
);
let result = wait_runnable_result(
path.clone(),
is_flow,
Some(&args),
self.fetch_authed(db, Some("init".to_string())).await?,
db,
&self.workspace_id,
&self.path,
)
.await?;
tracing::info!(
"Sending runnable {path} (is_flow: {is_flow}) result to websocket {}",
self.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 handle(&self, db: &DB, args: PushArgsOwned) -> () {
if let Err(err) = run_job(db, self, args).await {
report_critical_error(
format!(
"Failed to trigger job from websocket {}: {:?}",
self.url, err
),
db.clone(),
Some(&self.workspace_id),
None,
)
.await;
};
}
async fn fetch_authed(
&self,
db: &DB,
username_override: Option<String>,
) -> error::Result<ApiAuthed> {
fetch_api_authed(
self.edited_by.clone(),
self.email.clone(),
&self.workspace_id,
db,
username_override,
)
.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",
#[derive(Deserialize)]
struct CaptureConfigForWebsocket {
trigger_config: SqlxJson<WebsocketTriggerConfig>,
path: String,
is_flow: bool,
workspace_id: String,
owner: String,
email: String,
}
impl CaptureConfigForWebsocket {
async fn maybe_listen_to_websocket(
self,
db: DB,
killpill_rx: tokio::sync::broadcast::Receiver<()>,
) -> () {
match sqlx::query_scalar!(
"UPDATE capture_config SET server_id = $1, last_server_ping = now() WHERE last_client_ping > NOW() - INTERVAL '10 seconds' AND workspace_id = $2 AND path = $3 AND is_flow = $4 AND trigger_kind = 'websocket' AND (server_id IS NULL OR last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') RETURNING true",
*INSTANCE_NAME,
self.workspace_id,
self.path,
self.is_flow,
).fetch_optional(&db).await {
Ok(has_lock) => {
if has_lock.flatten().unwrap_or(false) {
tokio::spawn(listen_to_websocket(WebsocketEnum::Capture(self), db, killpill_rx));
} else {
tracing::info!("Websocket {} already being listened to", self.trigger_config.url);
}
},
Err(err) => {
tracing::error!("Error acquiring lock for capture websocket {}: {:?}", self.path, err);
}
};
}
async fn update_ping(&self, db: &DB, error: Option<&str>) -> Option<()> {
match sqlx::query_scalar!(
"UPDATE capture_config SET last_server_ping = now(), error = $1 WHERE workspace_id = $2 AND path = $3 AND is_flow = $4 AND trigger_kind = 'websocket' AND server_id = $5 AND last_client_ping > NOW() - INTERVAL '10 seconds' RETURNING 1",
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(), Some(&ws_trigger.workspace_id), None).await;
self.workspace_id,
self.path,
self.is_flow,
*INSTANCE_NAME
).fetch_optional(db).await {
Ok(updated) => {
if updated.flatten().is_none() {
tracing::info!("Websocket capture {} changed, disabled, or deleted, stopping...", self.trigger_config.url);
return None;
}
},
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(),
Some(&ws_trigger.workspace_id),
None,
).await;
Err(err) => {
tracing::warn!("Error updating ping of capture websocket {}: {:?}", self.trigger_config.url, err);
}
};
Some(())
}
async fn handle(&self, db: &DB, args: PushArgsOwned) -> () {
if let Err(err) = insert_capture_payload(
db,
&self.workspace_id,
&self.path,
self.is_flow,
&TriggerKind::Websocket,
PushArgsOwned { args: args.args, extra: None },
args.extra.as_ref().map(to_raw_value),
&self.owner,
)
.await
{
tracing::error!("Error inserting capture payload: {:?}", err);
}
}
async fn get_url_from_runnable(
&self,
path: &str,
is_flow: bool,
db: &DB,
) -> error::Result<String> {
let url_runnable_args = self
.trigger_config
.url_runnable_args
.as_ref()
.map(to_raw_value);
get_url_from_runnable(
&path,
is_flow,
db,
self.fetch_authed(db, Some("url".to_string())).await?,
url_runnable_args.as_ref(),
&self.workspace_id,
&self.get_trigger_path(),
)
.await
}
async fn fetch_authed(
&self,
db: &DB,
username_override: Option<String>,
) -> error::Result<ApiAuthed> {
fetch_api_authed(
self.owner.clone(),
self.email.clone(),
&self.workspace_id,
db,
username_override,
)
.await
}
async fn disable_with_error(&self, db: &DB, error: String) -> () {
if let Err(err) = sqlx::query!(
"UPDATE capture_config SET error = $1, server_id = NULL, last_server_ping = NULL WHERE workspace_id = $2 AND path = $3 AND is_flow = $4 AND trigger_kind = 'websocket'",
error,
self.workspace_id,
self.path,
self.is_flow,
)
.execute(db).await {
tracing::error!("Could not disable websocket capture {} ({}) with err {}, disabling because of error {}", self.path, self.workspace_id, err, error);
}
}
fn get_trigger_path(&self) -> String {
format!(
"{}-{}",
if self.is_flow { "flow" } else { "script" },
self.path
)
}
}
enum WebsocketEnum {
Trigger(WebsocketTrigger),
Capture(CaptureConfigForWebsocket),
}
impl WebsocketEnum {
async fn update_ping(&self, db: &DB, error: Option<&str>) -> Option<()> {
match self {
WebsocketEnum::Trigger(ws) => ws.update_ping(db, error).await,
WebsocketEnum::Capture(capture) => capture.update_ping(db, error).await,
}
}
async fn get_url_from_runnable(
&self,
path: &str,
is_flow: bool,
db: &DB,
) -> error::Result<String> {
match self {
WebsocketEnum::Trigger(ws) => ws.get_url_from_runnable(path, is_flow, db).await,
WebsocketEnum::Capture(capture) => {
capture.get_url_from_runnable(path, is_flow, db).await
}
}
}
async fn disable_with_error(&self, db: &DB, error: String) -> () {
match self {
WebsocketEnum::Trigger(ws) => ws.disable_with_error(db, error).await,
WebsocketEnum::Capture(capture) => capture.disable_with_error(db, error).await,
}
}
}
async fn listen_to_websocket(
ws_trigger: WebsocketTrigger,
ws: WebsocketEnum,
db: DB,
mut killpill_rx: tokio::sync::broadcast::Receiver<()>,
) -> () {
if let None = update_ping(&db, &ws_trigger, Some("Connecting...")).await {
if let None = ws.update_ping(&db, Some("Connecting")).await {
return;
}
let url = ws_trigger.url.as_str();
let url = match &ws {
WebsocketEnum::Trigger(ws_trigger) => &ws_trigger.url,
WebsocketEnum::Capture(capture) => &capture.trigger_config.url,
};
let filters: Vec<Filter> = ws_trigger
.filters
.iter()
.filter_map(|m| serde_json::from_str(m.get()).ok())
.collect_vec();
let filters: Vec<Filter> = match &ws {
WebsocketEnum::Trigger(ws_trigger) => ws_trigger
.filters
.iter()
.filter_map(|m| serde_json::from_str(m.get()).ok())
.collect_vec(),
WebsocketEnum::Capture(_) => vec![],
};
loop {
let connect_url = if url.starts_with("$") {
let connect_url: Cow<str> = if url.starts_with("$") {
if url.starts_with("$flow:") || url.starts_with("$script:") {
let path = url.splitn(2, ':').nth(1).unwrap();
tokio::select! {
@@ -776,18 +1031,17 @@ async fn listen_to_websocket(
_ = killpill_rx.recv() => {
return;
},
_ = loop_ping(&db, &ws_trigger, Some(
_ = loop_ping(&db, &ws, Some(
"Waiting on runnable to return websocket URL..."
)) => {
return;
},
url_result = get_url_from_runnable(path, url.starts_with("$flow:"), &ws_trigger, &db) => match url_result {
Ok(url) => url,
url_result = ws.get_url_from_runnable(path, url.starts_with("$flow:"), &db) => match url_result {
Ok(url) => Cow::Owned(url),
Err(err) => {
disable_with_error(
&db,
&ws_trigger,
format!(
ws.disable_with_error(&db, format!(
"Error getting websocket URL from runnable after 5 tries: {:?}",
err
),
@@ -798,16 +1052,12 @@ async fn listen_to_websocket(
},
}
} else {
disable_with_error(
&db,
&ws_trigger,
format!("Invalid websocket runnable path: {}", url),
)
.await;
ws.disable_with_error(&db, format!("Invalid websocket runnable path: {}", url))
.await;
return;
}
} else {
url.to_string()
Cow::Borrowed(url)
};
tokio::select! {
@@ -815,14 +1065,14 @@ async fn listen_to_websocket(
_ = killpill_rx.recv() => {
return;
},
_ = loop_ping(&db, &ws_trigger, Some("Connecting...")) => {
_ = loop_ping(&db, &ws, Some("Connecting...")) => {
return;
},
connection = connect_async(connect_url) => {
connection = connect_async(connect_url.as_ref()) => {
match connection {
Ok((ws_stream, _)) => {
tracing::info!("Listening to websocket {}", url);
if let None = update_ping(&db, &ws_trigger, None).await {
if let None = ws.update_ping(&db, None).await {
return;
}
let (writer, mut reader) = ws_stream.split();
@@ -834,12 +1084,19 @@ async fn listen_to_websocket(
return;
}
_ = async {
if let Err(err) = send_initial_messages(&ws_trigger, writer, &db).await {
disable_with_error(&db, &ws_trigger, format!("Error sending initial messages: {:?}", err)).await;
} else {
tracing::debug!("Initial messages sent successfully to websocket {}", url);
// if initial messages sent successfully, wait forever
futures::future::pending::<()>().await;
match &ws {
WebsocketEnum::Trigger(ws_trigger) => {
if let Err(err) = ws_trigger.send_initial_messages(writer, &db).await {
ws_trigger.disable_with_error(&db, format!("Error sending initial messages: {:?}", err)).await;
} else {
tracing::debug!("Initial messages sent successfully to websocket {}", url);
// if initial messages sent successfully, wait forever
futures::future::pending::<()>().await;
}
},
WebsocketEnum::Capture(_) => {
futures::future::pending::<()>().await;
}
}
} => {
// was disabled => exit
@@ -852,7 +1109,7 @@ async fn listen_to_websocket(
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 {
if let None = ws.update_ping(&db, None).await {
return;
}
last_ping = tokio::time::Instant::now();
@@ -883,9 +1140,22 @@ async fn listen_to_websocket(
}
}
if should_handle {
if let Err(err) = run_job(&db, &ws_trigger, text).await {
report_critical_error(format!("Failed to trigger job from websocket {}: {:?}", ws_trigger.url, err), db.clone(), Some(&ws_trigger.workspace_id), None).await;
};
let args = HashMap::from([("msg".to_string(), to_raw_value(&text))]);
let extra = Some(HashMap::from([(
"wm_trigger".to_string(),
to_raw_value(&serde_json::json!({"kind": "websocket", "websocket": { "url": url }})),
)]));
let args = PushArgsOwned { args, extra };
match &ws {
WebsocketEnum::Trigger(ws_trigger) => {
ws_trigger.handle(&db, args).await;
},
WebsocketEnum::Capture(capture) => {
capture.handle(&db, args).await;
}
}
}
},
a @ _ => {
@@ -899,9 +1169,7 @@ async fn listen_to_websocket(
}
} else {
tracing::error!("Websocket {} closed", url);
if let None =
update_ping(&db, &ws_trigger, Some("Websocket closed")).await
{
if let None = ws.update_ping(&db, Some("Websocket closed")).await {
return;
}
tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
@@ -910,7 +1178,7 @@ async fn listen_to_websocket(
},
_ = tokio::time::sleep(tokio::time::Duration::from_secs(5)) => {
tracing::debug!("Sending ping to websocket {}", url);
if let None = update_ping(&db, &ws_trigger, None).await {
if let None = ws.update_ping(&db, None).await {
return;
}
last_ping = tokio::time::Instant::now();
@@ -924,9 +1192,7 @@ async fn listen_to_websocket(
}
Err(err) => {
tracing::error!("Error connecting to websocket {}: {:?}", url, err);
if let None =
update_ping(&db, &ws_trigger, Some(err.to_string().as_str())).await
{
if let None = ws.update_ping(&db, Some(err.to_string().as_str())).await {
return;
}
tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
@@ -937,16 +1203,7 @@ async fn listen_to_websocket(
}
}
async fn run_job(db: &DB, 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", "websocket": { "url": trigger.url }}),
),
)])),
};
async fn run_job(db: &DB, trigger: &WebsocketTrigger, args: PushArgsOwned) -> anyhow::Result<()> {
let label_prefix = Some(format!("ws-{}-", trigger.path));
let authed = fetch_api_authed(
backend/windmill-api/src/workspaces_extra.rs
+11
View File
@@ -103,6 +103,14 @@ pub(crate) async fn change_workspace_id(
.execute(&mut *tx)
.await?;
sqlx::query!(
"UPDATE capture_config SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
sqlx::query!(
"UPDATE completed_job SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
@@ -396,6 +404,9 @@ pub(crate) async fn delete_workspace(
sqlx::query!("DELETE FROM capture WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM capture_config WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM draft WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;