refactor: trigger listener (#6401)

* base

* base crate

* websocket updated

* crud refactored

* fix and nits

* fix compiler warning, nits and update repo ref

* fix oss

* fix compilation

* update ref

* fix

* update feature

* listener base

* add listener

* refactor logic done and implemented for postgres

* fix capture

* websocket

* implem for all triggers

* update sqlx,repo ref and fix

* fix oss

* unify struct fix websocket

* nits and fix oss runtime axum error

* perf cache query

* add listener module

* update .sqlx and cargo.toml

* fix import

* fix

* fix

* update repo ref

* fix

* update ref

* update ref

* fix

* fix

* revert

* update ref
This commit is contained in:
dieriba
2025-09-12 18:10:58 +00:00
committed by GitHub
parent f79a46621e
commit 0e82962545
111 changed files with 2641 additions and 6931 deletions
@@ -1,50 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n path,\n is_flow,\n workspace_id,\n owner,\n email,\n trigger_config as \"trigger_config!: _\"\n FROM\n capture_config\n WHERE\n trigger_kind = 'postgres' AND\n last_client_ping > NOW() - INTERVAL '10 seconds' AND\n trigger_config IS NOT NULL AND\n (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds')\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 2,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "owner",
"type_info": "Varchar"
},
{
"ordinal": 4,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "trigger_config!: _",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
false,
false,
false,
false,
true
]
},
"hash": "011c7638eeeda710deb86a216a9e10df9c3e9458e85bcdde466b01011a1f2ac2"
}
@@ -1,26 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "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",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Bool",
"Text"
]
},
"nullable": [
null
]
},
"hash": "031d0d70b0aff52feaad487bddb74e5ef0aaa2505facbea8c764003dfc8fffb1"
}
@@ -1,24 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n mqtt_trigger \n SET \n server_id = $1, \n last_server_ping = now(),\n error = 'Connecting...'\n WHERE \n enabled IS TRUE \n AND workspace_id = $2 \n AND path = $3 \n AND (last_server_ping IS NULL \n OR last_server_ping < now() - INTERVAL '15 seconds'\n ) \n RETURNING true\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "115544a96173f9cb1d27757e7b931fb27912cfd05ba768a42cf9b3dfd7205e9a"
}
@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE \n capture_config \n SET \n last_server_ping = NULL \n WHERE \n workspace_id = $1 AND \n path = $2 AND \n is_flow = $3 AND \n trigger_kind = 'postgres' AND \n server_id IS NULL\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "12e868b63a7c622c76713db5a5577a927efca4ae49a15c2b999e2410f2a312ff"
}
@@ -1,24 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE postgres_trigger \n SET \n server_id = $1, \n last_server_ping = now(),\n error = 'Connecting...'\n WHERE \n enabled IS TRUE \n AND workspace_id = $2 \n AND path = $3 \n AND (last_server_ping IS NULL \n OR last_server_ping < now() - INTERVAL '15 seconds'\n ) \n RETURNING true\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "1488e1b5007752e1ebae4235ad04c398fe6398745e16fd119008b8ea67662416"
}
@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n mqtt_trigger \n SET\n last_server_ping = NULL \n WHERE \n workspace_id = $1 \n AND path = $2 \n AND server_id IS NULL",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": []
},
"hash": "186aef850c2eeb89c186ac6b2934dd3a703e2b9428096801e1d2d61fdbb99c9e"
}
@@ -1,26 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n capture_config \n SET \n last_server_ping = now(), \n error = $1 \n WHERE \n workspace_id = $2 AND \n path = $3 AND \n is_flow = $4 AND \n trigger_kind = 'postgres' AND \n server_id = $5 AND \n last_client_ping > NOW() - INTERVAL '10 seconds' \n RETURNING 1\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Bool",
"Text"
]
},
"nullable": [
null
]
},
"hash": "1974bd65bbf40024773aad4dee1c50b12e110e76bb58e6de25bec094e758a71c"
}
@@ -1,50 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n path,\n is_flow,\n workspace_id,\n owner,\n email,\n trigger_config as \"trigger_config!: _\"\n FROM\n capture_config\n WHERE\n trigger_kind = 'sqs' AND\n last_client_ping > NOW() - INTERVAL '10 seconds' AND\n trigger_config IS NOT NULL AND\n (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds')\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 2,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "owner",
"type_info": "Varchar"
},
{
"ordinal": 4,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "trigger_config!: _",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
false,
false,
false,
false,
true
]
},
"hash": "1b33393fbbc7e681b4d355f6096d982b52b48c8c3da392b14963a1ec86811546"
}
@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE kafka_trigger SET last_server_ping = NULL WHERE workspace_id = $1 AND path = $2 AND server_id IS NULL",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": []
},
"hash": "1c37f91192aa4f535c7fff80fa809260b906dc988f44a0db60952a2bf8b1cdaf"
}
@@ -1,26 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n capture_config \n SET \n last_server_ping = now(), \n error = $1 \n WHERE \n workspace_id = $2 AND \n path = $3 AND \n is_flow = $4 AND \n trigger_kind = 'sqs' AND \n server_id = $5 AND \n last_client_ping > NOW() - INTERVAL '10 seconds' \n RETURNING 1\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Bool",
"Text"
]
},
"nullable": [
null
]
},
"hash": "1cad25c24d0f80d58a50d4da923f8b0672797299545ee0754b6ad74ece92c77e"
}
@@ -1,17 +0,0 @@
{
"db_name": "PostgreSQL",
"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 = 'kafka'",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "203fa78d423ec5a8c5ff6166aed591b28cbf9ea8f61d379b84ee6e14c033035d"
}
@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE \n capture_config \n SET \n last_server_ping = NULL \n WHERE \n workspace_id = $1 AND \n path = $2 AND \n is_flow = $3 AND \n trigger_kind = 'sqs' AND \n server_id IS NULL\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "20e01ecb5d4aa4c532a8f906365c776994ae2c990bb7bb307c52296b9543fbb8"
}
@@ -1,25 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n sqs_trigger\n SET \n last_server_ping = now(),\n error = $1\n WHERE\n workspace_id = $2\n AND path = $3\n AND server_id = $4 \n AND enabled IS TRUE\n RETURNING 1\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "22dcd953d900fb0ddbe2099ccee27bcf833f56475bbc1aa3c8f6d79f146f41bf"
}
@@ -1,150 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n mqtt_resource_path,\n subscribe_topics as \"subscribe_topics!: Vec<SqlxJson<SubscribeTopic>>\",\n v3_config as \"v3_config!: Option<SqlxJson<MqttV3Config>>\",\n v5_config as \"v5_config!: Option<SqlxJson<MqttV5Config>>\",\n client_version as \"client_version: _\",\n client_id,\n workspace_id,\n path,\n script_path,\n is_flow,\n edited_by,\n email,\n edited_at,\n server_id,\n last_server_ping,\n extra_perms,\n error,\n enabled,\n error_handler_path,\n error_handler_args as \"error_handler_args: _\",\n retry as \"retry: _\"\n FROM\n mqtt_trigger\n WHERE\n enabled IS TRUE\n AND (last_server_ping IS NULL OR\n last_server_ping < now() - interval '15 seconds'\n )\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "mqtt_resource_path",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "subscribe_topics!: Vec<SqlxJson<SubscribeTopic>>",
"type_info": "JsonbArray"
},
{
"ordinal": 2,
"name": "v3_config!: Option<SqlxJson<MqttV3Config>>",
"type_info": "Jsonb"
},
{
"ordinal": 3,
"name": "v5_config!: Option<SqlxJson<MqttV5Config>>",
"type_info": "Jsonb"
},
{
"ordinal": 4,
"name": "client_version: _",
"type_info": {
"Custom": {
"name": "mqtt_client_version",
"kind": {
"Enum": [
"v3",
"v5"
]
}
}
}
},
{
"ordinal": 5,
"name": "client_id",
"type_info": "Varchar"
},
{
"ordinal": 6,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 7,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 8,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 9,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 10,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 11,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 12,
"name": "edited_at",
"type_info": "Timestamptz"
},
{
"ordinal": 13,
"name": "server_id",
"type_info": "Varchar"
},
{
"ordinal": 14,
"name": "last_server_ping",
"type_info": "Timestamptz"
},
{
"ordinal": 15,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 16,
"name": "error",
"type_info": "Text"
},
{
"ordinal": 17,
"name": "enabled",
"type_info": "Bool"
},
{
"ordinal": 18,
"name": "error_handler_path",
"type_info": "Varchar"
},
{
"ordinal": 19,
"name": "error_handler_args: _",
"type_info": "Jsonb"
},
{
"ordinal": 20,
"name": "retry: _",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
false,
true,
true,
false,
true,
false,
false,
false,
false,
false,
false,
false,
true,
true,
false,
true,
false,
true,
true,
true
]
},
"hash": "2659fe2e121ac15da08030c9e72bdb79a580711cba2139f0cf901b30bb491fd5"
}
@@ -1,25 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n capture_config \n SET \n server_id = $1,\n last_server_ping = now(), \n error = 'Connecting...' \n WHERE \n last_client_ping > NOW() - INTERVAL '10 seconds' AND \n workspace_id = $2 AND \n path = $3 AND \n is_flow = $4 AND \n trigger_kind = 'gcp' AND \n (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') \n RETURNING true\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text",
"Bool"
]
},
"nullable": [
null
]
},
"hash": "299e16725162888c01712f371785199960264b54c1ddf928c0c654ab15176f63"
}
@@ -1,17 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n capture_config \n SET \n error = $1, \n server_id = NULL, \n last_server_ping = NULL \n WHERE \n workspace_id = $2 AND \n path = $3 AND \n is_flow = $4 AND \n trigger_kind = 'postgres'\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "29f096ec62c4abb1435a5667e2b30e9c1724e419cdc23ef1b300e84c02a20427"
}
@@ -1,166 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n gcp_resource_path,\n subscription_id,\n topic_id,\n workspace_id,\n delivery_type AS \"delivery_type: _\",\n delivery_config AS \"delivery_config: _\",\n subscription_mode AS \"subscription_mode: _\",\n path,\n script_path,\n is_flow,\n edited_by,\n email,\n edited_at,\n server_id,\n last_server_ping,\n extra_perms,\n error,\n enabled,\n error_handler_path,\n error_handler_args as \"error_handler_args: _\",\n retry as \"retry: _\",\n auto_acknowledge_msg\n FROM\n gcp_trigger\n WHERE\n delivery_type != 'push'::DELIVERY_MODE AND\n enabled IS TRUE\n AND (last_server_ping IS NULL OR\n last_server_ping < now() - interval '15 seconds'\n )\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "gcp_resource_path",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "subscription_id",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "topic_id",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 4,
"name": "delivery_type: _",
"type_info": {
"Custom": {
"name": "delivery_mode",
"kind": {
"Enum": [
"push",
"pull"
]
}
}
}
},
{
"ordinal": 5,
"name": "delivery_config: _",
"type_info": "Jsonb"
},
{
"ordinal": 6,
"name": "subscription_mode: _",
"type_info": {
"Custom": {
"name": "gcp_subscription_mode",
"kind": {
"Enum": [
"create_update",
"existing"
]
}
}
}
},
{
"ordinal": 7,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 8,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 9,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 10,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 11,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 12,
"name": "edited_at",
"type_info": "Timestamptz"
},
{
"ordinal": 13,
"name": "server_id",
"type_info": "Varchar"
},
{
"ordinal": 14,
"name": "last_server_ping",
"type_info": "Timestamptz"
},
{
"ordinal": 15,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 16,
"name": "error",
"type_info": "Text"
},
{
"ordinal": 17,
"name": "enabled",
"type_info": "Bool"
},
{
"ordinal": 18,
"name": "error_handler_path",
"type_info": "Varchar"
},
{
"ordinal": 19,
"name": "error_handler_args: _",
"type_info": "Jsonb"
},
{
"ordinal": 20,
"name": "retry: _",
"type_info": "Jsonb"
},
{
"ordinal": 21,
"name": "auto_acknowledge_msg",
"type_info": "Bool"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
false,
false,
false,
false,
true,
false,
false,
false,
false,
false,
false,
false,
true,
true,
false,
true,
false,
true,
true,
true,
true
]
},
"hash": "33aaf2dd14397d0b50b986ed55bc458337b85b6a9a9ab3d93d8f33d0f57e4b0f"
}
@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE \n capture_config \n SET \n last_server_ping = NULL \n WHERE \n workspace_id = $1 AND \n path = $2 AND \n is_flow = $3 AND \n trigger_kind = 'mqtt' AND \n server_id IS NULL\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "3e33469f448fa86c7e0d0deb65ee2c13b5dcc4cbc4fcb0fe2bde1fdcb6d20e74"
}
@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE kafka_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": "3e64c894c89ef82c4527180b82c4e82c1b7060ba0ca288dc6a7c7afcd76212e8"
}
@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n sqs_trigger \n SET \n enabled = FALSE, \n error = $1, \n server_id = NULL, \n last_server_ping = NULL \n WHERE \n workspace_id = $2 AND \n path = $3\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "3f67e7cf8d9f021a075f1c88703287a9f42e252a1246a584249a224afdbbbf8a"
}
@@ -1,50 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT path, is_flow, workspace_id, trigger_config as \"trigger_config!: _\", owner, email FROM capture_config WHERE trigger_kind = 'nats' AND last_client_ping > NOW() - INTERVAL '10 seconds' AND trigger_config IS NOT NULL AND (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": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 2,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "trigger_config!: _",
"type_info": "Jsonb"
},
{
"ordinal": 4,
"name": "owner",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "email",
"type_info": "Varchar"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
false,
false,
true,
false,
false
]
},
"hash": "42030372693d8d6b8d03947bb6702024cfa226735eeeb720bbab46bd0e3cedcc"
}
@@ -1,50 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "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 (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": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 2,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "trigger_config!: _",
"type_info": "Jsonb"
},
{
"ordinal": 4,
"name": "owner",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "email",
"type_info": "Varchar"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
false,
false,
true,
false,
false
]
},
"hash": "47560aa3d3af93167663e763ad873088eb4f57785df3c383bbf6e5ceab268980"
}
@@ -1,25 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n mqtt_trigger\n SET \n last_server_ping = now(),\n error = $1\n WHERE\n workspace_id = $2\n AND path = $3\n AND server_id = $4 \n AND enabled IS TRUE\n RETURNING 1\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "4844a797ddec207f1d2cbf41836e65d6435840f6d0740f300d7c2cf88f8fe46a"
}
@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE websocket_trigger SET last_server_ping = NULL WHERE workspace_id = $1 AND path = $2 AND server_id IS NULL",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": []
},
"hash": "51f37d683d5a48b96f6224111639c364ba9c41572eb6799f7c822c66d00d2500"
}
@@ -1,50 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n path,\n is_flow,\n workspace_id,\n owner,\n email,\n trigger_config as \"trigger_config!: _\"\n FROM\n capture_config\n WHERE\n trigger_kind = 'mqtt' AND\n last_client_ping > NOW() - INTERVAL '10 seconds' AND\n trigger_config IS NOT NULL AND\n (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds')\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 2,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "owner",
"type_info": "Varchar"
},
{
"ordinal": 4,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "trigger_config!: _",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
false,
false,
false,
false,
true
]
},
"hash": "548a1424a6b9ac9998b8fc5312bcff671ee7fe59154e9ef32800c117e64bce19"
}
@@ -1,138 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n aws_auth_resource_type AS \"aws_auth_resource_type: _\",\n queue_url,\n aws_resource_path,\n message_attributes,\n workspace_id,\n path,\n script_path,\n is_flow,\n edited_by,\n email,\n edited_at,\n server_id,\n last_server_ping,\n extra_perms,\n error,\n enabled,\n error_handler_path,\n error_handler_args as \"error_handler_args: _\",\n retry as \"retry: _\"\n FROM\n sqs_trigger\n WHERE\n enabled IS TRUE\n AND (last_server_ping IS NULL OR\n last_server_ping < now() - interval '15 seconds'\n )\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "aws_auth_resource_type: _",
"type_info": {
"Custom": {
"name": "aws_auth_resource_type",
"kind": {
"Enum": [
"oidc",
"credentials"
]
}
}
}
},
{
"ordinal": 1,
"name": "queue_url",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "aws_resource_path",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "message_attributes",
"type_info": "TextArray"
},
{
"ordinal": 4,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 6,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 7,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 8,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 9,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 10,
"name": "edited_at",
"type_info": "Timestamptz"
},
{
"ordinal": 11,
"name": "server_id",
"type_info": "Varchar"
},
{
"ordinal": 12,
"name": "last_server_ping",
"type_info": "Timestamptz"
},
{
"ordinal": 13,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 14,
"name": "error",
"type_info": "Text"
},
{
"ordinal": 15,
"name": "enabled",
"type_info": "Bool"
},
{
"ordinal": 16,
"name": "error_handler_path",
"type_info": "Varchar"
},
{
"ordinal": 17,
"name": "error_handler_args: _",
"type_info": "Jsonb"
},
{
"ordinal": 18,
"name": "retry: _",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
false,
false,
true,
false,
false,
false,
false,
false,
false,
false,
true,
true,
true,
true,
false,
true,
true,
true
]
},
"hash": "58ddf5c76455d30dc42f46a0553e761461c327e7502b707c2d2742946d7d1f4c"
}
@@ -15,7 +15,7 @@
]
},
"nullable": [
null
true
]
},
"hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55"
@@ -1,24 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n sqs_trigger \n SET \n server_id = $1, \n last_server_ping = now(),\n error = 'Connecting...'\n WHERE \n enabled IS TRUE \n AND workspace_id = $2 \n AND path = $3 \n AND (last_server_ping IS NULL \n OR last_server_ping < now() - INTERVAL '15 seconds'\n ) \n RETURNING true\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "5bd8ae8d694ac9f6afef762276141d95ba195f5543120e76e68f6e6715b65a71"
}
@@ -0,0 +1,45 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n capture_config\n SET \n last_server_ping = now(), error = $1\n WHERE \n workspace_id = $2 AND \n path = $3 AND \n is_flow = $4 AND \n trigger_kind = $5 AND \n server_id = $6 AND \n last_client_ping > NOW() - INTERVAL '10 seconds'\n RETURNING 1\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Bool",
{
"Custom": {
"name": "trigger_kind",
"kind": {
"Enum": [
"webhook",
"http",
"websocket",
"kafka",
"email",
"nats",
"postgres",
"sqs",
"mqtt",
"gcp"
]
}
}
},
"Text"
]
},
"nullable": [
null
]
},
"hash": "615d832a452a6c64de50cd0efada0be238fb16daacb3464bbcf47ca2e21bdaae"
}
@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n postgres_trigger \n SET\n last_server_ping = NULL \n WHERE \n workspace_id = $1 \n AND path = $2 \n AND server_id IS NULL",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": []
},
"hash": "61bed1bc6d3e6a3c1d640eeacc290a85d8b63ee36c39dfbf4348d120f6e561ae"
}
@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE \n capture_config \n SET \n last_server_ping = NULL \n WHERE \n workspace_id = $1 AND \n path = $2 AND \n is_flow = $3 AND \n trigger_kind = 'gcp' AND \n server_id IS NULL\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "673564e6c4dcf30dae3d7a75c397998ea800860ede856e5fc6cd1f57d5408333"
}
@@ -1,25 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n capture_config \n SET \n server_id = $1,\n last_server_ping = now(), \n error = 'Connecting...' \n WHERE \n last_client_ping > NOW() - INTERVAL '10 seconds' AND \n workspace_id = $2 AND \n path = $3 AND \n is_flow = $4 AND \n trigger_kind = 'sqs' AND \n (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') \n RETURNING true\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text",
"Bool"
]
},
"nullable": [
null
]
},
"hash": "6b776693091191f36eaf9e35fb3cabb01481a19bef5a956dc36cac41acad0e53"
}
@@ -1,25 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE kafka_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",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "6e70ebf078ac04a2933d2f83791973e8fc108d9f32be8a8501391052d76e191e"
}
@@ -1,25 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n capture_config \n SET \n server_id = $1,\n last_server_ping = now(), \n error = 'Connecting...' \n WHERE \n last_client_ping > NOW() - INTERVAL '10 seconds' AND \n workspace_id = $2 AND \n path = $3 AND \n is_flow = $4 AND \n trigger_kind = 'postgres' AND \n (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') \n RETURNING true\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text",
"Bool"
]
},
"nullable": [
null
]
},
"hash": "74d928f4c3f0de191f414471b9a4fbe9c20f9685b06ad5bbded424948b2dc88c"
}
@@ -1,17 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n capture_config \n SET \n error = $1, \n server_id = NULL, \n last_server_ping = NULL \n WHERE \n workspace_id = $2 AND \n path = $3 AND \n is_flow = $4 AND \n trigger_kind = 'sqs'\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "7e31c47e02492b74cbb5702dfc5ef9f4c8faa986843024a2cd58e212f706aae8"
}
@@ -1,25 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n postgres_trigger\n SET \n last_server_ping = now(),\n error = $1\n WHERE\n workspace_id = $2\n AND path = $3\n AND server_id = $4 \n AND enabled IS TRUE\n RETURNING 1\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "9116102c6ccad5b0d752d5d690c233dfe48062aef23072b4f4ae4ab5ca269082"
}
@@ -1,24 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n gcp_trigger \n SET \n server_id = $1, \n last_server_ping = now(),\n error = 'Connecting...'\n WHERE \n enabled IS TRUE \n AND workspace_id = $2 \n AND path = $3 \n AND (last_server_ping IS NULL \n OR last_server_ping < now() - INTERVAL '15 seconds'\n ) \n RETURNING true\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "92c2b66eb6287449f5f8cc9f8d1329f748006ab131489d2fcac21d32641bb633"
}
@@ -1,24 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE websocket_trigger SET server_id = $1, last_server_ping = now(), error = 'Connecting...' WHERE enabled IS TRUE AND workspace_id = $2 AND path = $3 AND (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') RETURNING true",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "92e60af0d3ae8c73d74ae68d70e20ae18f79cc84626097b766307cd42722baa3"
}
@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE capture_config SET last_server_ping = NULL WHERE workspace_id = $1 AND path = $2 AND is_flow = $3 AND trigger_kind = 'nats' AND server_id IS NULL",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "93d64930c74ccb1abdc9bda8287540a354f8eec49913aaf07fc1e737e7c93330"
}
@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n gcp_trigger \n SET\n last_server_ping = NULL \n WHERE \n workspace_id = $1 \n AND path = $2 \n AND server_id IS NULL",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": []
},
"hash": "999edc6f54a9efb6dc6237992dad59f418ed6fc2a98ddb9bbc33dce5f029d904"
}
@@ -1,25 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE capture_config SET server_id = $1, last_server_ping = now(), error = 'Connecting...' WHERE last_client_ping > NOW() - INTERVAL '10 seconds' AND workspace_id = $2 AND path = $3 AND is_flow = $4 AND trigger_kind = 'websocket' AND (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') RETURNING true",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text",
"Bool"
]
},
"nullable": [
null
]
},
"hash": "99a2c935acf5d6bbeb70ea1255679115b6e9042800d40899a36b3867049c5c46"
}
@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n sqs_trigger \n SET\n last_server_ping = NULL \n WHERE \n workspace_id = $1 \n AND path = $2 \n AND server_id IS NULL",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": []
},
"hash": "a5fbef9db2308920ea26f6154f0f3490f88ced636c9ebdecba2205f964b1d0f1"
}
@@ -1,26 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n capture_config \n SET \n last_server_ping = now(), \n error = $1 \n WHERE \n workspace_id = $2 AND \n path = $3 AND \n is_flow = $4 AND \n trigger_kind = 'gcp' AND \n server_id = $5 AND \n last_client_ping > NOW() - INTERVAL '10 seconds' \n RETURNING 1\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Bool",
"Text"
]
},
"nullable": [
null
]
},
"hash": "a7ffc5b983d365159ef379ec6a2ab0dc7217d1b3f86f362c213982984d2cf652"
}
@@ -0,0 +1,44 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n capture_config \n SET \n server_id = $1,\n last_server_ping = now(), \n error = 'Connecting...' \n WHERE \n last_client_ping > NOW() - INTERVAL '10 seconds' AND \n workspace_id = $2 AND \n path = $3 AND \n is_flow = $4 AND \n trigger_kind = $5 AND \n (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') \n RETURNING true\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text",
"Bool",
{
"Custom": {
"name": "trigger_kind",
"kind": {
"Enum": [
"webhook",
"http",
"websocket",
"kafka",
"email",
"nats",
"postgres",
"sqs",
"mqtt",
"gcp"
]
}
}
}
]
},
"nullable": [
null
]
},
"hash": "ac9037b8adce156b95390a0ffac04e38ab8474849e0cacb3be1443d7f3265d30"
}
@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE nats_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": "add48c8e7c6fa2c549ad6293cbee22889d35e919d3267c1d2a265d868fa8a7d1"
}
@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE capture_config SET last_server_ping = NULL WHERE workspace_id = $1 AND path = $2 AND is_flow = $3 AND trigger_kind = 'kafka' AND server_id IS NULL",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "b223c56f55a138abef8c8cb2df40472a87b7e267a143709728b09ac063c33b96"
}
@@ -1,134 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n workspace_id,\n path,\n nats_resource_path,\n subjects,\n stream_name,\n consumer_name,\n use_jetstream,\n script_path,\n is_flow,\n edited_by,\n email,\n edited_at,\n server_id,\n last_server_ping,\n extra_perms,\n error,\n enabled,\n error_handler_path,\n error_handler_args as \"error_handler_args: _\",\n retry as \"retry: _\"\n FROM nats_trigger\n WHERE enabled IS TRUE AND (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds')",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "nats_resource_path",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "subjects",
"type_info": "VarcharArray"
},
{
"ordinal": 4,
"name": "stream_name",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "consumer_name",
"type_info": "Varchar"
},
{
"ordinal": 6,
"name": "use_jetstream",
"type_info": "Bool"
},
{
"ordinal": 7,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 8,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 9,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 10,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 11,
"name": "edited_at",
"type_info": "Timestamptz"
},
{
"ordinal": 12,
"name": "server_id",
"type_info": "Varchar"
},
{
"ordinal": 13,
"name": "last_server_ping",
"type_info": "Timestamptz"
},
{
"ordinal": 14,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 15,
"name": "error",
"type_info": "Text"
},
{
"ordinal": 16,
"name": "enabled",
"type_info": "Bool"
},
{
"ordinal": 17,
"name": "error_handler_path",
"type_info": "Varchar"
},
{
"ordinal": 18,
"name": "error_handler_args: _",
"type_info": "Jsonb"
},
{
"ordinal": 19,
"name": "retry: _",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
false,
false,
false,
true,
true,
false,
false,
false,
false,
false,
false,
true,
true,
false,
true,
false,
true,
true,
true
]
},
"hash": "bcc4f786fa2771a4a265aec8d1c403e660002503aad860caa12ebee2dbab3f0f"
}
@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n postgres_trigger \n SET \n enabled = FALSE, \n error = $1, \n server_id = NULL, \n last_server_ping = NULL \n WHERE \n workspace_id = $2 AND \n path = $3\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "bfc534d87d701d7ac78cc97d0054d829165ba3f22fba75c3161e4cddb72264ee"
}
@@ -1,50 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n path,\n is_flow,\n workspace_id,\n owner,\n email,\n trigger_config as \"trigger_config!: _\"\n FROM\n capture_config\n WHERE\n trigger_kind = 'gcp' AND\n last_client_ping > NOW() - INTERVAL '10 seconds' AND\n trigger_config IS NOT NULL AND\n trigger_config->>'delivery_type' IS DISTINCT FROM 'push' AND\n (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds')\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 2,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "owner",
"type_info": "Varchar"
},
{
"ordinal": 4,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "trigger_config!: _",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
false,
false,
false,
false,
true
]
},
"hash": "c0e6dbce7a401b06e1bf45155c3f81572818177a6024e743b12f558b46edf74c"
}
@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n mqtt_trigger \n SET \n enabled = FALSE, \n error = $1, \n server_id = NULL, \n last_server_ping = NULL \n WHERE \n workspace_id = $2 AND \n path = $3\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "c5063e79aafa70b974276f5bea43ad71135d44b2e84c9efac43a6678f0cf9a18"
}
@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE capture_config SET last_server_ping = NULL WHERE workspace_id = $1 AND path = $2 AND is_flow = $3 AND trigger_kind = 'websocket' AND server_id IS NULL",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "cba3bfb174829ee3b08ea195831fdba4335ca221a5fd25c72519ac376c593e44"
}
@@ -1,16 +0,0 @@
{
"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"
}
@@ -1,24 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE nats_trigger SET server_id = $1, last_server_ping = now(), error = 'Connecting...' WHERE enabled IS TRUE AND workspace_id = $2 AND path = $3 AND (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') RETURNING true",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "cd5d62d456b74237b941bc72ea8de007185263167bf1dfa2e469d319cc4da674"
}
@@ -1,122 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n workspace_id,\n path,\n script_path,\n replication_slot_name,\n publication_name,\n is_flow,\n edited_by,\n email,\n edited_at,\n server_id,\n last_server_ping,\n extra_perms,\n error,\n enabled,\n postgres_resource_path,\n error_handler_path,\n error_handler_args as \"error_handler_args: _\",\n retry as \"retry: _\"\n FROM\n postgres_trigger\n WHERE\n enabled IS TRUE\n AND (last_server_ping IS NULL OR\n last_server_ping < now() - interval '15 seconds'\n )\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "replication_slot_name",
"type_info": "Varchar"
},
{
"ordinal": 4,
"name": "publication_name",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 6,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 7,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 8,
"name": "edited_at",
"type_info": "Timestamptz"
},
{
"ordinal": 9,
"name": "server_id",
"type_info": "Varchar"
},
{
"ordinal": 10,
"name": "last_server_ping",
"type_info": "Timestamptz"
},
{
"ordinal": 11,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 12,
"name": "error",
"type_info": "Text"
},
{
"ordinal": 13,
"name": "enabled",
"type_info": "Bool"
},
{
"ordinal": 14,
"name": "postgres_resource_path",
"type_info": "Varchar"
},
{
"ordinal": 15,
"name": "error_handler_path",
"type_info": "Varchar"
},
{
"ordinal": 16,
"name": "error_handler_args: _",
"type_info": "Jsonb"
},
{
"ordinal": 17,
"name": "retry: _",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
false,
false,
false,
false,
false,
false,
false,
false,
true,
true,
true,
true,
false,
false,
true,
true,
true
]
},
"hash": "cf76aedfdffc25057a64fc94fd6be0791d5970458810a2b465c27f02ad1d4b2f"
}
@@ -1,26 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n capture_config \n SET \n last_server_ping = now(), \n error = $1 \n WHERE \n workspace_id = $2 AND \n path = $3 AND \n is_flow = $4 AND \n trigger_kind = 'mqtt' AND \n server_id = $5 AND \n last_client_ping > NOW() - INTERVAL '10 seconds' \n RETURNING 1\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Bool",
"Text"
]
},
"nullable": [
null
]
},
"hash": "cff764601318bfecb8e0f15b64ff9f430d7b1efe8ba863cbcc4a49bab4951794"
}
@@ -1,25 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE capture_config SET server_id = $1, last_server_ping = now(), error = 'Connecting...' WHERE last_client_ping > NOW() - INTERVAL '10 seconds' AND workspace_id = $2 AND path = $3 AND is_flow = $4 AND trigger_kind = 'kafka' AND (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') RETURNING true",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text",
"Bool"
]
},
"nullable": [
null
]
},
"hash": "d06efdc24706e0d7479bffc0b19a0c5976ee60125289e9f7b0b04090bce4a3a3"
}
@@ -1,17 +0,0 @@
{
"db_name": "PostgreSQL",
"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'",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "d08f34000c3d96ccd0f44ca8520f966d751a4dda554d8215eedb8f65be98e100"
}
@@ -1,24 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE kafka_trigger SET server_id = $1, last_server_ping = now(), error = 'Connecting...' WHERE enabled IS TRUE AND workspace_id = $2 AND path = $3 AND (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') RETURNING true",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "d78ecf85c1e1e95650c380c9488aa90d8d8c5c76f3971484e4c515ff60293d3b"
}
@@ -1,50 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT path, is_flow, workspace_id, trigger_config as \"trigger_config!: _\", owner, email FROM capture_config WHERE trigger_kind = 'kafka' AND last_client_ping > NOW() - INTERVAL '10 seconds' AND trigger_config IS NOT NULL AND (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": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 2,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "trigger_config!: _",
"type_info": "Jsonb"
},
{
"ordinal": 4,
"name": "owner",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "email",
"type_info": "Varchar"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
false,
false,
true,
false,
false
]
},
"hash": "d8ea17fba0e417333e9c6cf82ad36a57830a791c8358e30e9be012b815f5e8e3"
}
@@ -1,26 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE capture_config SET last_server_ping = now(), error = $1 WHERE workspace_id = $2 AND path = $3 AND is_flow = $4 AND trigger_kind = 'kafka' AND server_id = $5 AND last_client_ping > NOW() - INTERVAL '10 seconds' RETURNING 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Bool",
"Text"
]
},
"nullable": [
null
]
},
"hash": "d9a6f75e4c4a1f61e55b313cc09bceffac637548841897341672da427a9140fc"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n capture_config \n SET \n error = $1, \n server_id = NULL, \n last_server_ping = NULL \n WHERE \n workspace_id = $2 AND \n path = $3 AND \n is_flow = $4 AND \n trigger_kind = 'gcp'\n ",
"query": "\n UPDATE \n capture_config \n SET \n error = $1, \n server_id = NULL, \n last_server_ping = NULL \n WHERE \n workspace_id = $2 AND \n path = $3 AND \n is_flow = $4 AND \n trigger_kind = $5\n ",
"describe": {
"columns": [],
"parameters": {
@@ -8,10 +8,29 @@
"Text",
"Text",
"Text",
"Bool"
"Bool",
{
"Custom": {
"name": "trigger_kind",
"kind": {
"Enum": [
"webhook",
"http",
"websocket",
"kafka",
"email",
"nats",
"postgres",
"sqs",
"mqtt",
"gcp"
]
}
}
}
]
},
"nullable": []
},
"hash": "43c8cd9f8560412bb06d9966ccfa2524943ba277c5f9a37c06bd5ee14fd46bda"
"hash": "db7b39335049f7b5fbb1ba2b99618eeeccdd4b7e14a0c0077af9d978f99ae899"
}
@@ -0,0 +1,35 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n capture_config\n SET \n last_server_ping = NULL\n WHERE \n workspace_id = $1 AND \n path = $2 AND \n is_flow = $3 AND \n trigger_kind = $4 AND \n server_id IS NULL\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Bool",
{
"Custom": {
"name": "trigger_kind",
"kind": {
"Enum": [
"webhook",
"http",
"websocket",
"kafka",
"email",
"nats",
"postgres",
"sqs",
"mqtt",
"gcp"
]
}
}
}
]
},
"nullable": []
},
"hash": "dcaf17a826e8f4cba4145abcf72bf749ad1d4381fa3b9df8b5bf534f9c13692e"
}
@@ -1,25 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE nats_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",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "dcf03ff4b922b93be37d2be5da6884ae3e8c6cf7eaa2d9c62056366eb42f2276"
}
@@ -1,25 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n gcp_trigger\n SET \n last_server_ping = now(),\n error = $1\n WHERE\n workspace_id = $2\n AND path = $3\n AND server_id = $4 \n AND enabled IS TRUE\n RETURNING 1\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "df2a426658f0a36683cc2163fe70912d0221d191b8ab091601cb53cdd932ad58"
}
@@ -1,26 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE capture_config SET last_server_ping = now(), error = $1 WHERE workspace_id = $2 AND path = $3 AND is_flow = $4 AND trigger_kind = 'nats' AND server_id = $5 AND last_client_ping > NOW() - INTERVAL '10 seconds' RETURNING 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Bool",
"Text"
]
},
"nullable": [
null
]
},
"hash": "e37ba13aa3174931f0bfcff26dbc141fe8a346ab6a1a3bc924794bfe3c9af306"
}
@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE nats_trigger SET last_server_ping = NULL WHERE workspace_id = $1 AND path = $2 AND server_id IS NULL",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": []
},
"hash": "e38a66d15382703b99054a9b609af9f3854bb65393d33b265476c0e68aec4f61"
}
@@ -1,17 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n capture_config \n SET \n error = $1, \n server_id = NULL, \n last_server_ping = NULL \n WHERE \n workspace_id = $2 AND \n path = $3 AND \n is_flow = $4 AND \n trigger_kind = 'mqtt'\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "f050be8da0d99aa1af64f5d32a8df01c5a59b036b2669878cc557537efb492c8"
}
@@ -1,25 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE capture_config SET server_id = $1, last_server_ping = now(), error = 'Connecting...' WHERE last_client_ping > NOW() - INTERVAL '10 seconds' AND workspace_id = $2 AND path = $3 AND is_flow = $4 AND trigger_kind = 'nats' AND (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') RETURNING true",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text",
"Bool"
]
},
"nullable": [
null
]
},
"hash": "f16b00bad2880f896f4452e5894fe101cd5c2fe3e4c143c0b979668143d85dd2"
}
@@ -1,25 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n capture_config \n SET \n server_id = $1,\n last_server_ping = now(), \n error = 'Connecting...' \n WHERE \n last_client_ping > NOW() - INTERVAL '10 seconds' AND \n workspace_id = $2 AND \n path = $3 AND \n is_flow = $4 AND \n trigger_kind = 'mqtt' AND \n (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') \n RETURNING true\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text",
"Bool"
]
},
"nullable": [
null
]
},
"hash": "fc932ecf697b921fe9143c36872e9cc4e9aca7f9129b0466aacae51334a3db96"
}
@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n gcp_trigger \n SET \n enabled = FALSE, \n error = $1, \n server_id = NULL, \n last_server_ping = NULL \n WHERE \n workspace_id = $2 AND \n path = $3\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "fd0a0ca4a107dc813240ad71f372ab7e0bf26431158fbfb5f0023ed847ca7dc5"
}
@@ -1,25 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "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",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "febd70e6d5304ad912c232d3d8fc3f8ce0133d8a1a9e5ca1124eff9a988dc2d7"
}
@@ -1,17 +0,0 @@
{
"db_name": "PostgreSQL",
"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 = 'nats'",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "ffb6b2a40b605f9c55826a0e519af80b100d2ffa58ec567223e27099d364fcc8"
}
+1 -1
View File
@@ -1 +1 @@
ef50587d21cc765c2f32ad95622f84d68d1e91a3
f860277e8a5719c59d8961856fd84e3be5c3a83b
+2 -2
View File
@@ -46,10 +46,10 @@ use serde::de::DeserializeOwned;
use windmill_common::error::Error;
#[cfg(all(feature = "enterprise", feature = "kafka", feature = "private"))]
use crate::kafka_triggers_ee::KafkaTriggerConfigConnection;
use crate::triggers::kafka::KafkaTriggerConfigConnection;
#[cfg(feature = "mqtt_trigger")]
use crate::mqtt_triggers::{MqttClientVersion, MqttV3Config, MqttV5Config, SubscribeTopic};
use crate::triggers::mqtt::{MqttClientVersion, MqttV3Config, MqttV5Config, SubscribeTopic};
#[cfg(all(feature = "enterprise", feature = "nats", feature = "private"))]
use crate::triggers::nats::NatsTriggerConfigConnection;
+1 -4
View File
@@ -11,10 +11,7 @@ use {crate::db::ApiAuthed, windmill_common::DB};
#[cfg(not(feature = "private"))]
use anyhow::anyhow;
#[cfg(all(feature = "enterprise", not(feature = "private")))]
use std::sync::Arc;
#[cfg(all(feature = "enterprise", not(feature = "private")))]
use tokio::sync::RwLock;
use {std::sync::Arc, tokio::sync::RwLock};
#[cfg(not(feature = "private"))]
pub async fn validate_license_key(_license_key: String) -> anyhow::Result<(String, bool)> {
// Implementation is not open source
@@ -1,120 +0,0 @@
#[cfg(feature = "private")]
#[allow(unused)]
pub use crate::gcp_triggers_ee::*;
#[cfg(not(feature = "private"))]
use {
crate::{db::DB, trigger_helpers::TriggerJobArgs},
serde::{Deserialize, Serialize},
serde_json::value::RawValue,
sqlx::FromRow,
std::collections::HashMap,
windmill_common::{triggers::TriggerKind, utils::empty_as_none, worker::to_raw_value},
};
#[cfg(not(feature = "private"))]
type SqlxJson<T> = sqlx::types::Json<T>;
#[derive(sqlx::Type, Debug, Deserialize, Serialize)]
#[serde(rename_all(serialize = "lowercase", deserialize = "lowercase"))]
#[sqlx(type_name = "DELIVERY_MODE", rename_all = "lowercase")]
#[allow(unused)]
#[cfg(not(feature = "private"))]
pub enum DeliveryType {
Pull,
Push,
}
#[cfg(not(feature = "private"))]
impl Default for DeliveryType {
fn default() -> Self {
Self::Pull
}
}
#[derive(FromRow, Deserialize, Serialize, Debug)]
#[allow(unused)]
#[cfg(not(feature = "private"))]
pub struct PushConfig {
#[serde(deserialize_with = "empty_as_none")]
route_path: Option<String>,
#[serde(deserialize_with = "empty_as_none")]
audience: Option<String>,
authenticate: bool,
base_endpoint: String,
}
#[derive(Default, Debug, Serialize, Deserialize)]
#[allow(unused)]
#[cfg(not(feature = "private"))]
pub struct CreateUpdateConfig {
pub delivery_type: DeliveryType,
#[serde(default, deserialize_with = "empty_as_none")]
pub subscription_id: Option<String>,
pub delivery_config: Option<SqlxJson<PushConfig>>,
}
#[derive(Debug, Deserialize, Serialize)]
#[cfg(not(feature = "private"))]
pub struct ExistingGcpSubscription {
pub subscription_id: String,
pub base_endpoint: String,
}
#[derive(Debug, Deserialize, Serialize, sqlx::Type)]
#[serde(rename_all = "snake_case")]
#[sqlx(type_name = "GCP_SUBSCRIPTION_MODE", rename_all = "snake_case")]
#[cfg(not(feature = "private"))]
pub enum SubscriptionMode {
Existing,
CreateUpdate,
}
#[cfg(not(feature = "private"))]
pub fn start_consuming_gcp_pubsub_event(
_db: DB,
mut _killpill_rx: tokio::sync::broadcast::Receiver<()>,
) -> () {
// implementation is not open source
}
#[cfg(not(feature = "private"))]
#[derive(FromRow, Deserialize, Serialize, Debug)]
pub struct GcpTrigger {
pub gcp_resource_path: String,
pub subscription_id: String,
pub delivery_type: DeliveryType,
pub delivery_config: Option<SqlxJson<PushConfig>>,
pub subscription_mode: SubscriptionMode,
pub topic_id: String,
pub path: String,
pub script_path: String,
pub is_flow: bool,
pub workspace_id: String,
pub edited_by: String,
pub email: String,
pub edited_at: chrono::DateTime<chrono::Utc>,
pub extra_perms: Option<serde_json::Value>,
pub error: Option<String>,
pub server_id: Option<String>,
pub last_server_ping: Option<chrono::DateTime<chrono::Utc>>,
pub enabled: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub error_handler_path: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub error_handler_args: Option<SqlxJson<HashMap<String, Box<RawValue>>>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub retry: Option<SqlxJson<windmill_common::flows::Retry>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub auto_acknowledge_msg: Option<bool>,
}
#[cfg(not(feature = "private"))]
impl TriggerJobArgs<String> for GcpTrigger {
fn v1_payload_fn(payload: String) -> HashMap<String, Box<RawValue>> {
HashMap::from([("payload".to_string(), to_raw_value(&payload))])
}
fn trigger_kind() -> TriggerKind {
TriggerKind::Gcp
}
}
@@ -1,55 +0,0 @@
#[cfg(feature = "private")]
#[allow(unused)]
pub use crate::kafka_triggers_ee::*;
#[cfg(not(feature = "private"))]
use crate::db::DB;
#[cfg(not(feature = "private"))]
use serde::{Deserialize, Serialize};
#[derive(Serialize, Deserialize)]
#[cfg(not(feature = "private"))]
pub struct KafkaResourceSecurity {}
#[cfg(not(feature = "private"))]
pub fn start_kafka_consumers(
_db: DB,
mut _killpill_rx: tokio::sync::broadcast::Receiver<()>,
) -> () {
// implementation is not open source
}
#[derive(Serialize, Deserialize)]
#[cfg(not(feature = "private"))]
pub enum KafkaTriggerConfigConnection {}
#[derive(Serialize, Clone)]
#[cfg(not(feature = "private"))]
pub struct KafkaTrigger {
pub workspace_id: String,
pub path: String,
pub kafka_resource_path: String,
pub group_id: String,
pub topics: Vec<String>,
pub script_path: String,
pub is_flow: bool,
pub edited_by: String,
pub email: String,
pub edited_at: chrono::DateTime<chrono::Utc>,
#[serde(skip_serializing_if = "Option::is_none")]
pub server_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub last_server_ping: Option<chrono::DateTime<chrono::Utc>>,
pub extra_perms: serde_json::Value,
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
pub enabled: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub error_handler_path: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub error_handler_args: Option<
sqlx::types::Json<std::collections::HashMap<String, Box<serde_json::value::RawValue>>>,
>,
#[serde(skip_serializing_if = "Option::is_none")]
pub retry: Option<sqlx::types::Json<windmill_common::flows::Retry>>,
}
+3 -68
View File
@@ -20,7 +20,7 @@ use crate::smtp_server_oss::SmtpServer;
#[cfg(feature = "mcp")]
use crate::mcp::{extract_and_store_workspace_id, setup_mcp_server, shutdown_mcp_server};
use http::StatusCode;
use crate::triggers::start_all_listeners;
#[cfg(feature = "mcp")]
use rmcp::transport::streamable_http_server::session::local::LocalSessionManager;
use tower_http::catch_panic::CatchPanicLayer;
@@ -100,14 +100,9 @@ mod inkeep_oss;
mod inputs;
mod integration;
mod live_migrations;
#[cfg(feature = "postgres_trigger")]
mod postgres_triggers;
#[cfg(all(feature = "private", feature = "parquet"))]
pub mod s3_proxy_ee;
mod s3_proxy_oss;
mod trigger_helpers;
pub mod openapi;
mod approvals;
@@ -115,10 +110,6 @@ mod approvals;
pub mod apps_ee;
#[cfg(feature = "enterprise")]
mod apps_oss;
#[cfg(all(feature = "enterprise", feature = "gcp_trigger", feature = "private"))]
pub mod gcp_triggers_ee;
#[cfg(all(feature = "enterprise", feature = "gcp_trigger"))]
mod gcp_triggers_oss;
#[cfg(all(feature = "enterprise", feature = "private"))]
pub mod git_sync_ee;
#[cfg(feature = "enterprise")]
@@ -129,16 +120,6 @@ pub mod job_helpers_ee;
mod job_helpers_oss;
pub mod job_metrics;
pub mod jobs;
#[cfg(all(feature = "enterprise", feature = "kafka", feature = "private"))]
pub mod kafka_triggers_ee;
#[cfg(all(feature = "enterprise", feature = "kafka"))]
mod kafka_triggers_oss;
#[cfg(feature = "mqtt_trigger")]
mod mqtt_triggers;
#[cfg(all(feature = "enterprise", feature = "nats", feature = "private"))]
pub mod nats_triggers_ee;
#[cfg(all(feature = "enterprise", feature = "nats"))]
mod nats_triggers_oss;
#[cfg(all(feature = "oauth2", feature = "private"))]
pub mod oauth2_ee;
#[cfg(feature = "oauth2")]
@@ -164,10 +145,6 @@ mod slack_approvals;
pub mod smtp_server_ee;
#[cfg(feature = "smtp")]
mod smtp_server_oss;
#[cfg(all(feature = "enterprise", feature = "sqs_trigger", feature = "private"))]
pub mod sqs_triggers_ee;
#[cfg(all(feature = "enterprise", feature = "sqs_trigger"))]
mod sqs_triggers_oss;
#[cfg(feature = "private")]
pub mod teams_approvals_ee;
mod teams_approvals_oss;
@@ -191,8 +168,6 @@ mod utils;
pub mod var_resource_cache;
mod variables;
pub mod webhook_util;
#[cfg(feature = "websocket")]
mod websocket_triggers;
mod workers;
mod workspaces;
#[cfg(feature = "private")]
@@ -391,47 +366,7 @@ pub async fn run_server(
let triggers_service = triggers::generate_trigger_routers();
if !*CLOUD_HOSTED && server_mode && !mcp_mode {
#[cfg(feature = "websocket")]
{
let ws_killpill_rx = killpill_rx.resubscribe();
websocket_triggers::start_websockets(db.clone(), ws_killpill_rx);
}
#[cfg(all(feature = "enterprise", feature = "kafka"))]
{
let kafka_killpill_rx = killpill_rx.resubscribe();
kafka_triggers_oss::start_kafka_consumers(db.clone(), kafka_killpill_rx);
}
#[cfg(all(feature = "enterprise", feature = "nats"))]
{
let nats_killpill_rx = killpill_rx.resubscribe();
nats_triggers_oss::start_nats_consumers(db.clone(), nats_killpill_rx);
}
#[cfg(feature = "postgres_trigger")]
{
let db_killpill_rx = killpill_rx.resubscribe();
postgres_triggers::start_database(db.clone(), db_killpill_rx);
}
#[cfg(feature = "mqtt_trigger")]
{
let mqtt_killpill_rx = killpill_rx.resubscribe();
mqtt_triggers::start_mqtt_consumer(db.clone(), mqtt_killpill_rx);
}
#[cfg(all(feature = "enterprise", feature = "sqs_trigger"))]
{
let sqs_killpill_rx = killpill_rx.resubscribe();
sqs_triggers_oss::start_sqs(db.clone(), sqs_killpill_rx);
}
#[cfg(all(feature = "enterprise", feature = "gcp_trigger"))]
{
let gcp_killpill_rx = killpill_rx.resubscribe();
gcp_triggers_oss::start_consuming_gcp_pubsub_event(db.clone(), gcp_killpill_rx);
}
start_all_listeners(db.clone(), &killpill_rx);
}
let listener = tokio::net::TcpListener::bind(addr)
@@ -740,7 +675,7 @@ pub async fn run_server(
let app = app.layer(CatchPanicLayer::custom(|err| {
tracing::error!("panic in handler, returning 500: {:?}", err);
Response::builder()
.status(StatusCode::INTERNAL_SERVER_ERROR)
.status(http::StatusCode::INTERNAL_SERVER_ERROR)
.body(Body::from("Internal Server Error"))
.unwrap()
}));
File diff suppressed because it is too large Load Diff
@@ -1,54 +0,0 @@
#[cfg(feature = "private")]
#[allow(unused)]
pub use crate::nats_triggers_ee::*;
#[cfg(not(feature = "private"))]
use crate::db::DB;
#[cfg(not(feature = "private"))]
use serde::{Deserialize, Serialize};
#[cfg(not(feature = "private"))]
#[derive(Serialize, Deserialize)]
pub struct NatsResourceAuth {}
#[cfg(not(feature = "private"))]
pub fn start_nats_consumers(_db: DB, mut _killpill_rx: tokio::sync::broadcast::Receiver<()>) -> () {
// implementation is not open source
}
#[derive(Serialize, Deserialize)]
#[cfg(not(feature = "private"))]
pub enum NatsTriggerConfigConnection {}
#[derive(Serialize, Clone)]
#[cfg(not(feature = "private"))]
pub struct NatsTrigger {
pub workspace_id: String,
pub path: String,
pub nats_resource_path: String,
pub subjects: Vec<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub stream_name: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub consumer_name: Option<String>,
pub use_jetstream: bool,
pub script_path: String,
pub is_flow: bool,
pub edited_by: String,
pub email: String,
pub edited_at: chrono::DateTime<chrono::Utc>,
#[serde(skip_serializing_if = "Option::is_none")]
pub server_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub last_server_ping: Option<chrono::DateTime<chrono::Utc>>,
pub extra_perms: serde_json::Value,
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
pub enabled: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub error_handler_path: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub error_handler_args: Option<serde_json::Value>,
#[serde(skip_serializing_if = "Option::is_none")]
pub retry: Option<serde_json::Value>,
}
@@ -1,244 +0,0 @@
use crate::{
db::{ApiAuthed, DB},
resources::try_get_resource_from_db_as,
};
use native_tls::{Certificate, TlsConnector};
use pg_escape::quote_identifier;
use rust_postgres::{config::SslMode, Client, Config, NoTls};
use rust_postgres_native_tls::MakeTlsConnector;
use serde::{Deserialize, Serialize};
use serde_json::value::RawValue;
use sqlx::FromRow;
use std::collections::HashMap;
use windmill_common::{
db::UserDB,
error::{to_anyhow, Error, Result},
utils::empty_as_none,
};
mod bool;
mod converter;
mod hex;
mod relation;
mod replication_message;
mod trigger;
pub use trigger::start_database;
const ERROR_REPLICATION_SLOT_NOT_EXISTS: &str = r#"The replication slot associated with this trigger no longer exists. Recreate a new replication slot or select an existing one in the advanced tab, or delete and recreate a new trigger"#;
const ERROR_PUBLICATION_NAME_NOT_EXISTS: &str = r#"The publication associated with this trigger no longer exists. Recreate a new publication or select an existing one in the advanced tab, or delete and recreate a new trigger"#;
fn build_tls_connector(
ssl_mode: SslMode,
root_certificate_pem: Option<&String>,
) -> Result<Option<MakeTlsConnector>> {
let get_tls_builder_for_verify = |root_certificate: Option<&String>| {
let mut builder = TlsConnector::builder();
if let Some(root_certificate) = root_certificate {
let root_certificate_pem =
Certificate::from_pem(root_certificate.as_bytes()).map_err(to_anyhow)?;
builder.add_root_certificate(root_certificate_pem);
}
Ok::<_, Error>(builder)
};
let connector = match ssl_mode {
SslMode::Disable => return Ok(None),
SslMode::Require | SslMode::Prefer => {
let mut builder = TlsConnector::builder();
builder.danger_accept_invalid_certs(true);
builder.danger_accept_invalid_hostnames(true);
builder
}
SslMode::VerifyCa => {
let mut builder = get_tls_builder_for_verify(root_certificate_pem)?;
builder.danger_accept_invalid_hostnames(true);
builder
}
SslMode::VerifyFull => {
let builder = get_tls_builder_for_verify(root_certificate_pem)?;
builder
}
_ => unreachable!(),
};
Ok(Some(MakeTlsConnector::new(
connector.build().map_err(to_anyhow)?,
)))
}
#[derive(FromRow, Serialize, Deserialize, Debug)]
pub struct Postgres {
pub user: String,
pub password: String,
pub host: String,
pub port: Option<u16>,
pub dbname: String,
#[serde(default)]
pub sslmode: String,
#[serde(default, deserialize_with = "empty_as_none")]
pub root_certificate_pem: Option<String>,
}
pub async fn get_raw_postgres_connection(
database: &Postgres,
logical_mode: bool,
) -> Result<Client> {
let ssl_mode = match database.sslmode.as_ref() {
"disable" => SslMode::Disable,
"" | "prefer" | "allow" => SslMode::Prefer,
"require" => SslMode::Require,
"verify-ca" => SslMode::VerifyCa,
"verify-full" => SslMode::VerifyFull,
ssl_mode => {
return Err(Error::BadRequest(
format!("Invalid ssl mode for postgres: {}, please put a valid ssl_mode among the following available ssl mode: ['disable', 'allow', 'prefer', 'verify-ca', 'verify-full']", ssl_mode),
))
}
};
let mut config = Config::new();
config
.dbname(&database.dbname)
.host(&database.host)
.user(&database.user)
.ssl_mode(ssl_mode);
if logical_mode {
config.replication_mode(rust_postgres::config::ReplicationMode::Logical);
}
if let Some(port) = database.port {
config.port(port);
};
if !database.password.is_empty() {
config.password(&database.password);
}
let connector = build_tls_connector(ssl_mode, database.root_certificate_pem.as_ref())?;
let client = if let Some(connector) = connector {
let (client, connection) = config.connect(connector).await.map_err(to_anyhow)?;
tokio::spawn(async move {
if let Err(e) = connection.await {
tracing::debug!("{:#?}", e);
};
tracing::info!("Successfully Connected into database");
});
client
} else {
let (client, connection) = config.connect(NoTls).await.map_err(to_anyhow)?;
tokio::spawn(async move {
if let Err(e) = connection.await {
tracing::debug!("{:#?}", e);
};
tracing::info!("Successfully Connected into database");
});
client
};
Ok(client)
}
pub async fn get_pg_connection(
authed: ApiAuthed,
user_db: Option<UserDB>,
db: &DB,
postgres_resource_path: &str,
w_id: &str,
logical_mode: bool,
) -> Result<Client> {
let database =
try_get_resource_from_db_as::<Postgres>(&authed, user_db, db, postgres_resource_path, w_id)
.await?;
Ok(get_raw_postgres_connection(&database, logical_mode).await?)
}
pub async fn get_default_pg_connection(
authed: ApiAuthed,
user_db: Option<UserDB>,
db: &DB,
postgres_resource_path: &str,
w_id: &str,
) -> Result<Client> {
get_pg_connection(authed, user_db, db, postgres_resource_path, w_id, false).await
}
#[derive(FromRow, Deserialize, Serialize, Debug)]
pub struct PostgresTrigger {
pub path: String,
pub script_path: String,
pub is_flow: bool,
pub workspace_id: String,
pub edited_by: String,
pub email: String,
pub edited_at: chrono::DateTime<chrono::Utc>,
#[serde(skip_serializing_if = "Option::is_none")]
pub extra_perms: Option<serde_json::Value>,
pub postgres_resource_path: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub server_id: Option<String>,
pub replication_slot_name: String,
pub publication_name: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub last_server_ping: Option<chrono::DateTime<chrono::Utc>>,
pub enabled: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub error_handler_path: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub error_handler_args: Option<sqlx::types::Json<HashMap<String, Box<RawValue>>>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub retry: Option<sqlx::types::Json<windmill_common::flows::Retry>>,
}
pub async fn drop_logical_replication_slot(pg_connection: &Client, slot_name: &str) -> Result<()> {
let row = pg_connection
.query_opt(
r#"
SELECT
active_pid
FROM
pg_replication_slots
WHERE
slot_name = $1
"#,
&[&slot_name],
)
.await
.map_err(to_anyhow)?;
let active_pid = row.map(|r| r.get::<_, Option<i32>>(0)).flatten();
if let Some(pid) = active_pid {
pg_connection
.execute("SELECT pg_terminate_backend($1)", &[&pid])
.await
.map_err(to_anyhow)?;
}
pg_connection
.execute("SELECT pg_drop_replication_slot($1)", &[&slot_name])
.await
.map_err(to_anyhow)?;
Ok(())
}
pub async fn drop_publication(pg_connection: &Client, publication_name: &str) -> Result<()> {
let mut query = String::from("DROP PUBLICATION IF EXISTS ");
let quoted_publication_name = quote_identifier(publication_name);
query.push_str(&quoted_publication_name);
pg_connection
.execute(&query, &[])
.await
.map_err(to_anyhow)?;
Ok(())
}
File diff suppressed because it is too large Load Diff
@@ -1,44 +0,0 @@
#[cfg(feature = "private")]
#[allow(unused)]
pub use crate::sqs_triggers_ee::*;
#[cfg(not(feature = "private"))]
use crate::db::DB;
#[cfg(not(feature = "private"))]
use serde::{Deserialize, Serialize};
#[cfg(not(feature = "private"))]
use windmill_common::auth::aws::AwsAuthResourceType;
#[cfg(not(feature = "private"))]
pub fn start_sqs(_db: DB, mut _killpill_rx: tokio::sync::broadcast::Receiver<()>) -> () {
// implementation is not open source
}
#[derive(Debug, Clone, Deserialize, Serialize)]
#[cfg(not(feature = "private"))]
pub struct SqsTrigger {
pub queue_url: String,
pub aws_auth_resource_type: AwsAuthResourceType,
pub aws_resource_path: String,
pub message_attributes: Option<Vec<String>>,
pub path: String,
pub script_path: String,
pub is_flow: bool,
pub workspace_id: String,
pub edited_by: String,
pub email: String,
pub edited_at: chrono::DateTime<chrono::Utc>,
pub extra_perms: Option<serde_json::Value>,
pub error: Option<String>,
pub server_id: Option<String>,
pub last_server_ping: Option<chrono::DateTime<chrono::Utc>>,
pub enabled: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub error_handler_path: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub error_handler_args: Option<
sqlx::types::Json<std::collections::HashMap<String, Box<serde_json::value::RawValue>>>,
>,
#[serde(skip_serializing_if = "Option::is_none")]
pub retry: Option<sqlx::types::Json<windmill_common::flows::Retry>>,
}
-808
View File
@@ -1,808 +0,0 @@
use anyhow::Context;
use axum::response::IntoResponse;
use http::StatusCode;
use serde::Deserialize;
use serde_json::value::RawValue;
use std::collections::HashMap;
use uuid::Uuid;
use windmill_common::{
db::{UserDB, UserDbWithAuthed},
error::Result,
flows::{FlowModuleValue, Retry},
get_latest_deployed_hash_for_path, get_latest_flow_version_info_for_path,
jobs::{get_has_preprocessor_from_content_and_lang, script_path_to_payload, JobPayload},
scripts::{get_full_hub_script_by_path, ScriptHash, ScriptLang},
triggers::{
HubOrWorkspaceId, RunnableFormat, RunnableFormatVersion, TriggerKind,
RUNNABLE_FORMAT_VERSION_CACHE,
},
users::username_to_permissioned_as,
utils::StripPath,
worker::to_raw_value,
FlowVersionInfo,
};
use windmill_queue::{push, PushArgs, PushArgsOwned, PushIsolationLevel};
#[cfg(feature = "enterprise")]
use crate::jobs::check_license_key_valid;
use crate::{
db::{ApiAuthed, DB},
jobs::{
check_tag_available_for_workspace, delete_job_metadata_after_use, result_to_response,
run_flow_by_path_inner, run_script_by_path_inner, run_wait_result_internal, RunJobQuery,
},
utils::check_scopes,
HTTP_CLIENT,
};
struct ScriptInfo {
has_preprocessor: Option<bool>,
language: ScriptLang,
content: String,
schema: Option<sqlx::types::Json<PartialSchema>>,
}
#[derive(Debug, Deserialize)]
struct PropertyDefinition {
r#type: Option<String>,
}
#[derive(Debug, Deserialize)]
struct PartialSchema {
properties: Option<HashMap<String, PropertyDefinition>>,
}
#[derive(Debug, Clone, Eq, PartialEq, Hash)]
pub enum RunnableId {
FlowPath(String),
ScriptId(ScriptId),
HubScript(String),
}
impl RunnableId {
#[allow(dead_code)]
pub fn from_script_hash(hash: ScriptHash) -> Self {
Self::ScriptId(ScriptId::ScriptHash(hash))
}
pub fn from_script_path(path: &str) -> Self {
if path.starts_with("hub/") {
Self::HubScript(path.to_string())
} else {
Self::ScriptId(ScriptId::ScriptPath(path.to_string()))
}
}
pub fn from_flow_path(path: &str) -> Self {
Self::FlowPath(path.to_string())
}
}
#[derive(Debug, Clone, Eq, PartialEq, Hash)]
#[allow(unused)]
pub enum ScriptId {
ScriptPath(String),
ScriptHash(ScriptHash),
}
impl ScriptId {
async fn get_script_hash(self, workspace_id: &str, db: &DB) -> Result<i64> {
let hash = match self {
ScriptId::ScriptPath(path) => {
let info = get_latest_deployed_hash_for_path(None, db.clone(), workspace_id, &path)
.await?;
info.hash
}
ScriptId::ScriptHash(hash) => hash.0,
};
Ok(hash)
}
}
async fn get_script_info(
db: &DB,
workspace_id: &str,
hash: i64,
) -> std::result::Result<ScriptInfo, sqlx::Error> {
sqlx::query_as!(ScriptInfo, "SELECT has_preprocessor, language as \"language: _\", content, schema as \"schema: _\" FROM script WHERE workspace_id = $1 AND hash = $2", workspace_id, hash)
.fetch_one(db)
.await
}
fn runnable_format_from_schema_without_preprocessor(
trigger_kind: &TriggerKind,
schema: Option<sqlx::types::Json<PartialSchema>>,
) -> RunnableFormat {
match trigger_kind {
TriggerKind::Mqtt
if schema.as_ref().is_some_and(|schema| {
schema.properties.as_ref().is_some_and(|properties| {
properties.iter().any(|(key, def)| {
key == "payload" && def.r#type.as_ref().is_some_and(|t| t == "array")
})
})
}) =>
{
RunnableFormat { version: RunnableFormatVersion::V1, has_preprocessor: false }
}
TriggerKind::Kafka | TriggerKind::Nats
if schema.as_ref().is_some_and(|schema| {
schema
.properties
.as_ref()
.is_some_and(|properties| properties.keys().any(|key| key == "msg"))
}) =>
{
RunnableFormat { version: RunnableFormatVersion::V1, has_preprocessor: false }
}
_ => RunnableFormat { version: RunnableFormatVersion::V2, has_preprocessor: false },
}
}
fn runnable_format_from_preprocessor_args(
args: Option<Vec<windmill_parser::Arg>>,
) -> RunnableFormat {
if let Some(args) = args {
if args.iter().any(|arg| arg.name == "wm_trigger")
|| (args.len() > 0 && args.iter().all(|arg| arg.name != "event"))
{
RunnableFormat { version: RunnableFormatVersion::V1, has_preprocessor: true }
} else {
RunnableFormat { version: RunnableFormatVersion::V2, has_preprocessor: true }
}
} else {
RunnableFormat { version: RunnableFormatVersion::V2, has_preprocessor: true }
}
}
enum PreprocessorInfo {
Preprocessor { content: String, language: ScriptLang },
NoPreprocessor { schema: Option<sqlx::types::Json<PartialSchema>> },
}
#[derive(Debug, Deserialize)]
struct FlowInfo {
preprocessor_module: Option<sqlx::types::Json<FlowModuleValue>>,
schema: Option<sqlx::types::Json<PartialSchema>>,
}
fn get_preprocessor_args_from_content_and_language(
content: &str,
language: &ScriptLang,
) -> Result<Option<Vec<windmill_parser::Arg>>> {
let args = match language {
ScriptLang::Bun | ScriptLang::Bunnative | ScriptLang::Deno | ScriptLang::Nativets => {
let args = windmill_parser_ts::parse_deno_signature(
&content,
true,
false,
Some("preprocessor".to_string()),
)?;
Some(args.args)
}
ScriptLang::Python3 => {
let args = windmill_parser_py::parse_python_signature(
&content,
Some("preprocessor".to_string()),
false,
)?;
Some(args.args)
}
_ => None,
};
Ok(args)
}
pub async fn get_runnable_format(
runnable_id: RunnableId,
workspace_id: &str,
db: &DB,
trigger_kind: &TriggerKind,
) -> Result<RunnableFormat> {
let (key, preprocessor_info) = match runnable_id {
RunnableId::HubScript(path) => {
let Some(version) = path.split("/").nth(1) else {
return Err(windmill_common::error::Error::internal_err(
"Invalid hub script path".to_string(),
));
};
let version = match version.parse::<i64>() {
Ok(version) => version,
Err(_) => {
return Err(windmill_common::error::Error::internal_err(
"Invalid hub script version".to_string(),
));
}
};
let key = (HubOrWorkspaceId::Hub, version, trigger_kind.clone());
let runnable_format = RUNNABLE_FORMAT_VERSION_CACHE.get(&key);
if let Some(runnable_format) = runnable_format {
tracing::debug!("Using cached runnable format for hub script {path}");
return Ok(runnable_format);
}
let hub_script =
get_full_hub_script_by_path(StripPath(path.to_string()), &HTTP_CLIENT, Some(db))
.await?;
let has_preprocessor = get_has_preprocessor_from_content_and_lang(
&hub_script.content,
&hub_script.language,
)?;
let partial_schema = serde_json::from_str(hub_script.schema.get())?;
(
key,
if has_preprocessor {
PreprocessorInfo::Preprocessor {
content: hub_script.content,
language: hub_script.language,
}
} else {
PreprocessorInfo::NoPreprocessor {
schema: Some(sqlx::types::Json(partial_schema)),
}
},
)
}
RunnableId::FlowPath(path) => {
let FlowVersionInfo { version, .. } =
get_latest_flow_version_info_for_path(None, &db, workspace_id, &path, true).await?;
let key = (
HubOrWorkspaceId::WorkspaceId(workspace_id.to_string()),
version,
trigger_kind.clone(),
);
let runnable_format = RUNNABLE_FORMAT_VERSION_CACHE.get(&key);
if let Some(runnable_format) = runnable_format {
tracing::debug!("Using cached runnable format for flow {path}");
return Ok(runnable_format);
}
let flow_info = sqlx::query_as!(
FlowInfo,
"SELECT
value->'preprocessor_module'->'value' as \"preprocessor_module: _\",
schema as \"schema: _\"
FROM flow_version
WHERE
path = $1
AND workspace_id = $2
ORDER BY created_at DESC
LIMIT 1",
path,
workspace_id,
)
.fetch_one(db)
.await?;
if let Some(preprocessor_module) = flow_info.preprocessor_module {
match preprocessor_module.0 {
FlowModuleValue::RawScript { content, language, .. } => {
(key, PreprocessorInfo::Preprocessor { content, language })
}
FlowModuleValue::Script { path, hash, .. } => {
let hash = if let Some(hash) = hash {
hash.0
} else {
let script_hash = get_latest_deployed_hash_for_path(
None,
db.clone(),
workspace_id,
&path,
)
.await?;
script_hash.hash
};
let script_info = get_script_info(db, workspace_id, hash).await?;
(
key,
PreprocessorInfo::Preprocessor {
content: script_info.content,
language: script_info.language,
},
)
}
_ => {
return Err(windmill_common::error::Error::internal_err(
"Unsupported preprocessor module".to_string(),
));
}
}
} else {
(
key,
PreprocessorInfo::NoPreprocessor { schema: flow_info.schema },
)
}
}
RunnableId::ScriptId(script_id) => {
let hash = script_id.get_script_hash(workspace_id, db).await?;
let key = (
HubOrWorkspaceId::WorkspaceId(workspace_id.to_string()),
hash,
trigger_kind.clone(),
);
let runnable_format = RUNNABLE_FORMAT_VERSION_CACHE.get(&key);
if let Some(runnable_format) = runnable_format {
tracing::debug!("Using cached runnable format for script {hash}");
return Ok(runnable_format);
}
let script_info = get_script_info(db, workspace_id, hash).await?;
if script_info.has_preprocessor.unwrap_or(false) {
(
key,
PreprocessorInfo::Preprocessor {
content: script_info.content,
language: script_info.language,
},
)
} else {
(
key,
PreprocessorInfo::NoPreprocessor { schema: script_info.schema },
)
}
}
};
let runnable_format = match preprocessor_info {
PreprocessorInfo::Preprocessor { content, language } => {
let args = get_preprocessor_args_from_content_and_language(&content, &language)?;
runnable_format_from_preprocessor_args(args)
}
PreprocessorInfo::NoPreprocessor { schema } => {
runnable_format_from_schema_without_preprocessor(trigger_kind, schema)
}
};
RUNNABLE_FORMAT_VERSION_CACHE.insert(key, runnable_format);
Ok(runnable_format)
}
#[allow(dead_code)]
pub trait TriggerJobArgs<T: Clone> {
fn v1_payload_fn(payload: T) -> HashMap<String, Box<RawValue>>;
fn v2_payload_fn(payload: T) -> HashMap<String, Box<RawValue>> {
Self::v1_payload_fn(payload)
}
fn trigger_kind() -> TriggerKind;
fn build_job_args_v2(
has_preprocessor: bool,
payload: T,
info: HashMap<String, Box<RawValue>>,
) -> PushArgsOwned {
let trigger_kind = Self::trigger_kind();
let mut args = Self::v2_payload_fn(payload);
if has_preprocessor {
args.insert("kind".to_string(), to_raw_value(&trigger_kind.to_key()));
args.extend(info);
let args = HashMap::from([("event".to_string(), to_raw_value(&args))]);
PushArgsOwned { args, extra: None }
} else {
PushArgsOwned { args, extra: None }
}
}
fn build_job_args_v1(
has_preprocessor: bool,
payload: T,
info: HashMap<String, Box<RawValue>>,
) -> PushArgsOwned {
let trigger_kind = Self::trigger_kind();
let trigger_key = trigger_kind.to_key();
let args = Self::v1_payload_fn(payload);
let extra = if has_preprocessor {
Some(HashMap::from([(
"wm_trigger".to_string(),
to_raw_value(&serde_json::json!({
"kind": trigger_key,
trigger_key: info
})),
)]))
} else {
None
};
PushArgsOwned { args, extra }
}
async fn build_job_args(
runnable_path: &str,
is_flow: bool,
w_id: &str,
db: &DB,
payload: T,
info: HashMap<String, Box<RawValue>>,
) -> Result<PushArgsOwned> {
let runnable_id = if is_flow {
RunnableId::from_flow_path(runnable_path)
} else {
RunnableId::from_script_path(runnable_path)
};
Self::build_job_args_from_runnable_id(runnable_id, w_id, db, payload, info).await
}
async fn build_job_args_from_runnable_id(
runnable_id: RunnableId,
w_id: &str,
db: &DB,
payload: T,
info: HashMap<String, Box<RawValue>>,
) -> Result<PushArgsOwned> {
let runnable_format =
get_runnable_format(runnable_id, w_id, db, &Self::trigger_kind()).await?;
match runnable_format {
RunnableFormat { version: RunnableFormatVersion::V1, has_preprocessor } => {
Ok(Self::build_job_args_v1(has_preprocessor, payload, info))
}
RunnableFormat { version: RunnableFormatVersion::V2, has_preprocessor } => {
Ok(Self::build_job_args_v2(has_preprocessor, payload, info))
}
}
}
fn build_capture_payloads(
payload: T,
info: HashMap<String, Box<RawValue>>,
) -> (PushArgsOwned, PushArgsOwned) {
let main_args = Self::build_job_args_v2(false, payload.clone(), info.clone());
let preprocessor_args = Self::build_job_args_v2(true, payload, info);
(main_args, preprocessor_args)
}
}
#[allow(dead_code)]
async fn trigger_runnable_inner(
db: &DB,
user_db: Option<UserDB>,
authed: ApiAuthed,
workspace_id: &str,
runnable_path: &str,
is_flow: bool,
args: PushArgsOwned,
retry: Option<&sqlx::types::Json<Retry>>,
error_handler_path: Option<&str>,
error_handler_args: Option<&sqlx::types::Json<HashMap<String, Box<RawValue>>>>,
trigger_path: String,
) -> Result<(Uuid, Option<bool>, Option<String>)> {
let user_db = user_db.unwrap_or_else(|| UserDB::new(db.clone()));
let (uuid, delete_after_use, early_return) = if is_flow {
let run_query = RunJobQuery::default();
let path = StripPath(runnable_path.to_string());
let (uuid, early_return) = run_flow_by_path_inner(
authed,
db.clone(),
user_db,
workspace_id.to_string(),
path,
run_query,
args,
)
.await?;
(uuid, None, early_return)
} else {
let (uuid, delete_after_use) = trigger_script_internal(
db,
user_db,
authed,
workspace_id,
runnable_path,
args,
retry,
error_handler_path,
error_handler_args,
trigger_path,
)
.await?;
(uuid, delete_after_use, None)
};
Ok((uuid, delete_after_use, early_return))
}
#[allow(dead_code)]
pub async fn trigger_runnable(
db: &DB,
user_db: Option<UserDB>,
authed: ApiAuthed,
workspace_id: &str,
runnable_path: &str,
is_flow: bool,
args: PushArgsOwned,
retry: Option<&sqlx::types::Json<Retry>>,
error_handler_path: Option<&str>,
error_handler_args: Option<&sqlx::types::Json<HashMap<String, Box<RawValue>>>>,
trigger_path: String,
) -> Result<axum::response::Response> {
let (uuid, _, _) = trigger_runnable_inner(
db,
user_db,
authed,
workspace_id,
runnable_path,
is_flow,
args,
retry,
error_handler_path,
error_handler_args,
trigger_path,
)
.await?;
Ok((StatusCode::CREATED, uuid.to_string()).into_response())
}
#[allow(dead_code)]
pub async fn trigger_runnable_and_wait_for_result(
db: &DB,
user_db: Option<UserDB>,
authed: ApiAuthed,
workspace_id: &str,
runnable_path: &str,
is_flow: bool,
args: PushArgsOwned,
retry: Option<&sqlx::types::Json<Retry>>,
error_handler_path: Option<&str>,
error_handler_args: Option<&sqlx::types::Json<HashMap<String, Box<RawValue>>>>,
trigger_path: String,
) -> Result<axum::response::Response> {
let username = authed.username.clone();
let (uuid, delete_after_use, early_return) = trigger_runnable_inner(
db,
user_db,
authed,
workspace_id,
runnable_path,
is_flow,
args,
retry,
error_handler_path,
error_handler_args,
trigger_path,
)
.await?;
let (result, success) =
run_wait_result_internal(db, uuid, workspace_id.to_string(), early_return, &username)
.await?;
if delete_after_use.unwrap_or(false) {
delete_job_metadata_after_use(&db, uuid).await?;
}
result_to_response(result, success)
}
#[allow(dead_code)]
pub async fn trigger_runnable_and_wait_for_raw_result(
db: &DB,
user_db: Option<UserDB>,
authed: ApiAuthed,
workspace_id: &str,
runnable_path: &str,
is_flow: bool,
args: PushArgsOwned,
retry: Option<&sqlx::types::Json<Retry>>,
error_handler_path: Option<&str>,
error_handler_args: Option<&sqlx::types::Json<HashMap<String, Box<RawValue>>>>,
trigger_path: String,
) -> Result<Box<RawValue>> {
let username = authed.username.clone();
let (uuid, delete_after_use, early_return) = trigger_runnable_inner(
db,
user_db,
authed,
workspace_id,
runnable_path,
is_flow,
args,
retry,
error_handler_path,
error_handler_args,
trigger_path,
)
.await?;
let (result, success) =
run_wait_result_internal(db, uuid, workspace_id.to_string(), early_return, &username)
.await
.with_context(|| {
format!(
"Error fetching job result for {} {}",
if is_flow { "flow" } else { "script" },
runnable_path
)
})?;
if delete_after_use.unwrap_or(false) {
delete_job_metadata_after_use(&db, uuid).await?;
}
if !success {
Err(windmill_common::error::Error::internal_err(format!(
"{} {runnable_path} failed: {:?}",
if is_flow { "Flow" } else { "Script" },
result
)))
} else {
Ok(result)
}
}
async fn trigger_script_internal(
db: &DB,
user_db: UserDB,
authed: ApiAuthed,
workspace_id: &str,
script_path: &str,
args: PushArgsOwned,
retry: Option<&sqlx::types::Json<Retry>>,
error_handler_path: Option<&str>,
error_handler_args: Option<&sqlx::types::Json<HashMap<String, Box<RawValue>>>>,
trigger_path: String,
) -> Result<(Uuid, Option<bool>)> {
if retry.is_none() && error_handler_path.is_none() {
let run_query = RunJobQuery::default();
let path = StripPath(script_path.to_string());
run_script_by_path_inner(
authed,
db.clone(),
user_db,
workspace_id.to_string(),
path,
run_query,
args,
)
.await
} else {
trigger_script_with_retry_and_error_handler(
db,
user_db,
authed,
workspace_id,
script_path,
args,
retry,
error_handler_path,
error_handler_args,
trigger_path,
)
.await
}
}
async fn trigger_script_with_retry_and_error_handler(
db: &DB,
user_db: UserDB,
authed: ApiAuthed,
workspace_id: &str,
script_path: &str,
args: PushArgsOwned,
retry: Option<&sqlx::types::Json<Retry>>,
error_handler_path: Option<&str>,
error_handler_args: Option<&sqlx::types::Json<HashMap<String, Box<RawValue>>>>,
trigger_path: String,
) -> Result<(Uuid, Option<bool>)> {
#[cfg(feature = "enterprise")]
check_license_key_valid().await?;
check_scopes(&authed, || format!("jobs:run:scripts:{script_path}"))?;
let retry = retry.map(|r| r.0.clone());
let error_handler_path = error_handler_path.map(|p| p.to_string());
let error_handler_args = error_handler_args.map(|args| args.0.clone());
let (job_payload, tag, delete_after_use, timeout, on_behalf_of) = {
let db_authed = UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() };
script_path_to_payload(
script_path,
Some(db_authed),
db.clone(),
&workspace_id,
Some(false),
)
.await?
};
check_tag_available_for_workspace(&db, &workspace_id, &tag, &authed).await?;
let (email, permissioned_as, push_authed, tx) =
if let Some(on_behalf_of) = on_behalf_of.as_ref() {
(
on_behalf_of.email.as_str(),
on_behalf_of.permissioned_as.clone(),
None,
PushIsolationLevel::IsolatedRoot(db.clone()),
)
} else {
(
authed.email.as_str(),
username_to_permissioned_as(&authed.username),
Some(authed.clone().into()),
PushIsolationLevel::Isolated(user_db, authed.clone().into()),
)
};
let push_args = PushArgs { args: &args.args, extra: args.extra };
let retryable_job_payload = match job_payload {
JobPayload::ScriptHash {
hash,
path,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
priority,
apply_preprocessor,
..
} => JobPayload::SingleScriptFlow {
path,
hash,
args: HashMap::from(&push_args),
retry,
error_handler_path,
error_handler_args,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
priority,
tag_override: tag.clone(),
apply_preprocessor,
trigger_path: Some(trigger_path),
},
_ => {
return Err(windmill_common::error::Error::internal_err(format!(
"Unsupported job payload: {:?}",
job_payload
)))
}
};
let (uuid, tx) = push(
&db,
tx,
&workspace_id,
retryable_job_payload,
push_args,
authed.display_username(),
email,
permissioned_as,
authed.token_prefix.as_deref(),
None,
None,
None,
None,
None,
None,
false,
false,
None,
true,
tag,
timeout,
None,
None,
push_authed.as_ref(),
false,
)
.await?;
tx.commit().await?;
Ok((uuid, delete_after_use))
}
@@ -1,3 +1,4 @@
#[allow(unused)]
#[cfg(feature = "private")]
pub use super::handler_ee::*;
@@ -29,7 +30,7 @@ impl TriggerCrud for GcpTrigger {
const SUPPORTS_TEST_CONNECTION: bool = false;
const ROUTE_PREFIX: &'static str = "/gcp_triggers";
const DEPLOYMENT_NAME: &'static str = "";
const IS_CLOUD_HOSTED: bool = false;
const IS_ALLOWED_ON_CLOUD: bool = false;
fn get_deployed_object(path: String) -> DeployedObject {
DeployedObject::GcpTrigger { path }
@@ -0,0 +1,41 @@
#[allow(unused)]
#[cfg(feature = "private")]
pub use super::listener_ee::*;
#[cfg(not(feature = "private"))]
use {
super::GcpTrigger,
crate::triggers::{listener::ListeningTrigger, Listener},
std::sync::Arc,
tokio::sync::RwLock,
windmill_common::{error::Result, jobs::JobTriggerKind, DB},
};
#[cfg(not(feature = "private"))]
#[async_trait::async_trait]
impl Listener for GcpTrigger {
type Consumer = ();
type Extra = ();
const JOB_TRIGGER_KIND: JobTriggerKind = JobTriggerKind::Gcp;
async fn get_consumer(
&self,
_db: &DB,
_listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
_err_message: Arc<RwLock<Option<String>>>,
_killpill_rx: tokio::sync::broadcast::Receiver<()>,
) -> Result<Option<Self::Consumer>> {
Ok(None)
}
async fn consume(
&self,
_db: &DB,
_consumer: Self::Consumer,
_listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
_err_message: Arc<RwLock<Option<String>>>,
_killpill_rx: tokio::sync::broadcast::Receiver<()>,
) {
()
}
}
@@ -2,6 +2,11 @@
mod handler_ee;
pub mod handler_oss;
#[cfg(feature = "private")]
mod listener_ee;
pub mod listener_oss;
#[cfg(feature = "private")]
mod mod_ee;
#[cfg(feature = "private")]
@@ -33,7 +33,7 @@ impl TriggerCrud for KafkaTrigger {
const SUPPORTS_TEST_CONNECTION: bool = false;
const ROUTE_PREFIX: &'static str = "/kafka_triggers";
const DEPLOYMENT_NAME: &'static str = "";
const IS_CLOUD_HOSTED: bool = false;
const IS_ALLOWED_ON_CLOUD: bool = false;
fn get_deployed_object(path: String) -> DeployedObject {
DeployedObject::KafkaTrigger { path }
@@ -0,0 +1,41 @@
#[allow(unused)]
#[cfg(feature = "private")]
pub use super::listener_ee::*;
#[cfg(not(feature = "private"))]
use {
super::KafkaTrigger,
crate::triggers::{listener::ListeningTrigger, Listener},
std::sync::Arc,
tokio::sync::RwLock,
windmill_common::{error::Result, jobs::JobTriggerKind, DB},
};
#[cfg(not(feature = "private"))]
#[async_trait::async_trait]
impl Listener for KafkaTrigger {
type Consumer = ();
type Extra = ();
const JOB_TRIGGER_KIND: JobTriggerKind = JobTriggerKind::Kafka;
async fn get_consumer(
&self,
_db: &DB,
_listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
_err_message: Arc<RwLock<Option<String>>>,
_killpill_rx: tokio::sync::broadcast::Receiver<()>,
) -> Result<Option<Self::Consumer>> {
Ok(None)
}
async fn consume(
&self,
_db: &DB,
_consumer: Self::Consumer,
_listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
_err_message: Arc<RwLock<Option<String>>>,
_killpill_rx: tokio::sync::broadcast::Receiver<()>,
) {
()
}
}
@@ -2,6 +2,10 @@
mod handler_ee;
pub mod handler_oss;
#[cfg(feature = "private")]
mod listener_ee;
pub mod listener_oss;
#[cfg(feature = "private")]
mod mod_ee;
#[cfg(feature = "private")]
@@ -0,0 +1,901 @@
use std::{collections::HashMap, fmt::Debug, sync::Arc};
use crate::{
capture::insert_capture_payload,
db::ApiAuthed,
triggers::{
handler::TriggerCrud,
trigger_helpers::{trigger_runnable, TriggerJobArgs},
Trigger, TriggerErrorHandling,
},
users::fetch_api_authed,
};
use async_trait::async_trait;
use itertools::Itertools;
use rand::seq::SliceRandom;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use serde_json::value::RawValue;
use sql_builder::SqlBuilder;
use sqlx::{FromRow, Row};
use tokio::sync::RwLock;
use windmill_common::{
error::{Error, Result},
jobs::JobTriggerKind,
triggers::TriggerKind,
utils::report_critical_error,
DB, INSTANCE_NAME,
};
#[allow(unused)]
#[async_trait]
pub trait Listener: TriggerCrud + TriggerJobArgs {
type Consumer: Send;
type Extra: Send + Sync;
//to use in next PR to add job trigger kind to eow
#[allow(unused)]
const JOB_TRIGGER_KIND: JobTriggerKind;
const EXTRA_TRIGGER_AND_WHERE_CLAUSE: &[&'static str] = &[];
const EXTRA_CAPTURE_AND_WHERE_CLAUSE: &[&'static str] = &[];
async fn get_consumer(
&self,
db: &DB,
listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
err_message: Arc<RwLock<Option<String>>>,
killpill_rx: tokio::sync::broadcast::Receiver<()>,
) -> Result<Option<Self::Consumer>>;
async fn consume(
&self,
db: &DB,
consumer: Self::Consumer,
listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
err_message: Arc<RwLock<Option<String>>>,
killpill_rx: tokio::sync::broadcast::Receiver<()>,
);
async fn fetch_enabled_unlistened_triggers(
&self,
db: &DB,
) -> Result<Vec<ListeningTrigger<Self::TriggerConfig>>> {
let mut fields = vec![
"workspace_id",
"path",
"script_path",
"is_flow",
"edited_by",
"email",
"edited_at",
"extra_perms",
];
if Self::SUPPORTS_SERVER_STATE {
fields.extend_from_slice(&["enabled", "server_id", "last_server_ping", "error"]);
}
fields.extend_from_slice(&["error_handler_path", "error_handler_args", "retry"]);
fields.extend_from_slice(Self::ADDITIONAL_SELECT_FIELDS);
let mut sqlb = SqlBuilder::select_from(Self::TABLE_NAME);
sqlb.fields(&fields).and_where("enabled IS TRUE").and_where(
"(last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds')",
);
for where_clause in Self::EXTRA_TRIGGER_AND_WHERE_CLAUSE {
sqlb.and_where(where_clause);
}
let sql = sqlb
.sql()
.map_err(|e| Error::InternalErr(format!("SQL error: {}", e)))?;
let triggers: Vec<Trigger<Self::TriggerConfig>> =
sqlx::query_as(&sql).fetch_all(db).await?;
let triggers = triggers
.into_iter()
.map(|trigger| ListeningTrigger {
path: trigger.base.path,
workspace_id: trigger.base.workspace_id,
is_flow: trigger.base.is_flow,
username: trigger.base.edited_by,
email: trigger.base.email,
script_path: trigger.base.script_path,
trigger_config: trigger.config,
error_handling: Some(trigger.error_handling),
trigger_mode: true,
})
.collect_vec();
Ok(triggers)
}
async fn fetch_unlistened_captures(
&self,
db: &DB,
) -> Result<Vec<ListeningTrigger<Self::TriggerConfig>>> {
let fields = vec![
"path",
"is_flow",
"workspace_id",
"owner AS username",
"email",
"trigger_config",
];
let mut sqlb = SqlBuilder::select_from("capture_config");
sqlb.fields(&fields)
.and_where(format!("trigger_kind = '{}'", Self::TRIGGER_KIND.to_key()))
.and_where("last_client_ping > NOW() - INTERVAL '10 seconds'")
.and_where("trigger_config IS NOT NULL")
.and_where(
"(last_server_ping IS NULL OR last_server_ping < NOW() - INTERVAL '15 seconds')",
);
for where_clause in Self::EXTRA_CAPTURE_AND_WHERE_CLAUSE {
sqlb.and_where(where_clause);
}
let sql = sqlb.sql().expect("failed to build SQL");
let captures: Vec<Capture<Self::TriggerConfig>> =
sqlx::query_as(&sql).fetch_all(db).await?;
let captures = captures
.into_iter()
.map(|capture| ListeningTrigger {
username: capture.username,
path: capture.path,
workspace_id: capture.workspace_id,
script_path: "".to_string(),
email: capture.email,
trigger_config: capture.trigger_config,
trigger_mode: false,
is_flow: capture.is_flow,
error_handling: None,
})
.collect_vec();
Ok(captures)
}
async fn cleanup(
&self,
_db: &DB,
_listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
) -> Result<()> {
Ok(())
}
async fn loop_ping(
&self,
db: &DB,
listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
status: Arc<RwLock<Option<String>>>,
error_message: Option<String>,
) {
update_rw_lock(status.clone(), error_message).await;
loop {
if let None = self
.update_ping(db, listening_trigger, status.read().await.as_deref())
.await
{
return;
}
tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
}
}
async fn update_ping(
&self,
db: &DB,
listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
error: Option<&str>,
) -> Option<()> {
if listening_trigger.trigger_mode {
self.update_trigger_ping(db, listening_trigger, error).await
} else {
self.update_capture_ping(db, listening_trigger, error).await
}
}
async fn update_ping_and_loop_ping_status(
&self,
db: &DB,
listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
loop_ping_status: Arc<RwLock<Option<String>>>,
error: Option<String>,
) -> Option<()> {
// update immediately the ping status and update the loop ping status so that the next loop pings will display the new status
update_rw_lock(loop_ping_status.clone(), error.clone()).await;
if let None = self
.update_ping(db, listening_trigger, error.as_deref())
.await
{
return None;
}
Some(())
}
async fn update_trigger_ping(
&self,
db: &DB,
listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
error: Option<&str>,
) -> Option<()> {
let updated = sqlx::query_scalar::<_, i32>(&format!(
r#"
UPDATE
{}
SET
last_server_ping = now(), error = $1
WHERE
workspace_id = $2 AND
path = $3 AND
server_id = $4 AND
enabled IS TRUE
RETURNING 1
"#,
Self::TABLE_NAME
))
.bind(error)
.bind(&listening_trigger.workspace_id)
.bind(&listening_trigger.path)
.bind(&*INSTANCE_NAME)
.fetch_optional(db)
.await;
self.handle_ping_result(updated, db, listening_trigger, "trigger")
.await
}
async fn update_capture_ping(
&self,
db: &DB,
listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
error: Option<&str>,
) -> Option<()> {
let updated = sqlx::query_scalar!(
r#"
UPDATE
capture_config
SET
last_server_ping = now(), error = $1
WHERE
workspace_id = $2 AND
path = $3 AND
is_flow = $4 AND
trigger_kind = $5 AND
server_id = $6 AND
last_client_ping > NOW() - INTERVAL '10 seconds'
RETURNING 1
"#,
error,
&listening_trigger.workspace_id,
&listening_trigger.path,
&listening_trigger.is_flow,
Self::TRIGGER_KIND as TriggerKind,
&*INSTANCE_NAME
)
.fetch_optional(db)
.await
.map(|result| result.flatten());
self.handle_ping_result(updated, db, listening_trigger, "capture")
.await
}
async fn handle_ping_result(
&self,
result: sqlx::Result<Option<i32>>,
db: &DB,
listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
entity_type: &str,
) -> Option<()> {
match result {
Ok(updated) => {
if updated.is_none() {
self.reset_ping_for_restart(db, listening_trigger).await;
tracing::info!(
"{} {} {} changed, disabled, or deleted, stopping...",
Self::TRIGGER_KIND,
entity_type,
listening_trigger.path
);
return None;
}
}
Err(error) => {
tracing::warn!(
"Error updating ping of {} {} {}: {:?}",
Self::TRIGGER_KIND,
entity_type,
&listening_trigger.path,
error
);
}
}
Some(())
}
async fn reset_ping_for_restart(
&self,
db: &DB,
listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
) {
if listening_trigger.trigger_mode {
let _ = sqlx::query(&format!(
r#"
UPDATE
{}
SET
last_server_ping = NULL
WHERE
workspace_id = $1 AND
path = $2 AND
server_id IS NULL
"#,
Self::TABLE_NAME
))
.bind(&listening_trigger.workspace_id)
.bind(&listening_trigger.path)
.execute(db)
.await;
} else {
let _ = sqlx::query!(
r#"
UPDATE
capture_config
SET
last_server_ping = NULL
WHERE
workspace_id = $1 AND
path = $2 AND
is_flow = $3 AND
trigger_kind = $4 AND
server_id IS NULL
"#,
&listening_trigger.workspace_id,
&listening_trigger.path,
&listening_trigger.is_flow,
Self::TRIGGER_KIND as TriggerKind
)
.execute(db)
.await;
}
}
async fn disable_with_error(
&self,
db: &DB,
listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
error: String,
) {
if listening_trigger.trigger_mode {
let report_status = sqlx::query(&format!(
r#"
UPDATE
{}
SET
enabled = FALSE,
error = $1,
server_id = NULL,
last_server_ping = NULL
WHERE
workspace_id = $2 AND
path = $3
"#,
Self::TABLE_NAME
))
.bind(&error)
.bind(&listening_trigger.workspace_id)
.bind(&listening_trigger.path)
.execute(db)
.await;
match report_status {
Ok(_) => {
report_critical_error(
format!(
"Disabling {} trigger {} because of error: {}",
Self::TRIGGER_KIND,
listening_trigger.path,
error
),
db.clone(),
Some(&listening_trigger.workspace_id),
None,
)
.await;
}
Err(disable_err) => {
report_critical_error(
format!("Could not disable {} trigger {} with err {}, disabling because of error {}", Self::TRIGGER_KIND, listening_trigger.path, disable_err, error),
db.clone(),
Some(&listening_trigger.workspace_id),
None,
).await;
}
}
return;
}
let report_status = sqlx::query!(
r#"
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 = $5
"#,
error,
listening_trigger.workspace_id,
listening_trigger.path,
listening_trigger.is_flow,
Self::TRIGGER_KIND as TriggerKind
)
.execute(db)
.await;
if let Err(disable_err) = report_status {
tracing::error!(
"Could not disable {} capture {} ({}) with err {}, disabling because of error {}",
Self::TRIGGER_KIND,
listening_trigger.path,
listening_trigger.workspace_id,
disable_err,
error
)
}
}
async fn handle_trigger(
&self,
db: &DB,
listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
payload: Self::Payload,
trigger_info: HashMap<String, Box<RawValue>>,
_extra: Option<Self::Extra>,
) -> Result<()> {
let args = Self::build_job_args(
&listening_trigger.script_path,
listening_trigger.is_flow,
&listening_trigger.workspace_id,
db,
payload,
trigger_info,
)
.await?;
let authed = listening_trigger
.authed(db, &Self::TRIGGER_KIND.to_string())
.await?;
let (retry, error_handler_path, error_handler_args) =
match listening_trigger.error_handling.as_ref() {
Some(error_handling) => (
error_handling.retry.as_ref(),
error_handling.error_handler_path.as_deref(),
error_handling.error_handler_args.as_ref(),
),
None => (None, None, None),
};
trigger_runnable(
db,
None,
authed,
&listening_trigger.workspace_id,
&listening_trigger.script_path,
listening_trigger.is_flow,
args,
retry,
error_handler_path.as_deref(),
error_handler_args,
format!("{}_trigger/{}", Self::TRIGGER_KIND, listening_trigger.path),
)
.await?;
Ok(())
}
async fn handle_event(
&self,
db: &DB,
listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
payload: Self::Payload,
trigger_info: HashMap<String, Box<RawValue>>,
extra: Option<Self::Extra>,
) -> Result<()> {
if listening_trigger.trigger_mode {
if let Err(err) = self
.handle_trigger(db, listening_trigger, payload, trigger_info, extra)
.await
{
report_critical_error(
format!(
"Failed to trigger job from {} event {}: {:?}",
Self::TRIGGER_KIND,
listening_trigger.path,
err
),
db.clone(),
Some(&listening_trigger.workspace_id),
None,
)
.await;
return Err(err);
};
return Ok(());
}
let (main_args, preprocessor_args) = Self::build_capture_payloads(&payload, trigger_info);
if let Err(err) = insert_capture_payload(
db,
&listening_trigger.workspace_id,
&listening_trigger.path,
listening_trigger.is_flow,
&Self::TRIGGER_KIND,
main_args,
preprocessor_args,
&listening_trigger.username,
)
.await
{
tracing::error!("Error inserting capture payload: {:?}", err);
return Err(err);
}
Ok(())
}
}
#[allow(unused)]
async fn listening<T: Listener>(
db: DB,
listener: T,
listening_trigger: ListeningTrigger<T::TriggerConfig>,
mut killpill_rx: tokio::sync::broadcast::Receiver<()>,
) {
let killpill_rx_consumer = killpill_rx.resubscribe();
let killpill_rx_get_consumer = killpill_rx.resubscribe();
let loop_ping_status = Arc::new(RwLock::new(None));
tokio::select! {
biased;
_ = killpill_rx.recv() => {
let _ = listener.cleanup(&db, &listening_trigger).await;
}
_ = listener.loop_ping(&db, &listening_trigger, loop_ping_status.clone(), Some("Connecting...".to_string())) => {
let _ = listener.cleanup(&db, &listening_trigger).await;
}
consumer = {
listener.get_consumer(&db, &listening_trigger, loop_ping_status.clone(), killpill_rx_get_consumer)
} => {
tokio::select! {
biased;
_ = killpill_rx.recv() => {
let _ = listener.cleanup(&db, &listening_trigger).await;
return;
}
_ = listener.loop_ping(&db, &listening_trigger, loop_ping_status.clone(), None) => {
let _ = listener.cleanup(&db, &listening_trigger).await;
return;
}
_ = async {
match consumer {
Ok(Some(consumer)) => {
listener.update_ping_and_loop_ping_status(&db, &listening_trigger, loop_ping_status.clone(), None).await;
let _ = listener.consume(&db, consumer, &listening_trigger, loop_ping_status.clone(), killpill_rx_consumer).await;
tracing::debug!("Stopping consumer for trigger");
}
Err(error) => {
tracing::warn!("Disabling trigger due to consumer error: {}", error);
listener.disable_with_error(&db, &listening_trigger, error.to_string()).await;
}
_ => {}
}
} => {
let _ = listener.cleanup(&db, &listening_trigger).await;
return;
}
}
}
}
}
#[allow(unused)]
async fn listen_to_unlistened_events<T: Copy + Listener>(
listener: T,
db: DB,
killpill_rx: &tokio::sync::broadcast::Receiver<()>,
) {
let unlistend_enabled_triggers = listener.fetch_enabled_unlistened_triggers(&db).await;
match unlistend_enabled_triggers {
Ok(mut unlistend_enabled_triggers) => {
unlistend_enabled_triggers.shuffle(&mut rand::rng());
for trigger in unlistend_enabled_triggers {
let has_lock = sqlx::query_scalar(&format!(
r#"
UPDATE
{}
SET
server_id = $1,
last_server_ping = now(),
error = 'Connecting...'
WHERE
enabled IS TRUE
AND workspace_id = $2
AND path = $3
AND (last_server_ping IS NULL
OR last_server_ping < now() - INTERVAL '15 seconds'
)
RETURNING true
"#,
T::TABLE_NAME,
))
.bind(&*INSTANCE_NAME)
.bind(&trigger.workspace_id)
.bind(&trigger.path)
.fetch_optional(&db)
.await;
match has_lock {
Ok(has_lock) => {
if has_lock.flatten().unwrap_or(false) {
tracing::info!(
"Spawning new task to listen for {} event",
T::TABLE_NAME
);
tokio::spawn({
let db = db.clone();
let killpill_rx = killpill_rx.resubscribe();
async move { listening(db, listener, trigger, killpill_rx).await }
});
} else {
tracing::info!(
"{} trigger {} already being listened to",
T::TRIGGER_KIND,
trigger.path
);
}
}
Err(err) => {
tracing::error!(
"Error acquiring lock for {} trigger {}: {:?}",
T::TRIGGER_KIND,
trigger.path,
err
);
}
};
}
}
Err(err) => {
tracing::error!("Error fetching {} triggers: {:?}", T::TRIGGER_KIND, err,);
}
}
let unlisted_captures = listener.fetch_unlistened_captures(&db).await;
match unlisted_captures {
Ok(unlistened_captures) => {
for capture in unlistened_captures {
let has_lock = sqlx::query_scalar!(
r#"
UPDATE
capture_config
SET
server_id = $1,
last_server_ping = now(),
error = 'Connecting...'
WHERE
last_client_ping > NOW() - INTERVAL '10 seconds' AND
workspace_id = $2 AND
path = $3 AND
is_flow = $4 AND
trigger_kind = $5 AND
(last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds')
RETURNING true
"#,
*INSTANCE_NAME,
&capture.workspace_id,
&capture.path,
&capture.is_flow,
T::TRIGGER_KIND as TriggerKind
)
.fetch_optional(&db)
.await;
match has_lock {
Ok(has_lock) => {
if has_lock.flatten().unwrap_or(false) {
tokio::spawn({
let db = db.clone();
let killpill_rx = killpill_rx.resubscribe();
async move { listening(db, listener, capture, killpill_rx).await }
});
} else {
tracing::info!(
"{} capture {} already being listened to",
T::TRIGGER_KIND.to_string(),
capture.path
);
}
}
Err(err) => {
tracing::error!(
"Error acquiring lock for capture {} {}: {:?}",
T::TRIGGER_KIND,
capture.path,
err
);
}
};
}
}
Err(err) => {
tracing::error!(
"Error fetching captures {} triggers: {:?}",
T::TRIGGER_KIND,
err
);
}
}
}
#[derive(Debug, Clone, Deserialize, Serialize)]
struct Capture<T>
where
T: for<'r> FromRow<'r, sqlx::postgres::PgRow>,
{
path: String,
is_flow: bool,
workspace_id: String,
username: String,
email: String,
#[serde(flatten)]
trigger_config: T,
}
impl<T> FromRow<'_, sqlx::postgres::PgRow> for Capture<T>
where
T: for<'r> FromRow<'r, sqlx::postgres::PgRow> + DeserializeOwned,
{
fn from_row(row: &sqlx::postgres::PgRow) -> std::result::Result<Self, sqlx::Error> {
let trigger_config_value = row.try_get("trigger_config")?;
let trigger_config: T = serde_json::from_value(trigger_config_value)
.map_err(|e| sqlx::Error::Decode(Box::new(e)))?;
Ok(Capture {
path: row.try_get("path")?,
is_flow: row.try_get("is_flow")?,
workspace_id: row.try_get("workspace_id")?,
username: row.try_get("username")?,
email: row.try_get("email")?,
trigger_config,
})
}
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct ListeningTrigger<T> {
pub path: String,
pub is_flow: bool,
pub workspace_id: String,
pub username: String,
pub email: String,
pub trigger_config: T,
pub script_path: String,
pub trigger_mode: bool,
pub error_handling: Option<TriggerErrorHandling>,
}
impl<T> ListeningTrigger<T> {
pub async fn authed(&self, db: &DB, username: &str) -> Result<ApiAuthed> {
fetch_api_authed(
self.username.clone(),
self.email.clone(),
&self.workspace_id,
db,
Some(format!("{}-{}", username, self.path)),
)
.await
}
}
#[allow(unused)]
pub async fn update_rw_lock<T>(lock: std::sync::Arc<tokio::sync::RwLock<T>>, value: T) -> () {
let mut w = lock.write().await;
*w = value;
}
#[allow(unused)]
fn listen_to<T: Copy + Listener>(
trigger: T,
db: DB,
mut killpill_rx: tokio::sync::broadcast::Receiver<()>,
) {
tokio::spawn(async move {
listen_to_unlistened_events(trigger, db.clone(), &killpill_rx).await;
loop {
tokio::select! {
biased;
_ = killpill_rx.recv() => {
return;
}
_ = tokio::time::sleep(tokio::time::Duration::from_secs(15)) => {
listen_to_unlistened_events(trigger, db.clone(), &killpill_rx).await
}
}
}
});
}
#[allow(unused)]
pub fn start_all_listeners(db: DB, killpill_rx: &tokio::sync::broadcast::Receiver<()>) {
tracing::info!("Starting trigger listeners based on available features...");
#[cfg(feature = "postgres_trigger")]
{
let postgres_killpill_rx = killpill_rx.resubscribe();
use crate::triggers::postgres::PostgresTrigger;
listen_to(PostgresTrigger, db.clone(), postgres_killpill_rx)
}
#[cfg(feature = "mqtt_trigger")]
{
let mqtt_killpill_rx = killpill_rx.resubscribe();
use crate::triggers::mqtt::MqttTrigger;
listen_to(MqttTrigger, db.clone(), mqtt_killpill_rx)
}
#[cfg(feature = "websocket")]
{
let mqtt_killpill_rx = killpill_rx.resubscribe();
use crate::triggers::websocket::WebsocketTrigger;
listen_to(WebsocketTrigger, db.clone(), mqtt_killpill_rx)
}
#[cfg(all(feature = "gcp_trigger", feature = "enterprise", feature = "private"))]
{
let gcp_killpill_rx = killpill_rx.resubscribe();
use crate::triggers::gcp::GcpTrigger;
listen_to(GcpTrigger, db.clone(), gcp_killpill_rx);
}
#[cfg(all(feature = "sqs_trigger", feature = "enterprise", feature = "private"))]
{
let gcp_killpill_rx = killpill_rx.resubscribe();
use crate::triggers::sqs::SqsTrigger;
listen_to(SqsTrigger, db.clone(), gcp_killpill_rx);
}
#[cfg(all(feature = "nats", feature = "enterprise", feature = "private"))]
{
let gcp_killpill_rx = killpill_rx.resubscribe();
use crate::triggers::nats::NatsTrigger;
listen_to(NatsTrigger, db.clone(), gcp_killpill_rx);
}
#[cfg(all(feature = "kafka", feature = "enterprise", feature = "private"))]
{
let gcp_killpill_rx = killpill_rx.resubscribe();
use crate::triggers::kafka::KafkaTrigger;
listen_to(KafkaTrigger, db.clone(), gcp_killpill_rx);
}
tracing::info!("All available trigger listeners have been started");
}
+4
View File
@@ -23,11 +23,15 @@ pub mod sqs;
pub mod websocket;
mod handler;
mod listener;
pub mod trigger_helpers;
#[allow(unused)]
pub(crate) use handler::TriggerCrud;
pub use handler::{generate_trigger_routers, get_triggers_count_internal, TriggersCount};
pub use listener::start_all_listeners;
#[allow(unused)]
pub(crate) use listener::Listener;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StandardTriggerQuery {
@@ -0,0 +1,374 @@
use std::{collections::HashMap, sync::Arc};
use async_trait::async_trait;
use bytes::Bytes;
use rumqttc::{
v5::{
mqttbytes::v5::PublishProperties, Event as V5Event, EventLoop as V5EventLoop,
Incoming as V5Incoming,
},
Event as V3Event, EventLoop as V3EventLoop, Incoming as V3Incoming,
};
use std::time::Duration;
use tokio::sync::RwLock;
use windmill_common::{
db::UserDB,
error::{to_anyhow, Error, Result},
jobs::JobTriggerKind,
worker::to_raw_value,
DB,
};
use crate::{
resources::try_get_resource_from_db_as,
triggers::{
listener::ListeningTrigger,
mqtt::{
MqttClientBuilder, MqttClientResult, MqttConfig, MqttResource, MqttTrigger,
V3MqttHandler, V5MqttHandler,
},
trigger_helpers::TriggerJobArgs,
Listener,
},
};
#[async_trait]
impl Listener for MqttTrigger {
type Consumer = MqttClientResult;
type Extra = ();
const JOB_TRIGGER_KIND: JobTriggerKind = JobTriggerKind::Mqtt;
async fn get_consumer(
&self,
db: &DB,
listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
_err_message: Arc<RwLock<Option<String>>>,
_killpill_rx: tokio::sync::broadcast::Receiver<()>,
) -> Result<Option<Self::Consumer>> {
let ListeningTrigger::<Self::TriggerConfig> { workspace_id, trigger_config, .. } =
listening_trigger;
let MqttConfig {
mqtt_resource_path,
subscribe_topics,
v3_config,
v5_config,
client_id,
client_version,
..
} = trigger_config;
let authed = listening_trigger
.authed(db, &Self::TRIGGER_KIND.to_string())
.await?;
let mqtt_resource = try_get_resource_from_db_as::<MqttResource>(
&authed,
Some(UserDB::new(db.clone())),
&db,
mqtt_resource_path,
workspace_id,
)
.await?;
let subscribe_topics = subscribe_topics
.iter()
.map(|topic| topic.0.clone())
.collect();
let client_builder = MqttClientBuilder::new(
mqtt_resource,
client_id.as_deref(),
subscribe_topics,
v3_config.as_ref().map(|c| &c.0),
v5_config.as_ref().map(|c| &c.0),
client_version.as_ref(),
);
let client_result = client_builder
.build_client()
.await
.map_err(|e| Error::BadConfig(format!("Failed to build MQTT client: {}", e)))?;
Ok(Some(client_result))
}
async fn consume(
&self,
db: &DB,
consumer: Self::Consumer,
listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
_err_message: Arc<RwLock<Option<String>>>,
_killpill_rx: tokio::sync::broadcast::Receiver<()>,
) {
tracing::info!(
"Starting to listen for MQTT trigger {}",
&listening_trigger.path
);
match consumer {
MqttClientResult::V3((v3_handler, event_loop)) => {
handle_event(&db, self, listening_trigger, v3_handler, event_loop).await
}
MqttClientResult::V5((v5_handler, event_loop)) => {
handle_event(&db, self, listening_trigger, v5_handler, event_loop).await
}
}
}
}
const TIMEOUT_DURATION: u64 = 10;
const CONNECTION_TIMEOUT: Duration = Duration::from_secs(TIMEOUT_DURATION);
fn convert_disconnect_packet_into_string(
disconnect: rumqttc::v5::mqttbytes::v5::Disconnect,
) -> String {
let err_message = disconnect
.properties
.map(|properties| properties.reason_string)
.flatten();
let reason_code = disconnect.reason_code as u8;
format!(
"Disconnected by the broker, reason code: {}, {}",
reason_code,
err_message
.map(|err| format!("message: {}", err))
.unwrap_or("".to_string())
)
}
#[async_trait]
pub trait EventLoop {
type Event;
type Error;
async fn poll(&mut self) -> Result<Self::Event>;
async fn verify_connection(&mut self) -> Result<()>;
}
#[async_trait]
impl EventLoop for V5EventLoop {
type Event = V5Event;
type Error = rumqttc::v5::ConnectionError;
async fn poll(&mut self) -> Result<Self::Event> {
self.poll().await.map_err(|err| to_anyhow(err).into())
}
async fn verify_connection(&mut self) -> Result<()> {
let start = std::time::Instant::now();
while start.elapsed() < CONNECTION_TIMEOUT {
match self.poll().await.map_err(to_anyhow)? {
Self::Event::Incoming(V5Incoming::ConnAck(_)) => return Ok(()),
Self::Event::Incoming(V5Incoming::Disconnect(disconnect)) => {
return Err(Error::BadConfig(convert_disconnect_packet_into_string(
disconnect,
)));
}
_ => continue,
}
}
Err(Error::BadConfig(format!(
"Timeout occurred while trying to connect to mqtt broker after {} seconds",
TIMEOUT_DURATION
)))
}
}
#[async_trait]
impl EventLoop for V3EventLoop {
type Event = V3Event;
type Error = rumqttc::ConnectionError;
async fn poll(&mut self) -> Result<Self::Event> {
self.poll().await.map_err(|err| to_anyhow(err).into())
}
async fn verify_connection(&mut self) -> Result<()> {
let start = std::time::Instant::now();
while start.elapsed() < CONNECTION_TIMEOUT {
match self.poll().await.map_err(to_anyhow)? {
Self::Event::Incoming(rumqttc::Packet::ConnAck(_)) => return Ok(()),
_ => continue,
}
}
Err(Error::BadConfig(format!(
"Timeout occurred while trying to connect to mqtt broker after {} seconds",
TIMEOUT_DURATION
)))
}
}
async fn handle_event<T, E, H>(
db: &DB,
listener: &T,
listening_trigger: &ListeningTrigger<T::TriggerConfig>,
handler: H,
mut event_loop: E,
) -> ()
where
T: Listener,
H: MqttEvent,
E: EventLoop<Event = H::Event>,
E::Error: ToString,
<T as TriggerJobArgs>::Payload: From<Bytes>,
{
loop {
let event = event_loop.poll().await;
match event {
Ok(event) => {
let publish_data = handler.handle_event(event);
if let Ok(Some((payload, publish_data))) = publish_data {
let trigger_info = HashMap::from([
("topic".to_string(), to_raw_value(&publish_data.topic)),
("retain".to_string(), to_raw_value(&publish_data.retain)),
("pkid".to_string(), to_raw_value(&publish_data.pkid)),
("qos".to_string(), to_raw_value(&publish_data.qos)),
(
"v5".to_string(),
to_raw_value(&publish_data.v5.map(|properties| {
serde_json::json!({
"payload_format_indicator": properties.payload_format_indicator,
"topic_alias": properties.topic_alias,
"response_topic": properties.response_topic,
"correlation_data": properties.correlation_data.as_deref(),
"user_properties": properties.user_properties,
"subscription_identifiers": properties.subscription_identifiers,
"content_type": properties.content_type,
})
})),
),
]);
let _ = listener
.handle_event(db, listening_trigger, payload.into(), trigger_info, None)
.await;
}
}
Err(err) => {
let error = err.to_string();
tracing::debug!("Error: {}", &err);
listener
.disable_with_error(db, listening_trigger, error)
.await;
return;
}
}
}
}
#[derive(Clone)]
#[allow(unused)]
pub struct PublishData {
topic: String,
retain: bool,
pkid: u16,
v5: Option<PublishProperties>,
qos: u8,
}
impl PublishData {
pub fn new(
topic: String,
retain: bool,
pkid: u16,
v5: Option<PublishProperties>,
qos: u8,
) -> PublishData {
PublishData { topic, retain, pkid, v5, qos }
}
}
trait MqttEvent {
type IncomingPacket;
type PublishPacket;
type Event;
fn handle_publish_packet(publish_packet: Self::PublishPacket) -> PublishData;
fn handle_event(&self, event: Self::Event) -> Result<Option<(Bytes, PublishData)>>;
}
impl MqttEvent for V5MqttHandler {
type IncomingPacket = V5Incoming;
type PublishPacket = rumqttc::v5::mqttbytes::v5::Publish;
type Event = V5Event;
fn handle_publish_packet(publish_packet: Self::PublishPacket) -> PublishData {
PublishData::new(
String::from_utf8(publish_packet.topic.as_ref().to_vec()).unwrap_or("".to_string()),
publish_packet.retain,
publish_packet.pkid,
publish_packet.properties,
publish_packet.qos as u8,
)
}
fn handle_event(&self, event: Self::Event) -> Result<Option<(Bytes, PublishData)>> {
tracing::debug!("Inside V5 event");
match event {
Self::Event::Incoming(packet) => match packet {
Self::IncomingPacket::Publish(publish_packet) => {
return Ok(Some((
publish_packet.payload.clone(),
Self::handle_publish_packet(publish_packet),
)))
}
Self::IncomingPacket::Disconnect(disconnect) => {
return Err(
anyhow::anyhow!(convert_disconnect_packet_into_string(disconnect)).into(),
);
}
packet => {
tracing::debug!("Received = {:#?}", packet);
}
},
Self::Event::Outgoing(packet) => {
tracing::debug!("Outgoing Received = {:#?}", packet);
}
}
Ok(None)
}
}
impl MqttEvent for V3MqttHandler {
type IncomingPacket = V3Incoming;
type PublishPacket = rumqttc::mqttbytes::v4::Publish;
type Event = V3Event;
fn handle_publish_packet(publish_packet: Self::PublishPacket) -> PublishData {
PublishData::new(
publish_packet.topic,
publish_packet.retain,
publish_packet.pkid,
None,
publish_packet.qos as u8,
)
}
fn handle_event(&self, event: Self::Event) -> Result<Option<(Bytes, PublishData)>> {
tracing::debug!("Inside V3 event");
match event {
Self::Event::Incoming(packet) => match packet {
Self::IncomingPacket::Publish(publish_packet) => {
return Ok(Some((
publish_packet.payload.clone(),
Self::handle_publish_packet(publish_packet),
)))
}
packet => {
tracing::debug!("Received = {:?}", packet);
}
},
Self::Event::Outgoing(packet) => {
tracing::debug!("Outgoing Received = {:?}", packet);
}
}
Ok(None)
}
}
@@ -11,7 +11,6 @@ use rumqttc::{
AsyncClient as V3AsyncClient, EventLoop as V3EventLoop, MqttOptions as V3MqttOptions,
QoS as V3QoS, SubscribeFilter, TlsConfiguration, Transport,
};
use crate::mqtt_triggers::EventLoop;
use serde::{Deserialize, Serialize};
use serde_json::value::RawValue;
use sqlx::{types::Json as SqlxJson, FromRow, Type};
@@ -22,9 +21,10 @@ use windmill_common::{
worker::to_raw_value,
};
use crate::triggers::{ trigger_helpers::TriggerJobArgs};
use crate::triggers::{mqtt::listener::EventLoop, trigger_helpers::TriggerJobArgs};
pub mod handler;
pub mod listener;
#[derive(Clone, Copy)]
pub struct MqttTrigger;
@@ -30,7 +30,7 @@ impl TriggerCrud for NatsTrigger {
const SUPPORTS_TEST_CONNECTION: bool = false;
const ROUTE_PREFIX: &'static str = "/nats_triggers";
const DEPLOYMENT_NAME: &'static str = "";
const IS_CLOUD_HOSTED: bool = false;
const IS_ALLOWED_ON_CLOUD: bool = false;
fn get_deployed_object(path: String) -> DeployedObject {
DeployedObject::NatsTrigger { path }
@@ -0,0 +1,41 @@
#[allow(unused)]
#[cfg(feature = "private")]
pub use super::listener_ee::*;
#[cfg(not(feature = "private"))]
use {
super::NatsTrigger,
crate::triggers::{listener::ListeningTrigger, Listener},
std::sync::Arc,
tokio::sync::RwLock,
windmill_common::{error::Result, jobs::JobTriggerKind, DB},
};
#[cfg(not(feature = "private"))]
#[async_trait::async_trait]
impl Listener for NatsTrigger {
type Consumer = ();
type Extra = ();
const JOB_TRIGGER_KIND: JobTriggerKind = JobTriggerKind::Nats;
async fn get_consumer(
&self,
_db: &DB,
_listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
_err_message: Arc<RwLock<Option<String>>>,
_killpill_rx: tokio::sync::broadcast::Receiver<()>,
) -> Result<Option<Self::Consumer>> {
Ok(None)
}
async fn consume(
&self,
_db: &DB,
_consumer: Self::Consumer,
_listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
_err_message: Arc<RwLock<Option<String>>>,
_killpill_rx: tokio::sync::broadcast::Receiver<()>,
) {
()
}
}
@@ -2,6 +2,10 @@
mod handler_ee;
pub mod handler_oss;
#[cfg(feature = "private")]
mod listener_ee;
pub mod listener_oss;
#[cfg(feature = "private")]
mod mod_ee;
#[cfg(feature = "private")]
@@ -3,10 +3,10 @@ use thiserror::Error;
/**
* This implementation is inspired by Postgres replication functionality
* from https://github.com/supabase/pg_replicate
*
* Original implementation:
*
* Original implementation:
* - https://github.dev/supabase/pg_replicate/blob/main/pg_replicate/src/conversions/bool.rs
*
*
*/
#[derive(Debug, Error)]

Some files were not shown because too many files have changed in this diff Show More