Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .sampo/changesets/event-uuid-timestamp-distinct-id.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
cargo/posthog-rs: major
---

Assign an event's uuid when the client takes the event, so a cloned or reused `Event` no longer sends one uuid twice. `Event::set_timestamp` no longer rejects future timestamps. `Event::new` with an empty or whitespace-only distinct ID creates an anonymous event.
6 changes: 6 additions & 0 deletions docs/migration-0.x-to-1.0.md
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,12 @@ Use the new `capture_ai` family for LLM analytics events (`$ai_generation`, `$ai

The backend decides which names belong on the AI lane and reports a misrouted or oversize event as a per-event `drop` inside a `200`. That verdict reaches an `on_error` hook (`CaptureFailure::endpoint()` says which lane) or the returned `CaptureSummary` for the immediate variants; without a hook the SDK logs one aggregate warning per batch with drop and retry totals. To log every drop and retry reason, call `CaptureFailure::verdict_summary()` in an `on_error` hook; it returns one line such as `drop/invalid_options=2, retry/not_persisted=1`.

### Event uuid, timestamp and distinct ID

- The SDK assigns an event's uuid when you hand the event to `capture` or `capture_immediate`, not in `Event::new`. Capturing a cloned or reused `Event` twice now sends two uuids. In 0.x both copies shared one uuid, and capture V1 rejects a request that contains a uuid twice. A uuid set with `Event::set_uuid` is still sent as given on every capture.
- `Event::set_timestamp` accepts timestamps in the future. In 0.x it returned `Error::InvalidTimestamp` for a timestamp more than 1 s ahead. PostHog clamps far-future timestamps, as it does for every other SDK. The method keeps its `Result` return type.
- `Event::new` with an empty or whitespace-only distinct ID creates an anonymous event, as `Event::new_anon` does. In 0.x it sent the empty ID, and PostHog dropped the event.

### Event options

Capture V1 reads per-event processing controls, such as person profile processing, from an `options` map that is sent next to `properties`. Set them with `Event::insert_option`, or with `CaptureExceptionOptions::option` for exceptions. They work the same on `capture` and `capture_ai`.
Expand Down
4 changes: 3 additions & 1 deletion src/capture_event.rs
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,9 @@ impl CaptureEvent {

Self {
event: event.event_name().to_string(),
uuid: event.uuid(),
// Every send path assigns the uuid first (`preprocess_capture_event`);
// the fallback only covers a direct build.
uuid: event.uuid().unwrap_or_else(Uuid::now_v7),
distinct_id: event.distinct_id().to_string(),
timestamp,
session_id,
Expand Down
2 changes: 2 additions & 0 deletions src/client/common.rs
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,8 @@ pub(super) fn preprocess_capture_event(
defaults: &CaptureDefaults,
hooks: &[BeforeSendHook],
) -> Option<Event> {
// Queued events already have a uuid from enqueue; this covers immediate sends.
event.ensure_uuid();
// SDK values fill only keys the caller left unset, and go in before the
// hooks so `before_send` sees them and can change or remove them.
apply_runtime_context(&mut event);
Expand Down
8 changes: 5 additions & 3 deletions src/client/transport.rs
Original file line number Diff line number Diff line change
Expand Up @@ -273,12 +273,13 @@ impl TransportHandle {
}
}

/// Stamp the capture (enqueue) time on the producer side — so a batched or
/// retried event records when it occurred, not when it was finally sent —
/// then hand it to the worker, releasing the reserved slot if the worker is
/// Stamp the capture (enqueue) time and the event uuid on the producer side —
/// so a batched or retried event records when it occurred, not when it was
/// finally sent — then hand it to the worker, releasing the reserved slot if the worker is
/// gone. The slot must already be reserved by the caller.
fn send_reserved(&self, mut event: Event) {
event.ensure_timestamp(self.clock.now_utc());
event.ensure_uuid();
if self
.tx
.send(Control::Capture {
Expand Down Expand Up @@ -318,6 +319,7 @@ impl TransportHandle {
&& try_reserve(&self.len, self.max_queue_size, &self.full_warned)
{
events[fitted].ensure_timestamp(self.clock.now_utc());
events[fitted].ensure_uuid();
fitted += 1;
}
events.truncate(fitted);
Expand Down
98 changes: 76 additions & 22 deletions src/event.rs
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
use std::collections::HashMap;

use chrono::{DateTime, Duration, NaiveDateTime, TimeZone, Utc};
use chrono::{DateTime, NaiveDateTime, TimeZone, Utc};
use serde::Serialize;
use tracing::warn;
use uuid::Uuid;
Expand Down Expand Up @@ -69,7 +69,9 @@ pub struct Event {
options: HashMap<String, serde_json::Value>,
groups: HashMap<String, String>,
timestamp: Option<NaiveDateTime>,
uuid: Uuid,
/// Assigned once when the client takes the event, unless set with
/// [`Event::set_uuid`], so a reused or cloned `Event` never resends a uuid.
uuid: Option<Uuid>,
/// When set, the capture pipeline trims this event's properties to
/// [`MINIMAL_FLAG_CALLED_EVENT_PROPERTIES`] as its final enrichment step.
/// Set only for minimized `$feature_flag_called` events; never serialized.
Expand All @@ -87,15 +89,21 @@ impl Event {
/// - `event`: Event name, such as `"user_signed_up"`.
/// - `distinct_id`: Stable user or account identifier. For backend events,
/// use the same distinct ID your frontend passes to `posthog.identify()`.
/// An empty or whitespace-only ID counts as missing: the event becomes
/// anonymous, as from [`Event::new_anon`].
pub fn new<S: Into<String>>(event: S, distinct_id: S) -> Self {
let distinct_id = distinct_id.into();
if distinct_id.trim().is_empty() {
return Self::new_anon(event);
}
Self {
event: event.into(),
distinct_id: distinct_id.into(),
distinct_id,
properties: HashMap::new(),
options: HashMap::new(),
groups: HashMap::new(),
timestamp: None,
uuid: Uuid::now_v7(),
uuid: None,
minimal_flag_called: false,
}
}
Expand Down Expand Up @@ -128,7 +136,7 @@ impl Event {
options,
groups: HashMap::new(),
timestamp: None,
uuid: Uuid::now_v7(),
uuid: None,
minimal_flag_called: false,
}
}
Expand Down Expand Up @@ -220,16 +228,12 @@ impl Event {
///
/// # Errors
///
/// Returns [`Error::InvalidTimestamp`] if the timestamp is in the future.
/// Never returns an error; PostHog clamps timestamps far in the future.
/// The `Result` stays so existing callers keep compiling.
pub fn set_timestamp<Tz>(&mut self, timestamp: DateTime<Tz>) -> Result<(), Error>
where
Tz: TimeZone,
{
if timestamp > Utc::now() + Duration::seconds(1) {
return Err(Error::InvalidTimestamp(String::from(
"Events cannot occur in the future",
)));
}
self.timestamp = Some(timestamp.naive_utc());
Ok(())
}
Expand Down Expand Up @@ -267,7 +271,7 @@ impl Event {
options: HashMap::new(),
groups: HashMap::new(),
timestamp: None,
uuid: Uuid::now_v7(),
uuid: None,
minimal_flag_called: false,
})
}
Expand Down Expand Up @@ -327,7 +331,7 @@ impl Event {
options: HashMap::new(),
groups: HashMap::new(),
timestamp: None,
uuid: Uuid::now_v7(),
uuid: None,
minimal_flag_called: false,
}))
}
Expand All @@ -344,9 +348,19 @@ impl Event {

/// Override the auto-generated UUID for this event.
///
/// Useful for deduplication when re-importing historical data.
/// Useful for deduplication when re-importing historical data. Every
/// capture of an event with this uuid sends the same uuid, and PostHog
/// rejects a request that contains one uuid twice.
pub fn set_uuid(&mut self, uuid: Uuid) {
self.uuid = uuid;
self.uuid = Some(uuid);
}

/// Assign a v7 uuid if the event has none; the client calls this once when
/// it takes the event.
pub(crate) fn ensure_uuid(&mut self) {
if self.uuid.is_none() {
self.uuid = Some(Uuid::now_v7());
}
}

/// Attach the flag state captured by a [`FeatureFlagEvaluations`] snapshot
Expand Down Expand Up @@ -381,7 +395,7 @@ impl Event {
&self.distinct_id
}

pub(crate) fn uuid(&self) -> Uuid {
pub(crate) fn uuid(&self) -> Option<Uuid> {
self.uuid
}

Expand Down Expand Up @@ -435,27 +449,67 @@ mod test {

use chrono::{DateTime, Utc};

use uuid::Uuid;

use super::Event;
use crate::constants::PROCESS_PERSON_PROFILE_OPT;
use crate::Error;

#[test]
fn test_timestamp_is_correctly_set() {
let mut event = Event::new_anon("test");
let ts = DateTime::parse_from_rfc3339("2023-01-01T10:00:00+03:00").unwrap();
event.set_timestamp(ts).expect("Date is not in the future");
event.set_timestamp(ts).unwrap();
let expected = DateTime::parse_from_rfc3339("2023-01-01T07:00:00Z").unwrap();
assert_eq!(event.timestamp.unwrap(), expected.naive_utc())
}

#[test]
fn test_timestamp_is_correctly_set_with_future_date() {
fn future_timestamp_is_kept_for_posthog_to_handle() {
let mut event = Event::new_anon("test");
let ts = Utc::now() + Duration::from_secs(60);
event
.set_timestamp(ts)
.expect_err("Date is in the future, should be rejected");
event.set_timestamp(ts).unwrap();
assert_eq!(event.timestamp, Some(ts.naive_utc()));
}

#[test]
fn blank_distinct_id_makes_an_anonymous_event() {
for blank in ["", " ", " \n\t "] {
let event = Event::new("test", blank);
let generated = Uuid::parse_str(event.distinct_id()).unwrap();
assert_eq!(generated.get_version_num(), 7);
assert_eq!(
event.options().get(PROCESS_PERSON_PROFILE_OPT),
Some(&serde_json::json!(false))
);
}
let event = Event::new("test", " user-1 ");
assert_eq!(event.distinct_id(), " user-1 ");
assert!(event.options().is_empty());
}

assert!(event.timestamp.is_none())
#[test]
fn uuid_is_assigned_once_per_event_unless_set() {
let template = Event::new("test", "user1");
assert_eq!(template.uuid(), None);

// Each clone handed to the client gets its own uuid, kept on reassignment.
let mut first = template.clone();
let mut second = template.clone();
first.ensure_uuid();
second.ensure_uuid();
let assigned = first.uuid().unwrap();
assert_eq!(assigned.get_version_num(), 7);
assert_ne!(second.uuid(), Some(assigned));
first.ensure_uuid();
assert_eq!(first.uuid(), Some(assigned));

// A caller's uuid is never replaced.
let caller = Uuid::now_v7();
let mut event = template;
event.set_uuid(caller);
event.ensure_uuid();
assert_eq!(event.uuid(), Some(caller));
}

#[test]
Expand Down
40 changes: 40 additions & 0 deletions tests/test_capture.rs
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@

mod common;

use std::collections::HashSet;
use std::time::Duration;

use httpmock::prelude::*;
Expand Down Expand Up @@ -52,6 +53,19 @@ fn retry_body_prunes_terminal_events(req: &HttpMockRequest) -> bool {
body.contains(PARTIAL_UUID_RETRY) && !body.contains(PARTIAL_UUID_DROP)
}

/// Matcher: the body carries two events with distinct uuids.
fn batch_has_two_distinct_uuids(req: &HttpMockRequest) -> bool {
let body: serde_json::Value = match serde_json::from_slice(req.body_ref()) {
Ok(body) => body,
Err(_) => return false,
};
let uuids: HashSet<&str> = body["batch"]
.as_array()
.map(|batch| batch.iter().filter_map(|e| e["uuid"].as_str()).collect())
.unwrap_or_default();
body["batch"].as_array().map(Vec::len) == Some(2) && uuids.len() == 2
}

/// Poll until a mock has been hit `want` times (for the autonomous retry path
/// that isn't driven by `flush()`).
fn wait_for_hits(mock: &httpmock::Mock, want: usize) {
Expand Down Expand Up @@ -95,6 +109,32 @@ async fn capture_single_event_success() {
mock.assert();
}

#[tokio::test]
async fn capturing_one_event_twice_sends_two_uuids() {
// A reused or cloned `Event` must not resend one uuid: PostHog rejects a
// request that contains a uuid twice.
let server = MockServer::start();
let mock = server.mock(|when, then| {
when.method(POST)
.path("/i/v1/analytics/events")
.is_true(batch_has_two_distinct_uuids);
then.status(200)
.header("content-type", "application/json")
.json_body(json!({ "results": {} }));
});

let client = create_capture_client(server.base_url()).await;
let event = Event::new("test_event", "user-1");
client.capture(event.clone());
client.capture(event.clone());
client.flush().await;
client
.capture_batch_immediate(vec![event.clone(), event], false)
.await
.unwrap();
mock.assert_calls(2);
}

#[tokio::test]
async fn capture_bearer_auth_header() {
let server = MockServer::start();
Expand Down
Loading