feat: add before_send hooks (#135)

* feat: add before_send hooks

* chore: simplify before_send hook type

* address pr review feedback

Author: marandaneto

.sampo/changesets/stalwart-forgemaster-aino.md

@@ -0,0 +1,5 @@
+---
+cargo/posthog-rs: minor
+---
+
+Add before_send hooks for mutating or dropping events before capture.

src/client/async_client.rs

@@ -23,12 +23,14 @@ use crate::feature_flags::{match_feature_flag, FeatureFlag, FeatureFlagsResponse
 use crate::local_evaluation::{AsyncFlagPoller, FlagCache, LocalEvaluationConfig, LocalEvaluator};
 use crate::{Error, Event};
 
+#[cfg(feature = "capture-v1")]
+use super::common::apply_capture_defaults;
 use super::common::{
-    already_reported, build_dedup_key, extract_flag_details, flag_called_event,
-    flag_event_dedup_cache, local_record, remote_record_from_detail, DetailedFlagsResponse,
-    FlagEventDedupCache,
+    already_reported, apply_before_send_hooks, build_dedup_key, extract_flag_details,
+    flag_called_event, flag_event_dedup_cache, local_record, remote_record_from_detail,
+    DetailedFlagsResponse, FlagEventDedupCache,
 };
-use super::ClientOptions;
+use super::{BeforeSendHook, ClientOptions};
 
 #[cfg(not(feature = "capture-v1"))]
 async fn check_response(response: reqwest::Response) -> Result<(), Error> {
@@ -64,6 +66,7 @@ struct AsyncFlagEventHost {
     http_client: HttpClient,
     options: ClientOptions,
     capture_url: String,
+    before_send: Vec<BeforeSendHook>,
     dedup_cache: FlagEventDedupCache,
     /// Tokio runtime handle captured at host construction (which always runs
     /// inside the runtime that hosts `evaluate_flags`). This lets snapshot
@@ -85,6 +88,7 @@ impl AsyncFlagEventHost {
             http_client,
             options: options.clone(),
             capture_url,
+            before_send: options.before_send.clone(),
             dedup_cache: flag_event_dedup_cache(),
             runtime: tokio::runtime::Handle::current(),
         }
@@ -146,6 +150,9 @@ impl AsyncFlagEventHost {
     #[cfg(not(feature = "capture-v1"))]
     fn spawn_ship_v0(&self, mut event: Event) {
         event.prepare_for_v0();
+        let Some(event) = apply_before_send_hooks(&self.before_send, event) else {
+            return;
+        };
         let inner_event = InnerEvent::new(event, self.options.api_key.clone());
         let payload = match serde_json::to_string(&inner_event) {
             Ok(p) => p,
@@ -284,6 +291,12 @@ impl Client {
 
         #[cfg(feature = "capture-v1")]
         {
+            let mut event = event;
+            let defaults = self.options.capture_defaults();
+            apply_capture_defaults(&mut event, &defaults);
+            let Some(event) = apply_before_send_hooks(&self.options.before_send, event) else {
+                return Ok(());
+            };
             return self.capture_v1(vec![event], false).await.map(|_| ());
         }
 
@@ -318,6 +331,17 @@ impl Client {
 
         #[cfg(feature = "capture-v1")]
         {
+            let defaults = self.options.capture_defaults();
+            let events: Vec<_> = events
+                .into_iter()
+                .filter_map(|mut event| {
+                    apply_capture_defaults(&mut event, &defaults);
+                    apply_before_send_hooks(&self.options.before_send, event)
+                })
+                .collect();
+            if events.is_empty() {
+                return Ok(());
+            }
             return self
                 .capture_v1(events, historical_migration)
                 .await
@@ -332,6 +356,9 @@ impl Client {
     async fn capture_v0(&self, mut event: Event) -> Result<(), Error> {
         let defaults = self.options.capture_defaults();
         super::v0_capture::prepare_event(&mut event, &defaults);
+        let Some(event) = apply_before_send_hooks(&self.options.before_send, event) else {
+            return Ok(());
+        };
         let payload =
             super::v0_capture::build_capture_payload(event, self.options.api_key.clone())?;
         let url = self.options.endpoints().build_url(Endpoint::Capture);
@@ -346,12 +373,16 @@ impl Client {
         historical_migration: bool,
     ) -> Result<(), Error> {
         let defaults = self.options.capture_defaults();
-        let payload = super::v0_capture::build_batch_payload(
+        let Some(payload) = super::v0_capture::build_batch_payload(
             events,
             self.options.api_key.clone(),
             historical_migration,
             &defaults,
-        )?;
+            &self.options.before_send,
+        )?
+        else {
+            return Ok(());
+        };
         let url = self.options.endpoints().build_url(Endpoint::Batch);
         let (body, encoding) = super::v0_capture::encode_body(&self.options, payload);
         self.send_v0_with_retry(&url, body, encoding).await

src/client/blocking.rs

@@ -23,12 +23,14 @@ use crate::feature_flags::{match_feature_flag, FeatureFlag, FeatureFlagsResponse
 use crate::local_evaluation::{FlagCache, FlagPoller, LocalEvaluationConfig, LocalEvaluator};
 use crate::{Error, Event};
 
+#[cfg(feature = "capture-v1")]
+use super::common::apply_capture_defaults;
 use super::common::{
-    already_reported, build_dedup_key, extract_flag_details, flag_called_event,
-    flag_event_dedup_cache, local_record, remote_record_from_detail, DetailedFlagsResponse,
-    FlagEventDedupCache,
+    already_reported, apply_before_send_hooks, build_dedup_key, extract_flag_details,
+    flag_called_event, flag_event_dedup_cache, local_record, remote_record_from_detail,
+    DetailedFlagsResponse, FlagEventDedupCache,
 };
-use super::ClientOptions;
+use super::{BeforeSendHook, ClientOptions};
 
 #[cfg(not(feature = "capture-v1"))]
 fn check_response(response: reqwest::blocking::Response) -> Result<(), Error> {
@@ -63,6 +65,7 @@ struct BlockingFlagEventHost {
     http_client: HttpClient,
     options: ClientOptions,
     capture_url: String,
+    before_send: Vec<BeforeSendHook>,
     dedup_cache: FlagEventDedupCache,
 }
 
@@ -78,6 +81,7 @@ impl BlockingFlagEventHost {
             http_client,
             options: options.clone(),
             capture_url,
+            before_send: options.before_send.clone(),
             dedup_cache: flag_event_dedup_cache(),
         }
     }
@@ -130,6 +134,9 @@ impl BlockingFlagEventHost {
     #[cfg(not(feature = "capture-v1"))]
     fn ship_event_v0(&self, mut event: Event) {
         event.prepare_for_v0();
+        let Some(event) = apply_before_send_hooks(&self.before_send, event) else {
+            return;
+        };
         let inner_event = InnerEvent::new(event, self.options.api_key.clone());
         let payload = match serde_json::to_string(&inner_event) {
             Ok(p) => p,
@@ -260,6 +267,12 @@ impl Client {
 
         #[cfg(feature = "capture-v1")]
         {
+            let mut event = event;
+            let defaults = self.options.capture_defaults();
+            apply_capture_defaults(&mut event, &defaults);
+            let Some(event) = apply_before_send_hooks(&self.options.before_send, event) else {
+                return Ok(());
+            };
             return self.capture_v1(vec![event], false).map(|_| ());
         }
 
@@ -296,6 +309,17 @@ impl Client {
 
         #[cfg(feature = "capture-v1")]
         {
+            let defaults = self.options.capture_defaults();
+            let events: Vec<_> = events
+                .into_iter()
+                .filter_map(|mut event| {
+                    apply_capture_defaults(&mut event, &defaults);
+                    apply_before_send_hooks(&self.options.before_send, event)
+                })
+                .collect();
+            if events.is_empty() {
+                return Ok(());
+            }
             return self.capture_v1(events, historical_migration).map(|_| ());
         }
 
@@ -307,6 +331,9 @@ impl Client {
     fn capture_v0(&self, mut event: Event) -> Result<(), Error> {
         let defaults = self.options.capture_defaults();
         super::v0_capture::prepare_event(&mut event, &defaults);
+        let Some(event) = apply_before_send_hooks(&self.options.before_send, event) else {
+            return Ok(());
+        };
         let payload =
             super::v0_capture::build_capture_payload(event, self.options.api_key.clone())?;
         let url = self.options.endpoints().build_url(Endpoint::Capture);
@@ -321,12 +348,16 @@ impl Client {
         historical_migration: bool,
     ) -> Result<(), Error> {
         let defaults = self.options.capture_defaults();
-        let payload = super::v0_capture::build_batch_payload(
+        let Some(payload) = super::v0_capture::build_batch_payload(
             events,
             self.options.api_key.clone(),
             historical_migration,
             &defaults,
-        )?;
+            &self.options.before_send,
+        )?
+        else {
+            return Ok(());
+        };
         let url = self.options.endpoints().build_url(Endpoint::Batch);
         let (body, encoding) = super::v0_capture::encode_body(&self.options, payload);
         self.send_v0_with_retry(&url, body, encoding)

src/client/common.rs

@@ -1,9 +1,12 @@
 use std::collections::{HashMap, HashSet};
 use std::sync::Mutex;
 
+use crate::client::BeforeSendHook;
+use crate::client::CaptureDefaults;
 use crate::feature_flag_evaluations::{EvaluatedFlagRecord, FlagCalledEventParams};
 use crate::feature_flags::{FeatureFlagsResponse, FlagDetail, FlagMetadata, FlagValue};
 use crate::Event;
+use tracing::error;
 
 /// Cap on the number of `distinct_id` entries in the `$feature_flag_called`
 /// dedup cache. On overflow the entire map is reset (matches the JS SDK).
@@ -15,6 +18,33 @@ pub(super) fn flag_event_dedup_cache() -> FlagEventDedupCache {
     Mutex::new(HashMap::new())
 }
 
+pub(super) fn apply_capture_defaults(event: &mut Event, defaults: &CaptureDefaults) {
+    if defaults.disable_geoip {
+        event.insert_prop_default("$geoip_disable", serde_json::Value::Bool(true));
+    }
+    if defaults.is_server {
+        event.insert_prop_default("$is_server", serde_json::Value::Bool(true));
+    }
+}
+
+pub(super) fn apply_before_send_hooks(hooks: &[BeforeSendHook], event: Event) -> Option<Event> {
+    let mut current = Some(event);
+
+    for hook in hooks {
+        let event = current.take().expect("event is present between hooks");
+        match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| hook.apply(event))) {
+            Ok(Some(next)) => current = Some(next),
+            Ok(None) => return None,
+            Err(_) => {
+                error!("panic in PostHog before_send hook; dropping event");
+                return None;
+            }
+        }
+    }
+
+    current
+}
+
 /// Returns `true` when the helper has already shipped this
 /// `(distinct_id, key, response)` combination and the caller should skip.
 pub(super) fn already_reported(
@@ -295,4 +325,44 @@ mod tests {
         assert_eq!(event.properties().get("$is_server"), Some(&json!(true)));
         assert_eq!(event.properties().get("$geoip_disable"), Some(&json!(true)));
     }
+
+    #[test]
+    fn before_send_hooks_mutate_and_drop_events() {
+        let options = crate::ClientOptionsBuilder::default()
+            .api_key("test-key".to_string())
+            .before_send(|mut event| {
+                event.insert_prop("from_hook", true).unwrap();
+                Some(event)
+            })
+            .before_send(|event| {
+                if event.event_name() == "drop" {
+                    None
+                } else {
+                    Some(event)
+                }
+            })
+            .build()
+            .unwrap();
+
+        let event = apply_before_send_hooks(&options.before_send, Event::new("keep", "user-1"))
+            .expect("event should be kept");
+        assert_eq!(event.properties().get("from_hook"), Some(&json!(true)));
+
+        assert!(
+            apply_before_send_hooks(&options.before_send, Event::new("drop", "user-1")).is_none()
+        );
+    }
+
+    #[test]
+    fn before_send_hook_panic_drops_event() {
+        let options = crate::ClientOptionsBuilder::default()
+            .api_key("test-key".to_string())
+            .before_send(|_event| panic!("boom"))
+            .build()
+            .unwrap();
+
+        assert!(
+            apply_before_send_hooks(&options.before_send, Event::new("test", "user-1")).is_none()
+        );
+    }
 }

src/client/mod.rs

@@ -1,4 +1,7 @@
+use std::sync::{Arc, Mutex};
+
 use crate::endpoints::{EndpointManager, DEFAULT_HOST};
+use crate::event::Event;
 use derive_builder::Builder;
 use tracing::warn;
 
@@ -49,6 +52,38 @@ pub use async_client::client;
 #[cfg(feature = "async-client")]
 pub use async_client::Client;
 
+type BeforeSendFn = dyn FnMut(Event) -> Option<Event> + Send + 'static;
+type SharedBeforeSendHook = Arc<Mutex<Box<BeforeSendFn>>>;
+
+/// Hook that can modify or discard events before they are sent.
+///
+/// Hooks run before serialization. Return `Some(event)` to continue sending the
+/// event, or `None` to drop it.
+///
+/// Hook panics are caught and cause the current event to be dropped. If a hook
+/// keeps mutable state, a panic can leave that state partially updated; the SDK
+/// recovers the hook mutex and subsequent events continue through the same hook.
+#[derive(Clone)]
+pub struct BeforeSendHook(SharedBeforeSendHook);
+
+impl BeforeSendHook {
+    /// Create a new before-send hook.
+    pub fn new<F>(hook: F) -> Self
+    where
+        F: FnMut(Event) -> Option<Event> + Send + 'static,
+    {
+        Self(Arc::new(Mutex::new(Box::new(hook))))
+    }
+
+    pub(crate) fn apply(&self, event: Event) -> Option<Event> {
+        let mut hook = self
+            .0
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        (hook)(event)
+    }
+}
+
 /// Configuration options for the PostHog client.
 ///
 /// Use [`ClientOptionsBuilder`] to construct options with custom settings, or
@@ -147,6 +182,10 @@ pub struct ClientOptions {
     #[builder(default, setter(strip_option))]
     pub(crate) capture_compression: Option<CaptureCompression>,
 
+    /// Hooks to modify, filter, or sample events before they are sent.
+    #[builder(default, setter(custom))]
+    pub(crate) before_send: Vec<BeforeSendHook>,
+
     /// Extra HTTP headers injected into every outbound capture request.
     /// Used by the SDK test harness adapter to attach `X-Test-Id` for
     /// parallel test isolation.
@@ -229,6 +268,21 @@ impl ClientOptions {
 }
 
 impl ClientOptionsBuilder {
+    /// Add a hook that can modify or discard events before they are sent.
+    ///
+    /// Hooks should avoid panicking. Panics are caught and drop the current event,
+    /// but any mutable state captured by the hook may be left partially updated
+    /// and will be reused on subsequent calls.
+    pub fn before_send<F>(&mut self, hook: F) -> &mut Self
+    where
+        F: FnMut(Event) -> Option<Event> + Send + 'static,
+    {
+        self.before_send
+            .get_or_insert_with(Vec::new)
+            .push(BeforeSendHook::new(hook));
+        self
+    }
+
     /// Build sanitized [`ClientOptions`].
     ///
     /// Missing or whitespace-only API keys are allowed and disable the client so