fix: Error handler now supports flows (windmill-labs#2707)

* fix: Error handler now supports flows

* Update README

* remove unused import

README.md

@@ -363,7 +363,7 @@ it being synced automatically everyday.
 | SMTP_PASSWORD                                 | None                                       | password for the smtp server to send invite emails                                                                                                                                                 | Server                |
 | SMTP_TLS_IMPLICIT                             | false                                      | https://docs.rs/mail-send/latest/mail_send/struct.SmtpClientBuilder.html#method.implicit_tlsemails                                                                                                 | Server                |
 | CREATE_WORKSPACE_REQUIRE_SUPERADMIN           | true                                       | If true, only superadmin can create workspaces                                                                                                                                                     | Server                |
-| GLOBAL_ERROR_HANDLER_PATH_IN_ADMINS_WORKSPACE | None                                       | Path to a script to run when a root job fails. The script will be run in and from the admins workspace                                                                                             | Server                |
+| GLOBAL_ERROR_HANDLER_PATH_IN_ADMINS_WORKSPACE | None                                       | Path to a script or flow to run when a root job fails. The path needs to be prefixed with either `script/` or `flow/` to indicate the kind of error handler being used (assuming `script/` by default). The error handler will be run in and from the admins workspace                                                                                             | Server                |
 | WHITELIST_ENVS                                | None                                       | List of envs variables, separated by a ',' that are whitelisted as being safe to passthrough the workers                                                                                           | Worker                |
 | SAML_METADATA                                 | None                                       | SAML Metadata URL to enable SAML SSO (EE only)                                                                                                                                                     | Server                |
 | SECRET_SALT                                   | None                                       | Secret Salt used for encryption and decryption of secrets. If defined, the secrets will not be decryptable unless the right salt is passed in, which is the case for the workers and the server    | Server + Worker       |

backend/windmill-queue/src/jobs.rs

@@ -50,8 +50,7 @@ use windmill_common::{
     },
     flows::{add_virtual_items_if_necessary, FlowModuleValue, FlowValue},
     jobs::{
-        get_payload_tag_from_prefixed_path, script_path_to_payload, CompletedJob, JobKind,
-        JobPayload, QueuedJob, RawCode,
+        get_payload_tag_from_prefixed_path, CompletedJob, JobKind, JobPayload, QueuedJob, RawCode,
     },
     oauth2::WORKSPACE_SLACK_BOT_TOKEN_PATH,
     schedule::Schedule,
@@ -579,12 +578,13 @@ pub async fn run_error_handler<
     is_global: bool,
 ) -> Result<(), Error> {
     let w_id = &queued_job.workspace_id;
-    let script_w_id = if is_global { "admins" } else { w_id }; // script workspace id
+    let handler_w_id = if is_global { "admins" } else { w_id }; // script workspace id
     let job_id = queued_job.id;
-    let (job_payload, tag) = script_path_to_payload(&error_handler_path, db, script_w_id).await?;
+    let (job_payload, tag) =
+        get_payload_tag_from_prefixed_path(&error_handler_path, db, handler_w_id).await?;
 
     let mut extra = HashMap::new();
-    extra.insert("workspace_id".to_string(), to_raw_value(&w_id));
+    extra.insert("workspace_id".to_string(), to_raw_value(&handler_w_id));
     extra.insert("job_id".to_string(), to_raw_value(&job_id));
     extra.insert("path".to_string(), to_raw_value(&queued_job.script_path));
     extra.insert(
@@ -604,7 +604,7 @@ pub async fn run_error_handler<
     // TODO(gbouv): REMOVE THIS after December 1st 2023 and ping users to re-save their error handlers
     if error_handler_path
         .to_string()
-        .eq("hub/5792/workspace-or-schedule-error-handler-slack")
+        .eq("script/hub/5792/workspace-or-schedule-error-handler-slack")
     {
         // default slack error handler being used -> we need to inject the slack token
         let slack_resource = format!("$res:{WORKSPACE_SLACK_BOT_TOKEN_PATH}");
@@ -628,7 +628,7 @@ pub async fn run_error_handler<
     let (uuid, tx) = push(
         &db,
         tx,
-        script_w_id,
+        handler_w_id,
         job_payload,
         PushArgs { extra, args: result.to_owned() },
         if is_global {
@@ -682,12 +682,19 @@ pub async fn send_error_to_global_handler<
     result: Json<&'a T>,
 ) -> Result<(), Error> {
     if let Some(ref global_error_handler) = *GLOBAL_ERROR_HANDLER_PATH_IN_ADMINS_WORKSPACE {
+        let prefixed_global_error_handler_path = if global_error_handler.starts_with("script/")
+            || global_error_handler.starts_with("flow/")
+        {
+            global_error_handler.clone()
+        } else {
+            format!("script/{}", global_error_handler)
+        };
         run_error_handler(
             rsmq,
             queued_job,
             db,
             result,
-            global_error_handler,
+            &prefixed_global_error_handler_path,
             None,
             true,
         )
@@ -754,7 +761,7 @@ pub async fn send_error_to_workspace_handler<
                 queued_job,
                 db,
                 result,
-                &error_handler.strip_prefix("script/").unwrap(),
+                &error_handler,
                 error_handler_extra_args,
                 false,
             )

frontend/src/lib/components/ErrorOrRecoveryHandler.svelte

@@ -6,7 +6,14 @@
 	import type { Schema, SupportedLanguage } from '$lib/common'
 	import { enterpriseLicense, workspaceStore } from '$lib/stores'
 	import { emptySchema, emptyString, sendUserToast, tryEvery } from '$lib/utils'
-	import { JobService, Script, ScriptService, WorkspaceService } from '$lib/gen'
+	import {
+		FlowService,
+		JobService,
+		Script,
+		ScriptService,
+		WorkspaceService,
+		type Flow
+	} from '$lib/gen'
 	import { inferArgs } from '$lib/infer'
 
 	import { CheckCircle2, Loader2, RotateCw, XCircle } from 'lucide-svelte'
@@ -86,6 +93,7 @@
 	}
 
 	async function loadHandlerScriptArgs(p: string, defaultArgs: string[] = []) {
+		console.log(p)
 		try {
 			let schema: Schema | undefined = emptySchema()
 			if (p.startsWith('hub/')) {
@@ -99,8 +107,11 @@
 					await inferArgs(hubScript.language as SupportedLanguage, hubScript.content ?? '', schema)
 				}
 			} else {
-				const script = await ScriptService.getScriptByPath({ workspace: $workspaceStore!, path: p })
-				schema = script.schema as Schema
+				let scriptOrFlow: Script | Flow =
+					customHandlerKind === 'script'
+						? await ScriptService.getScriptByPath({ workspace: $workspaceStore!, path: p })
+						: await FlowService.getFlowByPath({ workspace: $workspaceStore!, path: p })
+				schema = scriptOrFlow.schema as Schema
 			}
 			if (schema && schema.properties) {
 				for (let key in schema.properties) {

frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte

@@ -194,6 +194,7 @@
 		workspaceToDeployTo = settings.deploy_to
 		webhook = settings.webhook
 		openaiResourceInitialPath = settings.openai_resource_path
+		errorHandlerItemKind = settings.error_handler?.split('/')[0] as 'flow' | 'script'
 		errorHandlerScriptPath = (settings.error_handler ?? '').split('/').slice(1).join('/')
 		errorHandlerInitialScriptPath = errorHandlerScriptPath
 		errorHandlerMutedOnCancel = settings.error_handler_muted_on_cancel