feat: emit workflow errors earlier (#502)

crates/cli/src/commands/apply.rs

@@ -2,7 +2,7 @@ use anyhow::Result;
 use clap::Args;
 use indicatif::MultiProgress;
 
-use crate::{flags::GlobalFormatFlags, updater::Updater};
+use crate::{flags::GlobalFormatFlags, messenger_variant::create_emitter, updater::Updater};
 use marzano_messenger::emit::ApplyDetails;
 use serde::Serialize;
 use std::env::current_dir;
@@ -99,13 +99,24 @@ pub(crate) async fn run_apply(
                 .await;
 
         if let Some(custom_workflow) = custom_workflow {
+            let format = crate::flags::OutputFormat::from(flags);
+            let emitter = create_emitter(
+                &format,
+                marzano_messenger::output_mode::OutputMode::default(),
+                None,
+                false,
+                None,
+                None,
+                args.apply_pattern_args.visibility,
+            )
+            .await?;
+
             run_apply_migration(
                 custom_workflow,
                 paths,
                 ranges,
                 args.apply_migration_args,
-                flags,
-                args.apply_pattern_args.visibility,
+                emitter,
                 execution_id.clone(),
             )
             .instrument(tracing::span!(

crates/cli/src/commands/apply_migration.rs

@@ -1,5 +1,3 @@
-use crate::flags::GlobalFormatFlags;
-use crate::{flags::OutputFormat, messenger_variant::create_emitter};
 use marzano_messenger::emit::FlushableMessenger;
 
 #[cfg(not(feature = "workflows_v2"))]
@@ -71,26 +69,13 @@ pub(crate) async fn run_apply_migration(
     paths: Vec<PathBuf>,
     ranges: Option<Vec<marzano_util::diff::FileDiff>>,
     arg: ApplyMigrationArgs,
-    flags: &GlobalFormatFlags,
-    min_level: marzano_messenger::emit::VisibilityLevels,
+    mut emitter: crate::messenger_variant::MessengerVariant<'static>,
     execution_id: String,
 ) -> Result<PackagedWorkflowOutcome> {
     use crate::error::GoodError;
 
     let input = arg.get_payload()?;
 
-    let format = OutputFormat::from(flags);
-    let mut emitter = create_emitter(
-        &format,
-        marzano_messenger::output_mode::OutputMode::default(),
-        None,
-        false,
-        None,
-        None,
-        min_level,
-    )
-    .await?;
-
     emitter.start_workflow()?;
 
     let mut emitter = run_bin_workflow(

crates/cli/src/commands/plumbing.rs

@@ -8,7 +8,7 @@ use marzano_gritmodule::fetcher::KeepFetcherKind;
 use marzano_gritmodule::patterns_directory::PatternsDirectory;
 use marzano_gritmodule::searcher::find_grit_modules_dir;
 use marzano_gritmodule::utils::is_pattern_name;
-use marzano_messenger::emit::{ApplyDetails, VisibilityLevels};
+use marzano_messenger::emit::{ApplyDetails, Messager, VisibilityLevels};
 use serde::{Deserialize, Serialize};
 use std::env::current_dir;
 use std::io::{stdin, Read};
@@ -18,8 +18,9 @@ use tracing::Instrument as _;
 
 use crate::analytics::track_event_line;
 use crate::error::GoodError;
-use crate::flags::GlobalFormatFlags;
+use crate::flags::{GlobalFormatFlags, OutputFormat};
 use crate::lister::list_applyables;
+use crate::messenger_variant::create_emitter;
 use crate::resolver::{get_grit_files_from, resolve_from, Source};
 use crate::updater::Updater;
 
@@ -303,6 +304,18 @@ pub(crate) async fn run_plumbing(
             let execution_id = std::env::var("GRIT_EXECUTION_ID")
                 .unwrap_or_else(|_| uuid::Uuid::new_v4().to_string());
 
+            let format = OutputFormat::from(&parent);
+            let mut emitter = create_emitter(
+                &format,
+                marzano_messenger::output_mode::OutputMode::default(),
+                None,
+                false,
+                None,
+                None,
+                VisibilityLevels::default(),
+            )
+            .await?;
+
             let current_dir = current_dir()?;
             let mut updater = Updater::from_current_bin().await?;
             let auth = updater
@@ -315,15 +328,29 @@ pub(crate) async fn run_plumbing(
                 .await
                 .ok();
 
-            let custom_workflow =
-                crate::workflows::find_workflow_file_from(current_dir.clone(), &definition, auth)
-                    .instrument(tracing::span!(
-                        tracing::Level::INFO,
-                        "grit_marzano.find_workflow",
-                        "execution_id" = execution_id.as_str(),
-                    ))
-                    .await
-                    .context("Failed to find workflow file")?;
+            let custom_workflow = match crate::workflows::find_workflow_file_from(
+                current_dir.clone(),
+                &definition,
+                auth,
+            )
+            .instrument(tracing::span!(
+                tracing::Level::INFO,
+                "grit_marzano.find_workflow",
+                "execution_id" = execution_id.as_str(),
+            ))
+            .await
+            .context("Failed to find workflow file")
+            {
+                Ok(workflow) => workflow,
+                Err(e) => {
+                    let log = marzano_messenger::SimpleLogMessage::new_error(format!(
+                        "Failed to find workflow file: {}",
+                        e
+                    ));
+                    emitter.emit_log(&log)?;
+                    return Err(e);
+                }
+            };
 
             super::apply_migration::run_apply_migration(
                 custom_workflow,
@@ -336,8 +363,7 @@ pub(crate) async fn run_plumbing(
                     verbose: true,
                     watch: false,
                 },
-                &parent,
-                VisibilityLevels::default(),
+                emitter,
                 execution_id.clone(),
             )
             .instrument(tracing::span!(

crates/cli/src/commands/workflows.rs

@@ -1,7 +1,10 @@
 use clap::{Parser, Subcommand};
 use serde::Serialize;
 
-use super::{workflows_list::WorkflowsListArgs, workflows_watch::WorkflowWatchArgs, workflows_upload::WorkflowsUploadArgs};
+use super::{
+    workflows_list::WorkflowsListArgs, workflows_upload::WorkflowsUploadArgs,
+    workflows_watch::WorkflowWatchArgs,
+};
 
 #[derive(Parser, Debug, Serialize)]
 pub struct Workflows {

crates/cli/src/commands/workflows_upload.rs

@@ -4,9 +4,10 @@ use console::style;
 use serde::Serialize;
 use std::path::PathBuf;
 
-use crate::flags::GlobalFormatFlags;
+use crate::flags::{GlobalFormatFlags, OutputFormat};
 
 use crate::commands::apply_migration::{run_apply_migration, ApplyMigrationArgs};
+use crate::messenger_variant::create_emitter;
 use crate::workflows::fetch_remote_workflow;
 use marzano_messenger::emit::VisibilityLevels;
 
@@ -45,13 +46,23 @@ pub async fn run_upload_workflows(
     let execution_id =
         std::env::var("GRIT_EXECUTION_ID").unwrap_or_else(|_| uuid::Uuid::new_v4().to_string());
 
+    let format = OutputFormat::from(parent);
+    let emitter = create_emitter(
+        &format,
+        marzano_messenger::output_mode::OutputMode::default(),
+        None,
+        false,
+        None,
+        None,
+        VisibilityLevels::default(),
+    )
+    .await?;
     let result = run_apply_migration(
         workflow_info,
         vec![],
         None,
         apply_migration_args,
-        parent,
-        VisibilityLevels::default(),
+        emitter,
         execution_id,
     )
     .await?;

crates/marzano_messenger/src/logs.rs

@@ -10,6 +10,17 @@ pub struct SimpleLogMessage {
     pub step_id: Option<String>,
 }
 
+impl SimpleLogMessage {
+    pub fn new_error(message: String) -> Self {
+        SimpleLogMessage {
+            message,
+            level: AnalysisLogLevel::Error,
+            meta: None,
+            step_id: None,
+        }
+    }
+}
+
 /// Represents a raw log message that can be sent to the server,
 /// Some fields are double serialized to handle JSON columns
 #[derive(Debug, Serialize, Deserialize)]