attempt at doing the json first

jmarrero · jmarrero · commit 7c3f6bd66b16 · 2025-07-22T14:25:39.000-04:00
diff --git a/crates/lib/src/deploy.rs b/crates/lib/src/deploy.rs
@@ -21,7 +21,9 @@ use ostree_ext::ostree::{self, Sysroot};
 use ostree_ext::sysroot::SysrootLock;
 use ostree_ext::tokio_util::spawn_blocking_cancellable_flatten;
 
+use crate::progress_aggregator::ProgressAggregatorBuilder;
 use crate::progress_jsonl::{Event, ProgressWriter, SubTaskBytes, SubTaskStep};
+use crate::progress_renderer::ProgressFilter;
 use crate::spec::ImageReference;
 use crate::spec::{BootOrder, HostSpec};
 use crate::status::labels_of_config;
@@ -138,7 +140,7 @@ fn prefix_of_progress(p: &ImportProgress) -> &'static str {
     }
 }
 
-/// Write container fetch progress to standard output.
+/// Write container fetch progress using JSON-first architecture.
 async fn handle_layer_progress_print(
     mut layers: tokio::sync::mpsc::Receiver<ostree_container::store::ImportProgress>,
     mut layer_bytes: tokio::sync::watch::Receiver<Option<ostree_container::store::LayerProgress>>,
@@ -152,35 +154,23 @@ async fn handle_layer_progress_print(
 ) -> ProgressWriter {
     let start = std::time::Instant::now();
     let mut total_read = 0u64;
-    let bar = indicatif::MultiProgress::new();
-    if quiet {
-        bar.set_draw_target(indicatif::ProgressDrawTarget::hidden());
-    }
-    let layers_bar = bar.add(indicatif::ProgressBar::new(
-        n_layers_to_fetch.try_into().unwrap(),
-    ));
-    let byte_bar = bar.add(indicatif::ProgressBar::new(0));
-    // let byte_bar = indicatif::ProgressBar::new(0);
-    // byte_bar.set_draw_target(indicatif::ProgressDrawTarget::hidden());
-    layers_bar.set_style(
-        indicatif::ProgressStyle::default_bar()
-            .template("{prefix} {bar} {pos}/{len} {wide_msg}")
-            .unwrap(),
-    );
-    let taskname = "Fetching layers";
-    layers_bar.set_prefix(taskname);
-    layers_bar.set_message("");
-    byte_bar.set_prefix("Fetching");
-    byte_bar.set_style(
-        indicatif::ProgressStyle::default_bar()
-                .template(
-                    " └ {prefix} {bar} {binary_bytes}/{binary_total_bytes} ({binary_bytes_per_sec}) {wide_msg}",
-                )
-                .unwrap()
-        );
+
+    // Create JSON-first progress aggregator for pulling tasks
+    let visual_filter = if quiet {
+        None
+    } else {
+        Some(ProgressFilter::TasksMatching(vec!["pulling".to_string()]))
+    };
+
+    let mut aggregator = ProgressAggregatorBuilder::new()
+        .with_json(prog.clone())
+        .with_visual(visual_filter.unwrap_or(ProgressFilter::TasksMatching(vec![])))
+        .build();
 
     let mut subtasks = vec![];
     let mut subtask: SubTaskBytes = Default::default();
+    let mut current_layer_step = 0u64;
+
     loop {
         tokio::select! {
             // Always handle layer changes first.
@@ -192,12 +182,6 @@ async fn handle_layer_progress_print(
                     let short_digest = &layer.digest().digest()[0..21];
                     let layer_size = layer.size();
                     if l.is_starting() {
-                        // Reset the progress bar
-                        byte_bar.reset_elapsed();
-                        byte_bar.reset_eta();
-                        byte_bar.set_length(layer_size);
-                        byte_bar.set_message(format!("{layer_type} {short_digest}"));
-
                         subtask = SubTaskBytes {
                             subtask: layer_type.into(),
                             description: format!("{layer_type}: {short_digest}").clone().into(),
@@ -207,24 +191,26 @@ async fn handle_layer_progress_print(
                             bytes_total: layer_size,
                         };
                     } else {
-                        byte_bar.set_position(layer_size);
-                        layers_bar.inc(1);
                         total_read = total_read.saturating_add(layer_size);
+                        current_layer_step += 1;
                         // Emit an event where bytes == total to signal completion.
                         subtask.bytes = layer_size;
                         subtasks.push(subtask.clone());
-                        prog.send(Event::ProgressBytes {
+
+                        // Send progress event via JSON-first aggregator
+                        let event = Event::ProgressBytes {
                             task: "pulling".into(),
                             description: format!("Pulling Image: {digest}").into(),
                             id: (*digest).into(),
                             bytes_cached: bytes_total - bytes_to_download,
                             bytes: total_read,
                             bytes_total: bytes_to_download,
                             steps_cached: (layers_total - n_layers_to_fetch) as u64,
-                            steps: layers_bar.position(),
+                            steps: current_layer_step,
                             steps_total: n_layers_to_fetch as u64,
                             subtasks: subtasks.clone(),
-                        }).await;
+                        };
+                        let _ = aggregator.send_event(event).await;
                     }
                 } else {
                     // If the receiver is disconnected, then we're done
@@ -241,40 +227,42 @@ async fn handle_layer_progress_print(
                     bytes.as_ref().cloned()
                 };
                 if let Some(bytes) = bytes {
-                    byte_bar.set_position(bytes.fetched);
-                    subtask.bytes = byte_bar.position();
-                    prog.send_lossy(Event::ProgressBytes {
+                    subtask.bytes = bytes.fetched;
+
+                    // Send lossy progress event via JSON-first aggregator
+                    let event = Event::ProgressBytes {
                         task: "pulling".into(),
                         description: format!("Pulling Image: {digest}").into(),
                         id: (*digest).into(),
                         bytes_cached: bytes_total - bytes_to_download,
-                        bytes: total_read + byte_bar.position(),
+                        bytes: total_read + bytes.fetched,
                         bytes_total: bytes_to_download,
                         steps_cached: (layers_total - n_layers_to_fetch) as u64,
-                        steps: layers_bar.position(),
+                        steps: current_layer_step,
                         steps_total: n_layers_to_fetch as u64,
                         subtasks: subtasks.clone().into_iter().chain([subtask.clone()]).collect(),
-                    }).await;
+                    };
+                    let _ = aggregator.send_event(event).await;
                 }
             }
         }
     }
-    byte_bar.finish_and_clear();
-    layers_bar.finish_and_clear();
-    if let Err(e) = bar.clear() {
-        tracing::warn!("clearing bar: {e}");
-    }
+
+    // Finish progress aggregator
+    aggregator.finish();
+
     let end = std::time::Instant::now();
     let elapsed = end.duration_since(start);
     let persec = total_read as f64 / elapsed.as_secs_f64();
     let persec = indicatif::HumanBytes(persec as u64);
-    if let Err(e) = bar.println(&format!(
-        "Fetched layers: {} in {} ({}/s)",
-        indicatif::HumanBytes(total_read),
-        indicatif::HumanDuration(elapsed),
-        persec,
-    )) {
-        tracing::warn!("writing to stdout: {e}");
+
+    if !quiet {
+        println!(
+            "Fetched layers: {} in {} ({}/s)",
+            indicatif::HumanBytes(total_read),
+            indicatif::HumanDuration(elapsed),
+            persec,
+        );
     }
 
     // Since the progress notifier closed, we know import has started
diff --git a/crates/lib/src/lib.rs b/crates/lib/src/lib.rs
@@ -19,7 +19,9 @@ mod lints;
 mod lsm;
 pub(crate) mod metadata;
 mod podman;
+mod progress_aggregator;
 mod progress_jsonl;
+mod progress_renderer;
 mod reboot;
 mod reexec;
 pub mod spec;
diff --git a/crates/lib/src/progress_aggregator.rs b/crates/lib/src/progress_aggregator.rs
@@ -0,0 +1,162 @@
+//! Progress aggregator that implements JSON-first progress architecture
+//! All progress flows through JSON events before being rendered or output
+
+use anyhow::Result;
+use std::io::IsTerminal;
+
+use crate::progress_jsonl::{Event, ProgressWriter};
+use crate::progress_renderer::{ProgressFilter, ProgressRenderer};
+
+/// Unified progress system that emits JSON events and optionally renders them visually
+/// This implements the JSON-first architecture where indicatif becomes a consumer of JSON events
+pub struct ProgressAggregator {
+    json_writer: Option<ProgressWriter>,
+    renderer: Option<ProgressRenderer>,
+}
+
+impl ProgressAggregator {
+    /// Create a new progress aggregator
+    pub fn new(json_writer: Option<ProgressWriter>, visual_filter: Option<ProgressFilter>) -> Self {
+        let renderer = if std::io::stderr().is_terminal() && visual_filter.is_some() {
+            Some(ProgressRenderer::new(visual_filter.unwrap()))
+        } else {
+            None
+        };
+
+        Self {
+            json_writer,
+            renderer,
+        }
+    }
+
+    /// Send a progress event - this is the core method that implements JSON-first architecture
+    pub async fn send_event(&mut self, event: Event<'_>) -> Result<()> {
+        // 1. Always emit JSON first (if enabled)
+        if let Some(ref writer) = self.json_writer {
+            writer.send_lossy(event.clone()).await;
+        }
+
+        // 2. Then render visually (if enabled)
+        if let Some(ref mut renderer) = self.renderer {
+            renderer.handle_event(&event)?;
+        }
+
+        Ok(())
+    }
+
+    /// Finish all progress and clean up
+    pub fn finish(&mut self) {
+        if let Some(ref mut renderer) = self.renderer {
+            renderer.finish();
+        }
+    }
+}
+
+/// Helper to create progress aggregators for common use cases
+pub struct ProgressAggregatorBuilder {
+    json_writer: Option<ProgressWriter>,
+    visual_filter: Option<ProgressFilter>,
+}
+
+impl ProgressAggregatorBuilder {
+    pub fn new() -> Self {
+        Self {
+            json_writer: None,
+            visual_filter: None,
+        }
+    }
+
+    /// Enable JSON output to the given writer
+    pub fn with_json(mut self, writer: ProgressWriter) -> Self {
+        self.json_writer = Some(writer);
+        self
+    }
+
+    /// Enable visual progress with the given filter
+    pub fn with_visual(mut self, filter: ProgressFilter) -> Self {
+        self.visual_filter = Some(filter);
+        self
+    }
+
+    /// Build the aggregator
+    pub fn build(self) -> ProgressAggregator {
+        ProgressAggregator::new(self.json_writer, self.visual_filter)
+    }
+}
+
+impl Default for ProgressAggregatorBuilder {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::progress_jsonl::Event;
+    use std::borrow::Cow;
+
+    #[tokio::test]
+    async fn test_json_first_architecture() -> Result<()> {
+        // Create an aggregator that outputs both JSON and visual progress
+        let mut aggregator = ProgressAggregatorBuilder::new()
+            .with_visual(ProgressFilter::All)
+            .build();
+
+        // Send a progress event
+        let event = Event::ProgressBytes {
+            task: Cow::Borrowed("test"),
+            description: Cow::Borrowed("Testing progress"),
+            id: Cow::Borrowed("test-id"),
+            bytes_cached: 0,
+            bytes: 50,
+            bytes_total: 100,
+            steps_cached: 0,
+            steps: 1,
+            steps_total: 2,
+            subtasks: vec![],
+        };
+
+        aggregator.send_event(event).await?;
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_task_filtering() -> Result<()> {
+        // Create an aggregator that only shows "pulling" tasks
+        let mut aggregator = ProgressAggregatorBuilder::new()
+            .for_tasks(vec!["pulling".to_string()])
+            .build();
+
+        // Send a pulling event (should be shown)
+        let pulling_event = Event::ProgressBytes {
+            task: Cow::Borrowed("pulling"),
+            description: Cow::Borrowed("Pulling image"),
+            id: Cow::Borrowed("image:latest"),
+            bytes_cached: 0,
+            bytes: 25,
+            bytes_total: 100,
+            steps_cached: 0,
+            steps: 1,
+            steps_total: 3,
+            subtasks: vec![],
+        };
+
+        // Send an installing event (should be filtered out visually)
+        let installing_event = Event::ProgressSteps {
+            task: Cow::Borrowed("installing"),
+            description: Cow::Borrowed("Installing package"),
+            id: Cow::Borrowed("package"),
+            steps_cached: 0,
+            steps: 1,
+            steps_total: 5,
+            subtasks: vec![],
+        };
+
+        aggregator.send_event(pulling_event).await?;
+        aggregator.send_event(installing_event).await?;
+
+        Ok(())
+    }
+}
diff --git a/crates/lib/src/progress_renderer.rs b/crates/lib/src/progress_renderer.rs
