posthog_rs/
local_evaluation.rs

1use crate::client::{apply_on_error_hooks, get_default_user_agent, OnErrorHook};
2use crate::feature_flags::{
3    match_feature_flag, match_feature_flag_with_context, CohortDefinition, EvaluationContext,
4    FeatureFlag, FlagValue, InconclusiveMatchError,
5};
6use crate::{Error, LocalEvaluationFailure, PostHogError};
7use reqwest::header::{HeaderMap, ETAG, IF_NONE_MATCH, USER_AGENT};
8use reqwest::StatusCode;
9use serde::{Deserialize, Serialize};
10use std::collections::HashMap;
11use std::sync::atomic::{AtomicBool, Ordering};
12use std::sync::{Arc, RwLock};
13use std::time::Duration;
14use tracing::{debug, error, info, instrument, trace, warn};
15
16/// Extract the ETag header value from a response's headers.
17/// Returns None if the header is missing, invalid UTF-8, or empty.
18fn extract_etag(headers: &HeaderMap) -> Option<String> {
19    headers
20        .get(ETAG)
21        .and_then(|v| v.to_str().ok())
22        .filter(|s| !s.is_empty())
23        .map(|s| s.to_string())
24}
25
26/// Sleep up to `duration`, waking early when `stop_signal` is set. Returns
27/// `true` if a stop was requested (either already pending or observed while
28/// waiting). Polling in short steps keeps shutdown latency bounded even when
29/// the poll interval is large — a plain `sleep(poll_interval)` would make
30/// `stop`/`Drop` block for the remainder of the current interval.
31fn sleep_until_stop(stop_signal: &AtomicBool, duration: Duration) -> bool {
32    const STEP: Duration = Duration::from_millis(200);
33    let mut remaining = duration;
34    while !remaining.is_zero() {
35        if stop_signal.load(Ordering::Relaxed) {
36            return true;
37        }
38        let step = remaining.min(STEP);
39        std::thread::sleep(step);
40        remaining -= step;
41    }
42    stop_signal.load(Ordering::Relaxed)
43}
44
45/// Fire the `on_error` hooks for a failed definitions poll. The personal API
46/// key is never included — only the cause and HTTP status are surfaced.
47fn report_local_eval_error(hooks: &[OnErrorHook], status: Option<u16>, error: &Error) {
48    if hooks.is_empty() {
49        return;
50    }
51    let failure = PostHogError::LocalEvaluation(LocalEvaluationFailure { error, status });
52    apply_on_error_hooks(hooks, &failure);
53}
54
55/// Response from the PostHog local evaluation API.
56///
57/// Contains feature flag definitions, group type mappings, and cohort definitions
58/// that can be cached locally for flag evaluation without server round-trips.
59#[derive(Debug, Clone, Serialize, Deserialize)]
60pub struct LocalEvaluationResponse {
61    /// List of feature flag definitions
62    pub flags: Vec<FeatureFlag>,
63    /// Mapping from group type keys to their display names
64    #[serde(default)]
65    pub group_type_mapping: HashMap<String, String>,
66    /// Cohort definitions for evaluating cohort membership
67    #[serde(default)]
68    pub cohorts: HashMap<String, Cohort>,
69}
70
71/// A cohort definition for local evaluation.
72///
73/// Cohorts are groups of users defined by property filters, used for
74/// targeting feature flags to specific user segments.
75#[derive(Debug, Clone, Serialize, Deserialize)]
76pub struct Cohort {
77    /// Unique identifier for this cohort
78    pub id: String,
79    /// Human-readable name of the cohort
80    pub name: String,
81    /// Property filters that define cohort membership
82    pub properties: serde_json::Value,
83}
84
85/// Thread-safe cache for feature flag definitions.
86///
87/// Stores feature flags, group type mappings, and cohort definitions that have
88/// been fetched from the PostHog API. The cache is shared between the poller
89/// (which updates it) and the evaluator (which reads from it).
90#[derive(Clone)]
91pub struct FlagCache {
92    flags: Arc<RwLock<HashMap<String, FeatureFlag>>>,
93    group_type_mapping: Arc<RwLock<HashMap<String, String>>>,
94    cohorts: Arc<RwLock<HashMap<String, Cohort>>>,
95}
96
97impl Default for FlagCache {
98    fn default() -> Self {
99        Self::new()
100    }
101}
102
103impl FlagCache {
104    /// Create an empty shared flag cache.
105    pub fn new() -> Self {
106        Self {
107            flags: Arc::new(RwLock::new(HashMap::new())),
108            group_type_mapping: Arc::new(RwLock::new(HashMap::new())),
109            cohorts: Arc::new(RwLock::new(HashMap::new())),
110        }
111    }
112
113    /// Replace cached flags, group type mappings, and cohorts from a local
114    /// evaluation API response.
115    pub fn update(&self, response: LocalEvaluationResponse) {
116        let flag_count = response.flags.len();
117        let mut flags = self.flags.write().unwrap();
118        flags.clear();
119        for flag in response.flags {
120            flags.insert(flag.key.clone(), flag);
121        }
122
123        let mut mapping = self.group_type_mapping.write().unwrap();
124        *mapping = response.group_type_mapping;
125
126        let mut cohorts = self.cohorts.write().unwrap();
127        *cohorts = response.cohorts;
128
129        debug!(flag_count, "Updated flag cache");
130    }
131
132    /// Return a cached feature flag by key.
133    pub fn get_flag(&self, key: &str) -> Option<FeatureFlag> {
134        self.flags.read().unwrap().get(key).cloned()
135    }
136
137    /// Return all cached feature flag definitions.
138    pub fn get_all_flags(&self) -> Vec<FeatureFlag> {
139        self.flags.read().unwrap().values().cloned().collect()
140    }
141
142    /// Return a cached cohort by ID.
143    pub fn get_cohort(&self, id: &str) -> Option<Cohort> {
144        self.cohorts.read().unwrap().get(id).cloned()
145    }
146
147    /// Return all cached cohorts, keyed by cohort ID.
148    pub fn get_all_cohorts(&self) -> HashMap<String, Cohort> {
149        self.cohorts.read().unwrap().clone()
150    }
151
152    /// Get all cohorts as CohortDefinitions for evaluation context
153    pub fn get_cohort_definitions(&self) -> HashMap<String, CohortDefinition> {
154        self.cohorts
155            .read()
156            .unwrap()
157            .iter()
158            .map(|(k, v)| {
159                (
160                    k.clone(),
161                    CohortDefinition {
162                        id: v.id.clone(),
163                        properties: v.properties.clone(),
164                    },
165                )
166            })
167            .collect()
168    }
169
170    /// Get all flags as a HashMap for evaluation context
171    pub fn get_flags_map(&self) -> HashMap<String, FeatureFlag> {
172        self.flags.read().unwrap().clone()
173    }
174
175    /// Get the group type mapping (group type index → group type name).
176    pub fn get_group_type_mapping(&self) -> HashMap<String, String> {
177        self.group_type_mapping.read().unwrap().clone()
178    }
179
180    /// Remove all cached flags, group type mappings, and cohorts.
181    pub fn clear(&self) {
182        self.flags.write().unwrap().clear();
183        self.group_type_mapping.write().unwrap().clear();
184        self.cohorts.write().unwrap().clear();
185    }
186}
187
188/// Configuration for local flag evaluation.
189///
190/// Specifies the credentials and settings needed to fetch feature flag
191/// definitions from the PostHog API for local evaluation.
192#[derive(Clone)]
193pub struct LocalEvaluationConfig {
194    /// Personal API key for authentication (found in PostHog project settings)
195    pub personal_api_key: String,
196    /// Project API key to identify which project's flags to fetch
197    pub project_api_key: String,
198    /// PostHog API host URL (for example, `https://us.i.posthog.com`).
199    /// Use `https://eu.i.posthog.com` for EU-hosted projects.
200    pub api_host: String,
201    /// How often to poll for updated flag definitions
202    pub poll_interval: Duration,
203    /// Timeout for API requests
204    pub request_timeout: Duration,
205}
206
207/// Synchronous poller for feature flag definitions.
208///
209/// Runs a background thread that periodically fetches flag definitions from
210/// the PostHog API and updates the shared cache. Use this for blocking/sync
211/// applications. With the `async-client` feature enabled, use
212/// [`AsyncFlagPoller`] for async applications instead.
213pub struct FlagPoller {
214    config: LocalEvaluationConfig,
215    cache: FlagCache,
216    client: reqwest::blocking::Client,
217    stop_signal: Arc<AtomicBool>,
218    thread_handle: Option<std::thread::JoinHandle<()>>,
219    /// Observability hooks, injected by the client builder before `start`.
220    /// Kept here rather than on `LocalEvaluationConfig` so the public config
221    /// struct stays unchanged.
222    on_error: Vec<OnErrorHook>,
223}
224
225impl FlagPoller {
226    /// Create a synchronous flag definition poller.
227    ///
228    /// # Parameters
229    ///
230    /// - `config`: Credentials, host, polling interval, and request timeout.
231    /// - `cache`: Shared cache updated by the poller.
232    pub fn new(config: LocalEvaluationConfig, cache: FlagCache) -> Self {
233        let client = reqwest::blocking::Client::builder()
234            .timeout(config.request_timeout)
235            .build()
236            .unwrap();
237
238        Self {
239            config,
240            cache,
241            client,
242            stop_signal: Arc::new(AtomicBool::new(false)),
243            thread_handle: None,
244            on_error: Vec::new(),
245        }
246    }
247
248    /// Register `on_error` hooks. Called by the client builder before
249    /// [`FlagPoller::start`]; not part of the public flag-poller API.
250    // Only the blocking client (built when `async-client` is off) injects hooks.
251    #[cfg_attr(feature = "async-client", allow(dead_code))]
252    pub(crate) fn set_on_error(&mut self, hooks: Vec<OnErrorHook>) {
253        self.on_error = hooks;
254    }
255
256    /// Start the polling thread.
257    ///
258    /// Performs an initial synchronous load, then refreshes definitions in the
259    /// background until [`FlagPoller::stop`] is called or the poller is dropped.
260    pub fn start(&mut self) {
261        info!(
262            poll_interval_secs = self.config.poll_interval.as_secs(),
263            "Starting feature flag poller"
264        );
265
266        // Initial load
267        match self.load_flags() {
268            Ok(()) => info!("Initial flag definitions loaded successfully"),
269            Err(e) => warn!(error = %e, "Failed to load initial flags, will retry on next poll"),
270        }
271
272        let config = self.config.clone();
273        let cache = self.cache.clone();
274        let stop_signal = self.stop_signal.clone();
275        let on_error = self.on_error.clone();
276
277        let handle = std::thread::spawn(move || {
278            let client = reqwest::blocking::Client::builder()
279                .timeout(config.request_timeout)
280                .build()
281                .unwrap();
282
283            let mut last_etag: Option<String> = None;
284
285            loop {
286                if sleep_until_stop(&stop_signal, config.poll_interval) {
287                    debug!("Flag poller received stop signal");
288                    break;
289                }
290
291                let url = format!(
292                    "{}/flags/definitions/?send_cohorts",
293                    config.api_host.trim_end_matches('/')
294                );
295
296                let mut request = client
297                    .get(&url)
298                    .header(
299                        "Authorization",
300                        format!("Bearer {}", config.personal_api_key),
301                    )
302                    .header("X-PostHog-Project-Api-Key", &config.project_api_key)
303                    .header(USER_AGENT, get_default_user_agent());
304
305                if let Some(ref etag) = last_etag {
306                    request = request.header(IF_NONE_MATCH, etag.as_str());
307                }
308
309                match request.send() {
310                    Ok(response) => {
311                        let status = response.status();
312                        if status == StatusCode::NOT_MODIFIED {
313                            debug!("Flag definitions unchanged (304 Not Modified)");
314                        } else if status.is_success() {
315                            // Extract ETag before consuming the response body
316                            let new_etag = extract_etag(response.headers());
317
318                            match response.json::<LocalEvaluationResponse>() {
319                                Ok(data) => {
320                                    trace!("Successfully fetched flag definitions");
321                                    cache.update(data);
322                                    last_etag = new_etag;
323                                }
324                                Err(e) => {
325                                    warn!(error = %e, "Failed to parse flag response");
326                                    let err = Error::Serialization(e.to_string());
327                                    report_local_eval_error(&on_error, Some(status.as_u16()), &err);
328                                }
329                            }
330                        } else {
331                            warn!(status = %status, "Failed to fetch flags");
332                            let err = Error::Connection(format!("HTTP {}", status));
333                            report_local_eval_error(&on_error, Some(status.as_u16()), &err);
334                        }
335                    }
336                    Err(e) => {
337                        warn!(error = %e, "Failed to fetch flags");
338                        let err = Error::Connection(e.to_string());
339                        report_local_eval_error(&on_error, None, &err);
340                    }
341                }
342            }
343        });
344
345        self.thread_handle = Some(handle);
346    }
347
348    /// Load flags synchronously and update the cache once.
349    ///
350    /// # Errors
351    ///
352    /// Returns [`Error::Connection`] for request failures or non-success HTTP
353    /// statuses, and [`Error::Serialization`] when the response cannot be
354    /// parsed.
355    #[instrument(skip(self), level = "debug")]
356    pub fn load_flags(&self) -> Result<(), Error> {
357        let url = format!(
358            "{}/flags/definitions/?send_cohorts",
359            self.config.api_host.trim_end_matches('/')
360        );
361
362        let response = match self
363            .client
364            .get(&url)
365            .header(
366                "Authorization",
367                format!("Bearer {}", self.config.personal_api_key),
368            )
369            .header("X-PostHog-Project-Api-Key", &self.config.project_api_key)
370            .header(USER_AGENT, get_default_user_agent())
371            .send()
372        {
373            Ok(r) => r,
374            Err(e) => {
375                error!(error = %e, "Connection error loading flags");
376                let err = Error::Connection(e.to_string());
377                report_local_eval_error(&self.on_error, None, &err);
378                return Err(err);
379            }
380        };
381
382        if !response.status().is_success() {
383            let status = response.status();
384            error!(status = %status, "HTTP error loading flags");
385            let err = Error::Connection(format!("HTTP {}", status));
386            report_local_eval_error(&self.on_error, Some(status.as_u16()), &err);
387            return Err(err);
388        }
389
390        let status = response.status().as_u16();
391        let data = match response.json::<LocalEvaluationResponse>() {
392            Ok(d) => d,
393            Err(e) => {
394                error!(error = %e, "Failed to parse flag response");
395                let err = Error::Serialization(e.to_string());
396                report_local_eval_error(&self.on_error, Some(status), &err);
397                return Err(err);
398            }
399        };
400
401        self.cache.update(data);
402        Ok(())
403    }
404
405    /// Stop the polling thread and wait for it to exit.
406    pub fn stop(&mut self) {
407        debug!("Stopping flag poller");
408        self.stop_signal.store(true, Ordering::Relaxed);
409        if let Some(handle) = self.thread_handle.take() {
410            handle.join().ok();
411        }
412    }
413}
414
415impl Drop for FlagPoller {
416    fn drop(&mut self) {
417        self.stop();
418    }
419}
420
421/// Asynchronous poller for feature flag definitions.
422///
423/// Runs a tokio task that periodically fetches flag definitions from the
424/// PostHog API and updates the shared cache. Use this for async applications.
425/// For blocking/sync applications, use [`FlagPoller`] instead.
426#[cfg(feature = "async-client")]
427pub struct AsyncFlagPoller {
428    config: LocalEvaluationConfig,
429    cache: FlagCache,
430    client: reqwest::Client,
431    stop_signal: Arc<AtomicBool>,
432    task_handle: Option<tokio::task::JoinHandle<()>>,
433    is_running: Arc<tokio::sync::RwLock<bool>>,
434    /// Observability hooks, injected by the client builder before `start`.
435    /// Kept here rather than on `LocalEvaluationConfig` so the public config
436    /// struct stays unchanged.
437    on_error: Vec<OnErrorHook>,
438}
439
440#[cfg(feature = "async-client")]
441impl AsyncFlagPoller {
442    /// Create an asynchronous flag definition poller.
443    ///
444    /// # Parameters
445    ///
446    /// - `config`: Credentials, host, polling interval, and request timeout.
447    /// - `cache`: Shared cache updated by the poller.
448    pub fn new(config: LocalEvaluationConfig, cache: FlagCache) -> Self {
449        let client = reqwest::Client::builder()
450            .timeout(config.request_timeout)
451            .build()
452            .unwrap();
453
454        Self {
455            config,
456            cache,
457            client,
458            stop_signal: Arc::new(AtomicBool::new(false)),
459            task_handle: None,
460            is_running: Arc::new(tokio::sync::RwLock::new(false)),
461            on_error: Vec::new(),
462        }
463    }
464
465    /// Register `on_error` hooks. Called by the client builder before
466    /// [`AsyncFlagPoller::start`]; not part of the public flag-poller API.
467    pub(crate) fn set_on_error(&mut self, hooks: Vec<OnErrorHook>) {
468        self.on_error = hooks;
469    }
470
471    /// Start the polling task.
472    ///
473    /// Performs an initial async load, then refreshes definitions in the
474    /// background until [`AsyncFlagPoller::stop`] is called or the poller is
475    /// dropped.
476    pub async fn start(&mut self) {
477        // Check if already running
478        {
479            let mut is_running = self.is_running.write().await;
480            if *is_running {
481                debug!("Flag poller already running, skipping start");
482                return;
483            }
484            *is_running = true;
485        }
486
487        info!(
488            poll_interval_secs = self.config.poll_interval.as_secs(),
489            "Starting async feature flag poller"
490        );
491
492        // Initial load
493        match self.load_flags().await {
494            Ok(()) => info!("Initial flag definitions loaded successfully"),
495            Err(e) => warn!(error = %e, "Failed to load initial flags, will retry on next poll"),
496        }
497
498        let config = self.config.clone();
499        let cache = self.cache.clone();
500        let stop_signal = self.stop_signal.clone();
501        let is_running = self.is_running.clone();
502        let client = self.client.clone();
503        let on_error = self.on_error.clone();
504
505        let task = tokio::spawn(async move {
506            let mut interval = tokio::time::interval(config.poll_interval);
507            interval.tick().await; // Skip the first immediate tick
508
509            let mut last_etag: Option<String> = None;
510
511            loop {
512                tokio::select! {
513                    _ = interval.tick() => {
514                        if stop_signal.load(Ordering::Relaxed) {
515                            debug!("Async flag poller received stop signal");
516                            break;
517                        }
518
519                        let url = format!(
520                            "{}/flags/definitions/?send_cohorts",
521                            config.api_host.trim_end_matches('/')
522                        );
523
524                        let mut request = client
525                            .get(&url)
526                            .header("Authorization", format!("Bearer {}", config.personal_api_key))
527                            .header("X-PostHog-Project-Api-Key", &config.project_api_key)
528                            .header(USER_AGENT, get_default_user_agent());
529
530                        if let Some(ref etag) = last_etag {
531                            request = request.header(IF_NONE_MATCH, etag.as_str());
532                        }
533
534                        match request.send().await {
535                            Ok(response) => {
536                                let status = response.status();
537                                if status == StatusCode::NOT_MODIFIED {
538                                    debug!("Flag definitions unchanged (304 Not Modified)");
539                                } else if status.is_success() {
540                                    // Extract ETag before consuming the response body
541                                    let new_etag = extract_etag(response.headers());
542
543                                    match response.json::<LocalEvaluationResponse>().await {
544                                        Ok(data) => {
545                                            trace!("Successfully fetched flag definitions");
546                                            cache.update(data);
547                                            last_etag = new_etag;
548                                        }
549                                        Err(e) => {
550                                            warn!(error = %e, "Failed to parse flag response");
551                                            let err = Error::Serialization(e.to_string());
552                                            report_local_eval_error(&on_error, Some(status.as_u16()), &err);
553                                        }
554                                    }
555                                } else {
556                                    warn!(status = %status, "Failed to fetch flags");
557                                    let err = Error::Connection(format!("HTTP {}", status));
558                                    report_local_eval_error(&on_error, Some(status.as_u16()), &err);
559                                }
560                            }
561                            Err(e) => {
562                                warn!(error = %e, "Failed to fetch flags");
563                                let err = Error::Connection(e.to_string());
564                                report_local_eval_error(&on_error, None, &err);
565                            }
566                        }
567                    }
568                }
569            }
570
571            // Clear running flag when task exits
572            *is_running.write().await = false;
573        });
574
575        self.task_handle = Some(task);
576    }
577
578    /// Load flags asynchronously and update the cache once.
579    ///
580    /// # Errors
581    ///
582    /// Returns [`Error::Connection`] for request failures or non-success HTTP
583    /// statuses, and [`Error::Serialization`] when the response cannot be
584    /// parsed.
585    #[instrument(skip(self), level = "debug")]
586    pub async fn load_flags(&self) -> Result<(), Error> {
587        let url = format!(
588            "{}/flags/definitions/?send_cohorts",
589            self.config.api_host.trim_end_matches('/')
590        );
591
592        let response = match self
593            .client
594            .get(&url)
595            .header(
596                "Authorization",
597                format!("Bearer {}", self.config.personal_api_key),
598            )
599            .header("X-PostHog-Project-Api-Key", &self.config.project_api_key)
600            .header(USER_AGENT, get_default_user_agent())
601            .send()
602            .await
603        {
604            Ok(r) => r,
605            Err(e) => {
606                error!(error = %e, "Connection error loading flags");
607                let err = Error::Connection(e.to_string());
608                report_local_eval_error(&self.on_error, None, &err);
609                return Err(err);
610            }
611        };
612
613        if !response.status().is_success() {
614            let status = response.status();
615            error!(status = %status, "HTTP error loading flags");
616            let err = Error::Connection(format!("HTTP {}", status));
617            report_local_eval_error(&self.on_error, Some(status.as_u16()), &err);
618            return Err(err);
619        }
620
621        let status = response.status().as_u16();
622        let data = match response.json::<LocalEvaluationResponse>().await {
623            Ok(d) => d,
624            Err(e) => {
625                error!(error = %e, "Failed to parse flag response");
626                let err = Error::Serialization(e.to_string());
627                report_local_eval_error(&self.on_error, Some(status), &err);
628                return Err(err);
629            }
630        };
631
632        self.cache.update(data);
633        Ok(())
634    }
635
636    /// Stop the polling task.
637    pub async fn stop(&mut self) {
638        debug!("Stopping async flag poller");
639        self.stop_signal.store(true, Ordering::Relaxed);
640        if let Some(handle) = self.task_handle.take() {
641            handle.abort();
642        }
643        *self.is_running.write().await = false;
644    }
645
646    /// Check if the poller currently has a running background task.
647    pub async fn is_running(&self) -> bool {
648        *self.is_running.read().await
649    }
650}
651
652#[cfg(feature = "async-client")]
653impl Drop for AsyncFlagPoller {
654    fn drop(&mut self) {
655        // Abort the task if still running
656        if let Some(handle) = self.task_handle.take() {
657            handle.abort();
658        }
659    }
660}
661
662/// Evaluates feature flags using locally cached definitions.
663///
664/// The evaluator reads from a [`FlagCache`] to determine flag values without
665/// making network requests. Supports cohort membership checks and flag
666/// dependencies through the evaluation context.
667#[derive(Clone)]
668pub struct LocalEvaluator {
669    cache: FlagCache,
670}
671
672impl LocalEvaluator {
673    /// Create an evaluator backed by a shared [`FlagCache`].
674    pub fn new(cache: FlagCache) -> Self {
675        Self { cache }
676    }
677
678    /// Access the underlying flag cache (e.g. to read group type mappings).
679    pub fn cache(&self) -> &FlagCache {
680        &self.cache
681    }
682
683    /// Evaluate a feature flag locally with full context support.
684    ///
685    /// Supports cohort membership checks, flag dependency evaluation, and
686    /// group / mixed-targeting flags. `groups` and `group_properties` are
687    /// only consulted when the flag (or one of its conditions) targets a
688    /// group via `aggregation_group_type_index`; pass empty maps for
689    /// person-targeted flags.
690    ///
691    /// # Returns
692    ///
693    /// `Ok(Some(value))` when the flag is present and evaluated,
694    /// `Ok(None)` when the flag is absent from the cache.
695    ///
696    /// # Errors
697    ///
698    /// Returns [`InconclusiveMatchError`] when required properties, cohorts, or
699    /// dependent flags are unavailable locally.
700    #[instrument(
701        skip(self, person_properties, groups, group_properties),
702        level = "trace"
703    )]
704    pub fn evaluate_flag(
705        &self,
706        key: &str,
707        distinct_id: &str,
708        person_properties: &HashMap<String, serde_json::Value>,
709        groups: &HashMap<String, String>,
710        group_properties: &HashMap<String, HashMap<String, serde_json::Value>>,
711    ) -> Result<Option<FlagValue>, InconclusiveMatchError> {
712        match self.cache.get_flag(key) {
713            Some(flag) => {
714                // Build evaluation context with cohorts, flags, and group info
715                let cohorts = self.cache.get_cohort_definitions();
716                let flags = self.cache.get_flags_map();
717                let group_type_mapping = self.cache.get_group_type_mapping();
718
719                let ctx = EvaluationContext {
720                    cohorts: &cohorts,
721                    flags: &flags,
722                    distinct_id,
723                    groups,
724                    group_properties,
725                    group_type_mapping: &group_type_mapping,
726                };
727
728                let result = match_feature_flag_with_context(&flag, person_properties, &ctx);
729                trace!(key, ?result, "Local flag evaluation");
730                result.map(Some)
731            }
732            None => {
733                trace!(key, "Flag not found in local cache");
734                Ok(None)
735            }
736        }
737    }
738
739    /// Evaluate a feature flag locally without cohort or flag dependency
740    /// support.
741    ///
742    /// Use this when you know the flag doesn't have cohort or flag dependency
743    /// conditions.
744    ///
745    /// # Returns
746    ///
747    /// `Ok(Some(value))` when the flag is present and evaluated,
748    /// `Ok(None)` when the flag is absent from the cache.
749    ///
750    /// # Errors
751    ///
752    /// Returns [`InconclusiveMatchError`] when required properties are
753    /// unavailable locally.
754    #[instrument(
755        skip(self, person_properties, groups, group_properties),
756        level = "trace"
757    )]
758    pub fn evaluate_flag_simple(
759        &self,
760        key: &str,
761        distinct_id: &str,
762        person_properties: &HashMap<String, serde_json::Value>,
763        groups: &HashMap<String, String>,
764        group_properties: &HashMap<String, HashMap<String, serde_json::Value>>,
765    ) -> Result<Option<FlagValue>, InconclusiveMatchError> {
766        match self.cache.get_flag(key) {
767            Some(flag) => {
768                let group_type_mapping = self.cache.get_group_type_mapping();
769                let result = match_feature_flag(
770                    &flag,
771                    distinct_id,
772                    person_properties,
773                    groups,
774                    group_properties,
775                    &group_type_mapping,
776                );
777                trace!(key, ?result, "Local flag evaluation (simple)");
778                result.map(Some)
779            }
780            None => {
781                trace!(key, "Flag not found in local cache");
782                Ok(None)
783            }
784        }
785    }
786
787    /// Get all flags and evaluate them with full context support.
788    ///
789    /// The returned map is keyed by feature flag key. Each value can be an
790    /// inconclusive error if that particular flag could not be evaluated from
791    /// the supplied context.
792    #[instrument(
793        skip(self, person_properties, groups, group_properties),
794        level = "debug"
795    )]
796    pub fn evaluate_all_flags(
797        &self,
798        distinct_id: &str,
799        person_properties: &HashMap<String, serde_json::Value>,
800        groups: &HashMap<String, String>,
801        group_properties: &HashMap<String, HashMap<String, serde_json::Value>>,
802    ) -> HashMap<String, Result<FlagValue, InconclusiveMatchError>> {
803        let mut results = HashMap::new();
804
805        // Build evaluation context once for all flags
806        let cohorts = self.cache.get_cohort_definitions();
807        let flags = self.cache.get_flags_map();
808        let group_type_mapping = self.cache.get_group_type_mapping();
809
810        let ctx = EvaluationContext {
811            cohorts: &cohorts,
812            flags: &flags,
813            distinct_id,
814            groups,
815            group_properties,
816            group_type_mapping: &group_type_mapping,
817        };
818
819        for flag in self.cache.get_all_flags() {
820            let result = match_feature_flag_with_context(&flag, person_properties, &ctx);
821            results.insert(flag.key.clone(), result);
822        }
823
824        debug!(flag_count = results.len(), "Evaluated all local flags");
825        results
826    }
827}
posthog_rs/local_evaluation.rs

posthog_rs/
local_evaluation.rs
