statsig_rust/event_logging_adapter/
statsig_http_event_logging_adapter.rs1use 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, StatsigOptions, 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(sdk_key: &str, options: Option<&StatsigOptions>) -> Self {
30 let headers = StatsigMetadata::get_constant_request_headers(
31 sdk_key,
32 options.and_then(|opts| opts.service_name.as_deref()),
33 );
34
35 let log_event_url = options
36 .and_then(|opts| opts.log_event_url.as_ref())
37 .map(|u| u.to_string())
38 .unwrap_or_else(|| DEFAULT_LOG_EVENT_URL.to_string());
39
40 Self {
41 log_event_url,
42 network: NetworkClient::new(sdk_key, Some(headers), options),
43 }
44 }
45
46 pub async fn send_events_over_http(&self, request: &LogEventRequest) -> Result<(), StatsigErr> {
47 log_d!(
48 TAG,
49 "Logging Events ({}): {}",
50 &request.event_count,
51 json!(&request.payload).to_string()
52 );
53
54 let compression_format = get_compression_format();
55
56 let headers = HashMap::from([
58 (
59 "statsig-event-count".to_string(),
60 request.event_count.to_string(),
61 ),
62 (
63 "statsig-retry-count".to_string(),
64 request.retries.to_string(),
65 ),
66 ("Content-Encoding".to_owned(), compression_format.clone()),
67 ("Content-Type".to_owned(), "application/json".to_owned()),
68 ]);
69
70 let bytes = serde_json::to_vec(&request.payload)
72 .map_err(|e| StatsigErr::SerializationError(e.to_string()))?;
73 let compressed = match compress_data(&bytes) {
74 Ok(c) => c,
75 Err(e) => return Err(e),
76 };
77
78 let response = self
80 .network
81 .post(
82 RequestArgs {
83 url: self.log_event_url.clone(),
84 headers: Some(headers),
85 accept_gzip_response: true,
86 ..RequestArgs::new()
87 },
88 Some(compressed),
89 )
90 .await
91 .map_err(StatsigErr::NetworkError)?;
92
93 let mut res_data = match response.data {
94 Some(data) => data,
95 None => {
96 return Err(StatsigErr::NetworkError(NetworkError::RequestFailed(
97 self.log_event_url.clone(),
98 response.status_code,
99 "Empty response from network".to_string(),
100 )));
101 }
102 };
103
104 let result = res_data
105 .deserialize_into::<LogEventResult>()
106 .map(|result| result.success != Some(false))
107 .map_err(|e| {
108 StatsigErr::JsonParseError(stringify!(LogEventResult).to_string(), e.to_string())
109 })?;
110
111 if result {
112 Ok(())
113 } else {
114 Err(StatsigErr::LogEventError(
115 "Unsuccessful response from network".into(),
116 ))
117 }
118 }
119}
120
121#[async_trait]
122impl EventLoggingAdapter for StatsigHttpEventLoggingAdapter {
123 async fn start(&self, _statsig_runtime: &Arc<StatsigRuntime>) -> Result<(), StatsigErr> {
124 Ok(())
125 }
126
127 async fn log_events(&self, request: LogEventRequest) -> Result<bool, StatsigErr> {
128 match self.send_events_over_http(&request).await {
129 Ok(()) => Ok(true),
130 Err(e) => Err(e),
131 }
132 }
133
134 async fn shutdown(&self) -> Result<(), StatsigErr> {
135 Ok(())
136 }
137
138 fn should_schedule_background_flush(&self) -> bool {
139 true
140 }
141}
142
143#[cfg(not(feature = "with_zstd"))]
144#[tokio::test]
145async fn test_event_logging() {
146 use crate::log_event_payload::{LogEventPayload, LogEventRequest};
147 use std::env;
148
149 let sdk_key = env::var("test_api_key").expect("test_api_key environment variable not set");
150
151 let adapter = StatsigHttpEventLoggingAdapter::new(&sdk_key, None);
152
153 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"}}"#;
154 let payload = serde_json::from_str::<LogEventPayload>(payload_str).unwrap();
155
156 let request = LogEventRequest {
157 payload,
158 event_count: 1,
159 retries: 0,
160 };
161
162 let result = adapter.log_events(request).await;
163
164 assert!(result.is_ok(), "Error logging events: {:?}", result.err());
165}