posthog_rs/client/
async_client.rs

1use std::collections::{HashMap, HashSet};
2#[cfg(feature = "error-tracking")]
3use std::error::Error as StdError;
4use std::sync::{Arc, OnceLock};
5use std::time::Duration;
6
7#[cfg(feature = "capture-v1")]
8use chrono::Utc;
9use reqwest::header::USER_AGENT;
10use reqwest::{header::CONTENT_TYPE, Client as HttpClient};
11use serde_json::json;
12use tracing::{debug, instrument, trace, warn};
13#[cfg(feature = "capture-v1")]
14use uuid::Uuid;
15
16use super::get_default_user_agent;
17use crate::endpoints::Endpoint;
18#[cfg(feature = "error-tracking")]
19use crate::error_tracking::{build_exception_event, CaptureExceptionOptions};
20#[cfg(not(feature = "capture-v1"))]
21use crate::event::InnerEvent;
22#[cfg(feature = "capture-v1")]
23use crate::event_v1::CaptureResponse;
24use crate::feature_flag_evaluations::{
25    EvaluateFlagsOptions, EvaluatedFlagRecord, FeatureFlagEvaluations, FeatureFlagEvaluationsHost,
26    FlagCalledEventParams,
27};
28use crate::feature_flags::{match_feature_flag, FeatureFlag, FeatureFlagsResponse, FlagValue};
29use crate::local_evaluation::{AsyncFlagPoller, FlagCache, LocalEvaluationConfig, LocalEvaluator};
30use crate::{Error, Event};
31
32#[cfg(feature = "capture-v1")]
33use super::common::apply_capture_defaults;
34use super::common::{
35    already_reported, apply_before_send_hooks, build_dedup_key, extract_flag_details,
36    flag_called_event, flag_event_dedup_cache, local_record, remote_record_from_detail,
37    DetailedFlagsResponse, FlagEventDedupCache,
38};
39use super::{BeforeSendHook, ClientOptions};
40
41#[cfg(not(feature = "capture-v1"))]
42async fn check_response(response: reqwest::Response) -> Result<(), Error> {
43    let status = response.status().as_u16();
44    let body = response
45        .text()
46        .await
47        .unwrap_or_else(|_| "Unknown error".to_string());
48
49    match Error::from_http_response(status, body) {
50        Some(err) => Err(err),
51        None => Ok(()),
52    }
53}
54
55/// A [`Client`] facilitates interactions with the PostHog API over HTTP.
56pub struct Client {
57    options: ClientOptions,
58    client: HttpClient,
59    local_evaluator: Option<LocalEvaluator>,
60    _flag_poller: Option<AsyncFlagPoller>,
61    flag_event_host: OnceLock<Arc<dyn FeatureFlagEvaluationsHost>>,
62}
63
64/// Implementation of [`FeatureFlagEvaluationsHost`] that emits dedup-aware
65/// `$feature_flag_called` events through a clone of the async [`Client`]'s
66/// HTTP transport. The event ship is fire-and-forget: errors are logged at
67/// `debug` level but do not surface to the caller, matching the JS SDK.
68///
69/// With `capture-v1`, events ship to the V1 endpoint (single attempt);
70/// otherwise the legacy v0 `/i/v0/e/` path.
71struct AsyncFlagEventHost {
72    http_client: HttpClient,
73    options: ClientOptions,
74    capture_url: String,
75    // Read by the v0 ship path only; unused under capture-v1, where the
76    // flag-event path does not currently apply before_send hooks.
77    #[cfg_attr(feature = "capture-v1", allow(dead_code))]
78    before_send: Vec<BeforeSendHook>,
79    dedup_cache: FlagEventDedupCache,
80    /// Tokio runtime handle captured at host construction (which always runs
81    /// inside the runtime that hosts `evaluate_flags`). This lets snapshot
82    /// access methods spawn `$feature_flag_called` shipping from any thread —
83    /// including ones without an entered runtime — by routing through the
84    /// captured handle instead of the free `tokio::spawn` (which would panic).
85    runtime: tokio::runtime::Handle,
86}
87
88impl AsyncFlagEventHost {
89    fn from_options(options: &ClientOptions, http_client: HttpClient) -> Self {
90        #[cfg(feature = "capture-v1")]
91        let capture_url = options
92            .endpoints()
93            .build_custom_url(super::v1_capture::V1_CAPTURE_PATH);
94        #[cfg(not(feature = "capture-v1"))]
95        let capture_url = options.endpoints().build_url(Endpoint::Capture);
96        Self {
97            http_client,
98            options: options.clone(),
99            capture_url,
100            before_send: options.before_send.clone(),
101            dedup_cache: flag_event_dedup_cache(),
102            runtime: tokio::runtime::Handle::current(),
103        }
104    }
105
106    fn spawn_ship(&self, event: Event) {
107        if self.options.is_disabled() {
108            return;
109        }
110        #[cfg(feature = "capture-v1")]
111        self.spawn_ship_v1(event);
112        #[cfg(not(feature = "capture-v1"))]
113        self.spawn_ship_v0(event);
114    }
115
116    /// Single attempt, no retries — matches v0 flag-event semantics: losses
117    /// aren't worth retry traffic and shipping must never slow flag reads.
118    #[cfg(feature = "capture-v1")]
119    fn spawn_ship_v1(&self, event: Event) {
120        let (headers, body) =
121            match super::v1_capture::build_flag_event_request(&self.options, &event) {
122                Ok(parts) => parts,
123                Err(e) => {
124                    debug!(error = %e, "failed to serialize $feature_flag_called event");
125                    return;
126                }
127            };
128        let http_client = self.http_client.clone();
129        let url = self.capture_url.clone();
130        self.runtime.spawn(async move {
131            match http_client
132                .post(&url)
133                .headers(headers)
134                .body(body)
135                .send()
136                .await
137            {
138                Ok(resp) => {
139                    let status = resp.status().as_u16();
140                    if !(200..=299).contains(&status) {
141                        let message = resp
142                            .text()
143                            .await
144                            .unwrap_or_else(|_| "Unknown error".to_string());
145                        debug!(
146                            status,
147                            "$feature_flag_called event rejected by server: {message}"
148                        );
149                    }
150                }
151                Err(send_err) => {
152                    let message = send_err.to_string();
153                    debug!("failed to send $feature_flag_called event: {message}");
154                }
155            }
156        });
157    }
158
159    #[cfg(not(feature = "capture-v1"))]
160    fn spawn_ship_v0(&self, mut event: Event) {
161        event.prepare_for_v0();
162        let Some(event) = apply_before_send_hooks(&self.before_send, event) else {
163            return;
164        };
165        let inner_event = InnerEvent::new(event, self.options.api_key.clone());
166        let payload = match serde_json::to_string(&inner_event) {
167            Ok(p) => p,
168            Err(e) => {
169                debug!(error = %e, "failed to serialize $feature_flag_called event");
170                return;
171            }
172        };
173        let http_client = self.http_client.clone();
174        let url = self.capture_url.clone();
175        self.runtime.spawn(async move {
176            let response = match http_client
177                .post(&url)
178                .header(CONTENT_TYPE, "application/json")
179                .header(USER_AGENT, get_default_user_agent())
180                .body(payload)
181                .send()
182                .await
183            {
184                Ok(r) => r,
185                Err(send_err) => {
186                    let message = send_err.to_string();
187                    debug!("failed to send $feature_flag_called event: {message}");
188                    return;
189                }
190            };
191            if let Err(check_err) = check_response(response).await {
192                let message = check_err.to_string();
193                debug!("$feature_flag_called event rejected by server: {message}");
194            }
195        });
196    }
197}
198
199impl FeatureFlagEvaluationsHost for AsyncFlagEventHost {
200    fn capture_flag_called_event_if_needed(&self, params: FlagCalledEventParams) {
201        let dedup_key = build_dedup_key(&params.key, params.response.as_ref(), &params.groups);
202        if already_reported(&self.dedup_cache, &params.distinct_id, &dedup_key) {
203            return;
204        }
205
206        if let Some(event) =
207            flag_called_event(params, self.options.disable_geoip, self.options.is_server)
208        {
209            self.spawn_ship(event);
210        }
211    }
212
213    fn log_warning(&self, message: &str) {
214        // Surface filter-helper misuse via tracing — users can silence these
215        // with their tracing-subscriber level filter (e.g. `posthog_rs=error`).
216        warn!("{message}");
217    }
218}
219
220/// Construct an async PostHog client from an API key or [`ClientOptions`].
221///
222/// # Parameters
223///
224/// - `options`: Either a project API key (for example `"phc_..."`) or a
225///   configured [`ClientOptions`] value.
226///
227/// # Returns
228///
229/// A [`Client`] that performs capture and feature flag requests asynchronously.
230///
231/// # Remarks
232///
233/// This constructor is available with the default `async-client` feature and
234/// must be awaited. Passing a blank API key creates a disabled client.
235pub async fn client<C: Into<ClientOptions>>(options: C) -> Client {
236    let options = options.into().sanitize();
237    let client = HttpClient::builder()
238        .timeout(Duration::from_secs(options.request_timeout_seconds))
239        .build()
240        .unwrap(); // Unwrap here is as safe as `HttpClient::new`
241
242    let (local_evaluator, flag_poller) = if options.enable_local_evaluation
243        && !options.is_disabled()
244    {
245        if let Some(ref personal_key) = options.personal_api_key {
246            let cache = FlagCache::new();
247
248            let config = LocalEvaluationConfig {
249                personal_api_key: personal_key.clone(),
250                project_api_key: options.api_key.clone(),
251                api_host: options.endpoints().api_host(),
252                poll_interval: Duration::from_secs(options.poll_interval_seconds),
253                request_timeout: Duration::from_secs(options.request_timeout_seconds),
254            };
255
256            let mut poller = AsyncFlagPoller::new(config, cache.clone());
257            poller.start().await;
258
259            (Some(LocalEvaluator::new(cache)), Some(poller))
260        } else {
261            warn!("Local evaluation enabled but personal_api_key not set, falling back to API evaluation");
262            (None, None)
263        }
264    } else {
265        (None, None)
266    };
267
268    Client {
269        options,
270        client,
271        local_evaluator,
272        _flag_poller: flag_poller,
273        flag_event_host: OnceLock::new(),
274    }
275}
276
277impl Client {
278    /// Capture the provided event, sending it to PostHog.
279    ///
280    /// # Parameters
281    ///
282    /// - `event`: Event name, distinct ID, properties, timestamp, groups, and
283    ///   optional feature flag state to send.
284    ///
285    /// # Errors
286    ///
287    /// Returns [`Error::Serialization`] if the event cannot be serialized,
288    /// [`Error::Connection`] for transport or unexpected HTTP failures,
289    /// [`Error::RateLimit`] for HTTP 429, [`Error::BadRequest`] for HTTP 400 or
290    /// 413, and [`Error::ServerError`] for HTTP 5xx.
291    ///
292    /// # Remarks
293    ///
294    /// Disabled clients skip the request and return `Ok(())`.
295    #[instrument(skip(self, event), level = "debug")]
296    pub async fn capture(&self, event: Event) -> Result<(), Error> {
297        if self.options.is_disabled() {
298            trace!("Client is disabled, skipping capture");
299            return Ok(());
300        }
301
302        #[cfg(feature = "capture-v1")]
303        {
304            let mut event = event;
305            let defaults = self.options.capture_defaults();
306            apply_capture_defaults(&mut event, &defaults);
307            let Some(event) = apply_before_send_hooks(&self.options.before_send, event) else {
308                return Ok(());
309            };
310            return self.capture_v1(vec![event], false).await.map(|_| ());
311        }
312
313        #[cfg(not(feature = "capture-v1"))]
314        self.capture_v0(event).await
315    }
316
317    /// Capture a Rust error personlessly, sending it to PostHog Error Tracking.
318    ///
319    /// The error's type, message, and full `source()` chain are sent as
320    /// `$exception_list`, with a stacktrace of the capture site attached to
321    /// the first entry (see `ErrorTrackingOptions::capture_stacktrace`).
322    ///
323    /// Accepts any [`std::error::Error`], including `&dyn Error`. A
324    /// `Box<dyn Error>` does not implement `Error` itself, so pass the
325    /// dereferenced trait object: `capture_exception(&*boxed)`.
326    ///
327    /// To associate the exception with a person or attach custom properties,
328    /// groups, a fingerprint, or a severity level, use
329    /// [`Client::capture_exception_with`].
330    ///
331    /// # Examples
332    ///
333    /// ```no_run
334    /// # async fn example() -> Result<(), posthog_rs::Error> {
335    /// let client = posthog_rs::client("phc_project_api_key").await;
336    /// let error = std::io::Error::other("checkout failed");
337    ///
338    /// client.capture_exception(&error).await?;
339    /// # Ok(())
340    /// # }
341    /// ```
342    #[cfg(feature = "error-tracking")]
343    pub async fn capture_exception<E>(&self, error: &E) -> Result<(), Error>
344    where
345        E: StdError + ?Sized,
346    {
347        self.capture_exception_with(error, CaptureExceptionOptions::default())
348            .await
349    }
350
351    /// Capture a Rust error with optional context, sending it to PostHog
352    /// Error Tracking.
353    ///
354    /// Set [`CaptureExceptionOptions::distinct_id`] to associate the exception
355    /// with a person; without it the exception is captured personlessly.
356    ///
357    /// # Examples
358    ///
359    /// ```no_run
360    /// # async fn example() -> Result<(), posthog_rs::Error> {
361    /// use posthog_rs::CaptureExceptionOptions;
362    ///
363    /// let client = posthog_rs::client("phc_project_api_key").await;
364    /// let error = std::io::Error::other("checkout failed");
365    ///
366    /// client
367    ///     .capture_exception_with(
368    ///         &error,
369    ///         CaptureExceptionOptions::new()
370    ///             .distinct_id("user-123")
371    ///             .property("route", "/checkout")?,
372    ///     )
373    ///     .await?;
374    /// # Ok(())
375    /// # }
376    /// ```
377    #[cfg(feature = "error-tracking")]
378    pub async fn capture_exception_with<E>(
379        &self,
380        error: &E,
381        options: CaptureExceptionOptions,
382    ) -> Result<(), Error>
383    where
384        E: StdError + ?Sized,
385    {
386        if self.options.is_disabled() {
387            trace!("Client is disabled, skipping exception capture");
388            return Ok(());
389        }
390
391        self.capture(build_exception_event(
392            error,
393            options,
394            self.options.error_tracking(),
395        )?)
396        .await
397    }
398
399    /// Capture a collection of events with a single request.
400    ///
401    /// Events are sent to the `/batch/` endpoint.
402    ///
403    /// # Parameters
404    ///
405    /// - `events`: Events to send in the batch.
406    /// - `historical_migration`: Set to `true` to route events to the
407    ///   historical ingestion topic, bypassing the main pipeline.
408    ///
409    /// # Errors
410    ///
411    /// Returns the same error categories as [`Client::capture`].
412    pub async fn capture_batch(
413        &self,
414        events: Vec<Event>,
415        historical_migration: bool,
416    ) -> Result<(), Error> {
417        if self.options.is_disabled() {
418            return Ok(());
419        }
420        if events.is_empty() {
421            return Ok(());
422        }
423
424        #[cfg(feature = "capture-v1")]
425        {
426            let defaults = self.options.capture_defaults();
427            let events: Vec<_> = events
428                .into_iter()
429                .filter_map(|mut event| {
430                    apply_capture_defaults(&mut event, &defaults);
431                    apply_before_send_hooks(&self.options.before_send, event)
432                })
433                .collect();
434            if events.is_empty() {
435                return Ok(());
436            }
437            return self
438                .capture_v1(events, historical_migration)
439                .await
440                .map(|_| ());
441        }
442
443        #[cfg(not(feature = "capture-v1"))]
444        self.capture_batch_v0(events, historical_migration).await
445    }
446
447    #[cfg(not(feature = "capture-v1"))]
448    async fn capture_v0(&self, mut event: Event) -> Result<(), Error> {
449        let defaults = self.options.capture_defaults();
450        super::v0_capture::prepare_event(&mut event, &defaults);
451        let Some(event) = apply_before_send_hooks(&self.options.before_send, event) else {
452            return Ok(());
453        };
454        let payload =
455            super::v0_capture::build_capture_payload(event, self.options.api_key.clone())?;
456        let url = self.options.endpoints().build_url(Endpoint::Capture);
457        let (body, encoding) = super::v0_capture::encode_body(&self.options, payload);
458        self.send_v0_with_retry(&url, body, encoding).await
459    }
460
461    #[cfg(not(feature = "capture-v1"))]
462    async fn capture_batch_v0(
463        &self,
464        events: Vec<Event>,
465        historical_migration: bool,
466    ) -> Result<(), Error> {
467        let defaults = self.options.capture_defaults();
468        let Some(payload) = super::v0_capture::build_batch_payload(
469            events,
470            self.options.api_key.clone(),
471            historical_migration,
472            &defaults,
473            &self.options.before_send,
474        )?
475        else {
476            return Ok(());
477        };
478        let url = self.options.endpoints().build_url(Endpoint::Batch);
479        let (body, encoding) = super::v0_capture::encode_body(&self.options, payload);
480        self.send_v0_with_retry(&url, body, encoding).await
481    }
482
483    /// POST `body` to `url`, retrying transient failures (transport errors and
484    /// 408/429/500/502/503/504) up to `max_capture_attempts`. The body is built
485    /// once by the caller and resent byte-for-byte, so a retried event keeps
486    /// its UUID and timestamp — which dedup relies on. The retry decision is the
487    /// shared sans-IO logic in [`super::retry`]; this loop is just the transport.
488    /// When `encoding` is `Some`, the request advertises that `Content-Encoding`
489    /// and a matching `compression=<token>` query param (capture reads the query
490    /// param on v0, not the header).
491    #[cfg(not(feature = "capture-v1"))]
492    async fn send_v0_with_retry(
493        &self,
494        url: &str,
495        body: Vec<u8>,
496        encoding: Option<&'static str>,
497    ) -> Result<(), Error> {
498        use super::retry::{v0_after_response, v0_after_transport_error, Step};
499
500        // v0 capture/batch URLs carry no query string, so capture reads the
501        // compression hint from this param (it does not consult Content-Encoding).
502        let url = match encoding {
503            Some(token) => format!("{url}?compression={token}"),
504            None => url.to_string(),
505        };
506        let mut attempt: u32 = 1;
507        loop {
508            let mut request = self
509                .client
510                .post(&url)
511                .header(CONTENT_TYPE, "application/json")
512                .header(USER_AGENT, get_default_user_agent())
513                .body(body.clone());
514            if let Some(token) = encoding {
515                request = request.header(reqwest::header::CONTENT_ENCODING, token);
516            }
517            let request = super::v0_capture::apply_extra_headers(&self.options, request);
518
519            let step = match request.send().await {
520                Err(e) => v0_after_transport_error(&self.options, attempt, e.to_string()),
521                Ok(response) => {
522                    let status = response.status().as_u16();
523                    let retry_after = super::retry::parse_retry_after(response.headers());
524                    let body = response
525                        .text()
526                        .await
527                        .unwrap_or_else(|_| "Unknown error".to_string());
528                    v0_after_response(&self.options, attempt, status, retry_after, &body)
529                }
530            };
531
532            match step {
533                Step::Done => return Ok(()),
534                Step::Fail(e) => return Err(e),
535                Step::Backoff(delay) => {
536                    tokio::time::sleep(delay).await;
537                    attempt += 1;
538                }
539            }
540        }
541    }
542
543    #[cfg(feature = "capture-v1")]
544    async fn capture_v1(
545        &self,
546        events: Vec<Event>,
547        historical_migration: bool,
548    ) -> Result<CaptureResponse, Error> {
549        use super::v1_capture::{self, Step};
550        use crate::event_v1::V1BatchRequestRef;
551
552        let request_id = Uuid::now_v7();
553        let created_at = Utc::now().to_rfc3339();
554        let mut attempt: u32 = 1;
555        let defaults = self.options.capture_defaults();
556        let mut pending = v1_capture::build_events(&events, &defaults);
557        let mut final_results = HashMap::new();
558        let historical_migration = historical_migration.then_some(true);
559        let url = self
560            .options
561            .endpoints()
562            .build_custom_url(v1_capture::V1_CAPTURE_PATH);
563
564        loop {
565            let req = V1BatchRequestRef {
566                created_at: &created_at,
567                historical_migration,
568                batch: &pending,
569            };
570            let payload =
571                serde_json::to_vec(&req).map_err(|e| Error::Serialization(e.to_string()))?;
572            let mut headers = v1_capture::build_headers(&self.options, &request_id, attempt);
573            let body =
574                v1_capture::maybe_compress(self.options.capture_compression, &mut headers, payload);
575
576            let step = match self
577                .client
578                .post(&url)
579                .headers(headers)
580                .body(body)
581                .send()
582                .await
583            {
584                Err(e) => v1_capture::after_transport_error(
585                    &self.options,
586                    &request_id,
587                    attempt,
588                    e.to_string(),
589                ),
590                Ok(resp) => {
591                    let status = resp.status().as_u16();
592                    let retry_after = v1_capture::parse_retry_after(resp.headers());
593                    let text = resp
594                        .text()
595                        .await
596                        .unwrap_or_else(|_| "Unknown error".to_string());
597                    v1_capture::after_response(
598                        &self.options,
599                        &request_id,
600                        attempt,
601                        status,
602                        retry_after,
603                        &text,
604                        &mut pending,
605                        &mut final_results,
606                    )
607                }
608            };
609
610            match step {
611                Step::Done => break,
612                Step::Fail(e) => return Err(e),
613                Step::Backoff(d) => {
614                    attempt += 1;
615                    tokio::time::sleep(d).await;
616                }
617            }
618        }
619
620        Ok(CaptureResponse {
621            results: final_results,
622        })
623    }
624
625    /// Get all remote feature flags and payloads for a user.
626    ///
627    /// For new code, prefer [`Client::evaluate_flags`] so flag reads are
628    /// deduplicated and can be attached to captured events with
629    /// [`Event::with_flags`](crate::Event::with_flags).
630    ///
631    /// # Parameters
632    ///
633    /// - `distinct_id`: User distinct ID.
634    /// - `groups`: Optional group keys for group-targeted flags.
635    /// - `person_properties`: Optional person properties for release
636    ///   conditions.
637    /// - `group_properties`: Optional group properties for group-targeted
638    ///   release conditions.
639    ///
640    /// # Returns
641    ///
642    /// A tuple of `(feature_flags, feature_flag_payloads)`, each keyed by flag
643    /// key. Disabled clients return two empty maps.
644    ///
645    /// # Errors
646    ///
647    /// Returns [`Error::Connection`] for request failures or non-success HTTP
648    /// statuses, and [`Error::Serialization`] when the response cannot be
649    /// parsed.
650    #[must_use = "feature flags result should be used"]
651    pub async fn get_feature_flags<S: Into<String>>(
652        &self,
653        distinct_id: S,
654        groups: Option<HashMap<String, String>>,
655        person_properties: Option<HashMap<String, serde_json::Value>>,
656        group_properties: Option<HashMap<String, HashMap<String, serde_json::Value>>>,
657    ) -> Result<
658        (
659            HashMap<String, FlagValue>,
660            HashMap<String, serde_json::Value>,
661        ),
662        Error,
663    > {
664        if self.options.is_disabled() {
665            trace!("Client is disabled, skipping feature flags request");
666            return Ok((HashMap::new(), HashMap::new()));
667        }
668
669        let flags_endpoint = self.options.endpoints().build_url(Endpoint::Flags);
670
671        let mut payload = json!({
672            "api_key": self.options.api_key,
673            "distinct_id": distinct_id.into(),
674        });
675
676        if let Some(groups) = groups {
677            payload["groups"] = json!(groups);
678        }
679
680        if let Some(person_properties) = person_properties {
681            payload["person_properties"] = json!(person_properties);
682        }
683
684        if let Some(group_properties) = group_properties {
685            payload["group_properties"] = json!(group_properties);
686        }
687
688        // Add geoip disable parameter if configured
689        if self.options.disable_geoip {
690            payload["disable_geoip"] = json!(true);
691        }
692
693        let response = self
694            .client
695            .post(&flags_endpoint)
696            .header(CONTENT_TYPE, "application/json")
697            .header(USER_AGENT, get_default_user_agent())
698            .json(&payload)
699            .timeout(Duration::from_secs(
700                self.options.feature_flags_request_timeout_seconds,
701            ))
702            .send()
703            .await
704            .map_err(|e| Error::Connection(e.to_string()))?;
705
706        if !response.status().is_success() {
707            let status = response.status();
708            let text = response
709                .text()
710                .await
711                .unwrap_or_else(|_| "Unknown error".to_string());
712            return Err(Error::Connection(format!(
713                "API request failed with status {status}: {text}"
714            )));
715        }
716
717        let flags_response = response.json::<FeatureFlagsResponse>().await.map_err(|e| {
718            Error::Serialization(format!("Failed to parse feature flags response: {e}"))
719        })?;
720
721        Ok(flags_response.normalize())
722    }
723
724    /// Get a specific feature flag value for a user.
725    ///
726    /// # Parameters
727    ///
728    /// - `key`: Feature flag key.
729    /// - `distinct_id`: User distinct ID.
730    /// - `groups`: Optional group keys for group-targeted flags.
731    /// - `person_properties`: Optional person properties for release
732    ///   conditions.
733    /// - `group_properties`: Optional group properties for group-targeted
734    ///   release conditions.
735    ///
736    /// # Returns
737    ///
738    /// `Ok(Some(value))` when the flag is returned, `Ok(None)` when it is not
739    /// returned or local-only evaluation cannot resolve it.
740    ///
741    /// # Errors
742    ///
743    /// Returns errors from remote `/flags` requests or response parsing.
744    #[must_use = "feature flag result should be used"]
745    #[instrument(skip_all, level = "debug")]
746    #[deprecated(
747        since = "0.6.0",
748        note = "Use Client::evaluate_flags() to fetch a snapshot, then call .get_flag(key) on it. \
749                The snapshot deduplicates $feature_flag_called events and supports attaching \
750                rich metadata to captured events via Event::with_flags()."
751    )]
752    pub async fn get_feature_flag<K: Into<String>, D: Into<String>>(
753        &self,
754        key: K,
755        distinct_id: D,
756        groups: Option<HashMap<String, String>>,
757        person_properties: Option<HashMap<String, serde_json::Value>>,
758        group_properties: Option<HashMap<String, HashMap<String, serde_json::Value>>>,
759    ) -> Result<Option<FlagValue>, Error> {
760        let key_str = key.into();
761        let distinct_id_str = distinct_id.into();
762
763        // Try local evaluation first if available
764        if let Some(ref evaluator) = self.local_evaluator {
765            let empty_props = HashMap::new();
766            let empty_groups: HashMap<String, String> = HashMap::new();
767            let empty_group_props: HashMap<String, HashMap<String, serde_json::Value>> =
768                HashMap::new();
769            let props = person_properties.as_ref().unwrap_or(&empty_props);
770            let groups_ref = groups.as_ref().unwrap_or(&empty_groups);
771            let group_props_ref = group_properties.as_ref().unwrap_or(&empty_group_props);
772            match evaluator.evaluate_flag(
773                &key_str,
774                &distinct_id_str,
775                props,
776                groups_ref,
777                group_props_ref,
778            ) {
779                Ok(Some(value)) => {
780                    debug!(flag = %key_str, ?value, "Flag evaluated locally");
781                    return Ok(Some(value));
782                }
783                Ok(None) => {
784                    if self.options.local_evaluation_only {
785                        debug!(flag = %key_str, "Flag not found locally, skipping remote fallback");
786                        return Ok(None);
787                    }
788                    debug!(flag = %key_str, "Flag not found locally, falling back to API");
789                }
790                Err(e) => {
791                    if self.options.local_evaluation_only {
792                        debug!(flag = %key_str, error = %e.message, "Inconclusive local evaluation, skipping remote fallback");
793                        return Ok(None);
794                    }
795                    debug!(flag = %key_str, error = %e.message, "Inconclusive local evaluation, falling back to API");
796                }
797            }
798        }
799
800        // Fall back to API
801        trace!(flag = %key_str, "Fetching flag from API");
802        let (feature_flags, _payloads) = self
803            .get_feature_flags(distinct_id_str, groups, person_properties, group_properties)
804            .await?;
805        Ok(feature_flags.get(&key_str).cloned())
806    }
807
808    /// Check if a feature flag is enabled for a user.
809    ///
810    /// # Returns
811    ///
812    /// `true` for `FlagValue::Boolean(true)` or any multivariate variant,
813    /// `false` for disabled or missing flags.
814    ///
815    /// # Errors
816    ///
817    /// Returns errors from [`Client::get_feature_flag`].
818    #[must_use = "feature flag enabled check result should be used"]
819    #[deprecated(
820        since = "0.6.0",
821        note = "Use Client::evaluate_flags() to fetch a snapshot, then call .is_enabled(key) \
822                on it. The snapshot deduplicates $feature_flag_called events and supports \
823                attaching rich metadata to captured events via Event::with_flags()."
824    )]
825    #[allow(deprecated)] // calls deprecated get_feature_flag internally
826    pub async fn is_feature_enabled<K: Into<String>, D: Into<String>>(
827        &self,
828        key: K,
829        distinct_id: D,
830        groups: Option<HashMap<String, String>>,
831        person_properties: Option<HashMap<String, serde_json::Value>>,
832        group_properties: Option<HashMap<String, HashMap<String, serde_json::Value>>>,
833    ) -> Result<bool, Error> {
834        let flag_value = self
835            .get_feature_flag(
836                key.into(),
837                distinct_id.into(),
838                groups,
839                person_properties,
840                group_properties,
841            )
842            .await?;
843        Ok(match flag_value {
844            Some(FlagValue::Boolean(b)) => b,
845            Some(FlagValue::String(_)) => true, // Variants are considered enabled
846            None => false,
847        })
848    }
849
850    /// Get a feature flag payload for a user.
851    ///
852    /// # Parameters
853    ///
854    /// - `key`: Feature flag key.
855    /// - `distinct_id`: User distinct ID.
856    ///
857    /// # Returns
858    ///
859    /// The JSON payload for the flag, if one was returned. This method does not
860    /// emit `$feature_flag_called` events.
861    ///
862    /// # Errors
863    ///
864    /// Returns [`Error::Connection`] for request failures and
865    /// [`Error::Serialization`] when the response cannot be parsed.
866    #[must_use = "feature flag payload result should be used"]
867    #[deprecated(
868        since = "0.6.0",
869        note = "Use Client::evaluate_flags() to fetch a snapshot, then call \
870                .get_flag_payload(key) on it. Reading the payload from a snapshot is \
871                event-free, matching this method's behavior, and avoids the per-call \
872                /flags request."
873    )]
874    pub async fn get_feature_flag_payload<K: Into<String>, D: Into<String>>(
875        &self,
876        key: K,
877        distinct_id: D,
878    ) -> Result<Option<serde_json::Value>, Error> {
879        if self.options.is_disabled() {
880            trace!("Client is disabled, skipping feature flag payload request");
881            return Ok(None);
882        }
883
884        let key_str = key.into();
885        let flags_endpoint = self.options.endpoints().build_url(Endpoint::Flags);
886
887        let mut payload = json!({
888            "api_key": self.options.api_key,
889            "distinct_id": distinct_id.into(),
890        });
891
892        // Add geoip disable parameter if configured
893        if self.options.disable_geoip {
894            payload["disable_geoip"] = json!(true);
895        }
896
897        let response = self
898            .client
899            .post(&flags_endpoint)
900            .header(CONTENT_TYPE, "application/json")
901            .header(USER_AGENT, get_default_user_agent())
902            .json(&payload)
903            .timeout(Duration::from_secs(
904                self.options.feature_flags_request_timeout_seconds,
905            ))
906            .send()
907            .await
908            .map_err(|e| Error::Connection(e.to_string()))?;
909
910        if !response.status().is_success() {
911            return Ok(None);
912        }
913
914        let flags_response: FeatureFlagsResponse = response
915            .json()
916            .await
917            .map_err(|e| Error::Serialization(format!("Failed to parse response: {e}")))?;
918
919        let (_flags, payloads) = flags_response.normalize();
920        Ok(payloads.get(&key_str).cloned())
921    }
922
923    /// Evaluate a supplied feature flag definition locally.
924    ///
925    /// `groups` and `group_properties` are only consulted when the flag (or one
926    /// of its conditions) targets a group; pass empty maps for person flags.
927    ///
928    /// # Parameters
929    ///
930    /// - `flag`: Feature flag definition to evaluate.
931    /// - `distinct_id`: User distinct ID.
932    /// - `person_properties`: Person properties available to release
933    ///   conditions.
934    /// - `groups`: Group keys for group-targeted flags.
935    /// - `group_properties`: Group properties for group-targeted release
936    ///   conditions.
937    ///
938    /// # Errors
939    ///
940    /// Returns [`Error::InconclusiveMatch`] when the flag cannot be evaluated
941    /// locally with the supplied context.
942    #[allow(clippy::too_many_arguments)]
943    pub fn evaluate_feature_flag_locally(
944        &self,
945        flag: &FeatureFlag,
946        distinct_id: &str,
947        person_properties: &HashMap<String, serde_json::Value>,
948        groups: &HashMap<String, String>,
949        group_properties: &HashMap<String, HashMap<String, serde_json::Value>>,
950    ) -> Result<FlagValue, Error> {
951        let group_type_mapping = self
952            .local_evaluator
953            .as_ref()
954            .map(|ev| ev.cache().get_group_type_mapping())
955            .unwrap_or_default();
956        match_feature_flag(
957            flag,
958            distinct_id,
959            person_properties,
960            groups,
961            group_properties,
962            &group_type_mapping,
963        )
964        .map_err(|e| Error::InconclusiveMatch(e.message))
965    }
966
967    /// Evaluate feature flags for `distinct_id`, returning a
968    /// [`FeatureFlagEvaluations`] snapshot.
969    ///
970    /// Each `is_enabled` / `get_flag` call on the returned snapshot fires a
971    /// dedup-aware `$feature_flag_called` event with full metadata, and the
972    /// snapshot can be passed to [`Event::with_flags`] so a downstream
973    /// [`Client::capture`] inherits `$feature/<key>` and `$active_feature_flags`
974    /// without an extra `/flags` request.
975    ///
976    /// # Parameters
977    ///
978    /// - `distinct_id`: User distinct ID. Empty values return an empty snapshot.
979    /// - `options`: Optional groups, properties, GeoIP override, local-only
980    ///   mode, and flag-key filtering.
981    ///
982    /// # Errors
983    ///
984    /// Returns [`Error::Connection`] or [`Error::Serialization`] when remote
985    /// evaluation is required and the `/flags` request fails before any local
986    /// results are available.
987    ///
988    /// [`Event::with_flags`]: crate::Event::with_flags
989    pub async fn evaluate_flags<S: Into<String>>(
990        &self,
991        distinct_id: S,
992        options: EvaluateFlagsOptions,
993    ) -> Result<FeatureFlagEvaluations, Error> {
994        let distinct_id: String = distinct_id.into();
995        let host = self.flag_event_host();
996
997        if distinct_id.is_empty() || self.options.is_disabled() {
998            return Ok(FeatureFlagEvaluations::empty(host));
999        }
1000
1001        let mut records: HashMap<String, EvaluatedFlagRecord> = HashMap::new();
1002        let mut locally_evaluated_keys: HashSet<String> = HashSet::new();
1003
1004        if let Some(evaluator) = &self.local_evaluator {
1005            let person_props_owned = options.person_properties.clone().unwrap_or_default();
1006            let groups_owned = options.groups.clone().unwrap_or_default();
1007            let group_props_owned = options.group_properties.clone().unwrap_or_default();
1008            let local_results = evaluator.evaluate_all_flags(
1009                &distinct_id,
1010                &person_props_owned,
1011                &groups_owned,
1012                &group_props_owned,
1013            );
1014            for (key, result) in local_results {
1015                if let Some(filter) = &options.flag_keys {
1016                    if !filter.iter().any(|k| k == &key) {
1017                        continue;
1018                    }
1019                }
1020                if let Ok(value) = result {
1021                    records.insert(key.clone(), local_record(value));
1022                    locally_evaluated_keys.insert(key);
1023                }
1024            }
1025        }
1026
1027        let mut request_id: Option<String> = None;
1028        let mut errors_while_computing = false;
1029        let mut quota_limited = false;
1030
1031        // Skip the remote round-trip when local evaluation has already covered
1032        // every requested flag. Without `flag_keys` we have to assume the caller
1033        // wants every flag the project has and still hit `/flags` to discover
1034        // any not loaded by the poller.
1035        let local_covers_request = options
1036            .flag_keys
1037            .as_ref()
1038            .is_some_and(|keys| keys.iter().all(|k| locally_evaluated_keys.contains(k)));
1039
1040        if !options.only_evaluate_locally && !local_covers_request {
1041            // Don't lose successful local evaluations if `/flags` fails — degrade
1042            // to a snapshot built from the local results we already have. The
1043            // alternative (returning Err) wastes useful data and surprises
1044            // callers who would otherwise get partial coverage.
1045            match self.fetch_flag_details(&distinct_id, &options).await {
1046                Ok(response) => {
1047                    request_id = response.request_id;
1048                    errors_while_computing = response.errors_while_computing_flags;
1049                    quota_limited = response.quota_limited;
1050                    for (key, detail) in response.flags {
1051                        if locally_evaluated_keys.contains(&key) {
1052                            continue;
1053                        }
1054                        records.insert(key, remote_record_from_detail(detail));
1055                    }
1056                }
1057                Err(e) => {
1058                    if records.is_empty() {
1059                        return Err(e);
1060                    }
1061                    debug!(
1062                        error = e.to_string(),
1063                        local_count = records.len(),
1064                        "/flags fetch failed; returning snapshot from local results only"
1065                    );
1066                    errors_while_computing = true;
1067                }
1068            }
1069        }
1070
1071        Ok(FeatureFlagEvaluations::new(
1072            host,
1073            distinct_id,
1074            records,
1075            options.groups.unwrap_or_default(),
1076            options.disable_geoip,
1077            request_id,
1078            None,
1079            errors_while_computing,
1080            quota_limited,
1081        ))
1082    }
1083
1084    fn flag_event_host(&self) -> Arc<dyn FeatureFlagEvaluationsHost> {
1085        self.flag_event_host
1086            .get_or_init(|| {
1087                Arc::new(AsyncFlagEventHost::from_options(
1088                    &self.options,
1089                    self.client.clone(),
1090                )) as Arc<dyn FeatureFlagEvaluationsHost>
1091            })
1092            .clone()
1093    }
1094
1095    async fn fetch_flag_details(
1096        &self,
1097        distinct_id: &str,
1098        options: &EvaluateFlagsOptions,
1099    ) -> Result<DetailedFlagsResponse, Error> {
1100        let flags_endpoint = self.options.endpoints().build_url(Endpoint::Flags);
1101
1102        let mut payload = json!({
1103            "api_key": self.options.api_key,
1104            "distinct_id": distinct_id,
1105        });
1106        if let Some(groups) = &options.groups {
1107            payload["groups"] = json!(groups);
1108        }
1109        if let Some(person_properties) = &options.person_properties {
1110            payload["person_properties"] = json!(person_properties);
1111        }
1112        if let Some(group_properties) = &options.group_properties {
1113            payload["group_properties"] = json!(group_properties);
1114        }
1115        let effective_disable_geoip = options.disable_geoip.unwrap_or(self.options.disable_geoip);
1116        if effective_disable_geoip {
1117            payload["disable_geoip"] = json!(true);
1118        }
1119        if let Some(flag_keys) = &options.flag_keys {
1120            payload["flag_keys_to_evaluate"] = json!(flag_keys);
1121        }
1122
1123        let response = self
1124            .client
1125            .post(&flags_endpoint)
1126            .header(CONTENT_TYPE, "application/json")
1127            .header(USER_AGENT, get_default_user_agent())
1128            .json(&payload)
1129            .timeout(Duration::from_secs(
1130                self.options.feature_flags_request_timeout_seconds,
1131            ))
1132            .send()
1133            .await
1134            .map_err(|e| Error::Connection(e.to_string()))?;
1135
1136        if !response.status().is_success() {
1137            let status = response.status();
1138            let text = response
1139                .text()
1140                .await
1141                .unwrap_or_else(|_| "Unknown error".to_string());
1142            return Err(Error::Connection(format!(
1143                "API request failed with status {status}: {text}"
1144            )));
1145        }
1146
1147        let parsed = response.json::<FeatureFlagsResponse>().await.map_err(|e| {
1148            Error::Serialization(format!("Failed to parse feature flags response: {e}"))
1149        })?;
1150        Ok(extract_flag_details(parsed))
1151    }
1152}
posthog_rs/client/async_client.rs

posthog_rs/client/
async_client.rs
