feat: .cat --follow builtin for xs exec (#130)

cablehead · web-flow · commit b1c16e614847 · 2025-10-17T16:19:14.000-04:00
diff --git a/src/api.rs b/src/api.rs
@@ -607,7 +607,7 @@ async fn handle_exec(store: &Store, body: hyper::body::Incoming) -> HTTPResult {
     // Add context-specific commands
     engine
         .add_commands(vec![
-            Box::new(nu::commands::cat_command::CatCommand::new(
+            Box::new(nu::commands::cat_stream_command::CatStreamCommand::new(
                 store.clone(),
                 context_id,
             )),
@@ -794,10 +794,12 @@ mod tests {
         // Add context-specific commands
         engine
             .add_commands(vec![
-                Box::new(crate::nu::commands::cat_command::CatCommand::new(
-                    store.clone(),
-                    context_id,
-                )),
+                Box::new(
+                    crate::nu::commands::cat_stream_command::CatStreamCommand::new(
+                        store.clone(),
+                        context_id,
+                    ),
+                ),
                 Box::new(crate::nu::commands::head_command::HeadCommand::new(
                     store.clone(),
                     context_id,
diff --git a/src/nu/commands/cat_stream_command.rs b/src/nu/commands/cat_stream_command.rs
@@ -0,0 +1,151 @@
+use nu_engine::CallExt;
+use nu_protocol::engine::{Call, Command, EngineState, Stack};
+use nu_protocol::{
+    Category, ListStream, PipelineData, ShellError, Signature, SyntaxShape, Type, Value,
+};
+use std::time::Duration;
+
+use crate::store::{FollowOption, ReadOptions, Store};
+
+#[derive(Clone)]
+pub struct CatStreamCommand {
+    store: Store,
+    context_id: scru128::Scru128Id,
+}
+
+impl CatStreamCommand {
+    pub fn new(store: Store, context_id: scru128::Scru128Id) -> Self {
+        Self { store, context_id }
+    }
+}
+
+impl Command for CatStreamCommand {
+    fn name(&self) -> &str {
+        ".cat"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build(".cat")
+            .input_output_types(vec![(Type::Nothing, Type::Any)])
+            .switch("follow", "long poll for new events", Some('f'))
+            .named(
+                "pulse",
+                SyntaxShape::Int,
+                "interval in ms for synthetic xs.pulse events",
+                Some('p'),
+            )
+            .switch("tail", "start at end of stream", Some('t'))
+            .switch("detail", "include all frame fields", Some('d'))
+            .switch("all", "read across all contexts", Some('a'))
+            .named(
+                "limit",
+                SyntaxShape::Int,
+                "limit the number of frames to retrieve",
+                None,
+            )
+            .named(
+                "last-id",
+                SyntaxShape::String,
+                "start from a specific frame ID",
+                None,
+            )
+            .named("topic", SyntaxShape::String, "filter by topic", Some('T'))
+            .category(Category::Experimental)
+    }
+
+    fn description(&self) -> &str {
+        "Reads the event stream and returns frames (streaming version)"
+    }
+
+    fn run(
+        &self,
+        engine_state: &EngineState,
+        stack: &mut Stack,
+        call: &Call,
+        _input: PipelineData,
+    ) -> Result<PipelineData, ShellError> {
+        let follow = call.has_flag(engine_state, stack, "follow")?;
+        let pulse: Option<i64> = call.get_flag(engine_state, stack, "pulse")?;
+        let tail = call.has_flag(engine_state, stack, "tail")?;
+        let detail = call.has_flag(engine_state, stack, "detail")?;
+        let all = call.has_flag(engine_state, stack, "all")?;
+        let limit: Option<i64> = call.get_flag(engine_state, stack, "limit")?;
+        let last_id: Option<String> = call.get_flag(engine_state, stack, "last-id")?;
+        let topic: Option<String> = call.get_flag(engine_state, stack, "topic")?;
+
+        // Parse last_id
+        let last_id: Option<scru128::Scru128Id> = last_id
+            .as_deref()
+            .map(|s| {
+                s.parse().map_err(|e| ShellError::GenericError {
+                    error: "Invalid last-id".into(),
+                    msg: format!("Failed to parse Scru128Id: {e}"),
+                    span: Some(call.head),
+                    help: None,
+                    inner: vec![],
+                })
+            })
+            .transpose()?;
+
+        // For non-follow mode, always use async path for consistency
+        // The store.read() will handle topic filtering correctly
+
+        // Build ReadOptions for async path (follow mode or no topic filter)
+        let options = ReadOptions::builder()
+            .follow(if let Some(pulse_ms) = pulse {
+                FollowOption::WithHeartbeat(Duration::from_millis(pulse_ms as u64))
+            } else if follow {
+                FollowOption::On
+            } else {
+                FollowOption::Off
+            })
+            .tail(tail)
+            .maybe_last_id(last_id)
+            .maybe_limit(limit.map(|l| l as usize))
+            .maybe_context_id(if all { None } else { Some(self.context_id) })
+            .maybe_topic(topic.clone())
+            .build();
+
+        let store = self.store.clone();
+        let span = call.head;
+        let signals = engine_state.signals().clone();
+
+        // Create channel for async -> sync bridge
+        let (tx, rx) = std::sync::mpsc::channel();
+
+        // Spawn thread to handle async store.read()
+        std::thread::spawn(move || {
+            let rt = tokio::runtime::Runtime::new().expect("Failed to create tokio runtime");
+            rt.block_on(async move {
+                let mut receiver = store.read(options).await;
+
+                while let Some(frame) = receiver.recv().await {
+                    // Convert frame to Nu value
+                    let mut value = crate::nu::util::frame_to_value(&frame, span);
+
+                    // Filter fields if not --detail
+                    if !detail {
+                        value = match value {
+                            Value::Record { val, .. } => {
+                                let mut filtered = val.into_owned();
+                                filtered.remove("context_id");
+                                filtered.remove("ttl");
+                                Value::record(filtered, span)
+                            }
+                            v => v,
+                        };
+                    }
+
+                    if tx.send(value).is_err() {
+                        break;
+                    }
+                }
+            });
+        });
+
+        // Create ListStream from channel
+        let stream = ListStream::new(std::iter::from_fn(move || rx.recv().ok()), span, signals);
+
+        Ok(PipelineData::ListStream(stream, None))
+    }
+}
diff --git a/src/nu/commands/mod.rs b/src/nu/commands/mod.rs
@@ -2,6 +2,7 @@ pub mod append_command;
 pub mod append_command_buffered;
 pub mod cas_command;
 pub mod cat_command;
+pub mod cat_stream_command;
 pub mod get_command;
 pub mod head_command;
 pub mod remove_command;
diff --git a/src/nu/util.rs b/src/nu/util.rs
@@ -49,6 +49,16 @@ pub fn frame_to_value(frame: &Frame, span: Span) -> Value {
         record.push("meta", json_to_value(meta, span));
     }
 
+    if let Some(ttl) = &frame.ttl {
+        let ttl_str = match ttl {
+            crate::store::TTL::Forever => "forever".to_string(),
+            crate::store::TTL::Ephemeral => "ephemeral".to_string(),
+            crate::store::TTL::Time(duration) => format!("{}s", duration.as_secs()),
+            crate::store::TTL::Head(n) => format!("head:{}", n),
+        };
+        record.push("ttl", Value::string(ttl_str, span));
+    }
+
     Value::record(record, span)
 }
 
diff --git a/tests/integration.rs b/tests/integration.rs
@@ -415,6 +415,167 @@ async fn test_exec_bytestream_behavior() {
     assert!(!output_str.starts_with('{')); // Not a JSON object
 }
 
+#[tokio::test]
+async fn test_exec_cat_streaming() {
+    let temp_dir = TempDir::new().expect("Failed to create temp dir");
+    let store_path = temp_dir.path();
+
+    let mut child = spawn_xs_supervisor(store_path).await;
+
+    let sock_path = store_path.join("sock");
+    let start = std::time::Instant::now();
+    while !sock_path.exists() {
+        if start.elapsed() > Duration::from_secs(5) {
+            panic!("Timeout waiting for sock file");
+        }
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+    tokio::time::sleep(Duration::from_millis(500)).await;
+
+    // Append initial test data
+    cmd!(cargo_bin("xs"), "append", store_path, "stream.test")
+        .stdin_bytes(b"initial")
+        .run()
+        .unwrap();
+
+    // Test 1: .cat without --follow (snapshot mode)
+    let output = cmd!(
+        cargo_bin("xs"),
+        "exec",
+        store_path,
+        ".cat --topic stream.test"
+    )
+    .read()
+    .unwrap();
+
+    let frames: Vec<serde_json::Value> = output
+        .lines()
+        .map(|l| serde_json::from_str(l).unwrap())
+        .collect();
+    assert_eq!(frames.len(), 1);
+    assert_eq!(frames[0]["topic"], "stream.test");
+
+    // Test 2: .cat --follow streams new frames
+    let mut follow_child = tokio::process::Command::new(cargo_bin("xs"))
+        .arg("exec")
+        .arg(store_path)
+        .arg(".cat --topic stream.test --follow")
+        .stdout(std::process::Stdio::piped())
+        .spawn()
+        .unwrap();
+
+    let stdout = follow_child.stdout.take().unwrap();
+    let mut reader = tokio::io::BufReader::new(stdout);
+
+    // Read initial frame (historical)
+    let mut line = String::new();
+    let result = tokio::time::timeout(Duration::from_secs(1), reader.read_line(&mut line))
+        .await
+        .expect("Timeout reading initial frame")
+        .expect("Failed to read initial frame");
+    assert!(result > 0, "Should read initial frame");
+    let initial_frame: serde_json::Value = serde_json::from_str(&line.trim()).unwrap();
+    assert_eq!(initial_frame["topic"], "stream.test");
+
+    // Read threshold frame (indicates caught up to real-time)
+    line.clear();
+    let result = tokio::time::timeout(Duration::from_secs(1), reader.read_line(&mut line))
+        .await
+        .expect("Timeout reading threshold")
+        .expect("Failed to read threshold");
+    assert!(result > 0, "Should read threshold frame");
+    let threshold_frame: serde_json::Value = serde_json::from_str(&line.trim()).unwrap();
+    assert_eq!(
+        threshold_frame["topic"], "xs.threshold",
+        "Should receive threshold frame indicating caught up"
+    );
+
+    // Append new frame while following
+    cmd!(cargo_bin("xs"), "append", store_path, "stream.test")
+        .stdin_bytes(b"streamed")
+        .run()
+        .unwrap();
+
+    // Should receive new frame via streaming
+    line.clear();
+    let result = tokio::time::timeout(Duration::from_secs(2), reader.read_line(&mut line))
+        .await
+        .expect("Timeout reading streamed frame")
+        .expect("Failed to read streamed frame");
+    assert!(result > 0, "Should read streamed frame");
+    let streamed_frame: serde_json::Value = serde_json::from_str(&line.trim()).unwrap();
+    assert_eq!(streamed_frame["topic"], "stream.test");
+
+    // Test 3: .cat --tail starts at end (skip for now - can block)
+    follow_child.kill().await.unwrap();
+
+    // Test 4: .cat --limit respects limit
+    let output = cmd!(
+        cargo_bin("xs"),
+        "exec",
+        store_path,
+        ".cat --topic stream.test --limit 1"
+    )
+    .read()
+    .unwrap();
+
+    eprintln!(
+        "Output from .cat --topic stream.test --limit 1: '{}'",
+        output
+    );
+    assert!(!output.trim().is_empty(), "Output should not be empty");
+
+    let frames: Vec<serde_json::Value> = output
+        .lines()
+        .filter(|l| !l.is_empty())
+        .map(|l| serde_json::from_str(l).expect(&format!("Failed to parse JSON: {}", l)))
+        .collect();
+    assert_eq!(frames.len(), 1, "Should respect --limit flag");
+
+    // Test 5: .cat --detail includes context_id and ttl
+    let output = cmd!(
+        cargo_bin("xs"),
+        "exec",
+        store_path,
+        ".cat --topic stream.test --limit 1 --detail"
+    )
+    .read()
+    .unwrap();
+
+    let frame: serde_json::Value = serde_json::from_str(output.lines().next().unwrap()).unwrap();
+    assert!(
+        frame.get("context_id").is_some(),
+        "Should include context_id with --detail"
+    );
+    assert!(
+        frame.get("ttl").is_some(),
+        "Should include ttl with --detail"
+    );
+
+    // Test 6: Without --detail, context_id and ttl are filtered
+    let output = cmd!(
+        cargo_bin("xs"),
+        "exec",
+        store_path,
+        ".cat --topic stream.test --limit 1"
+    )
+    .read()
+    .unwrap();
+
+    let frame: serde_json::Value = serde_json::from_str(output.lines().next().unwrap()).unwrap();
+    assert!(
+        frame.get("context_id").is_none(),
+        "Should not include context_id without --detail"
+    );
+    assert!(
+        frame.get("ttl").is_none(),
+        "Should not include ttl without --detail"
+    );
+
+    // Clean up
+    child.kill().await.unwrap();
+}
+
 #[tokio::test]
 async fn test_iroh_networking() {
     let temp_dir = TempDir::new().expect("Failed to create temp dir");
