diff --git a/crates/trapfall-core/src/store.rs b/crates/trapfall-core/src/store.rs index e3d1f61..e2f692e 100644 --- a/crates/trapfall-core/src/store.rs +++ b/crates/trapfall-core/src/store.rs @@ -133,6 +133,14 @@ impl Store { // ── Transactions ────────────────────────────────────────────────────── + pub async fn insert_transaction( + &self, + project_id: &str, + transaction: &trapfall_proto::Transaction, + ) -> Result { + self.db.insert_transaction(project_id, transaction).await + } + pub async fn list_transactions( &self, project_id: &str, diff --git a/crates/trapfalld/src/server.rs b/crates/trapfalld/src/server.rs index ca80c9f..530203b 100644 --- a/crates/trapfalld/src/server.rs +++ b/crates/trapfalld/src/server.rs @@ -409,10 +409,12 @@ async fn ingest_envelope( return StatusCode::BAD_REQUEST; } }; - // TODO(#237): Persist transactions — for now, acknowledge and log. - // Transaction-only envelopes return 200 OK but data is not yet stored. - if !parsed.transactions.is_empty() { - tracing::info!("Skipping {} transactions (not yet persisted, see #237)", parsed.transactions.len()); + // Persist transactions + for txn in &parsed.transactions { + match store.insert_transaction(&project.id, txn).await { + Ok(id) => tracing::info!(id = %id, name = %txn.transaction, "Stored transaction"), + Err(e) => tracing::warn!(error = %e, name = %txn.transaction, "Failed to insert transaction"), + } } // Persist session aggregates diff --git a/crates/trapfalld/tests/integration.rs b/crates/trapfalld/tests/integration.rs index 9c06e5c..67a4a01 100644 --- a/crates/trapfalld/tests/integration.rs +++ b/crates/trapfalld/tests/integration.rs @@ -37,6 +37,13 @@ fn make_envelope_body(exception_type: &str, message: &str) -> Vec { format!("{envelope_header}\n{item_header}\n{event_json}").into_bytes() } +fn make_transaction_envelope_body() -> Vec { + let envelope_header = r#"{"event_id":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}"#; + let item_header = r#"{"type":"transaction","length":0}"#; + let txn_json = r#"{"event_id":"cccccccccccccccccccccccccccccccccccc","level":"info","transaction":"GET /api/health","start_timestamp":1703894474.296,"timestamp":1703894474.891,"release":"1.0.0","environment":"production"}"#; + format!("{envelope_header}\n{item_header}\n{txn_json}").into_bytes() +} + fn make_state(store: Store, rate_limiter: RateLimiter) -> AppState { let (tx, rx) = tokio::sync::mpsc::channel(100); // Keep rx alive by spawning a dummy consumer @@ -737,3 +744,74 @@ async fn change_password_weak_new_returns_400() { let resp = app.oneshot(req).await.unwrap(); assert_eq!(resp.status(), StatusCode::BAD_REQUEST); } + +#[tokio::test] +async fn ingest_persists_transactions_and_returns_via_api() { + let store = test_store().await; + let project_id = seed_project(&store).await; + let state = make_state(store.clone(), RateLimiter::default()); + let app = router(state); + + // Send a transaction-only envelope + let body = make_transaction_envelope_body(); + let req = Request::builder() + .method("POST") + .uri(format!("/api/{project_id}/envelope/")) + .header("content-type", "application/octet-stream") + .header("authorization", "Bearer abc123") + .body(Body::from(body)) + .unwrap(); + + let resp = app.oneshot(req).await.unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + + // Verify transaction is stored + let count = store.count_transactions(&project_id).await.unwrap(); + assert_eq!(count, 1, "expected 1 transaction after ingest"); + + let rows = store.list_transactions(&project_id, 10, 0).await.unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].name, "GET /api/health"); + assert_eq!(rows[0].release.as_deref(), Some("1.0.0")); + assert_eq!(rows[0].environment.as_deref(), Some("production")); +} + +#[tokio::test] +async fn store_insert_transaction_directly() { + let store = test_store().await; + let project_id = seed_project(&store).await; + + let txn = trapfall_proto::Transaction { + event_id: "e-direct".into(), + level: trapfall_proto::Level::Info, + transaction: "GET /api/direct".into(), + start_timestamp: 1703894474.296, + timestamp: 1703894474.891, + release: Some("2.0.0".into()), + environment: Some("staging".into()), + spans: vec![], + contexts: None, + request: None, + tags: None, + extra: None, + }; + + let id = store.insert_transaction(&project_id, &txn).await.unwrap(); + assert!(!id.is_empty(), "insert_transaction should return an ID"); + + let count = store.count_transactions(&project_id).await.unwrap(); + assert_eq!(count, 1, "direct insert should persist"); + + let rows = store.list_transactions(&project_id, 10, 0).await.unwrap(); + assert_eq!(rows[0].name, "GET /api/direct"); +} + +#[tokio::test] +async fn parse_transaction_envelope() { + let body = make_transaction_envelope_body(); + let parsed = trapfall_ingest::parse_envelope(&body, None); + assert!(parsed.is_ok(), "envelope should parse: {:?}", parsed.err()); + let result = parsed.unwrap(); + assert_eq!(result.transactions.len(), 1, "expected 1 transaction parsed, got {}", result.transactions.len()); + assert_eq!(result.transactions[0].transaction, "GET /api/health"); +}