diff --git a/backend/.sqlx/query-04f8d738b1073b8c58db0965e8fdcdb872dce2c1872359c1bebb553a29ba1637.json b/backend/.sqlx/query-04f8d738b1073b8c58db0965e8fdcdb872dce2c1872359c1bebb553a29ba1637.json new file mode 100644 index 0000000000..b593669d35 --- /dev/null +++ b/backend/.sqlx/query-04f8d738b1073b8c58db0965e8fdcdb872dce2c1872359c1bebb553a29ba1637.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE \n workspace_settings\n SET\n error_handler = NULL,\n error_handler_extra_args = NULL,\n error_handler_muted_on_cancel = NULL\n WHERE \n workspace_id = $1\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [] + }, + "hash": "04f8d738b1073b8c58db0965e8fdcdb872dce2c1872359c1bebb553a29ba1637" +} diff --git a/backend/.sqlx/query-0d8415a79d8312db4d46da46fdf348ab17dee7b8358b6719fa399511962bc3e9.json b/backend/.sqlx/query-0d8415a79d8312db4d46da46fdf348ab17dee7b8358b6719fa399511962bc3e9.json new file mode 100644 index 0000000000..2040b0ed88 --- /dev/null +++ b/backend/.sqlx/query-0d8415a79d8312db4d46da46fdf348ab17dee7b8358b6719fa399511962bc3e9.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE \n nats_trigger\n SET\n nats_resource_path = $1,\n subjects = $2,\n stream_name = $3,\n consumer_name = $4,\n use_jetstream = $5,\n script_path = $6,\n path = $7,\n is_flow = $8,\n edited_by = $9,\n email = $10,\n edited_at = now(),\n server_id = NULL,\n error = NULL,\n error_handler_path = $13,\n error_handler_args = $14,\n retry = $15\n WHERE\n workspace_id = $11 AND path = $12\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "VarcharArray", + "Varchar", + "Varchar", + "Bool", + "Varchar", + "Varchar", + "Bool", + "Varchar", + "Varchar", + "Text", + "Text", + "Varchar", + "Jsonb", + "Jsonb" + ] + }, + "nullable": [] + }, + "hash": "0d8415a79d8312db4d46da46fdf348ab17dee7b8358b6719fa399511962bc3e9" +} diff --git a/backend/.sqlx/query-2549197c6750bb1a20041b2d7c2654a788a32a9282ed314c2b3bcafe6550efca.json b/backend/.sqlx/query-2549197c6750bb1a20041b2d7c2654a788a32a9282ed314c2b3bcafe6550efca.json new file mode 100644 index 0000000000..b80ece371b --- /dev/null +++ b/backend/.sqlx/query-2549197c6750bb1a20041b2d7c2654a788a32a9282ed314c2b3bcafe6550efca.json @@ -0,0 +1,17 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE \n workspace_settings\n SET\n error_handler = $1,\n error_handler_extra_args = $2,\n error_handler_muted_on_cancel = $3\n WHERE \n workspace_id = $4\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Json", + "Bool", + "Text" + ] + }, + "nullable": [] + }, + "hash": "2549197c6750bb1a20041b2d7c2654a788a32a9282ed314c2b3bcafe6550efca" +} diff --git a/backend/.sqlx/query-35b211d19e53da4b64b0bd097284de3236ab939e47a1fc2b15ffc9607b552f8d.json b/backend/.sqlx/query-35b211d19e53da4b64b0bd097284de3236ab939e47a1fc2b15ffc9607b552f8d.json new file mode 100644 index 0000000000..aa52f752ae --- /dev/null +++ b/backend/.sqlx/query-35b211d19e53da4b64b0bd097284de3236ab939e47a1fc2b15ffc9607b552f8d.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE \n websocket_trigger\n SET\n url = $1,\n script_path = $2,\n path = $3,\n is_flow = $4,\n filters = $5,\n initial_messages = $6,\n url_runnable_args = $7,\n edited_by = $8,\n email = $9,\n can_return_message = $10,\n edited_at = now(),\n server_id = NULL,\n error = NULL,\n error_handler_path = $13,\n error_handler_args = $14,\n retry = $15\n WHERE\n workspace_id = $11 AND path = $12\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Varchar", + "Bool", + "JsonbArray", + "JsonbArray", + "Jsonb", + "Varchar", + "Varchar", + "Bool", + "Text", + "Text", + "Varchar", + "Jsonb", + "Jsonb" + ] + }, + "nullable": [] + }, + "hash": "35b211d19e53da4b64b0bd097284de3236ab939e47a1fc2b15ffc9607b552f8d" +} diff --git a/backend/.sqlx/query-bf1fba73d3109aae48bda524cfafdf7f5b9b0e56396aed1b83a2c235222e8983.json b/backend/.sqlx/query-3ebf788b86d6aed59b8c61c08f1d9323e6c66107a1b7b37dadbbb7b91339e9d5.json similarity index 70% rename from backend/.sqlx/query-bf1fba73d3109aae48bda524cfafdf7f5b9b0e56396aed1b83a2c235222e8983.json rename to backend/.sqlx/query-3ebf788b86d6aed59b8c61c08f1d9323e6c66107a1b7b37dadbbb7b91339e9d5.json index b939715091..493a2ba2ec 100644 --- a/backend/.sqlx/query-bf1fba73d3109aae48bda524cfafdf7f5b9b0e56396aed1b83a2c235222e8983.json +++ b/backend/.sqlx/query-3ebf788b86d6aed59b8c61c08f1d9323e6c66107a1b7b37dadbbb7b91339e9d5.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT \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 replication_slot_name,\n publication_name,\n postgres_resource_path,\n error_handler_path,\n error_handler_args as \"error_handler_args: _\",\n retry as \"retry: _\" \n FROM postgres_trigger\n WHERE workspace_id = $1", + "query": "\n SELECT \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 replication_slot_name,\n publication_name,\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 workspace_id = $1\n ", "describe": { "columns": [ { @@ -120,5 +120,5 @@ true ] }, - "hash": "bf1fba73d3109aae48bda524cfafdf7f5b9b0e56396aed1b83a2c235222e8983" + "hash": "3ebf788b86d6aed59b8c61c08f1d9323e6c66107a1b7b37dadbbb7b91339e9d5" } diff --git a/backend/.sqlx/query-3f91433fc076f47dd3516b40c9337981c329ddc31132882f037eaf2709aa8805.json b/backend/.sqlx/query-3f91433fc076f47dd3516b40c9337981c329ddc31132882f037eaf2709aa8805.json deleted file mode 100644 index bbd1cbc510..0000000000 --- a/backend/.sqlx/query-3f91433fc076f47dd3516b40c9337981c329ddc31132882f037eaf2709aa8805.json +++ /dev/null @@ -1,26 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE kafka_trigger SET kafka_resource_path = $1, group_id = $2, topics = $3, script_path = $4, path = $5, is_flow = $6, edited_by = $7, email = $8, edited_at = now(), server_id = NULL, error = NULL, error_handler_path = $11, error_handler_args = $12, retry = $13\n WHERE workspace_id = $9 AND path = $10", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "VarcharArray", - "Varchar", - "Varchar", - "Bool", - "Varchar", - "Varchar", - "Text", - "Text", - "Varchar", - "Jsonb", - "Jsonb" - ] - }, - "nullable": [] - }, - "hash": "3f91433fc076f47dd3516b40c9337981c329ddc31132882f037eaf2709aa8805" -} diff --git a/backend/.sqlx/query-4479beaefc36afba093e5113bb330a87979623d69b94f274863318ed351a1dac.json b/backend/.sqlx/query-4479beaefc36afba093e5113bb330a87979623d69b94f274863318ed351a1dac.json deleted file mode 100644 index 02e3467f5c..0000000000 --- a/backend/.sqlx/query-4479beaefc36afba093e5113bb330a87979623d69b94f274863318ed351a1dac.json +++ /dev/null @@ -1,28 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE websocket_trigger SET url = $1, script_path = $2, path = $3, is_flow = $4, filters = $5, initial_messages = $6, url_runnable_args = $7, edited_by = $8, email = $9, can_return_message = $10, edited_at = now(), server_id = NULL, error = NULL, error_handler_path = $13, error_handler_args = $14, retry = $15\n WHERE workspace_id = $11 AND path = $12", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Varchar", - "Bool", - "JsonbArray", - "JsonbArray", - "Jsonb", - "Varchar", - "Varchar", - "Bool", - "Text", - "Text", - "Varchar", - "Jsonb", - "Jsonb" - ] - }, - "nullable": [] - }, - "hash": "4479beaefc36afba093e5113bb330a87979623d69b94f274863318ed351a1dac" -} diff --git a/backend/.sqlx/query-6489342f52a9e1e90466bad48e0a6079fcabd696b02d57e3dbe652d28b84f008.json b/backend/.sqlx/query-494d49e92dce7a1cd7074156704ccbbff020a2042409fc318e045852323cef5e.json similarity index 92% rename from backend/.sqlx/query-6489342f52a9e1e90466bad48e0a6079fcabd696b02d57e3dbe652d28b84f008.json rename to backend/.sqlx/query-494d49e92dce7a1cd7074156704ccbbff020a2042409fc318e045852323cef5e.json index fa87b125a2..25d6e0e6d3 100644 --- a/backend/.sqlx/query-6489342f52a9e1e90466bad48e0a6079fcabd696b02d57e3dbe652d28b84f008.json +++ b/backend/.sqlx/query-494d49e92dce7a1cd7074156704ccbbff020a2042409fc318e045852323cef5e.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT \n workspace_id,\n path,\n kafka_resource_path,\n group_id,\n topics,\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 kafka_trigger WHERE workspace_id = $1 AND path = $2", + "query": "SELECT \n workspace_id,\n path,\n kafka_resource_path,\n group_id,\n topics,\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 kafka_trigger \n WHERE \n workspace_id = $1 AND \n path = $2\n ", "describe": { "columns": [ { @@ -121,5 +121,5 @@ true ] }, - "hash": "6489342f52a9e1e90466bad48e0a6079fcabd696b02d57e3dbe652d28b84f008" + "hash": "494d49e92dce7a1cd7074156704ccbbff020a2042409fc318e045852323cef5e" } diff --git a/backend/.sqlx/query-4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0.json b/backend/.sqlx/query-4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0.json index 65fa6cc9eb..91512257cd 100644 --- a/backend/.sqlx/query-4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0.json +++ b/backend/.sqlx/query-4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0.json @@ -223,7 +223,8 @@ "ui", "postgres", "sqs", - "gcp" + "gcp", + "mqtt" ] } } diff --git a/backend/.sqlx/query-4bfa42153df81184e02ffc7ab6c8c591605211c070e52b2919cf9d7dede0ea16.json b/backend/.sqlx/query-4bfa42153df81184e02ffc7ab6c8c591605211c070e52b2919cf9d7dede0ea16.json new file mode 100644 index 0000000000..94ea76acc2 --- /dev/null +++ b/backend/.sqlx/query-4bfa42153df81184e02ffc7ab6c8c591605211c070e52b2919cf9d7dede0ea16.json @@ -0,0 +1,26 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE kafka_trigger\n SET\n kafka_resource_path = $1,\n group_id = $2,\n topics = $3,\n script_path = $4,\n path = $5,\n is_flow = $6,\n edited_by = $7,\n email = $8,\n edited_at = now(),\n server_id = NULL,\n error = NULL,\n error_handler_path = $11,\n error_handler_args = $12,\n retry = $13\n WHERE\n workspace_id = $9 AND path = $10\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "VarcharArray", + "Varchar", + "Varchar", + "Bool", + "Varchar", + "Varchar", + "Text", + "Text", + "Varchar", + "Jsonb", + "Jsonb" + ] + }, + "nullable": [] + }, + "hash": "4bfa42153df81184e02ffc7ab6c8c591605211c070e52b2919cf9d7dede0ea16" +} diff --git a/backend/.sqlx/query-8f511fd837a99fa37a95781e50f8a83eaf93651f82c36c9795637d3ad761d003.json b/backend/.sqlx/query-4c6fbed80ba44defe3d125c258bc3449b02925c8caf85d9203529471479fdde8.json similarity index 94% rename from backend/.sqlx/query-8f511fd837a99fa37a95781e50f8a83eaf93651f82c36c9795637d3ad761d003.json rename to backend/.sqlx/query-4c6fbed80ba44defe3d125c258bc3449b02925c8caf85d9203529471479fdde8.json index 8e37b499be..bb5c6096f8 100644 --- a/backend/.sqlx/query-8f511fd837a99fa37a95781e50f8a83eaf93651f82c36c9795637d3ad761d003.json +++ b/backend/.sqlx/query-4c6fbed80ba44defe3d125c258bc3449b02925c8caf85d9203529471479fdde8.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT \n workspace_id,\n path,\n kafka_resource_path,\n group_id,\n topics,\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 kafka_trigger\n WHERE workspace_id = $1", + "query": "SELECT \n workspace_id,\n path,\n kafka_resource_path,\n group_id,\n topics,\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 kafka_trigger\n WHERE workspace_id = $1", "describe": { "columns": [ { @@ -120,5 +120,5 @@ true ] }, - "hash": "8f511fd837a99fa37a95781e50f8a83eaf93651f82c36c9795637d3ad761d003" + "hash": "4c6fbed80ba44defe3d125c258bc3449b02925c8caf85d9203529471479fdde8" } diff --git a/backend/.sqlx/query-570a4684bbdb390e4652301aa8d2d09b6e09b258f748d48cf2d188465621ee04.json b/backend/.sqlx/query-570a4684bbdb390e4652301aa8d2d09b6e09b258f748d48cf2d188465621ee04.json deleted file mode 100644 index f9d6c86004..0000000000 --- a/backend/.sqlx/query-570a4684bbdb390e4652301aa8d2d09b6e09b258f748d48cf2d188465621ee04.json +++ /dev/null @@ -1,28 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE nats_trigger SET nats_resource_path = $1, subjects = $2, stream_name = $3, consumer_name = $4, use_jetstream = $5, script_path = $6, path = $7, is_flow = $8, edited_by = $9, email = $10, edited_at = now(), server_id = NULL, error = NULL, error_handler_path = $13, error_handler_args = $14, retry = $15\n WHERE workspace_id = $11 AND path = $12", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "VarcharArray", - "Varchar", - "Varchar", - "Bool", - "Varchar", - "Varchar", - "Bool", - "Varchar", - "Varchar", - "Text", - "Text", - "Varchar", - "Jsonb", - "Jsonb" - ] - }, - "nullable": [] - }, - "hash": "570a4684bbdb390e4652301aa8d2d09b6e09b258f748d48cf2d188465621ee04" -} diff --git a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json index 713ccb9dd3..36ddb8ab9f 100644 --- a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json +++ b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json @@ -15,7 +15,7 @@ ] }, "nullable": [ - null + true ] }, "hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55" diff --git a/backend/.sqlx/query-5af0299841a0313d0cc270e98b04b5b5968184ae53dc076b1a9bc76af35923da.json b/backend/.sqlx/query-711e665344873454428d84e35f300edb7e0ebad5de40ba5645c1ce3d4a6b84be.json similarity index 71% rename from backend/.sqlx/query-5af0299841a0313d0cc270e98b04b5b5968184ae53dc076b1a9bc76af35923da.json rename to backend/.sqlx/query-711e665344873454428d84e35f300edb7e0ebad5de40ba5645c1ce3d4a6b84be.json index ebd6dafb25..3abff2fa36 100644 --- a/backend/.sqlx/query-5af0299841a0313d0cc270e98b04b5b5968184ae53dc076b1a9bc76af35923da.json +++ b/backend/.sqlx/query-711e665344873454428d84e35f300edb7e0ebad5de40ba5645c1ce3d4a6b84be.json @@ -1,6 +1,6 @@ { "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 workspace_id = $1", + "query": "\n 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 \n nats_trigger\n WHERE \n workspace_id = $1\n ", "describe": { "columns": [ { @@ -132,5 +132,5 @@ true ] }, - "hash": "5af0299841a0313d0cc270e98b04b5b5968184ae53dc076b1a9bc76af35923da" + "hash": "711e665344873454428d84e35f300edb7e0ebad5de40ba5645c1ce3d4a6b84be" } diff --git a/backend/.sqlx/query-6e60068b582af46afbb93010355cb5e7cc038042a6fa7089f3f73cd137a63956.json b/backend/.sqlx/query-8960b73f0d3cbfa0729d24ecbe3f635592feee6a5724fdff662c6ac1f3c6ddc8.json similarity index 77% rename from backend/.sqlx/query-6e60068b582af46afbb93010355cb5e7cc038042a6fa7089f3f73cd137a63956.json rename to backend/.sqlx/query-8960b73f0d3cbfa0729d24ecbe3f635592feee6a5724fdff662c6ac1f3c6ddc8.json index 39510e0f99..1d33f1b2a0 100644 --- a/backend/.sqlx/query-6e60068b582af46afbb93010355cb5e7cc038042a6fa7089f3f73cd137a63956.json +++ b/backend/.sqlx/query-8960b73f0d3cbfa0729d24ecbe3f635592feee6a5724fdff662c6ac1f3c6ddc8.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT workspace_id, slack_team_id, teams_team_id, teams_team_name, slack_name, slack_command_script, teams_command_script, slack_email, auto_invite_domain, auto_invite_operator, auto_add, customer_id, plan, webhook, deploy_to, ai_config, error_handler, error_handler_extra_args, error_handler_muted_on_cancel, large_file_storage, ducklake, git_sync, deploy_ui, default_app, default_scripts, mute_critical_alerts, color, operator_settings, git_app_installations FROM workspace_settings WHERE workspace_id = $1", + "query": "\n SELECT \n workspace_id,\n slack_team_id,\n teams_team_id,\n teams_team_name,\n slack_name,\n slack_command_script,\n teams_command_script,\n slack_email,\n auto_invite_domain,\n auto_invite_operator,\n auto_add,\n customer_id,\n plan,\n webhook,\n deploy_to,\n ai_config,\n error_handler,\n error_handler_extra_args,\n error_handler_muted_on_cancel,\n large_file_storage,\n ducklake,\n git_sync,\n deploy_ui,\n default_app,\n default_scripts,\n mute_critical_alerts,\n color,\n operator_settings,\n git_app_installations\n FROM \n workspace_settings\n WHERE \n workspace_id = $1\n ", "describe": { "columns": [ { @@ -186,5 +186,5 @@ false ] }, - "hash": "6e60068b582af46afbb93010355cb5e7cc038042a6fa7089f3f73cd137a63956" + "hash": "8960b73f0d3cbfa0729d24ecbe3f635592feee6a5724fdff662c6ac1f3c6ddc8" } diff --git a/backend/.sqlx/query-09f43081a8b253d059d9d4302daa139cee6f055cdbafc213b2f992667f8ed578.json b/backend/.sqlx/query-89be9b595a892c82f2a9f3cfa4da213e360ebe8827d5b4680d3c227fb8ddec82.json similarity index 95% rename from backend/.sqlx/query-09f43081a8b253d059d9d4302daa139cee6f055cdbafc213b2f992667f8ed578.json rename to backend/.sqlx/query-89be9b595a892c82f2a9f3cfa4da213e360ebe8827d5b4680d3c227fb8ddec82.json index 9fb3f5ab6c..9b33a91ae2 100644 --- a/backend/.sqlx/query-09f43081a8b253d059d9d4302daa139cee6f055cdbafc213b2f992667f8ed578.json +++ b/backend/.sqlx/query-89be9b595a892c82f2a9f3cfa4da213e360ebe8827d5b4680d3c227fb8ddec82.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT \n workspace_id, \n workspaced_route,\n path, \n route_path, \n route_path_key, \n authentication_resource_path,\n script_path, \n is_flow, \n summary,\n description,\n edited_by, \n edited_at, \n email, \n extra_perms, \n is_async, \n authentication_method AS \"authentication_method: _\", \n http_method AS \"http_method: _\", \n static_asset_config AS \"static_asset_config: _\", \n is_static_website,\n wrap_body,\n raw_string,\n error_handler_path,\n error_handler_args as \"error_handler_args: _\",\n retry as \"retry: _\"\n FROM http_trigger\n WHERE workspace_id = $1\n ", + "query": "\n SELECT \n workspace_id, \n workspaced_route,\n path, \n route_path, \n route_path_key, \n authentication_resource_path,\n script_path, \n is_flow, \n summary,\n description,\n edited_by, \n edited_at, \n email, \n extra_perms, \n is_async, \n authentication_method AS \"authentication_method: _\", \n http_method AS \"http_method: _\", \n static_asset_config AS \"static_asset_config: _\", \n is_static_website,\n wrap_body,\n raw_string,\n error_handler_path,\n error_handler_args as \"error_handler_args: _\",\n retry as \"retry: _\"\n FROM \n http_trigger\n WHERE \n workspace_id = $1\n ", "describe": { "columns": [ { @@ -183,5 +183,5 @@ true ] }, - "hash": "09f43081a8b253d059d9d4302daa139cee6f055cdbafc213b2f992667f8ed578" + "hash": "89be9b595a892c82f2a9f3cfa4da213e360ebe8827d5b4680d3c227fb8ddec82" } diff --git a/backend/.sqlx/query-b7c3a66c3831eb5d145ff00807badae57bef81be051f150df754fd1444d7356d.json b/backend/.sqlx/query-acfe583fe17604ba72ba4800b62a72de0a9de0d58ef8c28dd709adf3be021597.json similarity index 83% rename from backend/.sqlx/query-b7c3a66c3831eb5d145ff00807badae57bef81be051f150df754fd1444d7356d.json rename to backend/.sqlx/query-acfe583fe17604ba72ba4800b62a72de0a9de0d58ef8c28dd709adf3be021597.json index e6ecb2b7d6..7d7eef1cab 100644 --- a/backend/.sqlx/query-b7c3a66c3831eb5d145ff00807badae57bef81be051f150df754fd1444d7356d.json +++ b/backend/.sqlx/query-acfe583fe17604ba72ba4800b62a72de0a9de0d58ef8c28dd709adf3be021597.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "WITH inserted_job AS (\n INSERT INTO v2_job (id, workspace_id, raw_code, raw_lock, raw_flow, tag, parent_job,\n created_by, permissioned_as, runnable_id, runnable_path, args, kind, trigger,\n script_lang, same_worker, pre_run_error, permissioned_as_email, visible_to_owner,\n flow_innermost_root_job, root_job, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id,\n cache_ttl, priority, trigger_kind, script_entrypoint_override, preprocessed)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,\n $19, $20, $38, $21, $22, $23, $24, $25, $26,\n CASE WHEN $14::VARCHAR IS NOT NULL THEN 'schedule'::job_trigger_kind END,\n ($12::JSONB)->>'_ENTRYPOINT_OVERRIDE', $27)\n ),\n inserted_runtime AS (\n INSERT INTO v2_job_runtime (id, ping) VALUES ($1, null)\n ),\n inserted_job_perms AS (\n INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id) \n values ($1, $32, $33, $34, $35, $36, $37, $2) \n ON CONFLICT (job_id) DO UPDATE SET email = $32, username = $33, is_admin = $34, is_operator = $35, folders = $36, groups = $37, workspace_id = $2\n )\n INSERT INTO v2_job_queue\n (workspace_id, id, running, scheduled_for, started_at, tag, priority)\n VALUES ($2, $1, $28, COALESCE($29, now()), CASE WHEN $27 THEN now() END, $30, $31)", + "query": "WITH inserted_job AS (\n INSERT INTO v2_job (id, workspace_id, raw_code, raw_lock, raw_flow, tag, parent_job,\n created_by, permissioned_as, runnable_id, runnable_path, args, kind, trigger,\n script_lang, same_worker, pre_run_error, permissioned_as_email, visible_to_owner,\n flow_innermost_root_job, root_job, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id,\n cache_ttl, priority, trigger_kind, script_entrypoint_override, preprocessed)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,\n $19, $20, $38, $21, $22, $23, $24, $25, $26, $39::job_trigger_kind,\n ($12::JSONB)->>'_ENTRYPOINT_OVERRIDE', $27)\n ),\n inserted_runtime AS (\n INSERT INTO v2_job_runtime (id, ping) VALUES ($1, null)\n ),\n inserted_job_perms AS (\n INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id) \n values ($1, $32, $33, $34, $35, $36, $37, $2) \n ON CONFLICT (job_id) DO UPDATE SET email = $32, username = $33, is_admin = $34, is_operator = $35, folders = $36, groups = $37, workspace_id = $2\n )\n INSERT INTO v2_job_queue\n (workspace_id, id, running, scheduled_for, started_at, tag, priority)\n VALUES ($2, $1, $28, COALESCE($29, now()), CASE WHEN $27 THEN now() END, $30, $31)", "describe": { "columns": [], "parameters": { @@ -98,10 +98,32 @@ "Bool", "JsonbArray", "TextArray", - "Uuid" + "Uuid", + { + "Custom": { + "name": "job_trigger_kind", + "kind": { + "Enum": [ + "webhook", + "http", + "websocket", + "kafka", + "email", + "nats", + "schedule", + "app", + "ui", + "postgres", + "sqs", + "gcp", + "mqtt" + ] + } + } + } ] }, "nullable": [] }, - "hash": "b7c3a66c3831eb5d145ff00807badae57bef81be051f150df754fd1444d7356d" + "hash": "acfe583fe17604ba72ba4800b62a72de0a9de0d58ef8c28dd709adf3be021597" } diff --git a/backend/.sqlx/query-d21e5be1ac26db926b7196316dae0c9fa82a865b95bb55de5370e4ee34889ba3.json b/backend/.sqlx/query-d21e5be1ac26db926b7196316dae0c9fa82a865b95bb55de5370e4ee34889ba3.json deleted file mode 100644 index 00f38b2f59..0000000000 --- a/backend/.sqlx/query-d21e5be1ac26db926b7196316dae0c9fa82a865b95bb55de5370e4ee34889ba3.json +++ /dev/null @@ -1,17 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE workspace_settings SET error_handler = $1, error_handler_extra_args = $2, error_handler_muted_on_cancel = $3 WHERE workspace_id = $4", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Json", - "Bool", - "Text" - ] - }, - "nullable": [] - }, - "hash": "d21e5be1ac26db926b7196316dae0c9fa82a865b95bb55de5370e4ee34889ba3" -} diff --git a/backend/.sqlx/query-d9308c0154e029f7568b1d021368ca07f4867d192aa58b3888ae532d1c040828.json b/backend/.sqlx/query-d9308c0154e029f7568b1d021368ca07f4867d192aa58b3888ae532d1c040828.json deleted file mode 100644 index 879c18a45a..0000000000 --- a/backend/.sqlx/query-d9308c0154e029f7568b1d021368ca07f4867d192aa58b3888ae532d1c040828.json +++ /dev/null @@ -1,14 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE workspace_settings SET error_handler = NULL, error_handler_extra_args = NULL WHERE workspace_id = $1", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [] - }, - "hash": "d9308c0154e029f7568b1d021368ca07f4867d192aa58b3888ae532d1c040828" -} diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 212fe1967e..f23987ab8b 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -77,7 +77,7 @@ mcp = ["windmill-api/mcp"] mqtt_trigger = ["windmill-api/mqtt_trigger"] sqs_trigger = ["windmill-api/sqs_trigger", "windmill-common/aws_auth", "windmill-api/openidconnect"] gcp_trigger = ["windmill-api/gcp_trigger"] -smtp = ["windmill-api/smtp", "windmill-common/smtp"] +smtp = ["windmill-api/smtp", "windmill-common/smtp", "windmill-queue/smtp"] license = ["windmill-api/license"] oauth2 = ["windmill-api/oauth2"] zip = ["windmill-api/zip"] diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 708beb4f29..eaf1fc90ae 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -dbc87a4966118e5486226d1a1863e4d97ba82ecc \ No newline at end of file +db17bc0b22a51192aeadd8a0eab44108b0725a5b \ No newline at end of file diff --git a/backend/migrations/20250723154518_add_mqtt_trigger_to_job_kind_trigger_enum_type.down.sql b/backend/migrations/20250723154518_add_mqtt_trigger_to_job_kind_trigger_enum_type.down.sql new file mode 100644 index 0000000000..d2f607c5b8 --- /dev/null +++ b/backend/migrations/20250723154518_add_mqtt_trigger_to_job_kind_trigger_enum_type.down.sql @@ -0,0 +1 @@ +-- Add down migration script here diff --git a/backend/migrations/20250723154518_add_mqtt_trigger_to_job_kind_trigger_enum_type.up.sql b/backend/migrations/20250723154518_add_mqtt_trigger_to_job_kind_trigger_enum_type.up.sql new file mode 100644 index 0000000000..d662e626bf --- /dev/null +++ b/backend/migrations/20250723154518_add_mqtt_trigger_to_job_kind_trigger_enum_type.up.sql @@ -0,0 +1,2 @@ +-- Add up migration script here +ALTER TYPE job_trigger_kind ADD VALUE IF NOT EXISTS 'mqtt'; \ No newline at end of file diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 09d104040f..831af7111a 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -2440,7 +2440,6 @@ paths: $ref: "#/components/schemas/ScriptArgs" error_handler_muted_on_cancel: type: boolean - responses: "200": description: status @@ -2449,6 +2448,7 @@ paths: schema: type: string + /w/{workspace}/workspaces/edit_large_file_storage_config: post: summary: edit large file storage settings @@ -15202,6 +15202,14 @@ components: - success - duration_ms + ErrorHandler: + type: string + enum: + - custom + - slack + - teams + - email + NewSchedule: type: object properties: diff --git a/backend/windmill-api/src/http_triggers.rs b/backend/windmill-api/src/http_triggers.rs index b6d5fff7f0..b8a52a58ba 100644 --- a/backend/windmill-api/src/http_triggers.rs +++ b/backend/windmill-api/src/http_triggers.rs @@ -12,7 +12,6 @@ use crate::{ users::fetch_api_authed, utils::{check_scopes, non_empty_str, ExpiringCacheEntry}, }; - use anyhow::anyhow; use axum::response::Response; use axum::{ @@ -197,38 +196,37 @@ async fn list_triggers( ) -> error::JsonResult> { let mut tx = user_db.begin(&authed).await?; let (per_page, offset) = paginate(Pagination { per_page: lst.per_page, page: lst.page }); - let mut sqlb = SqlBuilder::select_from("http_trigger") - .fields(&[ - "workspace_id", - "path", - "route_path", - "route_path_key", - "workspaced_route", - "wrap_body", - "raw_string", - "script_path", - "summary", - "description", - "is_flow", - "http_method", - "edited_by", - "email", - "edited_at", - "extra_perms", - "is_async", - "authentication_method", - "static_asset_config", - "is_static_website", - "authentication_resource_path", - "error_handler_path", - "error_handler_args", - "retry", - ]) - .order_by("edited_at", true) - .and_where("workspace_id = ?".bind(&w_id)) - .offset(offset) - .limit(per_page) - .clone(); + let mut sqlb = SqlBuilder::select_from("http_trigger"); + sqlb.fields(&[ + "workspace_id", + "path", + "route_path", + "route_path_key", + "workspaced_route", + "wrap_body", + "raw_string", + "script_path", + "summary", + "description", + "is_flow", + "http_method", + "edited_by", + "email", + "edited_at", + "extra_perms", + "is_async", + "authentication_method", + "static_asset_config", + "is_static_website", + "authentication_resource_path", + "error_handler_path", + "error_handler_args", + "retry", + ]) + .order_by("edited_at", true) + .and_where("workspace_id = ?".bind(&w_id)) + .offset(offset) + .limit(per_page); if let Some(path) = lst.path { sqlb.and_where_eq("script_path", "?".bind(&path)); } @@ -391,7 +389,7 @@ async fn create_trigger_inner( new_http_trigger.is_static_website, new_http_trigger.error_handler_path, new_http_trigger.error_handler_args as _, - new_http_trigger.retry as _, + new_http_trigger.retry as _ ) .execute(&mut *tx) .await?; diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 9a0f918978..4bb89e0260 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -16,6 +16,7 @@ use http::{HeaderMap, HeaderName}; use itertools::Itertools; use quick_cache::sync::Cache; use serde_json::value::RawValue; +use serde_json::Value; use sqlx::Pool; use std::collections::HashMap; use std::hash::{DefaultHasher, Hash, Hasher}; @@ -24,6 +25,8 @@ use std::str::FromStr; use std::time::Instant; use tokio::io::AsyncReadExt; use tower::ServiceBuilder; +#[cfg(feature = "smtp")] +use windmill_common::auth::is_super_admin_email; use windmill_common::auth::TOKEN_PREFIX_LEN; use windmill_common::error::JsonResult; use windmill_common::flow_status::{JobResult, RestartedFrom}; @@ -33,6 +36,8 @@ use windmill_common::jobs::{ }; use windmill_common::utils::WarnAfterExt; use windmill_common::worker::{Connection, CLOUD_HOSTED, TMP_DIR}; +#[cfg(feature = "smtp")] +use windmill_common::{email_oss::send_email_html, server::load_smtp_config}; use windmill_common::scripts::PREVIEW_IS_CODEBASE_HASH; use windmill_common::variables::get_workspace_key; @@ -66,7 +71,9 @@ use tower_http::cors::{Any, CorsLayer}; use urlencoding::encode; use windmill_audit::audit_oss::{audit_log, AuditAuthor}; use windmill_audit::ActionKind; + use windmill_common::worker::to_raw_value; + use windmill_common::{ cache, db::UserDB, @@ -85,7 +92,7 @@ use windmill_common::{ use windmill_common::{ get_latest_deployed_hash_for_path, get_latest_flow_version_info_for_path, - get_script_info_for_hash, FlowVersionInfo, ScriptHashInfo, BASE_URL, + get_script_info_for_hash, utils::empty_as_none, FlowVersionInfo, ScriptHashInfo, BASE_URL, }; use windmill_queue::{ cancel_job, get_result_and_success_by_id_from_flow, job_is_complete, push, PushArgs, @@ -234,6 +241,10 @@ pub fn workspaced_service() -> Router { ) .route("/run/dependencies", post(run_dependencies_job)) .route("/run/flow_dependencies", post(run_flow_dependencies_job)) + .route( + "/send_email_with_instance_smtp", + post(send_email_with_instance_smtp), + ) } pub fn workspace_unauthed_service() -> Router { @@ -1037,6 +1048,240 @@ impl<'a> GetQuery<'a> { } } +#[cfg(all(feature = "smtp", feature = "enterprise"))] +async fn send_workspace_trigger_failure_email_notification( + db: &Pool, + w_id: &str, + job_id: &Uuid, + trigger_path: Option<&str>, + runnable_path: Option<&str>, + email_recipients: &Vec, + error: &Value, +) -> Result { + let smtp_config = match load_smtp_config(db).await? { + Some(config) => config, + None => { + tracing::info!( + "SMTP not configured, skipping workspace trigger failure email notification" + ); + return Err(anyhow::anyhow!( + "SMTP not configured, skipping workspace trigger failure email notification" + ) + .into()); + } + }; + + let runnable_path = runnable_path.as_deref().unwrap_or("Unknown"); + + let (trigger_kind, trigger_path) = if let Some(trigger_path) = trigger_path.as_deref() { + match trigger_path.split_once('/') { + Some((trigger_kind, trigger_path)) => { + tracing::debug!( + "Workspace trigger job {} is a {:?} trigger", + &job_id, + trigger_kind + ); + (Some(trigger_kind), Some(trigger_path)) + } + _ => (None, None), + } + } else { + (None, None) + }; + + let base_url = BASE_URL.read().await; + let job_url = format!("{}/run/{}?workspace={}", base_url, &job_id, w_id); + let trigger_kind_str = trigger_kind.unwrap_or("Unknown").to_string().to_uppercase(); + + let subject = format!( + "Windmill Job Failed: {} in workspace {}", + runnable_path, w_id + ); + + let error_details = serde_json::to_string_pretty(&error) + .unwrap_or_else(|_| format!("Unable to serialize error: {:?}", error)); + + let trigger_info = if trigger_kind.is_some() && trigger_path.is_some() { + format!( + r#" +
+ Trigger path: {} +
+ +
+ Trigger Type: {} +
"#, + trigger_path.unwrap(), + trigger_kind_str + ) + } else { + String::new() + }; + + let email_title = if trigger_kind.is_some() { + format!("Windmill Trigger Job {} Failed", &job_id) + } else { + format!("Windmill Job {} Failed", &job_id) + }; + + let content = format!( + r#" + + + + + + + +

{}

+ +
+ Workspace: {} +
+{} +
+ Script/Flow Path: {} +
+ +
+ Job ID: {} +
+ +
+ Error Details: + 
{}
+
+ + View Job Details + + +"#, + email_title, + w_id, + trigger_info, + runnable_path, + &job_id, + error_details, + job_url + ); + + if let Err(e) = send_email_html( + &subject, + &content, + email_recipients.to_owned(), + smtp_config, + None, + ) + .await + { + let err_msg = format!( + "Failed to send workspace email notification for trigger failure (trigger kind: {}, job ID: {}): {}", + trigger_kind_str, + &job_id, + e + ); + return Err(Error::internal_err(err_msg)); + } + let success_msg = format!( + "Job ID '{}' failed. An email with error details has been sent to {:?}.", + &job_id, email_recipients + ); + tracing::info!("{}", &success_msg); + Ok(success_msg) +} + +#[derive(Debug, Serialize, Deserialize)] +struct SendEmail { + job_id: Uuid, + #[serde(default, deserialize_with = "empty_as_none")] + trigger_path: Option, + #[serde(default, deserialize_with = "empty_as_none")] + runnable_path: Option, + #[serde(default, deserialize_with = "empty_as_none")] + email_recipients: Option>, + error: Value, +} + +#[cfg(all(feature = "enterprise", feature = "smtp"))] +async fn send_email_with_instance_smtp( + authed: ApiAuthed, + Extension(db): Extension, + Path(w_id): Path, + Json(send_email): Json, +) -> error::Result> { + use windmill_common::jobs::EMAIL_ERROR_HANDLER_USER_EMAIL; + + if *CLOUD_HOSTED { + tracing::warn!( + "Workspace trigger failure email notification is not available for cloud hosted Windmill", + ); + return Err(anyhow::anyhow!("Feature not supported in cloud hosted windmill").into()); + } + + if send_email.email_recipients.is_none() { + use windmill_common::utils::report_critical_error; + + tracing::error!("No recipient to send the error"); + report_critical_error( + "No recipient to send the error".to_string(), + db.clone(), + Some(&w_id), + None, + ) + .await; + return Err(anyhow::anyhow!("No recipient to send the error").into()); + } + + if authed.email == EMAIL_ERROR_HANDLER_USER_EMAIL + || is_super_admin_email(&db, &authed.email).await? + { + let resp = send_workspace_trigger_failure_email_notification( + &db, + &w_id, + &send_email.job_id, + send_email.trigger_path.as_deref(), + send_email.runnable_path.as_deref(), + &send_email.email_recipients.unwrap(), + &send_email.error, + ) + .await?; + + return Ok(Json(resp)); + } + + return Err(Error::NotAuthorized( + "Only super admin or whitelisted token can access email workspace error handler feature" + .to_string(), + )); +} + +#[cfg(not(all(feature = "enterprise", feature = "smtp")))] +async fn send_email_with_instance_smtp( + _authed: ApiAuthed, + Extension(_db): Extension, + Path(_w_id): Path, + Json(_send_email): Json, +) -> error::Result> { + tracing::warn!("SMTP is not enabled, skipping workspace trigger failure email notification",); + + return Err(anyhow::anyhow!("SMTP is not enabled").into()); +} + #[cfg(all(feature = "enterprise", feature = "parquet"))] async fn get_logs_from_store( log_offset: i32, diff --git a/backend/windmill-api/src/mqtt_triggers.rs b/backend/windmill-api/src/mqtt_triggers.rs index ae62f662e7..5659c9b944 100644 --- a/backend/windmill-api/src/mqtt_triggers.rs +++ b/backend/windmill-api/src/mqtt_triggers.rs @@ -644,35 +644,34 @@ pub async fn list_mqtt_triggers( ) -> error::JsonResult> { let mut tx = user_db.begin(&authed).await?; let (per_page, offset) = paginate(Pagination { per_page: lst.per_page, page: lst.page }); - let mut sqlb = SqlBuilder::select_from("mqtt_trigger") - .fields(&[ - "mqtt_resource_path", - "subscribe_topics", - "v3_config", - "v5_config", - "client_version", - "client_id", - "workspace_id", - "path", - "script_path", - "is_flow", - "edited_by", - "email", - "edited_at", - "server_id", - "last_server_ping", - "extra_perms", - "error", - "enabled", - "error_handler_path", - "error_handler_args", - "retry", - ]) - .order_by("edited_at", true) - .and_where("workspace_id = ?".bind(&w_id)) - .offset(offset) - .limit(per_page) - .clone(); + let mut sqlb = SqlBuilder::select_from("mqtt_trigger"); + sqlb.fields(&[ + "mqtt_resource_path", + "subscribe_topics", + "v3_config", + "v5_config", + "client_version", + "client_id", + "workspace_id", + "path", + "script_path", + "is_flow", + "edited_by", + "email", + "edited_at", + "server_id", + "last_server_ping", + "extra_perms", + "error", + "enabled", + "error_handler_path", + "error_handler_args", + "retry", + ]) + .order_by("edited_at", true) + .and_where("workspace_id = ?".bind(&w_id)) + .offset(offset) + .limit(per_page); if let Some(path) = lst.path { sqlb.and_where_eq("script_path", "?".bind(&path)); } @@ -760,7 +759,9 @@ pub async fn update_mqtt_trigger( Json(mqtt_trigger): Json, ) -> error::Result { let workspace_path = path.to_path(); - check_scopes(&authed, || format!("mqtt_triggers:write:{}", workspace_path))?; + check_scopes(&authed, || { + format!("mqtt_triggers:write:{}", workspace_path) + })?; let EditMqttTrigger { mqtt_resource_path, diff --git a/backend/windmill-api/src/postgres_triggers/handler.rs b/backend/windmill-api/src/postgres_triggers/handler.rs index 44318200f4..71d53dcf8b 100644 --- a/backend/windmill-api/src/postgres_triggers/handler.rs +++ b/backend/windmill-api/src/postgres_triggers/handler.rs @@ -374,7 +374,9 @@ pub async fn create_postgres_trigger( Path(w_id): Path, Json(new_postgres_trigger): Json, ) -> Result<(StatusCode, String)> { - check_scopes(&authed, || format!("postgres_triggers:write:{}", new_postgres_trigger.path))?; + check_scopes(&authed, || { + format!("postgres_triggers:write:{}", new_postgres_trigger.path) + })?; if *CLOUD_HOSTED { return Err(error::Error::BadRequest( @@ -515,32 +517,31 @@ pub async fn list_postgres_triggers( ) -> error::JsonResult> { let mut tx = user_db.begin(&authed).await?; let (per_page, offset) = paginate(Pagination { per_page: lst.per_page, page: lst.page }); - let mut sqlb = SqlBuilder::select_from("postgres_trigger") - .fields(&[ - "workspace_id", - "path", - "script_path", - "is_flow", - "edited_by", - "email", - "edited_at", - "server_id", - "last_server_ping", - "extra_perms", - "error", - "enabled", - "postgres_resource_path", - "replication_slot_name", - "publication_name", - "error_handler_path", - "error_handler_args", - "retry", - ]) - .order_by("edited_at", true) - .and_where("workspace_id = ?".bind(&w_id)) - .offset(offset) - .limit(per_page) - .clone(); + let mut sqlb = SqlBuilder::select_from("postgres_trigger"); + sqlb.fields(&[ + "workspace_id", + "path", + "script_path", + "is_flow", + "edited_by", + "email", + "edited_at", + "server_id", + "last_server_ping", + "extra_perms", + "error", + "enabled", + "postgres_resource_path", + "replication_slot_name", + "publication_name", + "error_handler_path", + "error_handler_args", + "retry", + ]) + .order_by("edited_at", true) + .and_where("workspace_id = ?".bind(&w_id)) + .offset(offset) + .limit(per_page); if let Some(path) = lst.path { sqlb.and_where_eq("script_path", "?".bind(&path)); } @@ -1204,7 +1205,9 @@ pub async fn update_postgres_trigger( Json(postgres_trigger): Json, ) -> Result { let workspace_path = path.to_path(); - check_scopes(&authed, || format!("postgres_triggers:write:{}", workspace_path))?; + check_scopes(&authed, || { + format!("postgres_triggers:write:{}", workspace_path) + })?; let EditPostgresTrigger { replication_slot_name, @@ -1298,7 +1301,7 @@ pub async fn update_postgres_trigger( workspace_path, error_handler_path, error_handler_args as _, - retry as _, + retry as _ ) .execute(&mut *tx) .await?; diff --git a/backend/windmill-api/src/settings.rs b/backend/windmill-api/src/settings.rs index 59ec5e3569..cacbeb4174 100644 --- a/backend/windmill-api/src/settings.rs +++ b/backend/windmill-api/src/settings.rs @@ -32,7 +32,7 @@ use serde::Deserialize; use windmill_common::ee_oss::{send_critical_alert, CriticalAlertKind, CriticalErrorChannel}; use windmill_common::error::to_anyhow; use windmill_common::{ - email_oss::send_email, + email_oss::send_email_plain_text, error::{self, JsonResult, Result}, get_database_url, global_settings::{ @@ -105,7 +105,7 @@ pub async fn test_email( let to = test_email.to; let client_timeout = Duration::from_secs(3); - send_email( + send_email_plain_text( "Test email from Windmill", "Test email content", vec![to], diff --git a/backend/windmill-api/src/websocket_triggers.rs b/backend/windmill-api/src/websocket_triggers.rs index 87ecb87e09..ee32a1150f 100644 --- a/backend/windmill-api/src/websocket_triggers.rs +++ b/backend/windmill-api/src/websocket_triggers.rs @@ -24,7 +24,9 @@ use windmill_common::{ db::UserDB, error::{self, to_anyhow, JsonResult}, triggers::TriggerKind, - utils::{not_found_if_none, paginate, report_critical_error, Pagination, StripPath}, + utils::{ + empty_as_none, not_found_if_none, paginate, report_critical_error, Pagination, StripPath, + }, worker::{to_raw_value, CLOUD_HOSTED}, INSTANCE_NAME, }; @@ -64,6 +66,7 @@ struct NewWebsocketTrigger { initial_messages: Option>>, url_runnable_args: Option>, can_return_message: bool, + #[serde(default, deserialize_with = "empty_as_none")] error_handler_path: Option, error_handler_args: Option>>>, retry: Option>, @@ -152,13 +155,34 @@ async fn list_websocket_triggers( ) -> error::JsonResult> { let mut tx = user_db.begin(&authed).await?; let (per_page, offset) = paginate(Pagination { per_page: lst.per_page, page: lst.page }); - let mut sqlb = SqlBuilder::select_from("websocket_trigger") - .field("*") - .order_by("edited_at", true) - .and_where("workspace_id = ?".bind(&w_id)) - .offset(offset) - .limit(per_page) - .clone(); + let mut sqlb = SqlBuilder::select_from("websocket_trigger"); + + sqlb.fields(&[ + "workspace_id", + "path", + "url", + "script_path", + "is_flow", + "edited_by", + "email", + "edited_at", + "server_id", + "last_server_ping", + "extra_perms", + "error", + "enabled", + "filters", + "initial_messages", + "url_runnable_args", + "can_return_message", + "error_handler_path", + "error_handler_args", + "retry", + ]) + .order_by("edited_at", true) + .and_where("workspace_id = ?".bind(&w_id)) + .offset(offset) + .limit(per_page); if let Some(path) = lst.path { sqlb.and_where_eq("script_path", "?".bind(&path)); } @@ -188,14 +212,39 @@ async fn get_websocket_trigger( check_scopes(&authed, || format!("websocket_triggers:read:{}", path))?; let mut tx = user_db.begin(&authed).await?; let trigger = sqlx::query_as::<_, WebsocketTrigger>( - r#"SELECT * - FROM websocket_trigger - WHERE workspace_id = $1 AND path = $2"#, + r#" + SELECT + workspace_id, + path, + url, + script_path, + is_flow, + edited_by, + email, + edited_at, + server_id, + last_server_ping, + extra_perms, + error, + enabled, + filters, + initial_messages, + url_runnable_args, + can_return_message, + error_handler_path, + error_handler_args, + retry, + FROM + websocket_trigger + WHERE + workspace_id = $1 AND path = $2 + "#, ) .bind(w_id) .bind(path) .fetch_optional(&mut *tx) .await?; + tx.commit().await?; let trigger = not_found_if_none(trigger, "Trigger", path)?; @@ -317,8 +366,29 @@ async fn update_websocket_trigger( // important to update server_id to NULL to stop current websocket listener sqlx::query!( - "UPDATE websocket_trigger SET url = $1, script_path = $2, path = $3, is_flow = $4, filters = $5, initial_messages = $6, url_runnable_args = $7, edited_by = $8, email = $9, can_return_message = $10, edited_at = now(), server_id = NULL, error = NULL, error_handler_path = $13, error_handler_args = $14, retry = $15 - WHERE workspace_id = $11 AND path = $12", + " + UPDATE + websocket_trigger + SET + url = $1, + script_path = $2, + path = $3, + is_flow = $4, + filters = $5, + initial_messages = $6, + url_runnable_args = $7, + edited_by = $8, + email = $9, + can_return_message = $10, + edited_at = now(), + server_id = NULL, + error = NULL, + error_handler_path = $13, + error_handler_args = $14, + retry = $15 + WHERE + workspace_id = $11 AND path = $12 + ", ct.url, ct.script_path, ct.path, @@ -333,9 +403,10 @@ async fn update_websocket_trigger( path, ct.error_handler_path, ct.error_handler_args as _, - ct.retry as _, + ct.retry as _ ) - .execute(&mut *tx).await?; + .execute(&mut *tx) + .await?; audit_log( &mut *tx, @@ -546,18 +617,45 @@ async fn listen_to_unlistened_websockets( db: &DB, killpill_rx: &tokio::sync::broadcast::Receiver<()>, ) { - match sqlx::query_as::<_, WebsocketTrigger>( - r#"SELECT * - FROM websocket_trigger - WHERE enabled IS TRUE AND (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds')"# + let websocket_triggers = sqlx::query_as::<_, WebsocketTrigger>( + r#" + SELECT + workspace_id, + path, + url, + script_path, + is_flow, + edited_by, + email, + edited_at, + server_id, + last_server_ping, + extra_perms, + error, + enabled, + filters, + initial_messages, + url_runnable_args, + can_return_message, + error_handler_path, + error_handler_args, + retry + FROM websocket_trigger + WHERE + enabled IS TRUE + AND (last_server_ping IS NULL OR last_server_ping < now() - interval '15 seconds') + "#, ) .fetch_all(db) - .await - { + .await; + + match websocket_triggers { Ok(mut triggers) => { triggers.shuffle(&mut rand::rng()); for trigger in triggers { - trigger.maybe_listen_to_websocket(db.clone(), killpill_rx.resubscribe()).await; + trigger + .maybe_listen_to_websocket(db.clone(), killpill_rx.resubscribe()) + .await; } } Err(err) => { diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index 6f6650713e..acec13e5f0 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -381,6 +381,10 @@ pub struct EditErrorHandler { pub error_handler_muted_on_cancel: Option, } +lazy_static::lazy_static! { + pub static ref EMAIL_REGEXP: Regex = Regex::new(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$").unwrap(); +} + async fn list_pending_invites( authed: ApiAuthed, Extension(user_db): Extension, @@ -457,15 +461,51 @@ async fn get_settings( let mut tx = user_db.begin(&authed).await?; let settings = sqlx::query_as!( WorkspaceSettings, - "SELECT workspace_id, slack_team_id, teams_team_id, teams_team_name, slack_name, slack_command_script, teams_command_script, slack_email, auto_invite_domain, auto_invite_operator, auto_add, customer_id, plan, webhook, deploy_to, ai_config, error_handler, error_handler_extra_args, error_handler_muted_on_cancel, large_file_storage, ducklake, git_sync, deploy_ui, default_app, default_scripts, mute_critical_alerts, color, operator_settings, git_app_installations FROM workspace_settings WHERE workspace_id = $1", + r#" + SELECT + workspace_id, + slack_team_id, + teams_team_id, + teams_team_name, + slack_name, + slack_command_script, + teams_command_script, + slack_email, + auto_invite_domain, + auto_invite_operator, + auto_add, + customer_id, + plan, + webhook, + deploy_to, + ai_config, + error_handler, + error_handler_extra_args, + error_handler_muted_on_cancel, + large_file_storage, + ducklake, + git_sync, + deploy_ui, + default_app, + default_scripts, + mute_critical_alerts, + color, + operator_settings, + git_app_installations + FROM + workspace_settings + WHERE + workspace_id = $1 + "#, &w_id ) .fetch_optional(&mut *tx) .await .map_err(|e| Error::internal_err(format!("getting settings: {e:#}")))?; - tx.commit().await?; - let settings = not_found_if_none(settings, "workspace settings", &w_id)?; + tx.commit().await?; + + let settings = not_found_if_none(settings, "workspace settings", &w_id)?; Ok(Json(settings)) } @@ -1645,8 +1685,46 @@ async fn edit_error_handler( .await?; if let Some(error_handler) = &ee.error_handler { + match ee.error_handler_extra_args.as_ref() { + Some(extra_args) if extra_args.is_object() => { + let Ok(email_recipients) = serde_json::from_value::>>( + extra_args["email_recipients"].to_owned(), + ) else { + return Err(Error::BadRequest( + "Field `email_recipients` expected to be JSON array".to_string(), + )); + }; + + if let Some(email_recipients) = email_recipients { + for email in email_recipients { + if !EMAIL_REGEXP.is_match(&email) { + return Err(Error::BadRequest(format!( + "Invalid email format: {}", + email + ))); + } + } + } + } + None => {} + _ => { + return Err(Error::BadRequest( + "Field `error_handler_extra_args` expected to be JSON object".to_string(), + )) + } + } + sqlx::query!( - "UPDATE workspace_settings SET error_handler = $1, error_handler_extra_args = $2, error_handler_muted_on_cancel = $3 WHERE workspace_id = $4", + r#" + UPDATE + workspace_settings + SET + error_handler = $1, + error_handler_extra_args = $2, + error_handler_muted_on_cancel = $3 + WHERE + workspace_id = $4 + "#, error_handler, ee.error_handler_extra_args, ee.error_handler_muted_on_cancel.unwrap_or(false), @@ -1656,12 +1734,22 @@ async fn edit_error_handler( .await?; } else { sqlx::query!( - "UPDATE workspace_settings SET error_handler = NULL, error_handler_extra_args = NULL WHERE workspace_id = $1", - &w_id, + r#" + UPDATE + workspace_settings + SET + error_handler = NULL, + error_handler_extra_args = NULL, + error_handler_muted_on_cancel = NULL + WHERE + workspace_id = $1 + "#, + &w_id ) .execute(&mut *tx) .await?; } + audit_log( &mut *tx, &authed, diff --git a/backend/windmill-api/src/workspaces_export.rs b/backend/windmill-api/src/workspaces_export.rs index 88475dd129..5fe996dbe3 100644 --- a/backend/windmill-api/src/workspaces_export.rs +++ b/backend/windmill-api/src/workspaces_export.rs @@ -564,8 +564,10 @@ pub(crate) async fn tarball_workspace( error_handler_path, error_handler_args as "error_handler_args: _", retry as "retry: _" - FROM http_trigger - WHERE workspace_id = $1 + FROM + http_trigger + WHERE + workspace_id = $1 "#, &w_id ) @@ -649,7 +651,7 @@ pub(crate) async fn tarball_workspace( enabled, error_handler_path, error_handler_args as "error_handler_args: _", - retry as "retry: _" + retry as "retry: _" FROM kafka_trigger WHERE workspace_id = $1"#, &w_id @@ -760,7 +762,8 @@ pub(crate) async fn tarball_workspace( { let nats_triggers = sqlx::query_as!( crate::nats_triggers_oss::NatsTrigger, - r#"SELECT + r#" + SELECT workspace_id, path, nats_resource_path, @@ -781,8 +784,11 @@ pub(crate) async fn tarball_workspace( error_handler_path, error_handler_args as "error_handler_args: _", retry as "retry: _" - FROM nats_trigger - WHERE workspace_id = $1"#, + FROM + nats_trigger + WHERE + workspace_id = $1 + "#, &w_id ) .fetch_all(&mut *tx) @@ -801,7 +807,8 @@ pub(crate) async fn tarball_workspace( { let postgres_triggers = sqlx::query_as!( crate::postgres_triggers::PostgresTrigger, - r#"SELECT + r#" + SELECT workspace_id, path, script_path, @@ -819,9 +826,12 @@ pub(crate) async fn tarball_workspace( postgres_resource_path, error_handler_path, error_handler_args as "error_handler_args: _", - retry as "retry: _" - FROM postgres_trigger - WHERE workspace_id = $1"#, + retry as "retry: _" + FROM + postgres_trigger + WHERE + workspace_id = $1 + "#, &w_id ) .fetch_all(&mut *tx) diff --git a/backend/windmill-common/src/agent_workers.rs b/backend/windmill-common/src/agent_workers.rs index 52c7e2d70c..74626522e6 100644 --- a/backend/windmill-common/src/agent_workers.rs +++ b/backend/windmill-common/src/agent_workers.rs @@ -14,8 +14,6 @@ use reqwest_retry::{policies::ExponentialBackoff, RetryTransientMiddleware}; use crate::{jwt::decode_without_verify, worker::HttpClient}; lazy_static! { - pub static ref BASE_INTERNAL_URL: String = - std::env::var("BASE_INTERNAL_URL").unwrap_or("http://localhost:8080".to_string()); pub static ref AGENT_TOKEN: String = std::env::var("AGENT_TOKEN").unwrap_or_default(); pub static ref DECODED_AGENT_TOKEN: Option = { if AGENT_TOKEN.is_empty() { diff --git a/backend/windmill-common/src/email_oss.rs b/backend/windmill-common/src/email_oss.rs index e6a340523e..05eb6bf243 100644 --- a/backend/windmill-common/src/email_oss.rs +++ b/backend/windmill-common/src/email_oss.rs @@ -15,3 +15,25 @@ pub async fn send_email( ) -> crate::error::Result<()> { Ok(()) } + +#[cfg(not(feature = "private"))] +pub async fn send_email_html( + _subject: &str, + _content: &str, + _to: Vec, + _smtp: Smtp, + _client_timeout: Option, +) -> crate::error::Result<()> { + Ok(()) +} + +#[cfg(not(feature = "private"))] +pub async fn send_email_plain_text( + _subject: &str, + _content: &str, + _to: Vec, + _smtp: Smtp, + _client_timeout: Option, +) -> crate::error::Result<()> { + Ok(()) +} diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index 2b4a9a3067..c2be3abf6d 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -12,6 +12,8 @@ use uuid::Uuid; pub const ENTRYPOINT_OVERRIDE: &str = "_ENTRYPOINT_OVERRIDE"; pub const LARGE_LOG_THRESHOLD_SIZE: usize = 9000; +pub const EMAIL_ERROR_HANDLER_USER_EMAIL: &str = "email_error_handler@windmill.dev"; + use crate::{ apps::AppScriptId, auth::is_super_admin_email, diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 3076157c02..4b5408289f 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -135,6 +135,7 @@ lazy_static::lazy_static! { pub static ref BASE_URL: Arc> = Arc::new(RwLock::new("".to_string())); pub static ref IS_READY: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false); + pub static ref BASE_INTERNAL_URL: String = std::env::var("BASE_INTERNAL_URL").unwrap_or("http://localhost:8000".to_string()); pub static ref HUB_BASE_URL: Arc> = Arc::new(RwLock::new(DEFAULT_HUB_BASE_URL.to_string())); diff --git a/backend/windmill-common/src/utils.rs b/backend/windmill-common/src/utils.rs index 0abed36d9b..6486819e61 100644 --- a/backend/windmill-common/src/utils.rs +++ b/backend/windmill-common/src/utils.rs @@ -49,6 +49,7 @@ use std::sync::atomic::Ordering; use crate::worker::CLOUD_HOSTED; lazy_static::lazy_static! { + pub static ref HTTP_CLIENT: Client = reqwest::ClientBuilder::new() .user_agent("windmill/beta") .timeout(std::time::Duration::from_secs(20)) diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index e10a450055..5236bb8f75 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -26,13 +26,7 @@ use uuid::Uuid; use windmill_macros::annotations; use crate::{ - agent_workers::{PingJobStatusResponse, BASE_INTERNAL_URL}, - cache::{unwrap_or_error, RawNode, RawScript}, - error::{self, to_anyhow}, - global_settings::CUSTOM_TAGS_SETTING, - indexer::TantivyIndexerSettings, - server::Smtp, - KillpillSender, DB, + agent_workers::PingJobStatusResponse, cache::{unwrap_or_error, RawNode, RawScript}, error::{self, to_anyhow}, global_settings::CUSTOM_TAGS_SETTING, indexer::TantivyIndexerSettings, server::Smtp, KillpillSender, BASE_INTERNAL_URL, DB }; #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)] diff --git a/backend/windmill-queue/Cargo.toml b/backend/windmill-queue/Cargo.toml index 4458e8aa99..b50773f9c8 100644 --- a/backend/windmill-queue/Cargo.toml +++ b/backend/windmill-queue/Cargo.toml @@ -15,6 +15,7 @@ enterprise = ["windmill-common/enterprise"] cloud = [] benchmark = ["windmill-common/benchmark"] prometheus = ["dep:prometheus"] +smtp = [] [dependencies] windmill-audit.workspace = true diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 76a763f463..995c3ac554 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -33,8 +33,11 @@ use windmill_common::add_time; use windmill_common::auth::JobPerms; #[cfg(feature = "benchmark")] use windmill_common::bench::BenchmarkIter; +use windmill_common::jobs::EMAIL_ERROR_HANDLER_USER_EMAIL; use windmill_common::utils::now_from_db; use windmill_common::worker::{Connection, SCRIPT_TOKEN_EXPIRY}; +#[cfg(feature = "enterprise")] +use windmill_common::BASE_URL; use windmill_common::{ auth::{fetch_authed_from_permissioned_as, permissioned_as_to_username}, cache::{self, FlowData}, @@ -63,16 +66,12 @@ use windmill_common::{ use backon::ConstantBuilder; use backon::{BackoffBuilder, Retryable}; -#[cfg(feature = "enterprise")] -use windmill_common::BASE_URL; - -#[cfg(feature = "cloud")] -use windmill_common::users::SUPERADMIN_SYNC_EMAIL; - use crate::flow_status::{update_flow_status_in_progress, update_workflow_as_code_status}; use crate::jobs_oss::update_concurrency_counter; use crate::schedule::{get_schedule_opt, push_scheduled_job}; use crate::tags::per_workspace_tag; +#[cfg(feature = "cloud")] +use windmill_common::users::SUPERADMIN_SYNC_EMAIL; #[cfg(feature = "prometheus")] lazy_static::lazy_static! { @@ -121,6 +120,7 @@ const MAX_FREE_CONCURRENT_RUNS: i32 = 30; const ERROR_HANDLER_USERNAME: &str = "error_handler"; const SCHEDULE_ERROR_HANDLER_USERNAME: &str = "schedule_error_handler"; +const GLOBAL_ERROR_HANDLER_USERNAME: &str = "global"; #[cfg(feature = "enterprise")] const SCHEDULE_RECOVERY_HANDLER_USERNAME: &str = "schedule_recovery_handler"; const ERROR_HANDLER_USER_GROUP: &str = "g/error_handler"; @@ -1360,6 +1360,7 @@ async fn apply_completed_job_error_handlers {} - _ => { + err => { + tracing::error!( + "Could not run workspace error handler for job {}: {}", + &queued_job.id, + err + ); let base_url = BASE_URL.read().await; let w_id: &String = &queued_job.workspace_id; report_critical_error(format!( - "Could not push workspace error handler for failed job ({base_url}/run/{}?workspace={w_id}): {}", + "Failed to push workspace error handler job to handle failed job ({base_url}/run/{}?workspace={w_id}): {}", queued_job.id, err ), db.clone(), Some(&w_id), None) @@ -1485,8 +1491,16 @@ pub async fn report_error_to_workspace_handler_or_critical_side_channel( error_message: String, ) -> () { let w_id = &queued_job.workspace_id; - let (error_handler, error_handler_extra_args) = sqlx::query_as::<_, (Option, Option>>)>( - "SELECT error_handler, error_handler_extra_args FROM workspace_settings WHERE workspace_id = $1", + let row_result = sqlx::query_as::<_, (Option, Option>>)>( + r#" + SELECT + error_handler, + error_handler_extra_args + FROM + workspace_settings + WHERE + workspace_id = $1 + "#, ) .bind(&w_id) .fetch_optional(db) @@ -1495,6 +1509,8 @@ pub async fn report_error_to_workspace_handler_or_critical_side_channel( .flatten() .unwrap_or((None, None)); + let (error_handler, error_handler_extra_args) = row_result; + if let Some(error_handler) = error_handler { if let Err(err) = push_error_handler( db, @@ -1538,8 +1554,18 @@ pub async fn send_error_to_workspace_handler<'a, 'c, T: Serialize + Send + Sync> result: Json<&'a T>, ) -> Result<(), Error> { let w_id = &queued_job.workspace_id; - let (error_handler, error_handler_extra_args, error_handler_muted_on_cancel) = sqlx::query_as::<_, (Option, Option>>, bool)>( - "SELECT error_handler, error_handler_extra_args, error_handler_muted_on_cancel FROM workspace_settings WHERE workspace_id = $1", + + let row_result = sqlx::query_as::<_, (Option, Option>>, bool)>( + r#" + SELECT + error_handler, + error_handler_extra_args, + error_handler_muted_on_cancel + FROM + workspace_settings + WHERE + workspace_id = $1 + "#, ) .bind(&w_id) .fetch_optional(db) @@ -1547,6 +1573,8 @@ pub async fn send_error_to_workspace_handler<'a, 'c, T: Serialize + Send + Sync> .context("fetching error handler info from workspace_settings")? .ok_or_else(|| Error::internal_err(format!("no workspace settings for id {w_id}")))?; + let (error_handler, error_handler_extra_args, error_handler_muted_on_cancel) = row_result; + if is_canceled && error_handler_muted_on_cancel { return Ok(()); } @@ -1598,7 +1626,6 @@ pub async fn send_error_to_workspace_handler<'a, 'c, T: Serialize + Send + Sync> .await?; } } - Ok(()) } @@ -1846,6 +1873,71 @@ async fn apply_schedule_handlers<'a, 'c, T: Serialize + Send + Sync>( Ok(()) } +pub const ERROR_HANDLER_PATH_TEAMS: &str = "/workspace-or-schedule-error-handler-teams"; +pub const ERROR_HANDLER_PATH_SLACK: &str = "/workspace-or-schedule-error-handler-slack"; +pub const ERROR_HANDLER_PATH_EMAIL: &str = "/workspace-or-error-handler-email"; + +enum ErrorHandlerType { + Custom, + Teams, + Slack, + Email, +} + +impl ErrorHandlerType { + fn from_error_handler_path(error_handler_path: &str) -> Option { + let error_handler_path = if error_handler_path.starts_with("script/") { + error_handler_path.strip_prefix("script/").unwrap() + } else if error_handler_path.starts_with("flow/") { + error_handler_path.strip_prefix("flow/").unwrap() + } else { + error_handler_path + }; + + if let Some(from_hub) = error_handler_path.strip_prefix("hub/") { + let handler_type = if from_hub.ends_with(ERROR_HANDLER_PATH_TEAMS) { + ErrorHandlerType::Teams + } else if from_hub.ends_with(ERROR_HANDLER_PATH_SLACK) { + ErrorHandlerType::Slack + } else if from_hub.ends_with(ERROR_HANDLER_PATH_EMAIL) { + ErrorHandlerType::Email + } else { + return None; + }; + + return Some(handler_type); + } + + Some(ErrorHandlerType::Custom) + } +} + +fn get_email_and_permissioned_as( + error_handler_path: &str, + is_global_error_handler: bool, + is_schedule_error_handler: bool, +) -> (&'static str, String) { + let res = if is_global_error_handler { + (SUPERADMIN_SECRET_EMAIL, SUPERADMIN_SECRET_EMAIL.to_string()) + } else if is_schedule_error_handler { + ( + SCHEDULE_ERROR_HANDLER_USER_EMAIL, + ERROR_HANDLER_USER_GROUP.to_string(), + ) + } else { + let handler_type = ErrorHandlerType::from_error_handler_path(error_handler_path); + + let email = match handler_type { + Some(ErrorHandlerType::Email) => EMAIL_ERROR_HANDLER_USER_EMAIL, + _ => ERROR_HANDLER_USER_EMAIL, + }; + + (email, ERROR_HANDLER_USER_GROUP.to_string()) + }; + + res +} + pub async fn push_error_handler<'a, 'c, T: Serialize + Send + Sync>( db: &Pool, job_id: Uuid, @@ -1881,6 +1973,7 @@ pub async fn push_error_handler<'a, 'c, T: Serialize + Send + Sync>( extra.insert("is_flow".to_string(), to_raw_value(&is_flow)); extra.insert("started_at".to_string(), to_raw_value(&started_at)); extra.insert("email".to_string(), to_raw_value(&email)); + if let Some(failed_times) = failed_times { extra.insert("failed_times".to_string(), to_raw_value(&failed_times)); } @@ -1902,17 +1995,11 @@ pub async fn push_error_handler<'a, 'c, T: Serialize + Send + Sync>( on_behalf_of.email.as_str(), on_behalf_of.permissioned_as.clone(), ) - } else if is_global_error_handler { - (SUPERADMIN_SECRET_EMAIL, SUPERADMIN_SECRET_EMAIL.to_string()) - } else if is_schedule_error_handler { - ( - SCHEDULE_ERROR_HANDLER_USER_EMAIL, - ERROR_HANDLER_USER_GROUP.to_string(), - ) } else { - ( - ERROR_HANDLER_USER_EMAIL, - ERROR_HANDLER_USER_GROUP.to_string(), + get_email_and_permissioned_as( + on_failure_path, + is_global_error_handler, + is_schedule_error_handler, ) }; @@ -1924,7 +2011,7 @@ pub async fn push_error_handler<'a, 'c, T: Serialize + Send + Sync>( payload, PushArgs { extra: Some(extra), args: &result }, if is_global_error_handler { - "global" + GLOBAL_ERROR_HANDLER_USERNAME } else if is_schedule_error_handler { SCHEDULE_ERROR_HANDLER_USERNAME } else { @@ -2172,6 +2259,25 @@ pub enum JobTriggerKind { Gcp, } +impl std::fmt::Display for JobTriggerKind { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let kind = match self { + JobTriggerKind::Webhook => "webhook", + JobTriggerKind::Http => "http", + JobTriggerKind::Websocket => "websocket", + JobTriggerKind::Kafka => "kafka", + JobTriggerKind::Email => "email", + JobTriggerKind::Nats => "nats", + JobTriggerKind::Mqtt => "mqtt", + JobTriggerKind::Sqs => "sqs", + JobTriggerKind::Postgres => "postgres", + JobTriggerKind::Schedule => "schedule", + JobTriggerKind::Gcp => "gcp", + }; + write!(f, "{}", kind) + } +} + #[derive(sqlx::FromRow, Debug, Clone, Serialize, Deserialize)] pub struct MiniPulledJob { pub workspace_id: String, @@ -4532,6 +4638,12 @@ pub async fn push<'c, 'd>( // tracing::error!("Could not insert job_perms for job {job_id}: {err:#}"); // } + let trigger_kind = if schedule_path.is_some() { + Some(JobTriggerKind::Schedule) + } else { + None + }; + sqlx::query!( "WITH inserted_job AS ( INSERT INTO v2_job (id, workspace_id, raw_code, raw_lock, raw_flow, tag, parent_job, @@ -4540,8 +4652,7 @@ pub async fn push<'c, 'd>( flow_innermost_root_job, root_job, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, cache_ttl, priority, trigger_kind, script_entrypoint_override, preprocessed) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, - $19, $20, $38, $21, $22, $23, $24, $25, $26, - CASE WHEN $14::VARCHAR IS NOT NULL THEN 'schedule'::job_trigger_kind END, + $19, $20, $38, $21, $22, $23, $24, $25, $26, $39::job_trigger_kind, ($12::JSONB)->>'_ENTRYPOINT_OVERRIDE', $27) ), inserted_runtime AS ( @@ -4596,7 +4707,8 @@ pub async fn push<'c, 'd>( job_authed.is_operator, folders.as_slice(), job_authed.groups.as_slice(), - root_job.or(parent_job) + root_job.or(parent_job), + trigger_kind as Option, ) .execute(&mut *tx) .warn_after_seconds(1) diff --git a/frontend/src/lib/components/ErrorOrRecoveryHandler.svelte b/frontend/src/lib/components/ErrorOrRecoveryHandler.svelte index 9f4463018b..247dd05d6b 100644 --- a/frontend/src/lib/components/ErrorOrRecoveryHandler.svelte +++ b/frontend/src/lib/components/ErrorOrRecoveryHandler.svelte @@ -10,6 +10,8 @@ import { enterpriseLicense, workspaceStore } from '$lib/stores' import MsTeamsIcon from '$lib/components/icons/MSTeamsIcon.svelte' import { emptySchema, emptyString, sendUserToast, tryEvery } from '$lib/utils' + import Description from '$lib/components/Description.svelte' + import MultiSelect from '$lib/components/select/MultiSelect.svelte' import { FlowService, JobService, @@ -18,17 +20,17 @@ WorkspaceService, type Flow } from '$lib/gen' - import type { ListAvailableTeamsChannelsResponse } from '$lib/gen/types.gen' + import type { ErrorHandler, ListAvailableTeamsChannelsResponse } from '$lib/gen/types.gen' import { inferArgs } from '$lib/infer' import { hubBaseUrlStore } from '$lib/stores' import { CheckCircle2, Loader2, RotateCw, XCircle, RefreshCcw } from 'lucide-svelte' import { hubPaths } from '$lib/hub' + import { isCloudHosted } from '$lib/cloud' const slackRecoveryHandler = hubPaths.slackRecoveryHandler const slackHandlerScriptPath = hubPaths.slackErrorHandler const slackSuccessHandler = hubPaths.slackSuccessHandler - const teamsRecoveryHandler = hubPaths.teamsRecoveryHandler const teamsHandlerScriptPath = hubPaths.teamsErrorHandler const teamsSuccessHandler = hubPaths.teamsSuccessHandler @@ -38,7 +40,7 @@ isEditable: boolean toggleText?: string showScriptHelpText?: boolean - handlerSelected: 'custom' | 'slack' | 'teams' + handlerSelected: ErrorHandler handlerPath: string | undefined handlerExtraArgs: Record customScriptTemplate: string @@ -51,18 +53,17 @@ isEditable, toggleText = 'Enable', showScriptHelpText = false, - handlerSelected = $bindable(), + handlerSelected = $bindable('custom'), handlerPath = $bindable(), handlerExtraArgs = $bindable(), customScriptTemplate, customHandlerKind = $bindable('script'), - customTabTooltip + customTabTooltip, }: Props = $props() let customHandlerSchema: Schema | undefined = $state() let slackHandlerSchema: Schema | undefined = $state() let isFetching: boolean = $state(false) - let teams_channels: ListAvailableTeamsChannelsResponse = $state([]) let teams_team_name: string | undefined = $state(undefined) @@ -71,6 +72,8 @@ let connectionTestJob: { uuid: string; is_success: boolean; in_progress: boolean } | undefined = $state() + const EMAIL_RECIPIENTS_KEY = 'email_recipients' + const CHANNEL_KEY = 'channel' async function loadSlackResources() { const settings = await WorkspaceService.getSettings({ workspace: $workspaceStore! }) @@ -222,6 +225,14 @@ } } + function isEmailHandler(scriptPath: string | undefined) { + if (!scriptPath) { + return false + } + + return scriptPath.startsWith('hub/') && scriptPath.endsWith('/workspace-or-error-handler-email') + } + $effect(() => { if ($workspaceStore) { loadSlackResources() @@ -237,22 +248,27 @@ } }) - let lastHandlerSelected: 'slack' | 'teams' | 'custom' | undefined = $state(undefined) - let channelCache = $state({ + let lastHandlerSelected: ErrorHandler | undefined = $state(undefined) + let handlerCache = $state({ slack: undefined as string | undefined, - teams: undefined as string | undefined + teams: undefined as string | undefined, + email: undefined as string[] | undefined }) $effect(() => { if (lastHandlerSelected !== handlerSelected && lastHandlerSelected !== undefined) { - if (lastHandlerSelected === 'teams' || lastHandlerSelected === 'slack') { - channelCache[lastHandlerSelected] = handlerExtraArgs['channel'] + if (lastHandlerSelected != 'custom') { + const key = lastHandlerSelected === 'email' ? EMAIL_RECIPIENTS_KEY : CHANNEL_KEY + handlerCache[lastHandlerSelected] = handlerExtraArgs[key] } if (handlerSelected === 'custom') { - handlerExtraArgs['channel'] = '' + handlerExtraArgs[CHANNEL_KEY] = '' + handlerExtraArgs[EMAIL_RECIPIENTS_KEY] = [] handlerPath = undefined + } else if (handlerSelected === 'email') { + handlerExtraArgs[EMAIL_RECIPIENTS_KEY] = handlerCache[handlerSelected] ?? [] } else { - handlerExtraArgs['channel'] = channelCache[handlerSelected] ?? '' + handlerExtraArgs[CHANNEL_KEY] = handlerCache[handlerSelected] ?? '' } } @@ -263,6 +279,7 @@ handlerPath && !isSlackHandler(handlerPath) && !isTeamsHandler(handlerPath) && + !isEmailHandler(handlerPath) && loadHandlerScriptArgs(handlerPath, [ 'path', 'workspace_id', @@ -302,12 +319,19 @@ 'slack' ]).then((schema) => (slackHandlerSchema = schema)) }) + + $effect(() => { + if (handlerSelected === 'email') { + handlerPath = hubPaths.emailErrorHandler + } + })
Slack Teams + Email Custom {@render customTabTooltip?.()} @@ -334,8 +358,10 @@ size="xs" href={customScriptTemplate} disabled={!isEditable} - target="_blank">Create from template + Create from template + {/if}
{#if showScriptHelpText} @@ -563,4 +589,46 @@ {/if} {/if} {/if} +{:else if handlerSelected === 'email'} + {#if isCloudHosted()} + + Email notifications for trigger failures are only available in self-hosted Windmill instances. + + {:else} +
+ + Configure email addresses to receive notifications when jobs fail. This feature requires + SMTP to be configured. + +
+
+ handlerExtraArgs[EMAIL_RECIPIENTS_KEY] ?? [], + (recipients) => (handlerExtraArgs[EMAIL_RECIPIENTS_KEY] = recipients) + } + placeholder="Enter email addresses..." + onCreateItem={(email) => { + const emailRegex = /^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$/ + if (!emailRegex.test(email)) { + sendUserToast('Invalid email format', true) + return + } + const currentArray = handlerExtraArgs[EMAIL_RECIPIENTS_KEY] ?? [] + handlerExtraArgs[EMAIL_RECIPIENTS_KEY] = [...currentArray, email] + }} + class="w-full" + /> + {#if handlerExtraArgs[EMAIL_RECIPIENTS_KEY]?.length > 0} + + {handlerExtraArgs[EMAIL_RECIPIENTS_KEY]?.length} email{handlerExtraArgs[ + EMAIL_RECIPIENTS_KEY + ]?.length === 1 + ? '' + : 's'} configured + + {/if} +
+ {/if} {/if} diff --git a/frontend/src/lib/components/triggers/TriggerRetriesAndErrorHandler.svelte b/frontend/src/lib/components/triggers/TriggerRetriesAndErrorHandler.svelte index 24b6a3632f..c4d599652f 100644 --- a/frontend/src/lib/components/triggers/TriggerRetriesAndErrorHandler.svelte +++ b/frontend/src/lib/components/triggers/TriggerRetriesAndErrorHandler.svelte @@ -1,5 +1,5 @@