diff --git a/.github/workflows/sdk-spec.yml b/.github/workflows/sdk-spec.yml index 1f19dbb..7ea9946 100644 --- a/.github/workflows/sdk-spec.yml +++ b/.github/workflows/sdk-spec.yml @@ -11,7 +11,7 @@ permissions: jobs: sdk-spec: # Keep the ref in `uses:` and `spec-ref` in sync. - uses: Flagsmith/sdk-spec/.github/workflows/run-spec.yml@v0.3.0 + uses: Flagsmith/sdk-spec/.github/workflows/run-spec.yml@v0.7.0 with: - spec-ref: v0.3.0 + spec-ref: v0.7.0 language: rust diff --git a/sdk-spec/Cargo.toml b/sdk-spec/Cargo.toml index 73cb089..584632d 100644 --- a/sdk-spec/Cargo.toml +++ b/sdk-spec/Cargo.toml @@ -12,3 +12,4 @@ path = "steps.rs" flagsmith = { path = ".." } flagsmith-flag-engine = "0.6" serde = { version = "1.0", features = ["derive"] } +serde_json = "1.0" diff --git a/sdk-spec/steps.rs b/sdk-spec/steps.rs index 52321ce..c55ad07 100644 --- a/sdk-spec/steps.rs +++ b/sdk-spec/steps.rs @@ -5,10 +5,12 @@ use std::collections::HashMap; use flagsmith::error::Error; -use flagsmith::flagsmith::models::{Flags, SDKTrait}; +use flagsmith::flagsmith::events::EventOptions; +use flagsmith::flagsmith::models::{Flag, Flags, SDKTrait}; use flagsmith::{Flagsmith, FlagsmithOptions}; use flagsmith_flag_engine::types::FlagsmithValue; use serde::Serialize; +use serde_json::Value; #[derive(Serialize)] pub struct SpecFlag { @@ -17,15 +19,22 @@ pub struct SpecFlag { value: FlagsmithValue, } +fn as_spec_flag(flag: Flag) -> SpecFlag { + SpecFlag { + feature: flag.feature_name, + enabled: flag.enabled, + value: flag.value, + } +} + fn as_spec_flags(flags: Flags) -> Vec { - flags - .all_flags() + flags.all_flags().into_iter().map(as_spec_flag).collect() +} + +fn as_sdk_traits(traits: HashMap) -> Vec { + traits .into_iter() - .map(|flag| SpecFlag { - feature: flag.feature_name, - enabled: flag.enabled, - value: flag.value, - }) + .map(|(key, value)| SDKTrait::new(key, value)) .collect() } @@ -38,6 +47,21 @@ pub fn create_client(environment_key: String, api_url: String) -> Flagsmith { Flagsmith::new(environment_key, options) } +// Create the SDK client with events enabled +pub fn create_client_with_events( + environment_key: String, + api_url: String, + events_api_url: String, +) -> Flagsmith { + let options = FlagsmithOptions { + api_url, + enable_events: true, + events_api_url, + ..Default::default() + }; + Flagsmith::new(environment_key, options) +} + // When the application gets the environment flags pub fn get_environment_flags(flagsmith: &Flagsmith) -> Result, Error> { flagsmith.get_environment_flags().map(as_spec_flags) @@ -49,11 +73,53 @@ pub fn get_identity_flags( identifier: String, traits: HashMap, ) -> Result, Error> { - let traits = traits - .into_iter() - .map(|(key, value)| SDKTrait::new(key, value)) - .collect(); flagsmith - .get_identity_flags(&identifier, Some(traits), None) + .get_identity_flags(&identifier, Some(as_sdk_traits(traits)), None) .map(as_spec_flags) } + +// When the application gets the experiment flag "{feature}" for identity "{identifier}" with traits: +pub fn get_experiment_flag( + flagsmith: &Flagsmith, + feature: String, + identifier: String, + traits: HashMap, +) -> Result, Error> { + flagsmith + .get_experiment_flag(&feature, &identifier, Some(as_sdk_traits(traits))) + .map(|flag| vec![as_spec_flag(flag)]) +} + +// When the application tracks event "{event}" for identity "{identifier}" with value {value} and metadata: +pub fn track_event( + flagsmith: &Flagsmith, + event: String, + identifier: String, + value: FlagsmithValue, + metadata: HashMap, +) -> Result<(), Error> { + let options = EventOptions { + identifier: Some(identifier), + value: Some(value), + metadata: Some(metadata), + ..Default::default() + }; + flagsmith.track_event(&event, options) +} + +// When the application tracks an exposure of feature "{feature}" for identity "{identifier}" with value {value} and metadata: +pub fn track_exposure_event( + flagsmith: &Flagsmith, + feature: String, + identifier: String, + value: FlagsmithValue, + metadata: HashMap, +) -> Result<(), Error> { + let options = EventOptions { + identifier: Some(identifier), + value: Some(value), + metadata: Some(metadata), + ..Default::default() + }; + flagsmith.track_exposure_event(&feature, options) +} diff --git a/src/flagsmith/events.rs b/src/flagsmith/events.rs new file mode 100644 index 0000000..54a3550 --- /dev/null +++ b/src/flagsmith/events.rs @@ -0,0 +1,635 @@ +use super::models::SDKTrait; +use flagsmith_flag_engine::types::{FlagsmithValue, FlagsmithValueType}; +use log::{debug, warn}; +use reqwest::header::{self, HeaderMap}; +use serde::Serialize; +use std::collections::{HashMap, HashSet}; +use std::thread; +use std::time::{Duration, Instant}; + +pub const DEFAULT_EVENTS_API_URL: &str = "https://events.api.flagsmith.com/"; +pub const DEFAULT_EVENTS_FLUSH_INTERVAL_MILLS: u64 = 10 * 1000; +pub const DEFAULT_EVENTS_MAX_BUFFER_SIZE: usize = 1000; +/// The event recorded when an identity is exposed to a variant of a feature. +pub const FLAG_EXPOSURE_EVENT: &str = "$flag_exposure"; +const EVENTS_ENDPOINT: &str = "v1/events"; + +/// The optional fields of a tracked event. +#[derive(Clone, Debug, Default)] +pub struct EventOptions { + /// The identity the event is about. Required for exposure events. + pub identifier: Option, + /// The event's value, e.g. an order total, or the variant an identity was exposed to. + pub value: Option, + pub traits: Option>, + pub metadata: Option>, +} + +#[derive(Serialize, Clone, Debug, PartialEq)] +struct Event { + event: String, + feature_name: Option, + identifier: Option, + value: Option, + traits: Option>, + metadata: HashMap, + timestamp: i64, +} + +impl Event { + fn new(event: &str, feature_name: Option<&str>, options: EventOptions) -> Event { + let traits = options + .traits + .filter(|traits| !traits.is_empty()) + .map(|traits| { + traits + .into_iter() + .map(|t| (t.trait_key, t.trait_value)) + .collect() + }); + let mut metadata = options.metadata.unwrap_or_default(); + metadata.insert( + "sdk_version".to_string(), + serde_json::Value::from(env!("CARGO_PKG_VERSION")), + ); + Event { + event: event.to_string(), + feature_name: feature_name.map(str::to_string), + identifier: options.identifier, + value: options + .value + .filter(|value| value.value_type != FlagsmithValueType::None) + .map(|value| value.value), + traits, + metadata, + timestamp: chrono::Utc::now().timestamp_millis(), + } + } + + /// Who saw which variant of which feature, for exposure events only. + fn exposure_key(&self) -> Option { + (self.event == FLAG_EXPOSURE_EVENT).then(|| ExposureKey { + feature_name: self.feature_name.clone(), + identifier: self.identifier.clone(), + value: self.value.clone(), + }) + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +struct ExposureKey { + feature_name: Option, + identifier: Option, + value: Option, +} + +enum Message { + Track(Event), + Flush(flume::Sender<()>), +} + +enum SendOutcome { + Delivered, + Retryable, + Rejected, +} + +/// Events waiting to be sent. Equal exposures add no information, so only the first one +/// waiting is kept. +struct EventBuffer { + events: Vec, + exposures: HashSet, + max_size: usize, + /// A send failed: wait for the next timed flush rather than retry on every event. + held: bool, +} + +impl EventBuffer { + fn new(max_size: usize) -> EventBuffer { + EventBuffer { + events: vec![], + exposures: HashSet::new(), + max_size: max_size.max(1), + held: false, + } + } + + fn push(&mut self, event: Event) { + if let Some(key) = event.exposure_key() { + if !self.exposures.insert(key) { + debug!( + "Skipping duplicate {} event for feature {:?}", + FLAG_EXPOSURE_EVENT, event.feature_name + ); + return; + } + } + if self.events.len() >= self.max_size { + let oldest = self.events.remove(0); + if let Some(key) = oldest.exposure_key() { + self.exposures.remove(&key); + } + warn!("Events buffer is full; dropped the oldest event"); + } + self.events.push(event); + } + + fn is_full(&self) -> bool { + self.events.len() >= self.max_size + } + + fn clear(&mut self) { + self.events.clear(); + self.exposures.clear(); + } +} + +/// Buffers tracking and exposure events, and sends them to Flagsmith's events API in +/// batches: on a timer, when the buffer fills, on `flush`, and when dropped. +#[derive(Clone, Debug)] +pub struct EventProcessor { + tx: flume::Sender, +} + +impl EventProcessor { + pub fn new( + environment_key: &str, + events_api_url: &str, + timeout: Duration, + flush_interval: Duration, + max_buffer_size: usize, + ) -> Self { + let mut headers = HeaderMap::new(); + headers.insert( + "X-Environment-Key", + header::HeaderValue::from_str(environment_key).unwrap(), + ); + headers.insert(header::CONTENT_TYPE, "application/json".parse().unwrap()); + let user_agent = header::HeaderValue::from_str(&super::get_user_agent()).unwrap(); + headers.insert(header::USER_AGENT, user_agent.clone()); + // The events API records which SDK sent the events from this header. + headers.insert("Flagsmith-SDK-User-Agent", user_agent); + let client = reqwest::blocking::Client::builder() + .default_headers(headers) + .timeout(timeout) + .build() + .unwrap(); + let separator = if events_api_url.ends_with('/') { + "" + } else { + "/" + }; + let endpoint = format!("{events_api_url}{separator}{EVENTS_ENDPOINT}"); + + let (tx, rx) = flume::unbounded(); + thread::Builder::new() + .name("Event Processor".to_string()) + .spawn(move || { + let mut buffer = EventBuffer::new(max_buffer_size); + let mut last_flushed = Instant::now(); + loop { + let message = + rx.recv_timeout(flush_interval.saturating_sub(last_flushed.elapsed())); + match message { + Ok(Message::Track(event)) => { + buffer.push(event); + if buffer.is_full() && !buffer.held { + flush(&client, &endpoint, &mut buffer); + } + } + Ok(Message::Flush(done)) => { + flush(&client, &endpoint, &mut buffer); + let _ = done.send(()); + } + Err(flume::RecvTimeoutError::Timeout) => { + flush(&client, &endpoint, &mut buffer); + last_flushed = Instant::now(); + } + Err(flume::RecvTimeoutError::Disconnected) => { + debug!("Shutting down event processor"); + flush(&client, &endpoint, &mut buffer); + break; + } + } + } + }) + .expect("Failed to start event processor thread"); + + EventProcessor { tx } + } + + pub fn track_event(&self, event: &str, options: EventOptions) { + self.send(Event::new(event, None, options)); + } + + pub fn track_exposure_event(&self, feature_name: &str, options: EventOptions) { + self.send(Event::new(FLAG_EXPOSURE_EVENT, Some(feature_name), options)); + } + + /// Sends the buffered events, and returns once they have been sent, kept for a retry + /// or dropped. + pub fn flush(&self) { + let (done_tx, done_rx) = flume::bounded(1); + if self.tx.send(Message::Flush(done_tx)).is_ok() { + let _ = done_rx.recv(); + } + } + + fn send(&self, event: Event) { + if self.tx.send(Message::Track(event)).is_err() { + warn!("Event processor has stopped; dropped an event"); + } + } +} + +fn flush(client: &reqwest::blocking::Client, endpoint: &str, buffer: &mut EventBuffer) { + buffer.held = false; + if buffer.events.is_empty() { + return; + } + let count = buffer.events.len(); + match post_events(client, endpoint, &buffer.events) { + SendOutcome::Delivered => buffer.clear(), + SendOutcome::Retryable => { + warn!("Failed to send {count} events; keeping them for the next flush"); + buffer.held = true; + } + SendOutcome::Rejected => { + warn!("Events API rejected {count} events; dropping them"); + buffer.clear(); + } + } +} + +fn post_events( + client: &reqwest::blocking::Client, + endpoint: &str, + events: &[Event], +) -> SendOutcome { + let body = serde_json::json!({ "events": events }); + let response = match client.post(endpoint).body(body.to_string()).send() { + Ok(response) => response, + Err(e) => { + debug!("Error sending events: {e}"); + return SendOutcome::Retryable; + } + }; + let status = response.status(); + if status.is_success() { + let rejected = response + .json::() + .ok() + .and_then(|body| body["rejected"].as_array().map(Vec::len)) + .unwrap_or(0); + if rejected > 0 { + warn!("Events API rejected {rejected} of {} events", events.len()); + } + return SendOutcome::Delivered; + } + if status.is_server_error() + || status == reqwest::StatusCode::REQUEST_TIMEOUT + || status == reqwest::StatusCode::TOO_MANY_REQUESTS + { + debug!("Events API responded {status}"); + return SendOutcome::Retryable; + } + SendOutcome::Rejected +} + +#[cfg(test)] +mod tests { + use super::*; + use httpmock::prelude::*; + + const ENVIRONMENT_KEY: &str = "ser.test_environment_key"; + + fn string_value(value: &str) -> FlagsmithValue { + FlagsmithValue { + value: value.to_string(), + value_type: FlagsmithValueType::String, + } + } + + fn exposure(feature_name: &str, identifier: &str, value: &str) -> Event { + Event::new( + FLAG_EXPOSURE_EVENT, + Some(feature_name), + EventOptions { + identifier: Some(identifier.to_string()), + value: Some(string_value(value)), + ..Default::default() + }, + ) + } + + fn processor( + server: &MockServer, + flush_interval_mills: u64, + max_buffer_size: usize, + ) -> EventProcessor { + EventProcessor::new( + ENVIRONMENT_KEY, + &server.url("/"), + Duration::from_secs(10), + Duration::from_millis(flush_interval_mills), + max_buffer_size, + ) + } + + #[test] + fn event_carries_its_options_and_the_sdk_version() { + // Given + let options = EventOptions { + identifier: Some("user_123".to_string()), + value: Some(FlagsmithValue { + value: "42".to_string(), + value_type: FlagsmithValueType::Integer, + }), + traits: Some(vec![SDKTrait::new( + "plan".to_string(), + string_value("premium"), + )]), + metadata: Some(HashMap::from([( + "currency".to_string(), + serde_json::json!("GBP"), + )])), + }; + + // When + let event = Event::new("checkout_completed", None, options); + + // Then + let mut json = serde_json::to_value(&event).unwrap(); + assert!(json["timestamp"].as_i64().unwrap() > 1_577_836_800_000); + json.as_object_mut().unwrap().remove("timestamp"); + assert_eq!( + json, + serde_json::json!({ + "event": "checkout_completed", + "feature_name": null, + "identifier": "user_123", + "value": "42", + "traits": {"plan": "premium"}, + "metadata": {"currency": "GBP", "sdk_version": env!("CARGO_PKG_VERSION")}, + }) + ); + } + + #[test] + fn event_without_value_or_traits_sends_them_as_null() { + // When + let event = Event::new( + "signed_up", + None, + EventOptions { + value: Some(FlagsmithValue::default()), + traits: Some(vec![]), + ..Default::default() + }, + ); + + // Then + assert_eq!(event.value, None); + assert_eq!(event.traits, None); + assert_eq!(event.identifier, None); + } + + #[test] + fn buffer_keeps_one_of_equal_exposures() { + // Given + let mut buffer = EventBuffer::new(10); + + // When + buffer.push(exposure("banner", "user_123", "treatment")); + buffer.push(exposure("banner", "user_123", "treatment")); + buffer.push(exposure("banner", "user_123", "control")); + buffer.push(exposure("banner", "user_456", "treatment")); + + // Then + let values: Vec<_> = buffer + .events + .iter() + .map(|e| (e.identifier.clone().unwrap(), e.value.clone().unwrap())) + .collect(); + assert_eq!( + values, + vec![ + ("user_123".to_string(), "treatment".to_string()), + ("user_123".to_string(), "control".to_string()), + ("user_456".to_string(), "treatment".to_string()), + ] + ); + } + + #[test] + fn buffer_keeps_every_custom_event() { + // Given + let mut buffer = EventBuffer::new(10); + + // When + buffer.push(Event::new("clicked", None, EventOptions::default())); + buffer.push(Event::new("clicked", None, EventOptions::default())); + + // Then + assert_eq!(buffer.events.len(), 2); + } + + #[test] + fn full_buffer_drops_its_oldest_event() { + // Given + let mut buffer = EventBuffer::new(2); + buffer.push(exposure("banner", "user_1", "treatment")); + buffer.push(exposure("banner", "user_2", "treatment")); + + // When + buffer.push(exposure("banner", "user_3", "treatment")); + buffer.push(exposure("banner", "user_1", "treatment")); + + // Then + let identifiers: Vec<_> = buffer + .events + .iter() + .map(|e| e.identifier.clone().unwrap()) + .collect(); + assert_eq!(identifiers, vec!["user_3", "user_1"]); + } + + #[test] + fn flush_sends_tracked_events_to_the_events_api() { + // Given + let server = MockServer::start(); + let events_mock = server.mock(|when, then| { + when.method(POST) + .path("/v1/events") + .header("X-Environment-Key", ENVIRONMENT_KEY) + .header("Content-Type", "application/json") + .header("Flagsmith-SDK-User-Agent", super::super::get_user_agent()) + .json_body_partial( + serde_json::json!({"events": [ + {"event": "checkout_completed", "identifier": "user_123"}, + {"event": FLAG_EXPOSURE_EVENT, "feature_name": "banner", "value": "treatment"}, + ]}) + .to_string(), + ); + then.status(202) + .json_body(serde_json::json!({"accepted": 2, "rejected": []})); + }); + let processor = processor(&server, 60_000, 100); + processor.track_event( + "checkout_completed", + EventOptions { + identifier: Some("user_123".to_string()), + ..Default::default() + }, + ); + processor.track_exposure_event( + "banner", + EventOptions { + identifier: Some("user_123".to_string()), + value: Some(string_value("treatment")), + ..Default::default() + }, + ); + + // When + processor.flush(); + + // Then + events_mock.assert_hits(1); + } + + #[test] + fn flush_without_events_sends_nothing() { + // Given + let server = MockServer::start(); + let events_mock = server.mock(|when, then| { + when.method(POST).path("/v1/events"); + then.status(202); + }); + let processor = processor(&server, 60_000, 100); + + // When + processor.flush(); + + // Then + events_mock.assert_hits(0); + } + + #[test] + fn events_are_sent_on_the_flush_interval() { + // Given + let server = MockServer::start(); + let events_mock = server.mock(|when, then| { + when.method(POST).path("/v1/events"); + then.status(202); + }); + let processor = processor(&server, 50, 100); + + // When + processor.track_event("clicked", EventOptions::default()); + thread::sleep(Duration::from_millis(300)); + + // Then + events_mock.assert_hits(1); + } + + #[test] + fn events_are_sent_when_the_buffer_fills() { + // Given + let server = MockServer::start(); + let events_mock = server.mock(|when, then| { + when.method(POST).path("/v1/events"); + then.status(202); + }); + let processor = processor(&server, 60_000, 2); + + // When + processor.track_event("clicked", EventOptions::default()); + processor.track_event("clicked", EventOptions::default()); + thread::sleep(Duration::from_millis(100)); + + // Then + events_mock.assert_hits(1); + } + + #[test] + fn dropping_the_processor_sends_buffered_events() { + // Given + let server = MockServer::start(); + let events_mock = server.mock(|when, then| { + when.method(POST).path("/v1/events"); + then.status(202); + }); + let processor = processor(&server, 60_000, 100); + processor.track_event("clicked", EventOptions::default()); + + // When + drop(processor); + thread::sleep(Duration::from_millis(100)); + + // Then + events_mock.assert_hits(1); + } + + #[test] + fn events_are_kept_for_the_next_flush_when_the_events_api_is_unavailable() { + // Given + let server = MockServer::start(); + let mut unavailable_mock = server.mock(|when, then| { + when.method(POST).path("/v1/events"); + then.status(503); + }); + let processor = processor(&server, 60_000, 100); + processor.track_event("clicked", EventOptions::default()); + processor.flush(); + unavailable_mock.assert_hits(1); + unavailable_mock.delete(); + let events_mock = server.mock(|when, then| { + when.method(POST) + .path("/v1/events") + .json_body_partial(r#"{"events": [{"event": "clicked"}]}"#); + then.status(202); + }); + + // When + processor.flush(); + + // Then + events_mock.assert_hits(1); + } + + #[test] + fn events_rejected_by_the_events_api_are_dropped() { + // Given + let server = MockServer::start(); + let events_mock = server.mock(|when, then| { + when.method(POST).path("/v1/events"); + then.status(400); + }); + let processor = processor(&server, 60_000, 100); + processor.track_event("clicked", EventOptions::default()); + processor.flush(); + + // When + processor.flush(); + + // Then + events_mock.assert_hits(1); + } + + #[test] + fn flush_survives_an_unreachable_events_api() { + // Given + let processor = EventProcessor::new( + ENVIRONMENT_KEY, + "http://127.0.0.1:1/", + Duration::from_millis(100), + Duration::from_secs(60), + 100, + ); + processor.track_event("clicked", EventOptions::default()); + + // When / Then + processor.flush(); + } +} diff --git a/src/flagsmith/mod.rs b/src/flagsmith/mod.rs index 80f7038..9481cec 100644 --- a/src/flagsmith/mod.rs +++ b/src/flagsmith/mod.rs @@ -1,4 +1,5 @@ use self::analytics::AnalyticsProcessor; +use self::events::{EventOptions, EventProcessor, FLAG_EXPOSURE_EVENT}; use self::models::{Flag, Flags}; use super::error; use flagsmith_flag_engine::engine::get_evaluation_result; @@ -8,6 +9,7 @@ use flagsmith_flag_engine::engine_eval::{ use flagsmith_flag_engine::environments::Environment; use flagsmith_flag_engine::identities::Trait; use flagsmith_flag_engine::segments::Segment; +use flagsmith_flag_engine::types::{FlagsmithValue, FlagsmithValueType}; use log::debug; use models::SDKTrait; use reqwest::header::{self, HeaderMap}; @@ -19,6 +21,7 @@ use std::{thread, time::Duration}; mod analytics; +pub mod events; pub mod models; pub mod offline_handler; @@ -40,6 +43,11 @@ pub struct FlagsmithOptions { pub default_flag_handler: Option Flag>, pub offline_handler: Option>, pub offline_mode: bool, + /// Sends tracking and exposure events to Flagsmith's events API. + pub enable_events: bool, + pub events_api_url: String, + pub events_flush_interval_mills: u64, + pub events_max_buffer_size: usize, } impl Default for FlagsmithOptions { @@ -54,6 +62,10 @@ impl Default for FlagsmithOptions { default_flag_handler: None, offline_handler: None, offline_mode: false, + enable_events: false, + events_api_url: events::DEFAULT_EVENTS_API_URL.to_string(), + events_flush_interval_mills: events::DEFAULT_EVENTS_FLUSH_INTERVAL_MILLS, + events_max_buffer_size: events::DEFAULT_EVENTS_MAX_BUFFER_SIZE, } } } @@ -70,6 +82,7 @@ pub struct Flagsmith { // never take it. refresh_lock: Arc>, analytics_processor: Option, + event_processor: Option, _polling_thread_tx: SyncSender, // to trigger polling manager shutdown } @@ -127,6 +140,15 @@ impl Flagsmith { )), false => None, }; + let event_processor = flagsmith_options.enable_events.then(|| { + EventProcessor::new( + &environment_key, + &flagsmith_options.events_api_url, + timeout, + Duration::from_millis(flagsmith_options.events_flush_interval_mills), + flagsmith_options.events_max_buffer_size, + ) + }); // Put the environment model behind mutex to // to share it safely between threads @@ -146,6 +168,7 @@ impl Flagsmith { datastore: Arc::clone(&ds), refresh_lock: Arc::clone(&refresh_lock), analytics_processor, + event_processor, _polling_thread_tx: tx, }; @@ -258,6 +281,108 @@ impl Flagsmith { transient.unwrap_or(false), )); } + /// Returns the flag of a feature for an identity, and records the identity's exposure to + /// the variant it was served when it is enrolled in a running experiment on the feature. + /// Experiments are only known to remote evaluation. Requires `enable_events`. + pub fn get_experiment_flag( + &self, + feature_name: &str, + identifier: &str, + traits: Option>, + ) -> Result { + let event_processor = self.require_event_processor("use experiment flags")?; + let flag = self + .get_identity_flags(identifier, traits.clone(), None)? + .get_flag(feature_name)?; + let exposure = match (&flag.experiment, &flag.variant) { + _ if flag.is_default => Err("the flag is a default flag"), + _ if !flag.enabled => Err("the flag is disabled"), + (Some(experiment), Some(variant)) if experiment.in_experiment => { + Ok((experiment.id, variant.clone())) + } + (Some(experiment), None) if experiment.in_experiment => Err("the flag has no variant"), + _ => Err("the identity is not in a running experiment"), + }; + match exposure { + Ok((experiment_id, variant)) => event_processor.track_exposure_event( + feature_name, + EventOptions { + identifier: Some(identifier.to_string()), + value: Some(FlagsmithValue { + value: variant, + value_type: FlagsmithValueType::String, + }), + traits, + metadata: Some(HashMap::from([( + "experiment_id".to_string(), + json!(experiment_id), + )])), + }, + ), + Err(reason) => debug!( + "Not sending {} for feature {}: {}", + FLAG_EXPOSURE_EVENT, feature_name, reason + ), + } + return Ok(flag); + } + + /// Records an event, such as a conversion, for Flagsmith's events API. Names starting + /// with `$` are reserved. Requires `enable_events`. + pub fn track_event(&self, event: &str, options: EventOptions) -> Result<(), error::Error> { + let event_processor = self.require_event_processor("track events")?; + if event.starts_with('$') { + return Err(error::Error::new( + error::ErrorKind::FlagsmithClientError, + format!( + "Event names starting with \"$\" are reserved; use track_exposure_event to record {}.", + FLAG_EXPOSURE_EVENT + ), + )); + } + event_processor.track_event(event, options); + return Ok(()); + } + + /// Records that an identity was exposed to a variant of a feature evaluated elsewhere, + /// as a `$flag_exposure` event. Requires an identifier, and `enable_events`. + pub fn track_exposure_event( + &self, + feature_name: &str, + options: EventOptions, + ) -> Result<(), error::Error> { + let event_processor = self.require_event_processor("track exposure events")?; + if options + .identifier + .as_deref() + .unwrap_or("") + .trim() + .is_empty() + { + return Err(error::Error::new( + error::ErrorKind::FlagsmithClientError, + "An exposure event requires an identifier.".to_string(), + )); + } + event_processor.track_exposure_event(feature_name, options); + return Ok(()); + } + + /// Sends the buffered events, and returns once they have been sent, kept for a retry or + /// dropped. Does nothing without `enable_events`. + pub fn flush_events(&self) { + if let Some(event_processor) = &self.event_processor { + event_processor.flush(); + } + } + + fn require_event_processor(&self, action: &str) -> Result<&EventProcessor, error::Error> { + self.event_processor.as_ref().ok_or(error::Error::new( + error::ErrorKind::FlagsmithClientError, + format!("Events must be enabled to {action}."), + )) + } + // Returns a list of segments that the given identity is part of pub fn get_identity_segments( &self, diff --git a/src/flagsmith/models.rs b/src/flagsmith/models.rs index 2b9ad0d..87085fb 100644 --- a/src/flagsmith/models.rs +++ b/src/flagsmith/models.rs @@ -16,6 +16,21 @@ pub struct Flag { pub is_default: bool, pub feature_id: u32, pub feature_name: String, + /// The key of the multivariate variant the identity was served. Only set by remote + /// identity evaluation. + pub variant: Option, + /// The running experiment on the feature. Only set by remote identity evaluation. + pub experiment: Option, +} + +/// The running experiment a flag was evaluated under. +#[derive(Clone, Debug, Default, PartialEq, Deserialize)] +pub struct ExperimentMetadata { + pub id: u64, + pub name: String, + /// Whether the identity is enrolled in the experiment. Outside the experiment's + /// rollout, an identity is still served a variant. + pub in_experiment: bool, } impl Flag { @@ -26,6 +41,7 @@ impl Flag { is_default: false, feature_name: feature_state.feature.name, feature_id: feature_state.feature.id, + ..Default::default() }; } @@ -39,6 +55,8 @@ impl Flag { feature_name: flag_json["feature"]["name"].as_str()?.to_string(), feature_id: flag_json["feature"]["id"].as_u64()?.try_into().ok()?, value, + variant: flag_json["variant"].as_str().map(str::to_string), + experiment: serde_json::from_value(flag_json["metadata"]["experiment"].clone()).ok(), }; Some(flag) } @@ -129,6 +147,7 @@ impl Flags { enabled: flag_result.enabled, value: flag_result.value.clone(), feature_id: flag_result.metadata.feature_id, + ..Default::default() }; flags.insert(feature_name.clone(), flag); } @@ -286,6 +305,50 @@ mod tests { assert_eq!(flag.value, expected_value); } + #[test] + fn flag_from_api_flag_carries_the_variant_and_experiment_served_to_the_identity() { + // Given + let flag_json = serde_json::json!({ + "feature_state_value": "hello treatment", + "feature": {"name": "banner", "id": 1}, + "enabled": true, + "variant": "treatment", + "metadata": {"experiment": {"id": 7, "name": "banner experiment", "in_experiment": true}} + }); + + // When + let flag = Flag::from_api_flag(&flag_json).unwrap(); + + // Then + assert_eq!(flag.variant.as_deref(), Some("treatment")); + assert_eq!( + flag.experiment, + Some(ExperimentMetadata { + id: 7, + name: "banner experiment".to_string(), + in_experiment: true, + }) + ); + } + + #[test] + fn flag_from_api_flag_without_an_experiment_has_none() { + // Given + let flag_json = serde_json::json!({ + "feature_state_value": "hello", + "feature": {"name": "banner", "id": 1}, + "enabled": true, + "variant": null + }); + + // When + let flag = Flag::from_api_flag(&flag_json).unwrap(); + + // Then + assert_eq!(flag.variant, None); + assert_eq!(flag.experiment, None); + } + #[test] fn value_as_string() { // Give diff --git a/tests/integration_test.rs b/tests/integration_test.rs index 5698b26..70dc4c3 100644 --- a/tests/integration_test.rs +++ b/tests/integration_test.rs @@ -1,3 +1,4 @@ +use flagsmith::flagsmith::events::EventOptions; use flagsmith::flagsmith::models::SDKTrait; use flagsmith::flagsmith::offline_handler; use flagsmith::{Flagsmith, FlagsmithOptions}; @@ -6,6 +7,7 @@ use flagsmith_flag_engine::types::{FlagsmithValue, FlagsmithValueType}; use httpmock::prelude::*; use rstest::*; +use std::collections::HashMap; mod fixtures; @@ -17,6 +19,8 @@ use fixtures::identities_json; use fixtures::local_eval_flagsmith; use fixtures::mock_server; use fixtures::ENVIRONMENT_KEY; +use fixtures::FEATURE_1_ID; +use fixtures::FEATURE_1_NAME; use fixtures::SEGMENT_OVERRIDE_VALUE; #[rstest] @@ -906,3 +910,314 @@ fn test_malformed_environment_document_is_an_error_and_keeps_the_last_document( before ); } + +fn experiment_identities_json( + enabled: bool, + variant: Option<&str>, + in_experiment: bool, +) -> serde_json::Value { + serde_json::json!({ + "traits": [], + "flags": [{ + "feature": {"id": FEATURE_1_ID, "name": FEATURE_1_NAME, "type": "MULTIVARIATE"}, + "feature_state_value": "hello treatment", + "enabled": enabled, + "variant": variant, + "metadata": { + "experiment": {"id": 7, "name": "banner experiment", "in_experiment": in_experiment} + } + }] + }) +} + +fn events_flagsmith(mock_server: &MockServer) -> Flagsmith { + let flagsmith_options = FlagsmithOptions { + api_url: mock_server.url("/api/v1/"), + enable_events: true, + events_api_url: mock_server.url("/events/"), + ..Default::default() + }; + Flagsmith::new(ENVIRONMENT_KEY.to_string(), flagsmith_options) +} + +#[rstest] +fn test_get_experiment_flag_records_the_exposure_of_an_identity_in_a_running_experiment( + mock_server: MockServer, +) { + // Given + mock_server.mock(|when, then| { + when.method(POST).path("/api/v1/identities/"); + then.status(200) + .json_body(experiment_identities_json(true, Some("treatment"), true)); + }); + let events_mock = mock_server.mock(|when, then| { + when.method(POST) + .path("/events/v1/events") + .header("X-Environment-Key", ENVIRONMENT_KEY) + .json_body_partial( + serde_json::json!({"events": [{ + "event": "$flag_exposure", + "feature_name": FEATURE_1_NAME, + "identifier": "user_123", + "value": "treatment", + "traits": {"plan": "premium"}, + "metadata": {"experiment_id": 7}, + }]}) + .to_string(), + ); + then.status(202); + }); + let flagsmith = events_flagsmith(&mock_server); + let traits = vec![SDKTrait::new( + "plan".to_string(), + FlagsmithValue { + value: "premium".to_string(), + value_type: FlagsmithValueType::String, + }, + )]; + + // When + let flag = flagsmith + .get_experiment_flag(FEATURE_1_NAME, "user_123", Some(traits)) + .unwrap(); + flagsmith.flush_events(); + + // Then + assert_eq!(flag.value_as_string().unwrap(), "hello treatment"); + assert_eq!(flag.variant.as_deref(), Some("treatment")); + events_mock.assert_hits(1); +} + +#[rstest] +#[case::identity_outside_the_experiment(true, Some("treatment"), false)] +#[case::disabled_flag(false, Some("treatment"), true)] +#[case::flag_without_a_variant(true, None, true)] +fn test_get_experiment_flag_records_no_exposure_unless_an_enabled_flag_served_an_enrolled_identity_a_variant( + mock_server: MockServer, + #[case] enabled: bool, + #[case] variant: Option<&str>, + #[case] in_experiment: bool, +) { + // Given + mock_server.mock(|when, then| { + when.method(POST).path("/api/v1/identities/"); + then.status(200) + .json_body(experiment_identities_json(enabled, variant, in_experiment)); + }); + let events_mock = mock_server.mock(|when, then| { + when.method(POST).path("/events/v1/events"); + then.status(202); + }); + let flagsmith = events_flagsmith(&mock_server); + + // When + let flag = flagsmith + .get_experiment_flag(FEATURE_1_NAME, "user_123", None) + .unwrap(); + flagsmith.flush_events(); + + // Then + assert_eq!(flag.enabled, enabled); + events_mock.assert_hits(0); +} + +#[rstest] +fn test_get_experiment_flag_records_no_exposure_for_a_default_flag( + mock_server: MockServer, + default_flag_handler: fn(&str) -> flagsmith::Flag, +) { + // Given + mock_server.mock(|when, then| { + when.method(POST).path("/api/v1/identities/"); + then.status(500); + }); + let events_mock = mock_server.mock(|when, then| { + when.method(POST).path("/events/v1/events"); + then.status(202); + }); + let flagsmith_options = FlagsmithOptions { + api_url: mock_server.url("/api/v1/"), + enable_events: true, + events_api_url: mock_server.url("/events/"), + default_flag_handler: Some(default_flag_handler), + ..Default::default() + }; + let flagsmith = Flagsmith::new(ENVIRONMENT_KEY.to_string(), flagsmith_options); + + // When + let flag = flagsmith + .get_experiment_flag(FEATURE_1_NAME, "user_123", None) + .unwrap(); + flagsmith.flush_events(); + + // Then + assert!(flag.is_default); + events_mock.assert_hits(0); +} + +#[rstest] +fn test_experiment_flags_and_events_require_events_to_be_enabled(mock_server: MockServer) { + // Given + let flagsmith_options = FlagsmithOptions { + api_url: mock_server.url("/api/v1/"), + ..Default::default() + }; + let flagsmith = Flagsmith::new(ENVIRONMENT_KEY.to_string(), flagsmith_options); + + // When + let experiment_flag = flagsmith.get_experiment_flag(FEATURE_1_NAME, "user_123", None); + let event = flagsmith.track_event("checkout_completed", EventOptions::default()); + let exposure = flagsmith.track_exposure_event( + FEATURE_1_NAME, + EventOptions { + identifier: Some("user_123".to_string()), + ..Default::default() + }, + ); + + // Then + for error in [ + experiment_flag.unwrap_err(), + event.unwrap_err(), + exposure.unwrap_err(), + ] { + assert_eq!( + error.kind, + flagsmith::error::ErrorKind::FlagsmithClientError + ); + assert!(error.msg.starts_with("Events must be enabled")); + } +} + +#[rstest] +fn test_track_event_sends_the_event_to_flagsmith(mock_server: MockServer) { + // Given + let events_mock = mock_server.mock(|when, then| { + when.method(POST) + .path("/events/v1/events") + .json_body_partial( + serde_json::json!({"events": [{ + "event": "checkout_completed", + "feature_name": null, + "identifier": "user_123", + "value": "42", + "metadata": {"currency": "GBP"}, + }]}) + .to_string(), + ); + then.status(202); + }); + let flagsmith = events_flagsmith(&mock_server); + + // When + flagsmith + .track_event( + "checkout_completed", + EventOptions { + identifier: Some("user_123".to_string()), + value: Some(FlagsmithValue { + value: "42".to_string(), + value_type: FlagsmithValueType::Integer, + }), + metadata: Some(HashMap::from([( + "currency".to_string(), + serde_json::json!("GBP"), + )])), + ..Default::default() + }, + ) + .unwrap(); + flagsmith.flush_events(); + + // Then + events_mock.assert_hits(1); +} + +#[rstest] +fn test_track_event_rejects_reserved_event_names(mock_server: MockServer) { + // Given + let flagsmith = events_flagsmith(&mock_server); + + // When + let result = flagsmith.track_event("$flag_exposure", EventOptions::default()); + + // Then + let error = result.unwrap_err(); + assert_eq!( + error.kind, + flagsmith::error::ErrorKind::FlagsmithClientError + ); + assert!(error.msg.contains("reserved")); +} + +#[rstest] +fn test_track_exposure_event_sends_a_flag_exposure_event_to_flagsmith(mock_server: MockServer) { + // Given + let events_mock = mock_server.mock(|when, then| { + when.method(POST) + .path("/events/v1/events") + .json_body_partial( + serde_json::json!({"events": [{ + "event": "$flag_exposure", + "feature_name": "banner", + "identifier": "user_123", + "value": "control", + "metadata": {"experiment_id": 1}, + }]}) + .to_string(), + ); + then.status(202); + }); + let flagsmith = events_flagsmith(&mock_server); + + // When + flagsmith + .track_exposure_event( + "banner", + EventOptions { + identifier: Some("user_123".to_string()), + value: Some(FlagsmithValue { + value: "control".to_string(), + value_type: FlagsmithValueType::String, + }), + metadata: Some(HashMap::from([( + "experiment_id".to_string(), + serde_json::json!(1), + )])), + ..Default::default() + }, + ) + .unwrap(); + flagsmith.flush_events(); + + // Then + events_mock.assert_hits(1); +} + +#[rstest] +#[case::no_identifier(None)] +#[case::blank_identifier(Some(" "))] +fn test_track_exposure_event_requires_an_identifier( + mock_server: MockServer, + #[case] identifier: Option<&str>, +) { + // Given + let flagsmith = events_flagsmith(&mock_server); + + // When + let result = flagsmith.track_exposure_event( + "banner", + EventOptions { + identifier: identifier.map(str::to_string), + ..Default::default() + }, + ); + + // Then + let error = result.unwrap_err(); + assert_eq!( + error.kind, + flagsmith::error::ErrorKind::FlagsmithClientError + ); + assert_eq!(error.msg, "An exposure event requires an identifier."); +}