statsig_rust/event_logging_adapter/
statsig_http_event_logging_adapter.rs

1use crate::compression::compression_helper::{compress_data, get_compression_format};
2use crate::event_logging_adapter::EventLoggingAdapter;
3use crate::log_event_payload::LogEventRequest;
4use crate::networking::{NetworkClient, NetworkError, RequestArgs};
5use crate::statsig_metadata::StatsigMetadata;
6use crate::{log_d, StatsigErr, StatsigRuntime};
7use async_trait::async_trait;
8use serde::Deserialize;
9use serde_json::json;
10use std::collections::HashMap;
11use std::sync::Arc;
12
13const DEFAULT_LOG_EVENT_URL: &str = "https://prodregistryv2.org/v1/log_event";
14
15#[derive(Deserialize)]
16struct LogEventResult {
17    success: Option<bool>,
18}
19
20const TAG: &str = stringify!(StatsigHttpEventLoggingAdapter);
21
22pub struct StatsigHttpEventLoggingAdapter {
23    log_event_url: String,
24    network: NetworkClient,
25}
26
27impl StatsigHttpEventLoggingAdapter {
28    #[must_use]
29    pub fn new(
30        sdk_key: &str,
31        log_event_url: Option<&String>,
32        disable_network: Option<bool>,
33    ) -> Self {
34        let headers = StatsigMetadata::get_constant_request_headers(sdk_key);
35
36        let log_event_url = match log_event_url {
37            Some(u) => u,
38            _ => DEFAULT_LOG_EVENT_URL,
39        }
40        .to_string();
41        Self {
42            log_event_url,
43            network: NetworkClient::new(sdk_key, Some(headers), disable_network),
44        }
45    }
46
47    pub async fn send_events_over_http(
48        &self,
49        request: &LogEventRequest,
50    ) -> Result<bool, StatsigErr> {
51        log_d!(
52            TAG,
53            "Logging Events ({}): {}",
54            &request.event_count,
55            json!(&request.payload).to_string()
56        );
57
58        let compression_format = get_compression_format();
59
60        // Set headers
61        let headers = HashMap::from([
62            (
63                "statsig-event-count".to_string(),
64                request.event_count.to_string(),
65            ),
66            (
67                "statsig-retry-count".to_string(),
68                request.retries.to_string(),
69            ),
70            ("Content-Encoding".to_owned(), compression_format.clone()),
71            ("Content-Type".to_owned(), "application/json".to_owned()),
72        ]);
73
74        // Compress data before sending it
75        let bytes = serde_json::to_vec(&request.payload)
76            .map_err(|e| StatsigErr::SerializationError(e.to_string()))?;
77        let compressed = match compress_data(&bytes) {
78            Ok(c) => c,
79            Err(e) => return Err(e),
80        };
81
82        // Make request
83        let response = self
84            .network
85            .post(
86                RequestArgs {
87                    url: self.log_event_url.clone(),
88                    headers: Some(headers),
89                    accept_gzip_response: true,
90                    ..RequestArgs::new()
91                },
92                Some(compressed),
93            )
94            .await
95            .map_err(|err| StatsigErr::NetworkError(err, Some("Log event failure".into())))?;
96        let response_slice = match response.data {
97            Some(data) => data,
98            None => {
99                return Err(StatsigErr::NetworkError(
100                    NetworkError::RequestFailed,
101                    Some("Empty response from network".to_string()),
102                ));
103            }
104        };
105
106        serde_json::from_slice::<LogEventResult>(&response_slice)
107            .map(|result| result.success != Some(false))
108            .map_err(|e| {
109                StatsigErr::JsonParseError(stringify!(LogEventResult).to_string(), e.to_string())
110            })
111    }
112}
113
114#[async_trait]
115impl EventLoggingAdapter for StatsigHttpEventLoggingAdapter {
116    async fn start(&self, _statsig_runtime: &Arc<StatsigRuntime>) -> Result<(), StatsigErr> {
117        Ok(())
118    }
119
120    #[allow(clippy::manual_inspect)]
121    async fn log_events(&self, request: LogEventRequest) -> Result<bool, StatsigErr> {
122        match self.send_events_over_http(&request).await {
123            Ok(_) => Ok(true),
124            Err(StatsigErr::NetworkError(NetworkError::DisableNetworkOn, _)) => Ok(false),
125            Err(e) => Err(e),
126        } //TODO: surface retryable code status for the logger to know if it should put back into the queue
127    }
128
129    async fn shutdown(&self) -> Result<(), StatsigErr> {
130        Ok(())
131    }
132
133    fn should_schedule_background_flush(&self) -> bool {
134        true
135    }
136}
137
138#[cfg(not(feature = "with_zstd"))]
139#[tokio::test]
140async fn test_event_logging() {
141    use crate::log_event_payload::{LogEventPayload, LogEventRequest};
142    use std::env;
143
144    let sdk_key = env::var("test_api_key").expect("test_api_key environment variable not set");
145
146    let adapter = StatsigHttpEventLoggingAdapter::new(&sdk_key, None, None);
147
148    let payload_str = r#"{"events":[{"eventName":"statsig::config_exposure","metadata":{"config":"running_exp_in_unlayered_with_holdout","ruleID":"5suobe8yyvznqasn9Ph1dI"},"secondaryExposures":[{"gate":"global_holdout","gateValue":"false","ruleID":"3QoA4ncNdVGBaMt3N1KYjz:0.50:1"},{"gate":"exp_holdout","gateValue":"false","ruleID":"1rEqLOpCROaRafv7ubGgax"}],"time":1722386636538,"user":{"appVersion":null,"country":null,"custom":null,"customIDs":null,"email":"daniel@statsig.com","ip":null,"locale":null,"privateAttributes":null,"statsigEnvironment":null,"userAgent":null,"userID":"a-user"},"value":null}],"statsigMetadata":{"sdk_type":"statsig-server-core","sdk_version":"0.0.1"}}"#;
149    let payload = serde_json::from_str::<LogEventPayload>(payload_str).unwrap();
150
151    let request = LogEventRequest {
152        payload,
153        event_count: 1,
154        retries: 0,
155    };
156
157    let result = adapter.log_events(request).await;
158
159    assert!(result.is_ok(), "Error logging events: {:?}", result.err());
160}
statsig_rust/event_logging_adapter/statsig_http_event_logging_adapter.rs

statsig_rust/event_logging_adapter/
statsig_http_event_logging_adapter.rs
