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
7use reqwest::{header::CONTENT_TYPE, header::USER_AGENT, Client as HttpClient};
8use serde::Serialize;
9use serde_json::json;
10use tracing::{debug, instrument, trace, warn};
11
12use super::get_default_user_agent;
13use crate::endpoints::Endpoint;
14#[cfg(feature = "error-tracking")]
15use crate::error_tracking::{build_exception_event, CaptureExceptionOptions};
16use crate::feature_flag_evaluations::{
17    EvaluateFlagsOptions, EvaluatedFlagRecord, FeatureFlagEvaluations, FeatureFlagEvaluationsHost,
18    FlagCalledEventParams,
19};
20use crate::feature_flags::{match_feature_flag, FeatureFlag, FeatureFlagsResponse, FlagValue};
21use crate::local_evaluation::{AsyncFlagPoller, FlagCache, LocalEvaluationConfig, LocalEvaluator};
22use crate::{Error, Event};
23
24fn is_retryable_feature_flags_error(err: &reqwest::Error) -> bool {
25    if err.is_timeout() {
26        return true;
27    }
28
29    let mut source = std::error::Error::source(err);
30    while let Some(error) = source {
31        if let Some(io_error) = error.downcast_ref::<std::io::Error>() {
32            return matches!(
33                io_error.kind(),
34                std::io::ErrorKind::ConnectionReset
35                    | std::io::ErrorKind::TimedOut
36                    | std::io::ErrorKind::UnexpectedEof
37            );
38        }
39        source = std::error::Error::source(error);
40    }
41
42    !err.to_string()
43        .to_lowercase()
44        .contains("connection refused")
45}
46
47use super::common::{
48    already_reported, build_dedup_key, extract_flag_details, flag_called_event,
49    flag_event_dedup_cache, local_record, remote_record_from_detail, report_flags_error,
50    DetailedFlagsResponse, FlagEventDedupCache,
51};
52use super::transport::{Completion, Control, TransportHandle};
53use super::{CaptureSummary, ClientOptions};
54#[cfg(not(feature = "capture-v1"))]
55use reqwest::header::CONTENT_ENCODING;
56
57/// A [`Client`] facilitates interactions with the PostHog API over HTTP.
58pub struct Client {
59    options: ClientOptions,
60    client: HttpClient,
61    local_evaluator: Option<LocalEvaluator>,
62    _flag_poller: Option<AsyncFlagPoller>,
63    flag_event_host: OnceLock<Arc<dyn FeatureFlagEvaluationsHost>>,
64    /// Background event transport. `None` for disabled clients.
65    transport: Option<Arc<TransportHandle>>,
66}
67
68/// Implementation of [`FeatureFlagEvaluationsHost`] that emits dedup-aware
69/// `$feature_flag_called` events through the same background capture transport
70/// as any other event.
71struct AsyncFlagEventHost {
72    options: ClientOptions,
73    transport: Option<Arc<TransportHandle>>,
74    dedup_cache: FlagEventDedupCache,
75}
76
77impl AsyncFlagEventHost {
78    fn from_options(options: &ClientOptions, transport: Option<Arc<TransportHandle>>) -> Self {
79        Self {
80            options: options.clone(),
81            transport,
82            dedup_cache: flag_event_dedup_cache(),
83        }
84    }
85
86    fn enqueue(&self, event: Event) {
87        if let Some(transport) = &self.transport {
88            transport.enqueue(event);
89        }
90    }
91}
92
93impl FeatureFlagEvaluationsHost for AsyncFlagEventHost {
94    fn capture_flag_called_event_if_needed(&self, params: FlagCalledEventParams) {
95        let dedup_key = build_dedup_key(&params.key, params.response.as_ref(), &params.groups);
96        if already_reported(&self.dedup_cache, &params.distinct_id, &dedup_key) {
97            return;
98        }
99
100        if let Some(event) =
101            flag_called_event(params, self.options.disable_geoip, self.options.is_server)
102        {
103            self.enqueue(event);
104        }
105    }
106
107    fn log_warning(&self, message: &str) {
108        // Surface filter-helper misuse via tracing — users can silence these
109        // with their tracing-subscriber level filter (e.g. `posthog_rs=error`).
110        warn!("{message}");
111    }
112}
113
114/// Construct an async PostHog client from an API key or [`ClientOptions`].
115///
116/// # Parameters
117///
118/// - `options`: Either a project API key (for example `"phc_..."`) or a
119///   configured [`ClientOptions`] value.
120///
121/// # Returns
122///
123/// A [`Client`] that performs capture and feature flag requests asynchronously.
124///
125/// # Remarks
126///
127/// This constructor is available with the default `async-client` feature and
128/// must be awaited. Passing a blank API key creates a disabled client.
129pub async fn client<C: Into<ClientOptions>>(options: C) -> Client {
130    let options = options.into().sanitize();
131    let client = HttpClient::builder()
132        .timeout(Duration::from_secs(options.request_timeout_seconds))
133        .build()
134        .unwrap(); // Unwrap here is as safe as `HttpClient::new`
135
136    let (local_evaluator, flag_poller) =
137        if options.enable_local_evaluation && !options.is_disabled() {
138            if let Some(ref secret_key) = options.secret_key {
139                let cache = FlagCache::new();
140
141                let config = LocalEvaluationConfig {
142                    personal_api_key: secret_key.clone(),
143                    project_api_key: options.api_key.clone(),
144                    api_host: options.endpoints().api_host(),
145                    poll_interval: Duration::from_secs(options.poll_interval_seconds),
146                    request_timeout: Duration::from_secs(options.request_timeout_seconds),
147                };
148
149                let mut poller = AsyncFlagPoller::new(config, cache.clone());
150                poller.set_on_error(options.on_error.clone());
151                poller.start().await;
152
153                (Some(LocalEvaluator::new(cache)), Some(poller))
154            } else {
155                warn!(
156                "Local evaluation enabled but secret_key not set, falling back to API evaluation"
157            );
158                (None, None)
159            }
160        } else {
161            (None, None)
162        };
163
164    let transport = if options.is_disabled() {
165        None
166    } else {
167        Some(Arc::new(TransportHandle::spawn(options.clone())))
168    };
169
170    Client {
171        options,
172        client,
173        local_evaluator,
174        _flag_poller: flag_poller,
175        flag_event_host: OnceLock::new(),
176        transport,
177    }
178}
179
180impl Client {
181    /// Capture the provided event, sending it to PostHog.
182    ///
183    /// # Parameters
184    ///
185    /// - `event`: Event name, distinct ID, properties, timestamp, groups, and
186    ///   optional feature flag state to send.
187    ///
188    /// # Remarks
189    ///
190    /// Fire-and-forget: the event is handed to the background worker, which
191    /// batches, sends, and retries it. Returns once the event is queued — not
192    /// once it is delivered, and delivery failures are not surfaced to the
193    /// caller. Disabled clients and a full queue drop the event (the latter
194    /// with a single warning).
195    #[instrument(skip(self, event), level = "debug")]
196    pub fn capture(&self, event: Event) {
197        if let Some(transport) = &self.transport {
198            transport.enqueue(event);
199        }
200    }
201
202    /// Merge two distinct IDs onto the same person by sending a `$create_alias`
203    /// event.
204    ///
205    /// See <https://posthog.com/docs/product-analytics/identify#alias-assigning-multiple-distinct-ids-to-the-same-user>.
206    ///
207    /// # Parameters
208    ///
209    /// - `previous_id`: ID already known to PostHog, such as an anonymous ID.
210    /// - `distinct_id`: ID it should be merged into, such as a logged-in user ID.
211    ///
212    /// # Remarks
213    ///
214    /// Fire-and-forget, like [`Client::capture`]. A blank ID on either side
215    /// cannot describe a merge, so the event is dropped with a warning rather
216    /// than sent.
217    ///
218    /// # Examples
219    ///
220    /// ```no_run
221    /// # async fn example() {
222    /// let client = posthog_rs::client("phc_project_api_key").await;
223    ///
224    /// // The visitor browsed anonymously, then logged in.
225    /// client.alias("anon-abc123", "user-42");
226    /// # }
227    /// ```
228    pub fn alias<P: Into<String>, D: Into<String>>(&self, previous_id: P, distinct_id: D) {
229        if let Some(event) = Event::alias(previous_id.into(), distinct_id.into()) {
230            self.capture(event);
231        }
232    }
233
234    /// Create or update a group and set its properties by sending a `$groupidentify`
235    /// event.
236    ///
237    /// See <https://posthog.com/docs/product-analytics/group-analytics#setting-group-properties>.
238    ///
239    /// # Parameters
240    ///
241    /// - `group_type`: Group type, such as `"company"`, `"project"`, or `"organization"`.
242    /// - `group_key`: Unique identifier for the group, such as an ID in your database.
243    /// - `properties`: Any serializable object or JSON map representing group properties.
244    ///
245    /// # Remarks
246    ///
247    /// Fire-and-forget, like [`Client::capture`], for a blank `group_type` or
248    /// `group_key`: the event is dropped with a warning rather than sent.
249    ///
250    /// # Errors
251    ///
252    /// Returns [`Error::Serialization`] if `properties` fails to serialize to
253    /// JSON, or if it does not serialize to a JSON object (PostHog requires
254    /// `$group_set` to be an object).
255    ///
256    /// # Examples
257    ///
258    /// ```no_run
259    /// # async fn example() -> Result<(), posthog_rs::Error> {
260    /// use serde_json::json;
261    ///
262    /// let client = posthog_rs::client("phc_project_api_key").await;
263    ///
264    /// client.group_identify(
265    ///     "company",
266    ///     "company_id_in_your_db",
267    ///     json!({
268    ///         "name": "Awesome Inc.",
269    ///         "employees": 11,
270    ///     }),
271    /// )?;
272    /// # Ok(())
273    /// # }
274    /// ```
275    pub fn group_identify<T: Into<String>, K: Into<String>, P: Serialize>(
276        &self,
277        group_type: T,
278        group_key: K,
279        properties: P,
280    ) -> Result<(), Error> {
281        if let Some(event) = Event::group_identify(group_type.into(), group_key.into(), properties)?
282        {
283            self.capture(event);
284        }
285        Ok(())
286    }
287
288    /// Flush queued events, returning once the worker has attempted delivery of
289    /// everything queued before this call. Transient failures are kept for retry
290    /// (the call still returns without error). A no-op for disabled clients.
291    pub async fn flush(&self) {
292        let Some(transport) = &self.transport else {
293            return;
294        };
295        if transport.is_closed() {
296            return;
297        }
298        let (tx, rx) = tokio::sync::oneshot::channel();
299        if transport.send_control(Control::Flush(Completion::Async(tx))) {
300            let _ = rx.await;
301        }
302    }
303
304    /// Whether the client is disabled (no transport; capture is a no-op). Used
305    /// by the panic hook to skip building an event it could never send.
306    #[cfg(feature = "error-tracking")]
307    pub(crate) fn is_disabled(&self) -> bool {
308        self.options.is_disabled()
309    }
310
311    /// The client's Error Tracking options, used by the panic hook to build
312    /// panic exception events with the client's configured policy.
313    #[cfg(feature = "error-tracking")]
314    pub(crate) fn error_tracking_options(&self) -> &crate::error_tracking::ErrorTrackingOptions {
315        self.options.error_tracking()
316    }
317
318    /// Unbounded synchronous flush: blocks until the worker has attempted
319    /// delivery of everything queued. Test-only; the panic hook uses
320    /// `flush_blocking_timeout`.
321    #[cfg(test)]
322    pub(crate) fn flush_blocking(&self) {
323        if let Some(transport) = &self.transport {
324            transport.flush_blocking();
325        }
326    }
327
328    /// Synchronous, time-bounded flush for the panic hook: blocks (no runtime
329    /// needed) up to `timeout` for the worker to attempt delivery, then returns.
330    /// A no-op for disabled clients.
331    #[cfg(feature = "error-tracking")]
332    pub(crate) fn flush_blocking_timeout(&self, timeout: Duration) {
333        if let Some(transport) = &self.transport {
334            transport.flush_blocking_timeout(timeout);
335        }
336    }
337
338    /// True when the calling thread is this client's transport worker thread —
339    /// the panic hook skips capturing there.
340    #[cfg(feature = "error-tracking")]
341    pub(crate) fn on_transport_worker(&self) -> bool {
342        self.transport
343            .as_ref()
344            .is_some_and(|t| t.on_worker_thread())
345    }
346
347    /// Enqueue a panic `$exception` without the tracing `capture` performs:
348    /// `capture` is `#[instrument]` and its enqueue warns once on a full queue,
349    /// both of which run subscriber code — unsafe on the already-panicking
350    /// thread. The send still happens on the worker thread.
351    #[cfg(feature = "error-tracking")]
352    pub(crate) fn enqueue_panic_event(&self, event: Event) {
353        if let Some(transport) = &self.transport {
354            transport.enqueue_panic(event);
355        }
356    }
357
358    /// Flush, stop the background worker, and join it. Idempotent: subsequent
359    /// calls are no-ops. After shutdown, `capture` drops events. A no-op for
360    /// disabled clients. When called from a transport callback, queues shutdown
361    /// without waiting for or joining the current worker thread.
362    pub async fn shutdown(&self) {
363        let Some(transport) = &self.transport else {
364            return;
365        };
366        let on_worker = transport.on_worker_thread();
367        if transport.begin_close() {
368            let (tx, rx) = tokio::sync::oneshot::channel();
369            if transport.send_control(Control::Shutdown(Completion::Async(tx))) && !on_worker {
370                let _ = rx.await;
371            }
372        }
373        // Always join for external callers — even if this caller lost the
374        // `begin_close` race or its shutdown wait was cancelled — so every
375        // external shutdown/drop path waits for the worker and the flush stays
376        // durable. Joining from the worker itself would deadlock.
377        if !on_worker {
378            transport.join();
379        }
380    }
381
382    /// Capture a Rust error personlessly, sending it to PostHog Error Tracking.
383    ///
384    /// The error's type, message, and full `source()` chain are sent as
385    /// `$exception_list`, with a stacktrace of the capture site attached to
386    /// the first entry (see `ErrorTrackingOptions::capture_stacktrace`).
387    ///
388    /// Accepts any [`std::error::Error`], including `&dyn Error`. A
389    /// `Box<dyn Error>` does not implement `Error` itself, so pass the
390    /// dereferenced trait object: `capture_exception(&*boxed)`.
391    ///
392    /// To associate the exception with a person or attach custom properties,
393    /// groups, a fingerprint, or a severity level, use
394    /// [`Client::capture_exception_with`].
395    ///
396    /// # Examples
397    ///
398    /// ```no_run
399    /// # async fn example() -> Result<(), posthog_rs::Error> {
400    /// let client = posthog_rs::client("phc_project_api_key").await;
401    /// let error = std::io::Error::other("checkout failed");
402    ///
403    /// client.capture_exception(&error).await?;
404    /// # Ok(())
405    /// # }
406    /// ```
407    #[cfg(feature = "error-tracking")]
408    pub async fn capture_exception<E>(&self, error: &E) -> Result<(), Error>
409    where
410        E: StdError + ?Sized,
411    {
412        self.capture_exception_with(error, CaptureExceptionOptions::default())
413            .await
414    }
415
416    /// Capture a Rust error with optional context, sending it to PostHog
417    /// Error Tracking.
418    ///
419    /// Set [`CaptureExceptionOptions::distinct_id`] to associate the exception
420    /// with a person; without it the exception is captured personlessly.
421    ///
422    /// # Examples
423    ///
424    /// ```no_run
425    /// # async fn example() -> Result<(), posthog_rs::Error> {
426    /// use posthog_rs::CaptureExceptionOptions;
427    ///
428    /// let client = posthog_rs::client("phc_project_api_key").await;
429    /// let error = std::io::Error::other("checkout failed");
430    ///
431    /// client
432    ///     .capture_exception_with(
433    ///         &error,
434    ///         CaptureExceptionOptions::new()
435    ///             .distinct_id("user-123")
436    ///             .property("route", "/checkout")?,
437    ///     )
438    ///     .await?;
439    /// # Ok(())
440    /// # }
441    /// ```
442    #[cfg(feature = "error-tracking")]
443    pub async fn capture_exception_with<E>(
444        &self,
445        error: &E,
446        options: CaptureExceptionOptions,
447    ) -> Result<(), Error>
448    where
449        E: StdError + ?Sized,
450    {
451        if self.options.is_disabled() {
452            trace!("Client is disabled, skipping exception capture");
453            return Ok(());
454        }
455
456        self.capture(build_exception_event(
457            error,
458            options,
459            self.options.error_tracking(),
460        )?);
461        Ok(())
462    }
463
464    /// Capture a collection of events with a single request.
465    ///
466    /// Events are sent to the `/batch/` endpoint.
467    ///
468    /// # Parameters
469    ///
470    /// - `events`: Events to send in the batch.
471    /// - `historical_migration`: Set to `true` to route events to the
472    ///   historical ingestion topic, bypassing the main pipeline.
473    ///
474    /// # Remarks
475    ///
476    /// Fire-and-forget, like [`Client::capture`]. The batch is enqueued per event
477    /// rather than atomically, so if the bounded queue fills partway through, the
478    /// remaining events are dropped (with the usual single full-queue warning).
479    pub fn capture_batch(&self, events: Vec<Event>, historical_migration: bool) {
480        if let Some(transport) = &self.transport {
481            if historical_migration {
482                transport.enqueue_historical(events);
483            } else {
484                for event in events {
485                    transport.enqueue(event);
486                }
487            }
488        }
489    }
490
491    // ----- Immediate (inline) capture -------------------------------------
492    //
493    // `capture`/`capture_batch` above are fire-and-forget: they enqueue onto the
494    // background worker and never report the outcome. The `*_immediate` variants
495    // send inline and await a terminal result, for the rare caller that must know
496    // a batch persisted before advancing its own durable state (e.g. committing
497    // an upstream offset). They bypass the worker queue and do NOT fire `on_error`
498    // hooks — the returned `Result`/`CaptureSummary` is the delivery signal.
499
500    /// Capture a single event and await confirmation that the request completed.
501    ///
502    /// The immediate-delivery counterpart to [`Client::capture`]. This is a
503    /// convenience wrapper over [`Client::capture_batch_immediate`] with a
504    /// one-event batch; see it for full semantics.
505    #[must_use = "the delivery outcome should be inspected"]
506    pub async fn capture_immediate(&self, event: Event) -> Result<CaptureSummary, Error> {
507        self.capture_batch_immediate(vec![event], false).await
508    }
509
510    /// Capture a batch of events and await confirmation that the request
511    /// completed, returning a [`CaptureSummary`] describing the outcome.
512    ///
513    /// The immediate-delivery counterpart to [`Client::capture_batch`]. Prefer
514    /// the fire-and-forget [`Client::capture`]/[`Client::capture_batch`] for
515    /// normal analytics; reach for this only when the caller must know the batch
516    /// persisted before advancing its own durable state.
517    ///
518    /// # Parameters
519    ///
520    /// - `events`: Events to send in a single request.
521    /// - `historical_migration`: Route events to the historical ingestion topic.
522    ///
523    /// # Behavior
524    ///
525    /// Sends inline (bypassing the background worker) and retries transient
526    /// failures per the client's retry configuration. On the `capture-v1`
527    /// pipeline a returned `Ok` can still report unpersisted events — inspect
528    /// [`CaptureSummary::all_persisted`]. Does NOT fire `on_error` hooks: the
529    /// returned `Result` is the delivery signal. Disabled clients and an empty
530    /// (or fully `before_send`-filtered) batch return a default `CaptureSummary`.
531    ///
532    /// # Errors
533    ///
534    /// Returns [`Error`] when the request is rejected with a terminal status or
535    /// the retry budget is exhausted without a successful response.
536    #[must_use = "the delivery outcome should be inspected"]
537    #[instrument(
538        skip(self, events),
539        fields(event_count = events.len(), historical_migration),
540        level = "debug"
541    )]
542    pub async fn capture_batch_immediate(
543        &self,
544        events: Vec<Event>,
545        historical_migration: bool,
546    ) -> Result<CaptureSummary, Error> {
547        if self.options.is_disabled() || events.is_empty() {
548            return Ok(CaptureSummary::default());
549        }
550        self.send_immediate(events, historical_migration).await
551    }
552
553    /// Inline V1 capture: prepare once via the shared sans-IO helpers, then loop
554    /// send/classify, awaiting `tokio::time::sleep` between retries. The setup and
555    /// classification are shared with the blocking client; only this loop differs.
556    #[cfg(feature = "capture-v1")]
557    async fn send_immediate(
558        &self,
559        events: Vec<Event>,
560        historical_migration: bool,
561    ) -> Result<CaptureSummary, Error> {
562        use super::v1_capture::{self, Step};
563
564        let Some(mut prep) =
565            v1_capture::prepare_immediate(&self.options, events, historical_migration)
566        else {
567            return Ok(CaptureSummary::default());
568        };
569        let mut final_results = HashMap::new();
570        let mut attempt: u32 = 1;
571
572        loop {
573            let (headers, body) = v1_capture::build_attempt_parts(
574                &self.options,
575                &prep.request_id,
576                attempt,
577                &prep.created_at,
578                prep.historical_migration,
579                &prep.pending,
580            )?;
581
582            let step = match self
583                .client
584                .post(&prep.url)
585                .headers(headers)
586                .body(body)
587                .send()
588                .await
589            {
590                Err(e) => v1_capture::after_transport_error(
591                    &self.options,
592                    &prep.request_id,
593                    attempt,
594                    e.to_string(),
595                ),
596                Ok(response) => {
597                    let status = response.status().as_u16();
598                    let retry_after = v1_capture::parse_retry_after(response.headers());
599                    let text = response
600                        .text()
601                        .await
602                        .unwrap_or_else(|_| "Unknown error".to_string());
603                    v1_capture::after_response(
604                        &self.options,
605                        &prep.request_id,
606                        attempt,
607                        status,
608                        retry_after,
609                        &text,
610                        &mut prep.pending,
611                        &mut final_results,
612                    )
613                }
614            };
615
616            match step {
617                Step::Done => {
618                    return Ok(CaptureSummary::from_results(prep.submitted, final_results))
619                }
620                Step::Fail(e) => return Err(e),
621                Step::Backoff(delay) => {
622                    attempt += 1;
623                    tokio::time::sleep(delay).await;
624                }
625            }
626        }
627    }
628
629    /// Inline V0 capture: prepare the batch body once via the shared sans-IO
630    /// helpers, then loop send/classify. A `2xx` persists the whole batch.
631    #[cfg(not(feature = "capture-v1"))]
632    async fn send_immediate(
633        &self,
634        events: Vec<Event>,
635        historical_migration: bool,
636    ) -> Result<CaptureSummary, Error> {
637        use super::retry::{self, v0_after_response, v0_after_transport_error, Step};
638        use super::v0_capture;
639
640        let Some(prep) =
641            v0_capture::prepare_immediate(&self.options, events, historical_migration)?
642        else {
643            return Ok(CaptureSummary::default());
644        };
645
646        let mut attempt: u32 = 1;
647        loop {
648            let mut request = self
649                .client
650                .post(&prep.url)
651                .header(CONTENT_TYPE, "application/json")
652                .header(USER_AGENT, get_default_user_agent())
653                .body(prep.body.clone());
654            if let Some(token) = prep.encoding {
655                request = request.header(CONTENT_ENCODING, token);
656            }
657            #[cfg(feature = "test-harness")]
658            if let Some(ref extra) = self.options.extra_capture_headers {
659                for (k, v) in extra {
660                    request = request.header(k.as_str(), v.as_str());
661                }
662            }
663
664            let step = match request.send().await {
665                Err(e) => v0_after_transport_error(&self.options, attempt, e.to_string()),
666                Ok(response) => {
667                    let status = response.status().as_u16();
668                    let retry_after = retry::parse_retry_after(response.headers());
669                    let text = response
670                        .text()
671                        .await
672                        .unwrap_or_else(|_| "Unknown error".to_string());
673                    v0_after_response(&self.options, attempt, status, retry_after, &text)
674                }
675            };
676
677            match step {
678                Step::Done => return Ok(CaptureSummary::delivered(prep.kept)),
679                Step::Fail(e) => return Err(e),
680                Step::Backoff(delay) => {
681                    attempt += 1;
682                    tokio::time::sleep(delay).await;
683                }
684            }
685        }
686    }
687
688    /// Number of events accepted but not yet delivered or dropped — those still
689    /// in the channel, in the worker's current batch, or held for retry. Returns
690    /// 0 for a disabled client.
691    ///
692    /// Gated behind the `test-harness` feature: it exposes internal queue depth
693    /// for the SDK compliance harness and is not part of the normal public API.
694    #[cfg(feature = "test-harness")]
695    pub fn pending_events(&self) -> usize {
696        self.transport.as_ref().map_or(0, |t| t.pending())
697    }
698
699    /// Get all remote feature flags and payloads for a user.
700    ///
701    /// For new code, prefer [`Client::evaluate_flags`] so flag reads are
702    /// deduplicated and can be attached to captured events with
703    /// [`Event::with_flags`](crate::Event::with_flags).
704    ///
705    /// # Parameters
706    ///
707    /// - `distinct_id`: User distinct ID.
708    /// - `groups`: Optional group keys for group-targeted flags.
709    /// - `person_properties`: Optional person properties for release
710    ///   conditions.
711    /// - `group_properties`: Optional group properties for group-targeted
712    ///   release conditions.
713    ///
714    /// # Returns
715    ///
716    /// A tuple of `(feature_flags, feature_flag_payloads)`, each keyed by flag
717    /// key. Disabled clients return two empty maps.
718    ///
719    /// # Errors
720    ///
721    /// Returns [`Error::Connection`] for request failures or non-success HTTP
722    /// statuses, and [`Error::Serialization`] when the response cannot be
723    /// parsed.
724    #[must_use = "feature flags result should be used"]
725    pub async fn get_feature_flags<S: Into<String>>(
726        &self,
727        distinct_id: S,
728        groups: Option<HashMap<String, String>>,
729        person_properties: Option<HashMap<String, serde_json::Value>>,
730        group_properties: Option<HashMap<String, HashMap<String, serde_json::Value>>>,
731    ) -> Result<
732        (
733            HashMap<String, FlagValue>,
734            HashMap<String, serde_json::Value>,
735        ),
736        Error,
737    > {
738        if self.options.is_disabled() {
739            trace!("Client is disabled, skipping feature flags request");
740            return Ok((HashMap::new(), HashMap::new()));
741        }
742
743        let flags_endpoint = self.options.endpoints().build_url(Endpoint::Flags);
744
745        let mut payload = json!({
746            "api_key": self.options.api_key,
747            "distinct_id": distinct_id.into(),
748        });
749
750        if let Some(groups) = groups {
751            payload["groups"] = json!(groups);
752        }
753
754        if let Some(person_properties) = person_properties {
755            payload["person_properties"] = json!(person_properties);
756        }
757
758        if let Some(group_properties) = group_properties {
759            payload["group_properties"] = json!(group_properties);
760        }
761
762        // Add geoip disable parameter if configured
763        if self.options.disable_geoip {
764            payload["disable_geoip"] = json!(true);
765        }
766
767        let response = self
768            .send_feature_flags_request(&flags_endpoint, &payload)
769            .await?;
770
771        let distinct_id = payload.get("distinct_id").and_then(|v| v.as_str());
772        if !response.status().is_success() {
773            let status = response.status();
774            let text = response
775                .text()
776                .await
777                .unwrap_or_else(|_| "Unknown error".to_string());
778            let err = Error::Connection(format!("API request failed with status {status}: {text}"));
779            report_flags_error(
780                &self.options.on_error,
781                &flags_endpoint,
782                distinct_id,
783                Some(status.as_u16()),
784                Some(&text),
785                &err,
786            );
787            return Err(err);
788        }
789
790        let status = response.status().as_u16();
791        let flags_response = match response.json::<FeatureFlagsResponse>().await {
792            Ok(r) => r,
793            Err(e) => {
794                let err =
795                    Error::Serialization(format!("Failed to parse feature flags response: {e}"));
796                report_flags_error(
797                    &self.options.on_error,
798                    &flags_endpoint,
799                    distinct_id,
800                    Some(status),
801                    None,
802                    &err,
803                );
804                return Err(err);
805            }
806        };
807
808        Ok(flags_response.normalize())
809    }
810
811    /// Get a specific feature flag value for a user.
812    ///
813    /// # Parameters
814    ///
815    /// - `key`: Feature flag key.
816    /// - `distinct_id`: User distinct ID.
817    /// - `groups`: Optional group keys for group-targeted flags.
818    /// - `person_properties`: Optional person properties for release
819    ///   conditions.
820    /// - `group_properties`: Optional group properties for group-targeted
821    ///   release conditions.
822    ///
823    /// # Returns
824    ///
825    /// `Ok(Some(value))` when the flag is returned, `Ok(None)` when it is not
826    /// returned or local-only evaluation cannot resolve it.
827    ///
828    /// # Errors
829    ///
830    /// Returns errors from remote `/flags` requests or response parsing.
831    #[must_use = "feature flag result should be used"]
832    #[instrument(skip_all, level = "debug")]
833    #[deprecated(
834        since = "0.6.0",
835        note = "Use Client::evaluate_flags() to fetch a snapshot, then call .get_flag(key) on it. \
836                The snapshot deduplicates $feature_flag_called events and supports attaching \
837                rich metadata to captured events via Event::with_flags()."
838    )]
839    pub async fn get_feature_flag<K: Into<String>, D: Into<String>>(
840        &self,
841        key: K,
842        distinct_id: D,
843        groups: Option<HashMap<String, String>>,
844        person_properties: Option<HashMap<String, serde_json::Value>>,
845        group_properties: Option<HashMap<String, HashMap<String, serde_json::Value>>>,
846    ) -> Result<Option<FlagValue>, Error> {
847        let key_str = key.into();
848        let distinct_id_str = distinct_id.into();
849
850        // Try local evaluation first if available
851        if let Some(ref evaluator) = self.local_evaluator {
852            let empty_props = HashMap::new();
853            let empty_groups: HashMap<String, String> = HashMap::new();
854            let empty_group_props: HashMap<String, HashMap<String, serde_json::Value>> =
855                HashMap::new();
856            let mut local_props;
857            let props = if let Some(props) = person_properties.as_ref() {
858                local_props = props.clone();
859                local_props
860                    .entry("distinct_id".to_string())
861                    .or_insert_with(|| json!(distinct_id_str.clone()));
862                &local_props
863            } else {
864                local_props = empty_props;
865                local_props.insert("distinct_id".to_string(), json!(distinct_id_str.clone()));
866                &local_props
867            };
868            let groups_ref = groups.as_ref().unwrap_or(&empty_groups);
869            let group_props_ref = group_properties.as_ref().unwrap_or(&empty_group_props);
870            match evaluator.evaluate_flag(
871                &key_str,
872                &distinct_id_str,
873                props,
874                groups_ref,
875                group_props_ref,
876            ) {
877                Ok(Some(value)) => {
878                    debug!(flag = %key_str, ?value, "Flag evaluated locally");
879                    return Ok(Some(value));
880                }
881                Ok(None) => {
882                    if self.options.local_evaluation_only {
883                        debug!(flag = %key_str, "Flag not found locally, skipping remote fallback");
884                        return Ok(None);
885                    }
886                    debug!(flag = %key_str, "Flag not found locally, falling back to API");
887                }
888                Err(e) => {
889                    if self.options.local_evaluation_only {
890                        debug!(flag = %key_str, error = %e.message, "Inconclusive local evaluation, skipping remote fallback");
891                        return Ok(None);
892                    }
893                    debug!(flag = %key_str, error = %e.message, "Inconclusive local evaluation, falling back to API");
894                }
895            }
896        }
897
898        // Fall back to API
899        trace!(flag = %key_str, "Fetching flag from API");
900        let (feature_flags, _payloads) = self
901            .get_feature_flags(distinct_id_str, groups, person_properties, group_properties)
902            .await?;
903        Ok(feature_flags.get(&key_str).cloned())
904    }
905
906    /// Check if a feature flag is enabled for a user.
907    ///
908    /// # Returns
909    ///
910    /// `true` for `FlagValue::Boolean(true)` or any multivariate variant,
911    /// `false` for disabled or missing flags.
912    ///
913    /// # Errors
914    ///
915    /// Returns errors from [`Client::get_feature_flag`].
916    #[must_use = "feature flag enabled check result should be used"]
917    #[deprecated(
918        since = "0.6.0",
919        note = "Use Client::evaluate_flags() to fetch a snapshot, then call .is_enabled(key) \
920                on it. The snapshot deduplicates $feature_flag_called events and supports \
921                attaching rich metadata to captured events via Event::with_flags()."
922    )]
923    #[allow(deprecated)] // calls deprecated get_feature_flag internally
924    pub async fn is_feature_enabled<K: Into<String>, D: Into<String>>(
925        &self,
926        key: K,
927        distinct_id: D,
928        groups: Option<HashMap<String, String>>,
929        person_properties: Option<HashMap<String, serde_json::Value>>,
930        group_properties: Option<HashMap<String, HashMap<String, serde_json::Value>>>,
931    ) -> Result<bool, Error> {
932        let flag_value = self
933            .get_feature_flag(
934                key.into(),
935                distinct_id.into(),
936                groups,
937                person_properties,
938                group_properties,
939            )
940            .await?;
941        Ok(match flag_value {
942            Some(FlagValue::Boolean(b)) => b,
943            Some(FlagValue::String(_)) => true, // Variants are considered enabled
944            None => false,
945        })
946    }
947
948    /// Get a feature flag payload for a user.
949    ///
950    /// # Parameters
951    ///
952    /// - `key`: Feature flag key.
953    /// - `distinct_id`: User distinct ID.
954    ///
955    /// # Returns
956    ///
957    /// The JSON payload for the flag, if one was returned. This method does not
958    /// emit `$feature_flag_called` events.
959    ///
960    /// # Errors
961    ///
962    /// Returns [`Error::Connection`] for request failures and
963    /// [`Error::Serialization`] when the response cannot be parsed.
964    #[must_use = "feature flag payload result should be used"]
965    #[deprecated(
966        since = "0.6.0",
967        note = "Use Client::evaluate_flags() to fetch a snapshot, then call \
968                .get_flag_payload(key) on it. Reading the payload from a snapshot is \
969                event-free, matching this method's behavior, and avoids the per-call \
970                /flags request."
971    )]
972    pub async fn get_feature_flag_payload<K: Into<String>, D: Into<String>>(
973        &self,
974        key: K,
975        distinct_id: D,
976    ) -> Result<Option<serde_json::Value>, Error> {
977        if self.options.is_disabled() {
978            trace!("Client is disabled, skipping feature flag payload request");
979            return Ok(None);
980        }
981
982        let key_str = key.into();
983        let flags_endpoint = self.options.endpoints().build_url(Endpoint::Flags);
984
985        let mut payload = json!({
986            "api_key": self.options.api_key,
987            "distinct_id": distinct_id.into(),
988        });
989
990        // Add geoip disable parameter if configured
991        if self.options.disable_geoip {
992            payload["disable_geoip"] = json!(true);
993        }
994
995        let distinct_id = payload.get("distinct_id").and_then(|v| v.as_str());
996        let response = match self
997            .client
998            .post(&flags_endpoint)
999            .header(CONTENT_TYPE, "application/json")
1000            .header(USER_AGENT, get_default_user_agent())
1001            .json(&payload)
1002            .timeout(Duration::from_secs(
1003                self.options.feature_flags_request_timeout_seconds,
1004            ))
1005            .send()
1006            .await
1007        {
1008            Ok(r) => r,
1009            Err(e) => {
1010                let err = Error::Connection(e.to_string());
1011                report_flags_error(
1012                    &self.options.on_error,
1013                    &flags_endpoint,
1014                    distinct_id,
1015                    None,
1016                    None,
1017                    &err,
1018                );
1019                return Err(err);
1020            }
1021        };
1022
1023        if !response.status().is_success() {
1024            return Ok(None);
1025        }
1026
1027        let status = response.status().as_u16();
1028        let flags_response: FeatureFlagsResponse = match response.json().await {
1029            Ok(r) => r,
1030            Err(e) => {
1031                let err = Error::Serialization(format!("Failed to parse response: {e}"));
1032                report_flags_error(
1033                    &self.options.on_error,
1034                    &flags_endpoint,
1035                    distinct_id,
1036                    Some(status),
1037                    None,
1038                    &err,
1039                );
1040                return Err(err);
1041            }
1042        };
1043
1044        let (_flags, payloads) = flags_response.normalize();
1045        Ok(payloads.get(&key_str).cloned())
1046    }
1047
1048    /// Evaluate a supplied feature flag definition locally.
1049    ///
1050    /// `groups` and `group_properties` are only consulted when the flag (or one
1051    /// of its conditions) targets a group; pass empty maps for person flags.
1052    ///
1053    /// # Parameters
1054    ///
1055    /// - `flag`: Feature flag definition to evaluate.
1056    /// - `distinct_id`: User distinct ID.
1057    /// - `person_properties`: Person properties available to release
1058    ///   conditions.
1059    /// - `groups`: Group keys for group-targeted flags.
1060    /// - `group_properties`: Group properties for group-targeted release
1061    ///   conditions.
1062    ///
1063    /// # Errors
1064    ///
1065    /// Returns [`Error::InconclusiveMatch`] when the flag cannot be evaluated
1066    /// locally with the supplied context.
1067    #[allow(clippy::too_many_arguments)]
1068    pub fn evaluate_feature_flag_locally(
1069        &self,
1070        flag: &FeatureFlag,
1071        distinct_id: &str,
1072        person_properties: &HashMap<String, serde_json::Value>,
1073        groups: &HashMap<String, String>,
1074        group_properties: &HashMap<String, HashMap<String, serde_json::Value>>,
1075    ) -> Result<FlagValue, Error> {
1076        let group_type_mapping = self
1077            .local_evaluator
1078            .as_ref()
1079            .map(|ev| ev.cache().get_group_type_mapping())
1080            .unwrap_or_default();
1081        match_feature_flag(
1082            flag,
1083            distinct_id,
1084            person_properties,
1085            groups,
1086            group_properties,
1087            &group_type_mapping,
1088        )
1089        .map_err(|e| Error::InconclusiveMatch(e.message))
1090    }
1091
1092    /// Evaluate feature flags for `distinct_id`, returning a
1093    /// [`FeatureFlagEvaluations`] snapshot.
1094    ///
1095    /// Each `is_enabled` / `get_flag` call on the returned snapshot fires a
1096    /// dedup-aware `$feature_flag_called` event with full metadata, and the
1097    /// snapshot can be passed to [`Event::with_flags`] so a downstream
1098    /// [`Client::capture`] inherits `$feature/<key>` and `$active_feature_flags`
1099    /// without an extra `/flags` request.
1100    ///
1101    /// # Parameters
1102    ///
1103    /// - `distinct_id`: User distinct ID. Empty values return an empty snapshot.
1104    /// - `options`: Optional groups, properties, GeoIP override, local-only
1105    ///   mode, and flag-key filtering.
1106    ///
1107    /// # Errors
1108    ///
1109    /// Returns [`Error::Connection`] or [`Error::Serialization`] when remote
1110    /// evaluation is required and the `/flags` request fails before any local
1111    /// results are available.
1112    ///
1113    /// [`Event::with_flags`]: crate::Event::with_flags
1114    pub async fn evaluate_flags<S: Into<String>>(
1115        &self,
1116        distinct_id: S,
1117        options: EvaluateFlagsOptions,
1118    ) -> Result<FeatureFlagEvaluations, Error> {
1119        let distinct_id: String = distinct_id.into();
1120        let host = self.flag_event_host();
1121        if distinct_id.is_empty() || self.options.is_disabled() {
1122            return Ok(FeatureFlagEvaluations::empty(host));
1123        }
1124
1125        if options.flag_keys.as_ref().is_some_and(Vec::is_empty) {
1126            return Ok(FeatureFlagEvaluations::new(
1127                host,
1128                distinct_id,
1129                HashMap::new(),
1130                options.groups.unwrap_or_default(),
1131                options.disable_geoip,
1132                None,
1133                None,
1134                false,
1135                false,
1136            ));
1137        }
1138
1139        let mut options = options;
1140        options.groups.get_or_insert_with(HashMap::new);
1141        options.group_properties.get_or_insert_with(HashMap::new);
1142
1143        let mut records: HashMap<String, EvaluatedFlagRecord> = HashMap::new();
1144        let mut locally_evaluated_keys: HashSet<String> = HashSet::new();
1145
1146        if let Some(evaluator) = &self.local_evaluator {
1147            let mut person_props_owned = options.person_properties.clone().unwrap_or_default();
1148            person_props_owned
1149                .entry("distinct_id".to_string())
1150                .or_insert_with(|| json!(distinct_id.clone()));
1151            let groups_owned = options.groups.clone().unwrap_or_default();
1152            let group_props_owned = options.group_properties.clone().unwrap_or_default();
1153            let local_results = evaluator.evaluate_all_flags_with_details(
1154                &distinct_id,
1155                &person_props_owned,
1156                &groups_owned,
1157                &group_props_owned,
1158            );
1159            // Pin the gate from the poller's current definitions snapshot at the
1160            // point local evaluation succeeded, so it travels with these records
1161            // rather than being re-read from shared state at event time.
1162            let local_minimal_gate = evaluator.cache().minimal_flag_called_events();
1163            for (key, result) in local_results {
1164                if let Some(filter) = &options.flag_keys {
1165                    if !filter.iter().any(|k| k == &key) {
1166                        continue;
1167                    }
1168                }
1169                if let Ok(value) = result.result {
1170                    records.insert(
1171                        key.clone(),
1172                        local_record(
1173                            value,
1174                            result.payload,
1175                            result.has_experiment,
1176                            local_minimal_gate,
1177                        ),
1178                    );
1179                    locally_evaluated_keys.insert(key);
1180                }
1181            }
1182        }
1183
1184        let mut request_id: Option<String> = None;
1185        let mut errors_while_computing = false;
1186        let mut quota_limited = false;
1187
1188        // Skip the remote round-trip when local evaluation has already covered
1189        // every requested flag. Without `flag_keys` we have to assume the caller
1190        // wants every flag the project has and still hit `/flags` to discover
1191        // any not loaded by the poller.
1192        let local_covers_request = options
1193            .flag_keys
1194            .as_ref()
1195            .is_some_and(|keys| keys.iter().all(|k| locally_evaluated_keys.contains(k)));
1196
1197        if !options.only_evaluate_locally && !local_covers_request {
1198            // Don't lose successful local evaluations if `/flags` fails — degrade
1199            // to a snapshot built from the local results we already have. The
1200            // alternative (returning Err) wastes useful data and surprises
1201            // callers who would otherwise get partial coverage.
1202            match self.fetch_flag_details(&distinct_id, &options).await {
1203                Ok(response) => {
1204                    request_id = response.request_id;
1205                    errors_while_computing = response.errors_while_computing_flags;
1206                    quota_limited = response.quota_limited;
1207                    // The remote response is the source of these flags' values,
1208                    // so it is also the source of their minimization gate.
1209                    let remote_minimal_gate = response.minimal_flag_called_events;
1210                    for (key, detail) in response.flags {
1211                        if locally_evaluated_keys.contains(&key) {
1212                            continue;
1213                        }
1214                        records.insert(key, remote_record_from_detail(detail, remote_minimal_gate));
1215                    }
1216                }
1217                Err(e) => {
1218                    if records.is_empty() {
1219                        return Err(e);
1220                    }
1221                    debug!(
1222                        error = e.to_string(),
1223                        local_count = records.len(),
1224                        "/flags fetch failed; returning snapshot from local results only"
1225                    );
1226                    errors_while_computing = true;
1227                }
1228            }
1229        }
1230
1231        Ok(FeatureFlagEvaluations::new(
1232            host,
1233            distinct_id,
1234            records,
1235            options.groups.unwrap_or_default(),
1236            options.disable_geoip,
1237            request_id,
1238            None,
1239            errors_while_computing,
1240            quota_limited,
1241        ))
1242    }
1243
1244    fn flag_event_host(&self) -> Arc<dyn FeatureFlagEvaluationsHost> {
1245        self.flag_event_host
1246            .get_or_init(|| {
1247                Arc::new(AsyncFlagEventHost::from_options(
1248                    &self.options,
1249                    self.transport.clone(),
1250                )) as Arc<dyn FeatureFlagEvaluationsHost>
1251            })
1252            .clone()
1253    }
1254
1255    async fn send_feature_flags_request(
1256        &self,
1257        flags_endpoint: &str,
1258        payload: &serde_json::Value,
1259    ) -> Result<reqwest::Response, Error> {
1260        let mut attempt = 1;
1261        loop {
1262            let request = self
1263                .client
1264                .post(flags_endpoint)
1265                .header(CONTENT_TYPE, "application/json")
1266                .header(USER_AGENT, get_default_user_agent())
1267                .json(payload)
1268                .timeout(Duration::from_secs(
1269                    self.options.feature_flags_request_timeout_seconds,
1270                ));
1271            #[cfg(feature = "test-harness")]
1272            let request = {
1273                let mut request = request;
1274                if let Some(ref extra) = self.options.extra_capture_headers {
1275                    for (k, v) in extra {
1276                        request = request.header(k.as_str(), v.as_str());
1277                    }
1278                }
1279                request
1280            };
1281            let result = request.send().await;
1282
1283            match result {
1284                Ok(response) => match super::retry::feature_flags_after_response(
1285                    &self.options,
1286                    attempt,
1287                    response.status().as_u16(),
1288                ) {
1289                    super::retry::FeatureFlagsResponseStep::Backoff(delay) => {
1290                        tokio::time::sleep(delay).await;
1291                        attempt += 1;
1292                    }
1293                    super::retry::FeatureFlagsResponseStep::Done => return Ok(response),
1294                },
1295                Err(e) => {
1296                    let err_msg = e.to_string();
1297                    match super::retry::feature_flags_after_transport_error(
1298                        &self.options,
1299                        attempt,
1300                        is_retryable_feature_flags_error(&e),
1301                        err_msg,
1302                    ) {
1303                        super::retry::FeatureFlagsTransportStep::Backoff(delay) => {
1304                            tokio::time::sleep(delay).await;
1305                            attempt += 1;
1306                        }
1307                        super::retry::FeatureFlagsTransportStep::Fail(err) => {
1308                            report_flags_error(
1309                                &self.options.on_error,
1310                                flags_endpoint,
1311                                payload.get("distinct_id").and_then(|v| v.as_str()),
1312                                None,
1313                                None,
1314                                &err,
1315                            );
1316                            return Err(err);
1317                        }
1318                    }
1319                }
1320            }
1321        }
1322    }
1323
1324    async fn fetch_flag_details(
1325        &self,
1326        distinct_id: &str,
1327        options: &EvaluateFlagsOptions,
1328    ) -> Result<DetailedFlagsResponse, Error> {
1329        let flags_endpoint = self.options.endpoints().build_url(Endpoint::Flags);
1330
1331        let person_properties = options.person_properties.clone().unwrap_or_default();
1332        let groups = options.groups.clone().unwrap_or_default();
1333        let group_properties = options.group_properties.clone().unwrap_or_default();
1334        let effective_disable_geoip = options.disable_geoip.unwrap_or(self.options.disable_geoip);
1335
1336        let mut payload = json!({
1337            "api_key": self.options.api_key,
1338            "distinct_id": distinct_id,
1339            "groups": groups,
1340            "person_properties": person_properties,
1341            "group_properties": group_properties,
1342            "geoip_disable": effective_disable_geoip,
1343        });
1344        if let Some(flag_keys) = &options.flag_keys {
1345            payload["flag_keys_to_evaluate"] = json!(flag_keys);
1346        }
1347
1348        let response = self
1349            .send_feature_flags_request(&flags_endpoint, &payload)
1350            .await?;
1351
1352        if !response.status().is_success() {
1353            let status = response.status();
1354            let text = response
1355                .text()
1356                .await
1357                .unwrap_or_else(|_| "Unknown error".to_string());
1358            let err = Error::Connection(format!("API request failed with status {status}: {text}"));
1359            report_flags_error(
1360                &self.options.on_error,
1361                &flags_endpoint,
1362                Some(distinct_id),
1363                Some(status.as_u16()),
1364                Some(&text),
1365                &err,
1366            );
1367            return Err(err);
1368        }
1369
1370        let status = response.status().as_u16();
1371        let parsed = match response.json::<FeatureFlagsResponse>().await {
1372            Ok(p) => p,
1373            Err(e) => {
1374                let err =
1375                    Error::Serialization(format!("Failed to parse feature flags response: {e}"));
1376                report_flags_error(
1377                    &self.options.on_error,
1378                    &flags_endpoint,
1379                    Some(distinct_id),
1380                    Some(status),
1381                    None,
1382                    &err,
1383                );
1384                return Err(err);
1385            }
1386        };
1387        Ok(extract_flag_details(parsed))
1388    }
1389}
1390
1391impl Drop for Client {
1392    /// Best-effort flush and worker join on drop. A blocking drain (the async
1393    /// `shutdown` can't run in a destructor), so dropping a `Client` inside an
1394    /// async task blocks that executor thread until the drain completes (up to
1395    /// `shutdown_timeout_ms` plus any in-flight request) — prefer an explicit
1396    /// `shutdown().await` first, which makes this a no-op. A drop from a transport
1397    /// callback instead queues shutdown without waiting for or joining the current
1398    /// worker thread.
1399    fn drop(&mut self) {
1400        if let Some(transport) = &self.transport {
1401            transport.close_blocking();
1402        }
1403    }
1404}
1405
1406#[cfg(test)]
1407mod teardown_tests {
1408    use super::*;
1409    use std::sync::{mpsc, Mutex};
1410
1411    #[tokio::test]
1412    async fn shutdown_from_worker_callback_closes_without_blocking() {
1413        let client_slot = Arc::new(Mutex::new(None::<Arc<Client>>));
1414        let callback_slot = Arc::clone(&client_slot);
1415        let (shutdown_tx, shutdown_rx) = mpsc::channel();
1416        let options = crate::ClientOptionsBuilder::default()
1417            .api_key("phc_test".to_string())
1418            .host("http://localhost:0".to_string())
1419            .flush_at(1usize)
1420            .before_send(move |_| {
1421                let client = Arc::clone(
1422                    callback_slot
1423                        .lock()
1424                        .unwrap_or_else(|p| p.into_inner())
1425                        .as_ref()
1426                        .expect("client installed before capture"),
1427                );
1428                futures::executor::block_on(client.shutdown());
1429                shutdown_tx.send(()).unwrap();
1430                None
1431            })
1432            .build()
1433            .unwrap();
1434        let client = Arc::new(client(options).await);
1435        *client_slot.lock().unwrap_or_else(|p| p.into_inner()) = Some(Arc::clone(&client));
1436
1437        client.capture(Event::new("shutdown-in-callback", "user-1"));
1438        shutdown_rx
1439            .recv_timeout(Duration::from_secs(2))
1440            .expect("client shutdown blocked its transport worker");
1441        client_slot.lock().unwrap_or_else(|p| p.into_inner()).take();
1442
1443        // Reap the worker externally and verify repeated shutdown is safe.
1444        client.shutdown().await;
1445        client.shutdown().await;
1446        assert!(client.transport.as_ref().unwrap().is_closed());
1447    }
1448
1449    #[tokio::test]
1450    async fn drop_from_worker_callback_closes_without_blocking() {
1451        let client_slot = Arc::new(Mutex::new(None::<Client>));
1452        let callback_slot = Arc::clone(&client_slot);
1453        let (dropped_tx, dropped_rx) = mpsc::channel();
1454        let options = crate::ClientOptionsBuilder::default()
1455            .api_key("phc_test".to_string())
1456            .host("http://localhost:0".to_string())
1457            .flush_at(1usize)
1458            .before_send(move |_| {
1459                let client = callback_slot
1460                    .lock()
1461                    .unwrap_or_else(|p| p.into_inner())
1462                    .take()
1463                    .expect("client installed before capture");
1464                drop(client);
1465                dropped_tx.send(()).unwrap();
1466                None
1467            })
1468            .build()
1469            .unwrap();
1470        let client = client(options).await;
1471        let transport = Arc::clone(client.transport.as_ref().unwrap());
1472        *client_slot.lock().unwrap_or_else(|p| p.into_inner()) = Some(client);
1473
1474        client_slot
1475            .lock()
1476            .unwrap_or_else(|p| p.into_inner())
1477            .as_ref()
1478            .unwrap()
1479            .capture(Event::new("drop-in-callback", "user-1"));
1480        dropped_rx
1481            .recv_timeout(Duration::from_secs(2))
1482            .expect("client drop blocked its transport worker");
1483
1484        // Reap the worker from an external thread and verify repeated close is safe.
1485        transport.close_blocking();
1486        transport.close_blocking();
1487        assert!(transport.is_closed());
1488    }
1489}
1490
1491#[cfg(test)]
1492mod minimal_gate_tests {
1493    use super::*;
1494    use crate::client::minimal_gate_test_support::{definitions, RecordingHost};
1495
1496    fn test_client(cache: FlagCache, host: Arc<dyn FeatureFlagEvaluationsHost>) -> Client {
1497        let options = ClientOptions::from(("phc_test", "http://localhost:0"));
1498        let client = Client {
1499            options,
1500            client: HttpClient::builder().build().unwrap(),
1501            local_evaluator: Some(LocalEvaluator::new(cache)),
1502            _flag_poller: None,
1503            flag_event_host: OnceLock::new(),
1504            transport: None,
1505        };
1506        client
1507            .flag_event_host
1508            .set(host)
1509            .unwrap_or_else(|_| panic!("host already set"));
1510        client
1511    }
1512
1513    async fn evaluate(client: &Client) -> FeatureFlagEvaluations {
1514        client
1515            .evaluate_flags(
1516                "user-1",
1517                EvaluateFlagsOptions {
1518                    only_evaluate_locally: true,
1519                    ..Default::default()
1520                },
1521            )
1522            .await
1523            .expect("local evaluate_flags")
1524    }
1525
1526    /// The minimization gate must be pinned to the definitions snapshot that
1527    /// produced the flag value, not re-read from the shared cache when the
1528    /// deferred event finally fires. Mutating the cache in the gap between
1529    /// evaluation and event capture must not reshape the event.
1530    #[tokio::test]
1531    async fn local_gate_pinned_at_evaluation_survives_cache_mutation_to_off() {
1532        let cache = FlagCache::new();
1533        cache.update(definitions(Some(false), true)); // gate ON at evaluation
1534        let host = Arc::new(RecordingHost::default());
1535        let client = test_client(cache.clone(), Arc::clone(&host) as _);
1536
1537        let snapshot = evaluate(&client).await;
1538        // Poller refresh flips the gate OFF after the snapshot was produced.
1539        cache.update(definitions(Some(false), false));
1540
1541        assert!(snapshot.is_enabled("gated"));
1542        let captured = host.captured.lock().unwrap();
1543        assert_eq!(captured.len(), 1);
1544        assert!(
1545            captured[0].minimal,
1546            "event must reflect the gate pinned at evaluation (on), not the mutated cache (off)"
1547        );
1548    }
1549
1550    #[tokio::test]
1551    async fn local_gate_pinned_at_evaluation_survives_cache_mutation_to_on() {
1552        let cache = FlagCache::new();
1553        cache.update(definitions(Some(false), false)); // gate OFF at evaluation
1554        let host = Arc::new(RecordingHost::default());
1555        let client = test_client(cache.clone(), Arc::clone(&host) as _);
1556
1557        let snapshot = evaluate(&client).await;
1558        // Poller refresh flips the gate ON after the snapshot was produced.
1559        cache.update(definitions(Some(false), true));
1560
1561        assert!(snapshot.is_enabled("gated"));
1562        let captured = host.captured.lock().unwrap();
1563        assert_eq!(captured.len(), 1);
1564        assert!(
1565            !captured[0].minimal,
1566            "event must reflect the gate pinned at evaluation (off), not the mutated cache (on)"
1567        );
1568    }
1569
1570    #[tokio::test]
1571    async fn local_has_experiment_is_threaded_from_definitions() {
1572        let cache = FlagCache::new();
1573        cache.update(definitions(Some(false), true));
1574        let host = Arc::new(RecordingHost::default());
1575        let client = test_client(cache, Arc::clone(&host) as _);
1576
1577        assert!(evaluate(&client).await.is_enabled("gated"));
1578        let captured = host.captured.lock().unwrap();
1579        assert_eq!(
1580            captured[0].properties.get("$feature_flag_has_experiment"),
1581            Some(&serde_json::json!(false))
1582        );
1583        assert!(captured[0].minimal);
1584    }
1585}
1586
1587#[cfg(test)]
1588mod local_payload_tests {
1589    use super::*;
1590    use crate::client::local_payload_test_support::payload_definitions;
1591    use crate::client::minimal_gate_test_support::RecordingHost;
1592    use serde_json::json;
1593
1594    async fn snapshot() -> FeatureFlagEvaluations {
1595        let cache = FlagCache::new();
1596        cache.update(payload_definitions());
1597        let options = ClientOptions::from(("phc_test", "http://localhost:0"));
1598        let client = Client {
1599            options,
1600            client: HttpClient::builder().build().unwrap(),
1601            local_evaluator: Some(LocalEvaluator::new(cache)),
1602            _flag_poller: None,
1603            flag_event_host: OnceLock::new(),
1604            transport: None,
1605        };
1606        client
1607            .flag_event_host
1608            .set(Arc::new(RecordingHost::default()) as _)
1609            .unwrap_or_else(|_| panic!("host already set"));
1610        client
1611            .evaluate_flags(
1612                "user-1",
1613                EvaluateFlagsOptions {
1614                    only_evaluate_locally: true,
1615                    ..Default::default()
1616                },
1617            )
1618            .await
1619            .expect("local evaluate_flags")
1620    }
1621
1622    /// Payloads live in the definitions manifest, so local evaluation must
1623    /// surface them the same way `/flags` does — including the JSON decoding,
1624    /// or the same flag would yield different payloads depending on which path
1625    /// evaluated it.
1626    #[tokio::test]
1627    async fn local_evaluation_surfaces_payloads_matching_the_remote_shape() {
1628        let snapshot = snapshot().await;
1629
1630        assert_eq!(
1631            snapshot.get_flag_payload("json-string-payload"),
1632            Some(json!({"color": "blue"}))
1633        );
1634        assert_eq!(
1635            snapshot.get_flag_payload("parsed-payload"),
1636            Some(json!({"color": "blue"}))
1637        );
1638        assert_eq!(
1639            snapshot.get_flag_payload("quoted-string-payload"),
1640            Some(json!("just text"))
1641        );
1642        assert_eq!(
1643            snapshot.get_flag_payload("undecodable-payload"),
1644            Some(json!("not json"))
1645        );
1646    }
1647
1648    #[tokio::test]
1649    async fn local_payload_is_keyed_by_the_matched_variant() {
1650        let snapshot = snapshot().await;
1651
1652        assert_eq!(
1653            snapshot.get_flag("variant-payload"),
1654            Some(FlagValue::String("test".to_string()))
1655        );
1656        assert_eq!(
1657            snapshot.get_flag_payload("variant-payload"),
1658            Some(json!({"tier": 2}))
1659        );
1660    }
1661
1662    #[tokio::test]
1663    async fn local_payload_is_absent_without_a_matching_payload() {
1664        let snapshot = snapshot().await;
1665
1666        assert_eq!(snapshot.get_flag_payload("no-payload"), None);
1667        assert_eq!(snapshot.get_flag_payload("not-a-flag"), None);
1668
1669        // A missing key also yields `None`, so pin the flag down first:
1670        // it was evaluated, it evaluated false, and its "true" payload
1671        // stayed behind.
1672        assert_eq!(
1673            snapshot.get_flag("disabled-with-payload"),
1674            Some(FlagValue::Boolean(false))
1675        );
1676        assert_eq!(snapshot.get_flag_payload("disabled-with-payload"), None);
1677    }
1678}
posthog_rs/client/async_client.rs

posthog_rs/client/
async_client.rs
