mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-09-06 08:18:58 +00:00
JSONL analytics event file format (#100)
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
<details>
<summary>Ran 10 stages in 17m 42s for $3.98</summary>
| 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** |
</details>
<details>
<summary>Ran <code>ImplementAndSimplify.fabro</code> (13 nodes and 16
edges)</summary>
```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
}
```
</details>
⚒️ Generated with [Fabro](https://fabro.sh)
---------
Co-authored-by: Fabro <noreply@fabro.sh>
This commit is contained in:
brynary-fabro[bot]
GitHub
parent
d5bbc12261
commit
333787a417
2 changed files with 159 additions and 26 deletions
|
|
@ -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?;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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 <path>`) 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 <path>`) 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<String> = 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<serde_json::Value> {
|
||||
let mut batch = Vec::new();
|
||||
for line in content.lines() {
|
||||
let line = line.trim();
|
||||
if line.is_empty() {
|
||||
continue;
|
||||
}
|
||||
match serde_json::from_str::<serde_json::Map<String, serde_json::Value>>(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"));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue