feat: add generic metadata emitter (#382)

crates/cli/src/messenger_variant.rs

@@ -1,7 +1,9 @@
 use anyhow::{bail, Result};
 use marzano_core::api::AnalysisLog;
 use marzano_messenger::{
-    emit::Messager, output_mode::OutputMode, workflows::PackagedWorkflowOutcome,
+    emit::Messager,
+    output_mode::OutputMode,
+    workflows::{PackagedWorkflowOutcome, WorkflowMessenger},
 };
 use std::{
     io::{self, Write},
@@ -14,6 +16,8 @@ use cli_server::combined::CombinedMessenger;
 use cli_server::pubsub::GooglePubSubMessenger;
 #[cfg(feature = "remote_redis")]
 use cli_server::redis::RedisMessenger;
+#[cfg(feature = "server")]
+use cli_server::workflows::RemoteWorkflowMessenger;
 
 use crate::{
     flags::OutputFormat,
@@ -106,6 +110,33 @@ impl<'a> Messager for MessengerVariant<'a> {
     }
 }
 
+impl<'a> WorkflowMessenger for MessengerVariant<'a> {
+    fn save_metadata(
+        &mut self,
+        message: &marzano_messenger::workflows::SimpleWorkflowMessage,
+    ) -> anyhow::Result<()> {
+        match self {
+            MessengerVariant::Formatted(_)
+            | MessengerVariant::Transformed(_)
+            | MessengerVariant::JsonLine(_) => {
+                // These are local, so no need to save metadata
+                log::debug!(
+                    "Skipping save_metadata for local messenger: {} {:?}",
+                    message.kind,
+                    message.message
+                );
+                Ok(())
+            }
+            #[cfg(feature = "remote_redis")]
+            MessengerVariant::Redis(m) => m.save_metadata(message),
+            #[cfg(feature = "remote_pubsub")]
+            MessengerVariant::GooglePubSub(m) => m.save_metadata(message),
+            #[cfg(feature = "server")]
+            MessengerVariant::Combined(m) => m.save_metadata(message),
+        }
+    }
+}
+
 impl<'a> From<FormattedMessager<'a>> for MessengerVariant<'a> {
     fn from(value: FormattedMessager<'a>) -> Self {
         Self::Formatted(value)

crates/cli/src/workflows.rs

@@ -4,6 +4,7 @@ use console::style;
 use log::debug;
 use marzano_auth::env::{get_grit_api_url, ENV_VAR_GRIT_API_URL, ENV_VAR_GRIT_AUTH_TOKEN};
 use marzano_gritmodule::{fetcher::LocalRepo, searcher::find_grit_dir_from};
+use marzano_messenger::workflows::WorkflowMessenger;
 use marzano_messenger::{emit::Messager, workflows::PackagedWorkflowOutcome};
 use marzano_util::diff::FileDiff;
 use serde::Serialize;
@@ -53,7 +54,7 @@ pub async fn run_bin_workflow<M>(
     mut arg: WorkflowInputs,
 ) -> Result<(M, PackagedWorkflowOutcome)>
 where
-    M: Messager + Send + 'static,
+    M: Messager + WorkflowMessenger + Send + 'static,
 {
     let cwd = std::env::current_dir()?;
 

crates/cli_bin/tests/apply.rs

@@ -2822,7 +2822,7 @@ fn apply_remote_pattern() -> Result<()> {
     assert!(output.status.success(), "Command should have succeeded");
 
     let test_file = dir.join("test.js");
-    let content: String = fs_err::read_to_string(&test_file)?;
+    let content: String = fs_err::read_to_string(test_file)?;
     assert_snapshot!(content);
 
     Ok(())

crates/marzano_messenger/src/workflows.rs

@@ -6,3 +6,15 @@ pub struct PackagedWorkflowOutcome {
     pub success: bool,
     pub data: Option<serde_json::Value>,
 }
+
+/// Handle workflow-related messages
+pub trait WorkflowMessenger {
+    fn save_metadata(&mut self, message: &SimpleWorkflowMessage) -> anyhow::Result<()>;
+}
+
+/// Simple workflow message representation, mainly intended for RPC
+#[derive(Deserialize, Serialize, Debug)]
+pub struct SimpleWorkflowMessage {
+    pub kind: String,
+    pub message: serde_json::Value,
+}