From 333787a4172c7122bd81a06d3bfe8337288efaea Mon Sep 17 00:00:00 2001 From: "brynary-fabro[bot]" <265161896+brynary-fabro[bot]@users.noreply.github.com> Date: Thu, 19 Mar 2026 12:30:02 -0400 Subject: [PATCH] JSONL analytics event file format (#100) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This PR switches the analytics telemetry file format from single-JSON-per-file to JSONL (one JSON event per line), enabling future batching of multiple events into a single file and subprocess. Each file is now named `fabro-events-{uuid}.jsonl` instead of `fabro-event-{uuid}.json`, and the Segment API endpoint is updated from `/v1/track` to `/v1/batch`. The core changes are in `sender.rs`: `send(Track)` becomes `emit(&[Track])`, which serializes each track as a compact JSON line; `send_to_segment()` becomes `upload()`, which reads the JSONL file, parses each line, injects `"type": "track"`, and POSTs the batch to Segment. A new pure function `build_segment_batch()` is extracted for testability, handling empty content, malformed lines (skipped with a warning), and blank lines gracefully. The panic sender remains unchanged as single-JSON-per-file. The call sites in `main.rs` are updated to use the new `emit`/`upload` signatures, and comprehensive tests are added covering the batch builder (empty, single, multiple, malformed, blank lines), the emit no-op paths, and the upload no-op behavior when `SEGMENT_WRITE_KEY` is unset. ### Fabro Details
Ran 10 stages in 17m 42s for $3.98 | Stage | Duration | Cost | Retries | |---|---|---|---| | start | 0s | – | 0 | | toolchain | 0s | – | 0 | | preflight_compile | 1m 11s | – | 0 | | preflight_lint | 12s | – | 0 | | implement | 3m 57s | $0.93 | 0 | | simplify_opus | 3m 5s | $0.92 | 0 | | simplify_gemini | 3m 56s | $1.17 | 0 | | simplify_gpt | 3m 26s | $0.97 | 0 | | verify | 1m 16s | – | 0 | | fmt | 1s | – | 0 | | **Total** | **17m 42s** | **$3.98** | **0** |
Ran ImplementAndSimplify.fabro (13 nodes and 16 edges) ```dot digraph ImplementAndSimplify { graph [ goal="Implement and simplify", model_stylesheet=" * { backend: api; model: claude-opus-4-6;} " ] rankdir=LR start [shape=Mdiamond, label="Start"] exit [shape=Msquare, label="Exit"] toolchain [label="Toolchain", shape=parallelogram, script="command -v cargo >/dev/null || { curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y && sudo ln -sf $HOME/.cargo/bin/* /usr/local/bin/; }; cargo --version 2>&1", max_retries=0] preflight_compile [label="Preflight Compile", shape=parallelogram, script="cargo check -q --workspace 2>&1", max_retries=0] preflight_lint [label="Preflight Lint", shape=parallelogram, script="cargo clippy -q --workspace -- -D warnings 2>&1", max_retries=0] fix_lints [label="Fix Lints", prompt="The preflight lint step failed. Read the build output from context and fix all clippy lint warnings.", max_visits=3] implement [label="Implement", prompt="Read the plan file referenced in the goal and implement every step. Make all the code changes described in the plan. Use red/green TDD."] simplify_opus [label="Simplify (Opus)", prompt="@prompts/simplify.md"] simplify_gemini [label="Simplify (Gemini)", prompt="@prompts/simplify.md", model="gemini-3.1-pro-preview-customtools"] simplify_gpt [label="Simplify (GPT-54)", prompt="@prompts/simplify.md", model="gpt-54"] verify [label="Verify", shape=parallelogram, script="cargo clippy -q --workspace -- -D warnings 2>&1 && cargo nextest run --cargo-quiet --workspace --status-level fail 2>&1", goal_gate=true, retry_target="fixup"] fixup [label="Fixup", prompt="The verify step failed. Read the build output from context and fix all clippy lint warnings and test failures.", max_visits=3] fmt [label="Format", shape=parallelogram, script="cargo fmt --all 2>&1", max_retries=0] start -> toolchain toolchain -> preflight_compile [condition="outcome=success"] toolchain -> exit preflight_compile -> preflight_lint [condition="outcome=success"] preflight_compile -> exit preflight_lint -> implement [condition="outcome=success"] preflight_lint -> fix_lints fix_lints -> preflight_lint implement -> simplify_opus -> simplify_gemini -> simplify_gpt -> verify verify -> fmt [condition="outcome=success"] verify -> fixup fixup -> verify fmt -> exit } ```
⚒️ Generated with [Fabro](https://fabro.sh) --------- Co-authored-by: Fabro --- lib/crates/fabro-cli/src/main.rs | 4 +- lib/crates/fabro-util/src/telemetry/sender.rs | 181 +++++++++++++++--- 2 files changed, 159 insertions(+), 26 deletions(-) diff --git a/lib/crates/fabro-cli/src/main.rs b/lib/crates/fabro-cli/src/main.rs index 103977da2..daf6e3dd6 100644 --- a/lib/crates/fabro-cli/src/main.rs +++ b/lib/crates/fabro-cli/src/main.rs @@ -426,7 +426,7 @@ fn send_telemetry_event( }); let track = telemetry.build_track(event_name, properties); - fabro_util::telemetry::sender::send(track); + fabro_util::telemetry::sender::emit(&[track]); debug!( event = event_name, subcommand = command_name, @@ -908,7 +908,7 @@ async fn main_inner() -> (String, Result<()>) { } }, Command::SendAnalytics { path } => { - let result = fabro_util::telemetry::sender::send_to_segment(&path).await; + let result = fabro_util::telemetry::sender::upload(&path).await; let _ = std::fs::remove_file(&path); result?; } diff --git a/lib/crates/fabro-util/src/telemetry/sender.rs b/lib/crates/fabro-util/src/telemetry/sender.rs index bf89ce795..48025cab4 100644 --- a/lib/crates/fabro-util/src/telemetry/sender.rs +++ b/lib/crates/fabro-util/src/telemetry/sender.rs @@ -2,52 +2,96 @@ use std::path::Path; use base64::engine::general_purpose::STANDARD; use base64::Engine; +use uuid::Uuid; use super::event::Track; -const SEGMENT_API_URL: &str = "https://api.segment.io/v1/track"; +const SEGMENT_API_URL: &str = "https://api.segment.io/v1/batch"; const SEGMENT_WRITE_KEY: Option<&str> = option_env!("SEGMENT_WRITE_KEY"); -/// Serializes the track event to a temp file and spawns a detached subprocess -/// (`fabro __send_analytics `) to deliver it. This ensures the event is -/// sent even if the parent CLI process exits immediately. +/// Serializes the track events as JSONL to a temp file and spawns a detached +/// subprocess (`fabro __send_analytics `) to deliver them. This ensures +/// the events are sent even if the parent CLI process exits immediately. /// -/// No-ops if the SEGMENT_WRITE_KEY was not set at compile time. -pub fn send(track: Track) { +/// No-ops if the SEGMENT_WRITE_KEY was not set at compile time or `tracks` is empty. +pub fn emit(tracks: &[Track]) { if SEGMENT_WRITE_KEY.is_none() { - tracing::debug!("telemetry: no SEGMENT_WRITE_KEY, skipping send"); + tracing::debug!("telemetry: no SEGMENT_WRITE_KEY, skipping emit"); return; } - spawn_sender(track); + if tracks.is_empty() { + return; + } + + spawn_sender(tracks); } -fn spawn_sender(track: Track) { - let json = match serde_json::to_vec(&track) { - Ok(j) => j, - Err(_) => return, - }; +fn spawn_sender(tracks: &[Track]) { + let lines: Vec = tracks + .iter() + .filter_map(|t| serde_json::to_string(t).ok()) + .collect(); - let filename = format!("fabro-event-{}.json", track.message_id); - super::spawn::spawn_fabro_subcommand("__send_analytics", &filename, &json); + if lines.is_empty() { + return; + } + + let jsonl = lines.join("\n"); + let filename = format!("fabro-events-{}.jsonl", Uuid::new_v4()); + super::spawn::spawn_fabro_subcommand("__send_analytics", &filename, jsonl.as_bytes()); } -/// Reads a serialized track event from `path` and sends it to Segment. +/// Parse JSONL content into a Segment batch payload. +/// +/// Each non-empty line is parsed as JSON, has `"type": "track"` injected, +/// and is collected into a `{"batch": [...]}` wrapper. +/// Returns `None` if no valid events are found. +fn build_segment_batch(content: &str) -> Option { + let mut batch = Vec::new(); + for line in content.lines() { + let line = line.trim(); + if line.is_empty() { + continue; + } + match serde_json::from_str::>(line) { + Ok(mut map) => { + map.insert("type".into(), "track".into()); + batch.push(serde_json::Value::Object(map)); + } + Err(err) => { + tracing::warn!(%err, "skipping malformed JSONL line"); + } + } + } + + if batch.is_empty() { + return None; + } + + Some(serde_json::json!({ "batch": batch })) +} + +/// Reads a JSONL file of serialized track events from `path` and sends them +/// to Segment as a batch. /// Called by the `__send_analytics` subcommand. /// No-ops if `SEGMENT_WRITE_KEY` was not set at compile time. -pub async fn send_to_segment(path: &Path) -> anyhow::Result<()> { +pub async fn upload(path: &Path) -> anyhow::Result<()> { let write_key = SEGMENT_WRITE_KEY .ok_or_else(|| anyhow::anyhow!("SEGMENT_WRITE_KEY not set at compile time"))?; - let json = std::fs::read(path)?; - let track: Track = serde_json::from_slice(&json)?; + let content = std::fs::read_to_string(path)?; + let payload = match build_segment_batch(&content) { + Some(p) => p, + None => return Ok(()), + }; let auth = STANDARD.encode(format!("{write_key}:")); let resp = reqwest::Client::new() .post(SEGMENT_API_URL) .header("Authorization", format!("Basic {auth}")) - .json(&track) + .json(&payload) .send() .await?; @@ -64,10 +108,82 @@ mod tests { use crate::telemetry::event::User; use serde_json::json; + // -- Step 1: build_segment_batch tests -- + #[test] - fn send_noops_without_write_key() { + fn build_segment_batch_empty_content() { + assert!(build_segment_batch("").is_none()); + } + + #[test] + fn build_segment_batch_single_event() { + let line = r#"{"anonymousId":"abc","event":"Test","properties":{},"messageId":"m1"}"#; + let result = build_segment_batch(line).unwrap(); + + let batch = result["batch"].as_array().unwrap(); + assert_eq!(batch.len(), 1); + assert_eq!(batch[0]["type"], "track"); + assert_eq!(batch[0]["event"], "Test"); + assert_eq!(batch[0]["anonymousId"], "abc"); + } + + #[test] + fn build_segment_batch_multiple_events() { + let content = concat!( + r#"{"anonymousId":"a","event":"E1","properties":{},"messageId":"m1"}"#, + "\n", + r#"{"anonymousId":"b","event":"E2","properties":{},"messageId":"m2"}"#, + ); + let result = build_segment_batch(content).unwrap(); + + let batch = result["batch"].as_array().unwrap(); + assert_eq!(batch.len(), 2); + assert_eq!(batch[0]["type"], "track"); + assert_eq!(batch[0]["event"], "E1"); + assert_eq!(batch[1]["type"], "track"); + assert_eq!(batch[1]["event"], "E2"); + } + + #[test] + fn build_segment_batch_skips_malformed_lines() { + let content = concat!( + r#"{"anonymousId":"a","event":"Good","properties":{},"messageId":"m1"}"#, + "\n", + "this is not json", + ); + let result = build_segment_batch(content).unwrap(); + + let batch = result["batch"].as_array().unwrap(); + assert_eq!(batch.len(), 1); + assert_eq!(batch[0]["event"], "Good"); + } + + #[test] + fn build_segment_batch_all_malformed() { + let content = "not json\nalso not json\n"; + assert!(build_segment_batch(content).is_none()); + } + + #[test] + fn build_segment_batch_skips_blank_lines() { + let content = concat!( + "\n", + r#"{"anonymousId":"a","event":"E1","properties":{},"messageId":"m1"}"#, + "\n", + "\n", + ); + let result = build_segment_batch(content).unwrap(); + + let batch = result["batch"].as_array().unwrap(); + assert_eq!(batch.len(), 1); + } + + // -- Step 2: emit() tests -- + + #[test] + fn emit_noops_without_write_key() { // SEGMENT_WRITE_KEY is not set at compile time in tests, - // so send() should return immediately without spawning. + // so emit() should return immediately without spawning. let track = Track { user: User::AnonymousId { anonymous_id: "test".to_string(), @@ -80,7 +196,24 @@ mod tests { }; // This should not panic or require a tokio runtime - // because it returns before reaching tokio::spawn - send(track); + // because it returns before reaching spawn + emit(&[track]); + } + + #[test] + fn emit_noops_with_empty_tracks() { + emit(&[]); + } + + // -- Step 3: upload() tests -- + + #[test] + fn upload_noops_without_write_key() { + // SEGMENT_WRITE_KEY is not set at compile time in tests, so this should error. + let rt = tokio::runtime::Runtime::new().unwrap(); + let result = rt.block_on(upload(Path::new("/nonexistent"))); + assert!(result.is_err()); + let err_msg = result.unwrap_err().to_string(); + assert!(err_msg.contains("SEGMENT_WRITE_KEY not set")); } }