diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index 65e8b9ed14..e4328e31bc 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -1,5 +1,3 @@ -use crate::event_sink::AnalyticsEvent; -use crate::event_sink::AnalyticsEventSink; use crate::events::AppServerRpcTransport; use crate::events::GuardianReviewAnalyticsResult; use crate::events::GuardianReviewTrackContext; @@ -22,9 +20,11 @@ use crate::facts::TurnCodexErrorFact; use crate::facts::TurnProfileFact; use crate::facts::TurnResolvedConfigFact; use crate::facts::TurnTokenUsageFact; -use crate::local_sink::SharedLocalAnalyticsSink; -use crate::local_sink::local_analytics_sink_from_env; use crate::reducer::AnalyticsReducer; +use crate::sinks::AnalyticsEvent; +use crate::sinks::AnalyticsEventSink; +use crate::sinks::SharedLocalAnalyticsSink; +use crate::sinks::local_analytics_sink_from_env; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::InitializeParams; @@ -42,7 +42,7 @@ use std::sync::Mutex; use tokio::sync::mpsc; #[cfg(test)] -use crate::local_sink::local_analytics_sink_for_path; +use crate::sinks::local_analytics_sink_for_path; #[cfg(test)] use std::path::PathBuf; diff --git a/codex-rs/analytics/src/lib.rs b/codex-rs/analytics/src/lib.rs index 233b0442e2..07be168dae 100644 --- a/codex-rs/analytics/src/lib.rs +++ b/codex-rs/analytics/src/lib.rs @@ -1,10 +1,9 @@ mod accepted_lines; mod client; -mod event_sink; mod events; mod facts; -mod local_sink; mod reducer; +mod sinks; use std::time::SystemTime; use std::time::UNIX_EPOCH; @@ -50,9 +49,9 @@ pub use facts::TurnSteerRequestError; pub use facts::TurnSteerResult; pub use facts::TurnTokenUsageFact; pub use facts::build_track_events_context; -pub use local_sink::LOCAL_ANALYTICS_SCHEMA_VERSION; -pub use local_sink::LocalAnalyticsRecord; -pub use local_sink::LocalAnalyticsRecordType; +pub use sinks::LOCAL_ANALYTICS_SCHEMA_VERSION; +pub use sinks::LocalAnalyticsRecord; +pub use sinks::LocalAnalyticsRecordType; #[cfg(test)] mod analytics_client_tests; diff --git a/codex-rs/analytics/src/sinks/codex_backend.rs b/codex-rs/analytics/src/sinks/codex_backend.rs new file mode 100644 index 0000000000..1fc587cbd6 --- /dev/null +++ b/codex-rs/analytics/src/sinks/codex_backend.rs @@ -0,0 +1,87 @@ +use crate::events::TrackEventRequest; +use crate::events::TrackEventsRequest; +use codex_login::AuthManager; +use codex_login::CodexAuth; +use codex_login::default_client::create_client; +use std::time::Duration; + +const ANALYTICS_EVENTS_TIMEOUT: Duration = Duration::from_secs(10); + +pub(super) async fn write( + auth_manager: &AuthManager, + base_url: &str, + events: &[&TrackEventRequest], +) { + if events.is_empty() { + return; + } + + let Some(auth) = auth_manager.auth().await else { + return; + }; + if !auth.uses_codex_backend() { + return; + } + + let base_url = base_url.trim_end_matches('/'); + let url = format!("{base_url}/codex/analytics-events/events"); + for events in track_event_request_batches(events) { + send_track_events_request(&auth, &url, events).await; + } +} + +fn track_event_request_batches<'a>( + events: &'a [&'a TrackEventRequest], +) -> Vec<&'a [&'a TrackEventRequest]> { + let mut batches = Vec::new(); + let mut current_batch_start = 0; + + for (index, event) in events.iter().enumerate() { + if event.should_send_in_isolated_request() { + if current_batch_start < index { + batches.push(&events[current_batch_start..index]); + } + batches.push(&events[index..=index]); + current_batch_start = index + 1; + } + } + + if current_batch_start < events.len() { + batches.push(&events[current_batch_start..]); + } + + batches +} + +async fn send_track_events_request(auth: &CodexAuth, url: &str, events: &[&TrackEventRequest]) { + if events.is_empty() { + return; + } + + let payload = TrackEventsRequest { events }; + + let response = create_client() + .post(url) + .timeout(ANALYTICS_EVENTS_TIMEOUT) + .headers(codex_model_provider::auth_provider_from_auth(auth).to_auth_headers()) + .header("Content-Type", "application/json") + .json(&payload) + .send() + .await; + + match response { + Ok(response) if response.status().is_success() => {} + Ok(response) => { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + tracing::warn!("events failed with status {status}: {body}"); + } + Err(err) => { + tracing::warn!("failed to send events request: {err}"); + } + } +} + +#[cfg(test)] +#[path = "codex_backend_tests.rs"] +mod tests; diff --git a/codex-rs/analytics/src/event_sink_tests.rs b/codex-rs/analytics/src/sinks/codex_backend_tests.rs similarity index 100% rename from codex-rs/analytics/src/event_sink_tests.rs rename to codex-rs/analytics/src/sinks/codex_backend_tests.rs diff --git a/codex-rs/analytics/src/local_sink.rs b/codex-rs/analytics/src/sinks/local.rs similarity index 99% rename from codex-rs/analytics/src/local_sink.rs rename to codex-rs/analytics/src/sinks/local.rs index 2cb51feb0d..35f0eb69df 100644 --- a/codex-rs/analytics/src/local_sink.rs +++ b/codex-rs/analytics/src/sinks/local.rs @@ -149,5 +149,5 @@ fn string_field(value: Option<&JsonValue>, field: &str) -> Option { } #[cfg(test)] -#[path = "local_sink_tests.rs"] +#[path = "local_tests.rs"] mod tests; diff --git a/codex-rs/analytics/src/local_sink_tests.rs b/codex-rs/analytics/src/sinks/local_tests.rs similarity index 100% rename from codex-rs/analytics/src/local_sink_tests.rs rename to codex-rs/analytics/src/sinks/local_tests.rs diff --git a/codex-rs/analytics/src/event_sink.rs b/codex-rs/analytics/src/sinks/mod.rs similarity index 51% rename from codex-rs/analytics/src/event_sink.rs rename to codex-rs/analytics/src/sinks/mod.rs index 874e811816..8850b99b8a 100644 --- a/codex-rs/analytics/src/event_sink.rs +++ b/codex-rs/analytics/src/sinks/mod.rs @@ -1,14 +1,18 @@ -use crate::events::TrackEventRequest; -use crate::events::TrackEventsRequest; -use crate::local_sink::SharedLocalAnalyticsSink; -use crate::local_sink::append_codex_analytics_event_best_effort; -use codex_login::AuthManager; -use codex_login::CodexAuth; -use codex_login::default_client::create_client; -use std::sync::Arc; -use std::time::Duration; +mod codex_backend; +mod local; + +use crate::events::TrackEventRequest; +use codex_login::AuthManager; +use std::sync::Arc; + +pub use local::LOCAL_ANALYTICS_SCHEMA_VERSION; +pub use local::LocalAnalyticsRecord; +pub use local::LocalAnalyticsRecordType; +pub(crate) use local::SharedLocalAnalyticsSink; +#[cfg(test)] +pub(crate) use local::local_analytics_sink_for_path; +pub(crate) use local::local_analytics_sink_from_env; -const ANALYTICS_EVENTS_TIMEOUT: Duration = Duration::from_secs(10); const CODEX_ANALYTICS_EVENT_SINKS: AnalyticsEventSinkSet = AnalyticsEventSinkSet::CODEX_BACKEND.union(AnalyticsEventSinkSet::LOCAL); @@ -50,13 +54,13 @@ impl AnalyticsEventSink { AnalyticsEvent::CodexAnalytics(event) => event, }) .collect::>(); - send_track_events(auth_manager, base_url, &events).await; + codex_backend::write(auth_manager, base_url, &events).await; } Self::Local(sink) => { for event in writable_events(events, self.kind()) { match event { AnalyticsEvent::CodexAnalytics(event) => { - append_codex_analytics_event_best_effort(sink, event); + local::append_codex_analytics_event_best_effort(sink, event); } } } @@ -111,82 +115,3 @@ fn writable_events( .iter() .filter(move |event| event.writable_sinks().contains(sink)) } - -async fn send_track_events( - auth_manager: &AuthManager, - base_url: &str, - events: &[&TrackEventRequest], -) { - if events.is_empty() { - return; - } - - let Some(auth) = auth_manager.auth().await else { - return; - }; - if !auth.uses_codex_backend() { - return; - } - - let base_url = base_url.trim_end_matches('/'); - let url = format!("{base_url}/codex/analytics-events/events"); - for events in track_event_request_batches(events) { - send_track_events_request(&auth, &url, events).await; - } -} - -fn track_event_request_batches<'a>( - events: &'a [&'a TrackEventRequest], -) -> Vec<&'a [&'a TrackEventRequest]> { - let mut batches = Vec::new(); - let mut current_batch_start = 0; - - for (index, event) in events.iter().enumerate() { - if event.should_send_in_isolated_request() { - if current_batch_start < index { - batches.push(&events[current_batch_start..index]); - } - batches.push(&events[index..=index]); - current_batch_start = index + 1; - } - } - - if current_batch_start < events.len() { - batches.push(&events[current_batch_start..]); - } - - batches -} - -async fn send_track_events_request(auth: &CodexAuth, url: &str, events: &[&TrackEventRequest]) { - if events.is_empty() { - return; - } - - let payload = TrackEventsRequest { events }; - - let response = create_client() - .post(url) - .timeout(ANALYTICS_EVENTS_TIMEOUT) - .headers(codex_model_provider::auth_provider_from_auth(auth).to_auth_headers()) - .header("Content-Type", "application/json") - .json(&payload) - .send() - .await; - - match response { - Ok(response) if response.status().is_success() => {} - Ok(response) => { - let status = response.status(); - let body = response.text().await.unwrap_or_default(); - tracing::warn!("events failed with status {status}: {body}"); - } - Err(err) => { - tracing::warn!("failed to send events request: {err}"); - } - } -} - -#[cfg(test)] -#[path = "event_sink_tests.rs"] -mod tests;