fix: repair telemetry processing

crates/cli/src/analytics.rs

@@ -1,3 +1,4 @@
+use anyhow::Context;
 use anyhow::Result;
 use clap::Args;
 use lazy_static::lazy_static;
@@ -19,6 +20,16 @@ pub enum AnalyticsEventName {
     Errored,
 }
 
+impl<'a> From<&'a AnalyticsEvent<'a>> for AnalyticsEventName {
+    fn from(event: &'a AnalyticsEvent) -> Self {
+        match event {
+            AnalyticsEvent::Invoked(_) => AnalyticsEventName::Invoked,
+            AnalyticsEvent::Completed(_) => AnalyticsEventName::Completed,
+            AnalyticsEvent::Errored(_) => AnalyticsEventName::Errored,
+        }
+    }
+}
+
 #[derive(Debug, Serialize, Clone)]
 #[serde(untagged)]
 pub enum AnalyticsEvent<'a> {
@@ -148,6 +159,25 @@ async fn track_event_segment(
     Ok(())
 }
 
+pub async fn track_event_line(
+    line: &str,
+    command: String,
+    args: Vec<String>,
+    installation_id: Uuid,
+    user_id: Option<String>,
+) -> Result<()> {
+    let (name, json) = line
+        .split_once('\t')
+        .ok_or(anyhow::anyhow!("Invalid line, no tab found"))?;
+    let event = serde_json::from_str::<AnalyticsEventName>(name).context("Invalid event name")?;
+    let data = serde_json::from_str::<serde_json::Value>(json).context("Invalid event data")?;
+
+    track_event(event, data, command, args, installation_id, user_id).await;
+
+    println!("tracked event data: {:?}", name);
+    Ok(())
+}
+
 pub async fn track_event(
     analytics_event_name: AnalyticsEventName,
     analytics_event_data: serde_json::Value,

crates/cli/src/commands/mod.rs

@@ -33,7 +33,9 @@ pub(crate) mod workflows_list;
 pub(crate) mod docgen;
 
 use crate::{
-    analytics::{is_telemetry_disabled, AnalyticsEvent, CompletedEvent, ErroredEvent},
+    analytics::{
+        is_telemetry_disabled, AnalyticsEvent, AnalyticsEventName, CompletedEvent, ErroredEvent,
+    },
     flags::{GlobalFormatFlags, OutputFormat},
     updater::Updater,
 };
@@ -235,15 +237,21 @@ fn write_analytics_event(
     analytics_worker: Option<&mut ChildStdin>,
     analytics_event: &AnalyticsEvent,
 ) {
-    let serialized_name = serde_json::to_string(&analytics_event);
+    let serialized_name = serde_json::to_string(&AnalyticsEventName::from(analytics_event));
     let serialized_event = serde_json::to_string(&analytics_event);
     match (analytics_worker, serialized_name, serialized_event) {
         (Some(analytics_worker), Ok(serialized_name), Ok(serialized_event)) => {
             let data = format!("{}\t{}\n", serialized_name, serialized_event);
-            let _ = analytics_worker.write_all(data.as_bytes());
+            let res = analytics_worker.write_all(data.as_bytes());
+            if let Err(e) = res {
+                println!("Failed to write to analytics worker: {:?}", e);
+            }
+        }
+        (None, _, _) => {
+            // No analytics worker to send event to, do nothing
         }
         (worker, name_err, event_err) => {
-            debug!(
+            println!(
                 "Failed to serialize analytics event: {:?} {:?} {:?}",
                 worker, name_err, event_err
             );
@@ -261,9 +269,12 @@ pub async fn run_command() -> Result<()> {
     let mut updater = Updater::from_current_bin().await?;
     updater.dump().await?;
 
+    println!("Running command: {:?}", app.command);
+
     let mut analytics_worker =
         match maybe_spawn_analytics_worker(&app.command, &analytics_args, &updater) {
             Err(_e) => {
+                println!("Failed to start the analytics worker process");
                 // We failed to start the analytics worker process
                 None
             }
@@ -360,6 +371,7 @@ pub async fn run_command() -> Result<()> {
     };
 
     write_analytics_event(analytics_worker.as_mut(), &final_analytics_event);
+    println!("DONE!!!");
 
     res
 }

crates/cli/src/commands/plumbing.rs

@@ -11,7 +11,7 @@ use std::io::{stdin, Read};
 use std::path::Path;
 use std::path::PathBuf;
 
-use crate::analytics::AnalyticsEventName;
+use crate::analytics::{track_event_line, AnalyticsEventName};
 use crate::flags::GlobalFormatFlags;
 use crate::lister::list_applyables;
 use crate::resolver::{get_grit_files_from, resolve_from, Source};
@@ -173,20 +173,20 @@ pub(crate) async fn run_plumbing(
         }
         PlumbingArgs::Analytics { args, shared_args } => {
             let buffer = read_input(&shared_args)?;
-            let events_to_send = buffer.lines().filter_map(|line| {
-                // Split line in name and JSON
-                let (name, json) = line.split_once('\t')?;
-
-                Some(track_event(
-                    serde_json::from_str::<AnalyticsEventName>(name).ok()?,
-                    serde_json::from_str::<serde_json::Value>(json).ok()?,
+            for line in buffer.lines() {
+                let result = track_event_line(
+                    line,
                     args.command.clone(),
                     args.args.clone(),
                     args.installation_id,
                     args.user_id.clone(),
-                ))
-            });
-            join_all(events_to_send).await;
+                )
+                .await;
+                if let Err(e) = result {
+                    eprintln!("Error when processing {}: {:#}", line, e);
+                }
+            }
+
             Ok(())
         }
         PlumbingArgs::Check { args, shared_args } => {