From e7a6e081486344939479c951f0589480b4fb370f Mon Sep 17 00:00:00 2001 From: Nico Ritschel Date: Thu, 14 May 2026 08:47:31 -0700 Subject: [PATCH] Support PostHog session replay --- .github/workflows/ci.yml | 4 + README.md | 13 +- src/extractors.rs | 51 ++ src/lib.rs | 435 +++++++++++++++--- src/models.rs | 4 +- src/pipeline.rs | 16 +- tests/js_client/bun.lock | 32 +- tests/js_client/package.json | 4 + .../posthog_session_replay_playwright.js | 216 +++++++++ tests/js_client/verify_rrweb_playback.js | 124 +++++ tests/pipeline_e2e.rs | 21 +- tests/posthog_endpoints.rs | 173 ++++++- tests/posthog_js.rs | 89 +++- 13 files changed, 1098 insertions(+), 84 deletions(-) create mode 100644 tests/js_client/posthog_session_replay_playwright.js create mode 100644 tests/js_client/verify_rrweb_playback.js diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7cecc44..0234c87 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -49,6 +49,10 @@ jobs: - name: Install JS test dependencies run: bun install --cwd tests/js_client --frozen-lockfile + - name: Install Playwright browser + working-directory: tests/js_client + run: bunx playwright install --with-deps chromium + - name: Verify Docker Compose run: docker compose version diff --git a/README.md b/README.md index 1fe4479..90d91c2 100644 --- a/README.md +++ b/README.md @@ -6,9 +6,9 @@ Hogflare is a Cloudflare Workers ingestion layer for PostHog SDKs. It supports P #### What works today -- Ingestion endpoints: `/capture`, `/identify`, `/alias`, `/batch`, `/e`, `/engage`, `/groups` +- Ingestion endpoints: `/capture`, `/identify`, `/alias`, `/batch`, `/e`, `/engage`, `/groups`, `/s` - Persons and groups: `$set`, `$set_once`, `$unset`, aliasing, and group properties -- Feature flags: `/flags` and `/decide` are evaluated in the Worker (used by PostHog SDKs) +- SDK config and feature flags: `/array/:token/config`, `/flags`, and `/decide` are evaluated in the Worker - Request enrichment: Cloudflare IP/geo fields added when missing - Queryable people: append-only person snapshots can be written to a separate Iceberg table @@ -137,6 +137,7 @@ CLOUDFLARE_PIPELINE_TIMEOUT_SECS = "10" # POSTHOG_GROUP_TYPE_2 = "project" # POSTHOG_GROUP_TYPE_3 = "org" # POSTHOG_GROUP_TYPE_4 = "workspace" +# POSTHOG_SESSION_RECORDING_ENDPOINT = "/s/" [[durable_objects.bindings]] name = "PERSONS" @@ -532,6 +533,7 @@ Use `--skip-person-output` when resuming an event import after person rows were - `/e` (event payloads) - `/engage` - `/groups` +- `/s` (session replay payloads) ### Persons @@ -551,11 +553,14 @@ The Durable Object is the source of truth for the current person record. When `C ### Session replay -- `/s` stores raw session recording chunks only. +- SDK config advertises `sessionRecording: false` when `POSTHOG_SESSION_RECORDING_ENDPOINT` is unset, so the Worker can keep replay off remotely. Set `POSTHOG_SESSION_RECORDING_ENDPOINT=/s/` to turn replay on and route uploads through Hogflare's ingestion path. +- `/s` accepts PostHog replay payloads, including gzip/gzip-js compressed browser SDK requests. +- Modern `$snapshot` payloads are normalized to `$snapshot_items` rows before they are sent through Cloudflare Pipelines into R2. +- Legacy raw chunk payloads are still accepted as `$snapshot` rows. ### Feature flags -Feature flags are evaluated in the Worker and exposed via `/decide` and `/flags`. +Feature flags and SDK remote config are evaluated in the Worker and exposed via `/array/:token/config`, `/decide`, and `/flags`. Configuration is a JSON blob in `HOGFLARE_FEATURE_FLAGS`. It can be either: diff --git a/src/extractors.rs b/src/extractors.rs index dd464cb..0f1e161 100644 --- a/src/extractors.rs +++ b/src/extractors.rs @@ -41,6 +41,11 @@ pub struct PostHogBatchPayload { pub batch: BatchRequest, } +pub struct PostHogRawPayload { + pub items: Vec, + pub sent_at: Option>, +} + pub struct RequestEnrichment { properties: Map, } @@ -291,6 +296,52 @@ impl FromRequest for PostHogBatchPayload { } } +#[async_trait] +impl FromRequest for PostHogRawPayload { + type Rejection = PayloadExtractorError; + + async fn from_request( + request: Request, + state: &AppState, + ) -> Result { + let (parts, body) = request.into_parts(); + let headers = parts.headers; + let query = parts.uri.query().map(str::to_string); + let bytes = to_bytes(body, usize::MAX) + .await + .map_err(PayloadExtractorError::BodyRead)?; + + verify_signature(&headers, &bytes, state.signing_secret.as_deref())?; + let decoded = decode_content_encoding(&headers, &bytes)?; + let query_params = parse_query_params(query.as_deref())?; + let query_compression = query_param(&query_params, "compression") + .or_else(|| query_param(&query_params, "compression_method")); + let mut payloads = + parse_posthog_body_with_compression::(&headers, &decoded, query_compression)?; + + if let Some(api_key) = header_api_key(&headers) { + apply_api_key_to_values(&mut payloads, &api_key); + } + + Ok(PostHogRawPayload { + items: payloads, + sent_at: header_sent_at(&headers), + }) + } +} + +fn apply_api_key_to_values(items: &mut [Value], api_key: &str) { + for item in items { + let Value::Object(map) = item else { + continue; + }; + if map.get("api_key").is_none() && map.get("token").is_none() && map.get("$token").is_none() + { + map.insert("api_key".to_string(), Value::String(api_key.to_string())); + } + } +} + fn decode_content_encoding( headers: &HeaderMap, body: &[u8], diff --git a/src/lib.rs b/src/lib.rs index 32a44df..ca9658f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -8,10 +8,9 @@ pub mod models; pub mod persons; pub mod pipeline; -use std::sync::Arc; +use std::{collections::HashMap, sync::Arc}; use axum::{ - body::Bytes, extract::{Path, State}, http::{HeaderMap, StatusCode}, response::IntoResponse, @@ -21,14 +20,13 @@ use axum::{ use chrono::Utc; use config::{Config, ConfigError}; use extractors::{ - header_api_key, header_sent_at, verify_signature, ApplyApiKey, PostHogBatchPayload, - PostHogPayload, RequestEnrichment, + ApplyApiKey, PostHogBatchPayload, PostHogPayload, PostHogRawPayload, RequestEnrichment, }; use feature_flags::{FeatureFlagContext, FeatureFlagStore}; use groups::{GroupError, GroupStore, GroupTypeMap, NoopGroupStore}; use models::{ - AliasRequest, BatchRequest, CaptureRequest, DecideResponse, DecideSessionRecording, - EngageRequest, ErrorResponse, GroupIdentifyRequest, IdentifyRequest, PostHogResponse, + AliasRequest, BatchRequest, CaptureRequest, DecideResponse, EngageRequest, ErrorResponse, + GroupIdentifyRequest, IdentifyRequest, PostHogResponse, }; use persons::{ alias_from_request, update_from_capture, update_from_engage, update_from_identify, @@ -443,6 +441,8 @@ fn router(state: AppState) -> Router { .route("/decide", post(decide)) .route("/flags", post(flags)) .route("/flags/", post(flags)) + .route("/array/:token/config", get(array_config)) + .route("/array/:token/config.js", get(array_config_js)) .route("/s", post(session_recording)) .route("/s/", post(session_recording)) .route("/__debug/person/:id", get(debug_person)) @@ -1169,11 +1169,7 @@ async fn decide( response.config.api_token = api_key; response.feature_flags = feature_flags; response.feature_flag_payloads = feature_flag_payloads; - - if let Some(endpoint) = state.session_recording_endpoint.clone() { - response.session_recording.endpoint = Some(endpoint); - response.session_recording.proxy = true; - } + response.session_recording = decide_session_recording_config(&state); Ok(Json(response)) } @@ -1193,7 +1189,7 @@ struct FlagsResponse { #[serde(rename = "evaluatedAt")] evaluated_at: i64, #[serde(rename = "sessionRecording", skip_serializing_if = "Option::is_none")] - session_recording: Option, + session_recording: Option, #[serde(rename = "supportedCompression", skip_serializing_if = "Vec::is_empty")] supported_compression: Vec, } @@ -1216,13 +1212,7 @@ async fn flags( let evaluated_at = Utc::now().timestamp_millis(); if include_config { - let mut recording = DecideSessionRecording::default(); - recording.proxy = true; - if let Some(endpoint) = state.session_recording_endpoint.clone() { - recording.endpoint = Some(endpoint); - } - session_recording = Some(recording); - + session_recording = Some(decide_session_recording_config(&state)); supported_compression = vec!["gzip".to_string(), "gzip-js".to_string()]; } @@ -1239,53 +1229,390 @@ async fn flags( .into_response()) } +fn decide_session_recording_config(state: &AppState) -> Value { + let Some(endpoint) = state.session_recording_endpoint.as_ref() else { + return Value::Bool(false); + }; + + json!({ + "endpoint": endpoint, + "consoleLogRecordingEnabled": true, + "proxy": true + }) +} + #[cfg_attr(target_arch = "wasm32", worker::send)] -async fn session_recording( +async fn array_config( State(state): State, - enrichment: RequestEnrichment, - headers: HeaderMap, - body: Bytes, -) -> Result, AppError> { - let raw = body.to_vec(); - verify_signature(&headers, &raw, state.signing_secret.as_deref()).map_err(AppError::from)?; + Path(token): Path, +) -> Result, AppError> { + Ok(Json(remote_config_response(&state, &token))) +} - let sent_at = header_sent_at(&headers); - let payload: Value = - serde_json::from_slice(&raw).map_err(|err| AppError::InvalidPayload(err.to_string()))?; +#[cfg_attr(target_arch = "wasm32", worker::send)] +async fn array_config_js( + State(state): State, + Path(token): Path, +) -> Result { + let config = remote_config_response(&state, &token); + let token_json = + serde_json::to_string(&token).map_err(|err| AppError::InvalidPayload(err.to_string()))?; + let config_json = + serde_json::to_string(&config).map_err(|err| AppError::InvalidPayload(err.to_string()))?; + let body = format!( + "(function() {{\n window._POSTHOG_REMOTE_CONFIG = window._POSTHOG_REMOTE_CONFIG || {{}};\n window._POSTHOG_REMOTE_CONFIG[{token_json}] = {{\n config: {config_json},\n siteApps: []\n }}\n}})();\n" + ); - let api_key = header_api_key(&headers).or_else(|| { - payload - .get("token") - .and_then(Value::as_str) - .map(|s| s.to_string()) - }); + Ok(( + [("content-type", "application/javascript; charset=utf-8")], + body, + ) + .into_response()) +} + +fn remote_config_response(state: &AppState, token: &str) -> Value { + json!({ + "token": token, + "supportedCompression": ["gzip", "gzip-js"], + "hasFeatureFlags": !state.feature_flags.is_empty(), + "analytics": { "endpoint": "/e/" }, + "captureDeadClicks": false, + "capturePerformance": false, + "autocapture_opt_out": false, + "autocaptureExceptions": false, + "elementsChainAsString": true, + "errorTracking": { + "autocaptureExceptions": false, + "suppressionRules": [] + }, + "logs": { "captureConsoleLogs": false }, + "sessionRecording": session_recording_remote_config(state), + "heatmaps": false, + "conversations": false, + "surveys": false, + "productTours": false, + "defaultIdentifiedOnly": true + }) +} - let distinct_id = payload - .pointer("/data/metadata/distinct_id") - .or_else(|| payload.get("distinct_id")) - .and_then(Value::as_str) - .unwrap_or("session-recording") - .to_string(); +fn session_recording_remote_config(state: &AppState) -> Value { + let Some(endpoint) = state.session_recording_endpoint.as_ref() else { + return Value::Bool(false); + }; - let snapshot = ensure_person_snapshot(&state, &distinct_id).await?; - let (person_id, person_created_at, person_properties) = person_fields(&snapshot); + json!({ + "endpoint": endpoint, + "consoleLogRecordingEnabled": true, + "recorderVersion": "v2", + "sampleRate": null, + "minimumDurationMilliseconds": null, + "networkPayloadCapture": null, + "recordCanvas": false, + "canvasFps": null, + "canvasQuality": null, + "scriptConfig": { "script": "posthog-recorder" }, + "version": 1, + "urlTriggers": [], + "urlBlocklist": [], + "eventTriggers": [], + "triggerMatchType": null, + "masking": null, + "linkedFlag": null + }) +} - let event = PipelineEvent::from_session_recording(distinct_id, payload, api_key) +#[cfg_attr(target_arch = "wasm32", worker::send)] +async fn session_recording( + State(state): State, + enrichment: RequestEnrichment, + axum::extract::Query(query): axum::extract::Query, + payload: PostHogRawPayload, +) -> Result { + let sent_at = payload.sent_at.clone(); + let items = expand_session_recording_values(payload.items)?; + let mut events = Vec::new(); + let mut person_records = Vec::new(); + + for item in items { + let normalized = normalize_session_recording(item)?; + let snapshot = ensure_person_snapshot(&state, &normalized.distinct_id).await?; + let (person_id, person_created_at, person_properties) = person_fields(&snapshot); + + let event = PipelineEvent::from_session_recording( + normalized.distinct_id, + normalized.event, + normalized.properties, + normalized.api_key, + normalized.timestamp, + normalized.extra, + ) .with_team_id(state.posthog_team_id) .with_person(person_id, person_created_at, person_properties) .with_groups([None, None, None, None, None], None) - .with_sent_at(sent_at) + .with_sent_at(sent_at.clone()) .with_enrichment(enrichment.properties()); + push_person_record(&mut person_records, &snapshot, "session_recording", &event); + events.push(event); + } - send_person_records( - &state, - PersonPipelineRecord::from_snapshot(&snapshot, "session_recording", &event) - .into_iter() - .collect(), - ) - .await?; - state.pipeline.send(vec![event]).await?; - Ok(Json(PostHogResponse::success())) + send_person_records(&state, person_records).await?; + state.pipeline.send(events).await?; + + if query.beacon.as_deref() == Some("1") { + Ok(StatusCode::NO_CONTENT.into_response()) + } else { + Ok(Json(PostHogResponse::success()).into_response()) + } +} + +#[derive(Default, Deserialize)] +struct SessionRecordingQuery { + #[serde(default)] + beacon: Option, +} + +struct NormalizedSessionRecording { + event: String, + distinct_id: String, + api_key: Option, + timestamp: Option>, + properties: Value, + extra: HashMap, +} + +fn expand_session_recording_values(values: Vec) -> Result, AppError> { + let mut expanded = Vec::new(); + + for value in values { + let Value::Object(mut object) = value else { + return Err(AppError::InvalidPayload( + "expected session recording object".to_string(), + )); + }; + + let Some(batch_value) = object.remove("batch") else { + expanded.push(Value::Object(object)); + continue; + }; + + let batch = batch_value.as_array().ok_or_else(|| { + AppError::InvalidPayload("expected session recording batch array".to_string()) + })?; + let shared = object; + + for item in batch { + let mut item_object = item.as_object().cloned().ok_or_else(|| { + AppError::InvalidPayload("expected session recording batch item object".to_string()) + })?; + + for key in ["api_key", "token", "$token"] { + if item_object.get(key).is_none() { + if let Some(value) = shared.get(key) { + item_object.insert(key.to_string(), value.clone()); + } + } + } + + expanded.push(Value::Object(item_object)); + } + } + + if expanded.is_empty() { + return Err(AppError::InvalidPayload( + "empty session recording batch".to_string(), + )); + } + + Ok(expanded) +} + +fn normalize_session_recording(value: Value) -> Result { + let Value::Object(object) = value else { + return Err(AppError::InvalidPayload( + "expected session recording object".to_string(), + )); + }; + + if object + .get("properties") + .and_then(Value::as_object) + .is_none() + && (object.get("metadata").is_some() || object.get("chunk").is_some()) + { + return normalize_legacy_session_recording(object); + } + + normalize_modern_session_recording(object) +} + +fn normalize_legacy_session_recording( + mut object: serde_json::Map, +) -> Result { + let api_key = recording_api_key(&object, None) + .ok_or_else(|| AppError::InvalidPayload("missing session recording token".to_string()))?; + let distinct_id = object + .get("metadata") + .and_then(Value::as_object) + .and_then(|metadata| metadata.get("distinct_id")) + .or_else(|| object.get("distinct_id")) + .and_then(recording_id_string) + .ok_or_else(|| AppError::InvalidPayload("missing distinct_id".to_string()))?; + + object.remove("api_key"); + object.remove("token"); + object.remove("$token"); + + Ok(NormalizedSessionRecording { + event: "$snapshot".to_string(), + distinct_id, + api_key: Some(api_key), + timestamp: None, + properties: json!({ "data": Value::Object(object) }), + extra: HashMap::new(), + }) +} + +fn normalize_modern_session_recording( + object: serde_json::Map, +) -> Result { + let properties = object + .get("properties") + .and_then(Value::as_object) + .ok_or_else(|| AppError::InvalidPayload("missing recording properties".to_string()))?; + + let api_key = recording_api_key(&object, Some(properties)) + .ok_or_else(|| AppError::InvalidPayload("missing session recording token".to_string()))?; + let distinct_id = object + .get("distinct_id") + .or_else(|| object.get("$distinct_id")) + .or_else(|| properties.get("distinct_id")) + .or_else(|| properties.get("$distinct_id")) + .and_then(recording_id_string) + .ok_or_else(|| AppError::InvalidPayload("missing distinct_id".to_string()))?; + + let event = object + .get("event") + .and_then(Value::as_str) + .unwrap_or("$snapshot"); + if event != "$snapshot" && event != "$snapshot_items" { + return Err(AppError::InvalidPayload(format!( + "unsupported session recording event {event}" + ))); + } + + let session_id = properties + .get("$session_id") + .and_then(Value::as_str) + .filter(|value| valid_session_id(value)) + .ok_or_else(|| AppError::InvalidPayload("missing or invalid $session_id".to_string()))?; + let session_id_value = Value::String(session_id.to_string()); + let window_id_value = properties + .get("$window_id") + .cloned() + .unwrap_or_else(|| session_id_value.clone()); + let snapshot_source = properties + .get("$snapshot_source") + .cloned() + .unwrap_or_else(|| Value::String("web".to_string())); + let snapshot_library = properties + .get("$lib") + .and_then(Value::as_str) + .unwrap_or("web") + .to_string(); + + let snapshot_items = properties + .get("$snapshot_items") + .or_else(|| properties.get("$snapshot_data")) + .ok_or_else(|| AppError::InvalidPayload("missing $snapshot_data".to_string())) + .and_then(snapshot_items_from_value)?; + + let mut output = properties.clone(); + output.remove("$snapshot_data"); + output.insert( + "distinct_id".to_string(), + Value::String(distinct_id.clone()), + ); + output.insert("$session_id".to_string(), session_id_value); + output.insert("$window_id".to_string(), window_id_value); + output.insert("$snapshot_source".to_string(), snapshot_source); + output.insert("$snapshot_items".to_string(), Value::Array(snapshot_items)); + output.insert("$lib".to_string(), Value::String(snapshot_library)); + + let mut extra = HashMap::new(); + if let Some(offset) = object.get("offset").cloned() { + extra.insert("offset".to_string(), offset); + } + + Ok(NormalizedSessionRecording { + event: "$snapshot_items".to_string(), + distinct_id, + api_key: Some(api_key), + timestamp: object.get("timestamp").and_then(recording_timestamp), + properties: Value::Object(output), + extra, + }) +} + +fn recording_api_key( + object: &serde_json::Map, + properties: Option<&serde_json::Map>, +) -> Option { + ["api_key", "token", "$token"] + .into_iter() + .find_map(|key| object.get(key).and_then(Value::as_str)) + .or_else(|| { + properties.and_then(|props| { + ["api_key", "token", "$token"] + .into_iter() + .find_map(|key| props.get(key).and_then(Value::as_str)) + }) + }) + .map(str::to_string) +} + +fn recording_id_string(value: &Value) -> Option { + match value { + Value::String(value) => { + let value = value.trim(); + if value.is_empty() { + None + } else { + Some(value.replace('\0', "\u{FFFD}")) + } + } + Value::Number(_) | Value::Bool(_) => Some(value.to_string()), + _ => None, + } +} + +fn valid_session_id(value: &str) -> bool { + !value.is_empty() + && value.len() <= 70 + && value + .chars() + .all(|ch| ch.is_ascii_alphanumeric() || ch == '-') +} + +fn snapshot_items_from_value(value: &Value) -> Result, AppError> { + match value { + Value::Array(items) => Ok(items.clone()), + Value::Object(item) => Ok(vec![Value::Object(item.clone())]), + _ => Err(AppError::InvalidPayload( + "missing $snapshot_data".to_string(), + )), + } +} + +fn recording_timestamp(value: &Value) -> Option> { + match value { + Value::String(value) => chrono::DateTime::parse_from_rfc3339(value) + .ok() + .map(|timestamp| timestamp.with_timezone(&chrono::Utc)), + Value::Number(value) => value + .as_i64() + .and_then(chrono::DateTime::::from_timestamp_millis), + _ => None, + } } #[cfg_attr(target_arch = "wasm32", worker::send)] diff --git a/src/models.rs b/src/models.rs index f9406c3..8b97fc3 100644 --- a/src/models.rs +++ b/src/models.rs @@ -128,7 +128,7 @@ pub struct DecideResponse { #[serde(rename = "errorsWhileComputingFlags")] pub errors_while_computing_flags: bool, #[serde(rename = "sessionRecording")] - pub session_recording: DecideSessionRecording, + pub session_recording: Value, #[serde(rename = "supportedCompression")] pub supported_compression: Vec, } @@ -141,7 +141,7 @@ impl Default for DecideResponse { feature_flag_payloads: HashMap::new(), config: DecideConfig::default(), errors_while_computing_flags: false, - session_recording: DecideSessionRecording::default(), + session_recording: Value::Bool(false), supported_compression: vec!["gzip".to_string(), "gzip-js".to_string()], } } diff --git a/src/pipeline.rs b/src/pipeline.rs index a7ce5d8..acfeb42 100644 --- a/src/pipeline.rs +++ b/src/pipeline.rs @@ -451,18 +451,21 @@ impl PipelineEvent { pub fn from_session_recording( distinct_id: String, - payload: Value, + event: String, + properties: Value, api_key: Option, + timestamp: Option>, + extra: std::collections::HashMap, ) -> Self { Self { uuid: Uuid::new_v4().to_string(), team_id: None, source: "posthog", - event: "$snapshot".to_string(), + event, distinct_id, created_at: Utc::now(), - timestamp: None, - properties: Some(payload), + timestamp, + properties: Some(properties), context: None, person_id: None, person_created_at: None, @@ -474,7 +477,7 @@ impl PipelineEvent { group4: None, group_properties: None, api_key, - extra: std::collections::HashMap::new(), + extra, } } @@ -659,8 +662,11 @@ mod tests { let event = PipelineEvent::from_session_recording( "recording-user".to_string(), + "$snapshot".to_string(), payload.clone(), Some("phc_session".to_string()), + None, + std::collections::HashMap::new(), ); assert_eq!(event.event, "$snapshot"); diff --git a/tests/js_client/bun.lock b/tests/js_client/bun.lock index a60c411..26de61c 100644 --- a/tests/js_client/bun.lock +++ b/tests/js_client/bun.lock @@ -10,6 +10,10 @@ "posthog-node": "^5.34.1", }, "devDependencies": { + "fflate": "0.4.8", + "playwright": "1.60.0", + "rrweb": "2.0.0-alpha.4", + "rrweb-player": "1.0.0-alpha.4", "wrangler": "4.90.1", }, }, @@ -205,18 +209,28 @@ "@protobufjs/utf8": ["@protobufjs/utf8@1.1.1", "", {}, "sha512-oOAWABowe8EAbMyWKM0tYDKi8Yaox52D+HWZhAIJqQXbqe0xI/GV7FhLWqlEKreMkfDjshR5FKgi3mnle0h6Eg=="], + "@rrweb/types": ["@rrweb/types@2.0.0-alpha.20", "", {}, "sha512-RbnDgKxA/odwB1R4gF7eUUj+rdSrq6ROQJsnMw7MIsGzlbSYvJeZN8YY4XqU0G6sKJvXI6bSzk7w/G94jNwzhw=="], + "@sindresorhus/is": ["@sindresorhus/is@7.2.0", "", {}, "sha512-P1Cz1dWaFfR4IR+U13mqqiGsLFf1KbayybWwdd2vfctdV6hDpUkgCY0nKOLLTMSoRd/jJNjtbqzf13K8DCCXQw=="], "@speed-highlight/core": ["@speed-highlight/core@1.2.15", "", {}, "sha512-BMq1K3DsElxDWawkX6eLg9+CKJrTVGCBAWVuHXVUV2u0s2711qiChLSId6ikYPfxhdYocLNt3wWwSvDiTvFabw=="], + "@tsconfig/svelte": ["@tsconfig/svelte@1.0.13", "", {}, "sha512-5lYJP45Xllo4yE/RUBccBT32eBlRDbqN8r1/MIvQbKxW3aFqaYPCNgm8D5V20X4ShHcwvYWNlKg3liDh1MlBoA=="], + + "@types/css-font-loading-module": ["@types/css-font-loading-module@0.0.7", "", {}, "sha512-nl09VhutdjINdWyXxHWN/w9zlNCfr60JUqJbd24YXUuCwgeL0TpFSdElCwb6cxfB6ybE19Gjj4g0jsgkXxKv1Q=="], + "@types/node": ["@types/node@25.7.0", "", { "dependencies": { "undici-types": "~7.21.0" } }, "sha512-z+pdZyxE+RTQE9AcboAZCb4otwcrvgHD+GlBpPgn0emDVt0ohrTMhAwlr2Wd9nZ+nihhYFxO2pThz3C5qSu2Eg=="], "@types/trusted-types": ["@types/trusted-types@2.0.7", "", {}, "sha512-ScaPdn1dQczgbl0QFTeTOmVHFULt394XJgOQNoyVhZ6r2vLnMLJfBPd53SB52T/3G36VI1/g2MZaX0cwDuXsfw=="], + "@xstate/fsm": ["@xstate/fsm@1.6.5", "", {}, "sha512-b5o1I6aLNeYlU/3CPlj/Z91ybk1gUsKT+5NAJI+2W4UjvS5KLG28K9v5UvNoFVjHV8PajVZ00RH3vnjyQO7ZAw=="], + "agent-base": ["agent-base@7.1.4", "", {}, "sha512-MnA+YT8fwfJPgBx3m60MNqakm30XOkyIoH1y6huTQvC0PwZG7ki8NacLBcrPbNoo8vEZy7Jpuk7+jMO+CUovTQ=="], "asynckit": ["asynckit@0.4.0", "", {}, "sha512-Oei9OH4tRh0YqU3GxhX79dM/mwVgvbZJaSNaRk+bshkj0S5cfHcgYakreBjrHwatXKbz+IoIdYLxrKim2MjW0Q=="], + "base64-arraybuffer": ["base64-arraybuffer@1.0.2", "", {}, "sha512-I3yl4r9QB5ZRY3XuJVEPfc2XhZO6YweFPI+UovAzn+8/hb3oJ6lnysaFcjVpkCPfVWFUDvoZ8kmVDP7WyRtYtQ=="], + "blake3-wasm": ["blake3-wasm@2.1.5", "", {}, "sha512-F1+K8EbfOZE49dtoPtmxUQrpXaBIl3ICvasLh+nJta0xkz+9kF/7uet9fLnwKqhDrmj6g+6K3Tw9yQPUg2ka5g=="], "call-bind-apply-helpers": ["call-bind-apply-helpers@1.0.2", "", { "dependencies": { "es-errors": "^1.3.0", "function-bind": "^1.1.2" } }, "sha512-Sp1ablJ0ivDkSzjcaJdxEunN5/XvksFJ2sMBFfq6x0ryhQV/2b/KwFe21cMpmHtPOSij8K99/wSfoEuTObmuMQ=="], @@ -261,7 +275,7 @@ "form-data": ["form-data@4.0.4", "", { "dependencies": { "asynckit": "^0.4.0", "combined-stream": "^1.0.8", "es-set-tostringtag": "^2.1.0", "hasown": "^2.0.2", "mime-types": "^2.1.12" } }, "sha512-KrGhL9Q4zjj0kiUt5OO4Mr/A/jlI2jDYs5eHBpYHPcBEVSiipAvn2Ko2HnPe20rmcuuvMHNdZFp+4IlGTMF0Ow=="], - "fsevents": ["fsevents@2.3.3", "", { "os": "darwin" }, "sha512-5xoDfX+fL7faATnagmWPpbFtwh/R77WmMMqqHGS65C3vvB0YHrgF+B1YmZ3441tMj5n63k0212XNoJwzlhffQw=="], + "fsevents": ["fsevents@2.3.2", "", { "os": "darwin" }, "sha512-xiqMQR4xAeHTuB9uWm+fFRcIOgKBMiOBP+eXiyT7jsgVCq1bkVygt00oASowB7EdtpOHaaPgKt812P9ab+DDKA=="], "function-bind": ["function-bind@1.1.2", "", {}, "sha512-7XHNxH7qX9xG5mIwxkhumTox/MIRNcOgDrxWsMt2pAr23WHp6MrRlN7FBSFpCpr+oVO0F744iUgR82nJMfG2SA=="], @@ -303,6 +317,8 @@ "miniflare": ["miniflare@4.20260508.0", "", { "dependencies": { "@cspotcode/source-map-support": "0.8.1", "sharp": "^0.34.5", "undici": "7.24.8", "workerd": "1.20260508.1", "ws": "8.18.0", "youch": "4.1.0-beta.10" }, "bin": { "miniflare": "bootstrap.js" } }, "sha512-h3aG+PA8jEH76V4ZtBAbs3g7kjMfHJUF8hPvxeeajLTKwir+G+dqfBODg5yF9MT29LqrZKCRQRqzfHPWX4kCIg=="], + "mitt": ["mitt@3.0.1", "", {}, "sha512-vKivATfr97l2/QBCYAkXYDbrIWPM2IIKEl7YPhjCvKlG3kE2gm+uBo6nEXK3M5/Ffh/FLpKExzOQ3JJoJGFKBw=="], + "ms": ["ms@2.1.3", "", {}, "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA=="], "nwsapi": ["nwsapi@2.2.22", "", {}, "sha512-ujSMe1OWVn55euT1ihwCI1ZcAaAU3nxUiDwfDQldc51ZXaB9m2AyOn6/jh1BLe2t/G8xd6uKG1UBF2aZJeg2SQ=="], @@ -313,6 +329,10 @@ "pathe": ["pathe@2.0.3", "", {}, "sha512-WUjGcAqP1gQacoQe+OBJsFA7Ld4DyXuUIjZ5cc75cLHvJ7dtNsTugphxIADwspS+AraAUePCKrSVtPLFj/F88w=="], + "playwright": ["playwright@1.60.0", "", { "dependencies": { "playwright-core": "1.60.0" }, "optionalDependencies": { "fsevents": "2.3.2" }, "bin": { "playwright": "cli.js" } }, "sha512-hheHdokM8cdqCb0lcE3s+zT4t4W+vvjpGxsZlDnikarzx8tSzMebh3UiFtgqwFwnTnjYQcsyMF8ei2mCO/tpeA=="], + + "playwright-core": ["playwright-core@1.60.0", "", { "bin": { "playwright-core": "cli.js" } }, "sha512-9bW6zvX/m0lEbgTKJ6YppOKx8H3VOPBMOCFh2irXFOT4BbHgrx5hPjwJYLT40Lu+4qtD36qKc/Hn56StUW57IA=="], + "posthog-js": ["posthog-js@1.373.4", "", { "dependencies": { "@opentelemetry/api": "^1.9.0", "@opentelemetry/api-logs": "^0.208.0", "@opentelemetry/exporter-logs-otlp-http": "^0.208.0", "@opentelemetry/resources": "^2.2.0", "@opentelemetry/sdk-logs": "^0.208.0", "@posthog/core": "1.29.1", "@posthog/types": "1.373.4", "core-js": "^3.38.1", "dompurify": "^3.3.2", "fflate": "^0.4.8", "preact": "^10.28.2", "query-selector-shadow-dom": "^1.0.1", "web-vitals": "^5.1.0" } }, "sha512-6DueUS5KOGZlKsQlelLURmtp9PA52rdCxt/IvuoYKAErlkh2LczogMopUIpeqbwfOnGDQEf8gmDh4z/IFeU4CQ=="], "posthog-node": ["posthog-node@5.34.1", "", { "dependencies": { "@posthog/core": "1.29.1" }, "peerDependencies": { "rxjs": "^7.0.0" }, "optionalPeers": ["rxjs"] }, "sha512-kGl0kSfh2+Ey3KL5Sji3yv9W5xwPK9sTkINRoFqCh9fbYXWWY6Zwi5Psv2QmRcbYiMJBk/iecnoOKVDRRga6PA=="], @@ -331,8 +351,16 @@ "requires-port": ["requires-port@1.0.0", "", {}, "sha512-KigOCHcocU3XODJxsu8i/j8T9tzT4adHiecwORRQ0ZZFcp7ahwXuRU1m+yuO90C5ZUyGeGfocHDI14M3L3yDAQ=="], + "rrdom": ["rrdom@0.1.7", "", { "dependencies": { "rrweb-snapshot": "^2.0.0-alpha.4" } }, "sha512-ZLd8f14z9pUy2Hk9y636cNv5Y2BMnNEY99wxzW9tD2BLDfe1xFxtLjB4q/xCBYo6HRe0wofzKzjm4JojmpBfFw=="], + + "rrweb": ["rrweb@2.0.0-alpha.4", "", { "dependencies": { "@rrweb/types": "^2.0.0-alpha.4", "@types/css-font-loading-module": "0.0.7", "@xstate/fsm": "^1.4.0", "base64-arraybuffer": "^1.0.1", "fflate": "^0.4.4", "mitt": "^3.0.0", "rrdom": "^0.1.7", "rrweb-snapshot": "^2.0.0-alpha.4" } }, "sha512-wEHUILbxDPcNwkM3m4qgPgXAiBJyqCbbOHyVoNEVBJzHszWEFYyTbrZqUdeb1EfmTRC2PsumCIkVcomJ/xcOzA=="], + "rrweb-cssom": ["rrweb-cssom@0.7.1", "", {}, "sha512-TrEMa7JGdVm0UThDJSx7ddw5nVm3UJS9o9CCIZ72B1vSyEZoziDqBYP3XIoi/12lKrJR8rE3jeFHMok2F/Mnsg=="], + "rrweb-player": ["rrweb-player@1.0.0-alpha.4", "", { "dependencies": { "@tsconfig/svelte": "^1.0.0", "rrweb": "^2.0.0-alpha.4" } }, "sha512-Wlmn9GZ5Fdqa37vd3TzsYdLl/JWEvXNUrLCrYpnOwEgmY409HwVIvvA5aIo7k582LoKgdRCsB87N+f0oWAR0Kg=="], + + "rrweb-snapshot": ["rrweb-snapshot@2.0.0-alpha.4", "", {}, "sha512-KQ2OtPpXO5jLYqg1OnXS/Hf+EzqnZyP5A+XPqBCjYpj3XIje/Od4gdUwjbFo3cVuWq5Cw5Y1d3/xwgIS7/XpQQ=="], + "safer-buffer": ["safer-buffer@2.1.2", "", {}, "sha512-YZo3K82SD7Riyi0E1EQPojLz7kpepnSQI9IyPbHHg1XXXevb5dJI7tpyN2ADxGcQbHG7vcyRHk0cbwqcQriUtg=="], "saxes": ["saxes@6.0.0", "", { "dependencies": { "xmlchars": "^2.2.0" } }, "sha512-xAg7SOnEhrm5zI3puOOKyy1OMcMlIJZYNJY7xLBwSze0UjhPLnWfj2GF2EpT0jmzaJKIWKHLsaSSajf35bcYnA=="], @@ -400,5 +428,7 @@ "cssstyle/rrweb-cssom": ["rrweb-cssom@0.8.0", "", {}, "sha512-guoltQEx+9aMf2gDZ0s62EcV8lsXR+0w8915TC3ITdn2YueuNjdAYh/levpU9nFaoChh9RUS5ZdQMrKfVEN9tw=="], "miniflare/ws": ["ws@8.18.0", "", { "peerDependencies": { "bufferutil": "^4.0.1", "utf-8-validate": ">=5.0.2" }, "optionalPeers": ["bufferutil", "utf-8-validate"] }, "sha512-8VbfWfHLbbwu3+N6OKsOMpBdT4kXPDDB9cJk2bJ6mh9ucxdlnNvH1e+roYkKmN9Nxw2yjz7VzeO9oOz2zJ04Pw=="], + + "wrangler/fsevents": ["fsevents@2.3.3", "", { "os": "darwin" }, "sha512-5xoDfX+fL7faATnagmWPpbFtwh/R77WmMMqqHGS65C3vvB0YHrgF+B1YmZ3441tMj5n63k0212XNoJwzlhffQw=="], } } diff --git a/tests/js_client/package.json b/tests/js_client/package.json index b5f0ff5..d8591b5 100644 --- a/tests/js_client/package.json +++ b/tests/js_client/package.json @@ -10,6 +10,10 @@ "posthog-node": "^5.34.1" }, "devDependencies": { + "fflate": "0.4.8", + "playwright": "1.60.0", + "rrweb": "2.0.0-alpha.4", + "rrweb-player": "1.0.0-alpha.4", "wrangler": "4.90.1" } } diff --git a/tests/js_client/posthog_session_replay_playwright.js b/tests/js_client/posthog_session_replay_playwright.js new file mode 100644 index 0000000..3d10b3f --- /dev/null +++ b/tests/js_client/posthog_session_replay_playwright.js @@ -0,0 +1,216 @@ +import { createRequire } from 'module'; +import path from 'path'; +import { chromium } from 'playwright'; + +const require = createRequire(import.meta.url); + +const apiHost = process.env.HOGFLARE_HOST; +if (!apiHost) { + throw new Error('HOGFLARE_HOST must be provided'); +} + +const apiKey = process.env.HOGFLARE_API_KEY || 'phc_test_replay_key'; +const distinctId = process.env.HOGFLARE_DISTINCT_ID || 'js-replay-user'; +const posthogPackage = path.dirname(require.resolve('posthog-js/package.json')); +const posthogBundle = path.join(posthogPackage, 'dist', 'array.full.no-external.js'); + +const browser = await chromium.launch({ headless: true }); +const context = await browser.newContext({ + viewport: { width: 900, height: 600 }, +}); +const page = await context.newPage(); + +const consoleMessages = []; +const requests = []; +page.on('console', (message) => { + consoleMessages.push(`${message.type()}: ${message.text()}`); +}); +page.on('pageerror', (error) => { + consoleMessages.push(`pageerror: ${error.message}`); +}); +page.on('request', (request) => { + requests.push(`${request.method()} ${request.url()}`); +}); +page.on('requestfailed', (request) => { + requests.push(`FAILED ${request.method()} ${request.url()} ${request.failure()?.errorText || ''}`); +}); + +let replayStatus = null; +const replayResponse = new Promise((resolve) => { + page.on('response', (response) => { + if (response.url().includes('/s') && response.request().method() === 'POST') { + replayStatus = response.status(); + resolve(response.status()); + } + }); +}); + +await page.goto(`${apiHost}/healthz`); +await page.setContent(` + + + + Hogflare replay verification + + + +
+

Replay Ready

+ + +

Waiting for interaction

+
+ + + +`); +await page.addScriptTag({ path: posthogBundle }); + +await page.evaluate( + ({ apiHost, apiKey, distinctId }) => + new Promise((resolve) => { + window.posthog.init(apiKey, { + api_host: apiHost, + capture_pageview: false, + autocapture: false, + request_batching: false, + disable_session_recording: false, + disable_surveys: true, + opt_out_useragent_filter: true, + before_send: (event) => event, + bootstrap: { + distinctID: distinctId, + isIdentifiedID: true, + }, + loaded: () => resolve(true), + on_request_error: (error) => { + console.error('PostHog request error', error); + }, + }); + }), + { apiHost, apiKey, distinctId }, +); + +await page.evaluate(() => { + window.__hogflareReplayDebug = { + captureCalls: [], + sendRequests: [], + }; + + const originalCapture = window.posthog.capture.bind(window.posthog); + window.posthog.capture = (...args) => { + const call = { + event: args[0], + optionKeys: Object.keys(args[2] || {}), + url: args[2]?._url || null, + batchKey: args[2]?._batchKey || null, + isCapturing: window.posthog.is_capturing?.(), + requestBatching: window.posthog.config?.request_batching, + }; + window.__hogflareReplayDebug.captureCalls.push(call); + const result = originalCapture(...args); + call.returned = !!result; + call.resultEvent = result?.event || null; + call.resultKeys = result ? Object.keys(result) : []; + return result; + }; + + const originalSendRequest = window.posthog._send_request.bind(window.posthog); + window.posthog._send_request = (options) => { + window.__hogflareReplayDebug.sendRequests.push({ + method: options?.method, + url: options?.url, + batchKey: options?.batchKey || null, + compression: options?.compression || null, + event: Array.isArray(options?.data) ? 'array' : options?.data?.event || null, + }); + return originalSendRequest(options); + }; +}); + +await page.waitForFunction( + () => window.posthog?.sessionRecording && ['active', 'buffering'].includes(window.posthog.sessionRecording.status), + null, + { timeout: 15000 }, +); + +await page.fill('#name', 'Replay input value'); +await page.click('#record-button'); +await page.evaluate(() => { + const el = document.createElement('div'); + el.id = 'late-mutation'; + el.textContent = 'Late mutation marker'; + document.getElementById('app').appendChild(el); +}); + +await page.waitForFunction( + () => { + const lazyRecorder = window.posthog?.sessionRecording?._lazyLoadedSessionRecording; + return (lazyRecorder?._buffer?.data?.length || 0) > 0; + }, + null, + { timeout: 10000 }, +); +await page.waitForTimeout(500); + +const flushResult = await page.evaluate(() => { + const posthog = window.posthog; + const lazyRecorder = posthog?.sessionRecording?._lazyLoadedSessionRecording; + + if (lazyRecorder && typeof lazyRecorder._flushBuffer === 'function') { + const bufferLengthBeforeFlush = lazyRecorder._buffer?.data?.length || 0; + lazyRecorder._flushBuffer(); + return { + status: posthog.sessionRecording.status, + flushedVia: '_flushBuffer', + started: posthog.sessionRecordingStarted?.(), + bufferLengthBeforeFlush, + bufferLengthAfterFlush: lazyRecorder._buffer?.data?.length || 0, + debug: window.__hogflareReplayDebug, + }; + } + + posthog?.sessionRecording?.stopRecording?.(); + return { + status: posthog?.sessionRecording?.status, + flushedVia: 'stopRecording', + started: posthog?.sessionRecordingStarted?.(), + debug: window.__hogflareReplayDebug, + }; +}); + +const status = await Promise.race([ + replayResponse, + page.waitForTimeout(15000).then(() => null), +]); + +await browser.close(); + +if (status === null || replayStatus === null || replayStatus >= 400) { + throw new Error( + `Replay upload did not complete successfully. replayStatus=${replayStatus}, flush=${JSON.stringify( + flushResult, + )}, requests=${JSON.stringify(requests)}, console=${JSON.stringify(consoleMessages)}`, + ); +} + +console.log( + JSON.stringify({ + replayStatus, + flushResult, + requests, + consoleMessages, + }), +); diff --git a/tests/js_client/verify_rrweb_playback.js b/tests/js_client/verify_rrweb_playback.js new file mode 100644 index 0000000..972d859 --- /dev/null +++ b/tests/js_client/verify_rrweb_playback.js @@ -0,0 +1,124 @@ +import { createRequire } from 'module'; +import { gunzipSync, strFromU8, strToU8 } from 'fflate'; +import fs from 'fs'; +import path from 'path'; +import { chromium } from 'playwright'; + +const require = createRequire(import.meta.url); + +const replayEventsPath = process.env.HOGFLARE_REPLAY_EVENTS_FILE; +if (!replayEventsPath) { + throw new Error('HOGFLARE_REPLAY_EVENTS_FILE must be provided'); +} + +const events = JSON.parse(fs.readFileSync(replayEventsPath, 'utf8')); +if (!Array.isArray(events) || events.length === 0) { + throw new Error('Replay events file must contain a non-empty rrweb event array'); +} + +const gunzipJsonField = (value) => JSON.parse(strFromU8(gunzipSync(strToU8(value, true)))); + +const decodePostHogCompressedEvent = (event) => { + const decoded = structuredClone(event); + if (decoded?.cv !== '2024-10') { + return decoded; + } + + if (typeof decoded.data === 'string') { + decoded.data = gunzipJsonField(decoded.data); + } else if (decoded.data && typeof decoded.data === 'object') { + for (const field of ['adds', 'attributes', 'removes', 'texts']) { + if (typeof decoded.data[field] === 'string') { + decoded.data[field] = gunzipJsonField(decoded.data[field]); + } + } + } + + delete decoded.cv; + return decoded; +}; + +const playableEvents = events.map(decodePostHogCompressedEvent); +const eventTypes = new Set(playableEvents.map((event) => event?.type)); +if (!eventTypes.has(2)) { + throw new Error(`Replay events are missing an rrweb FullSnapshot event. types=${[...eventTypes].join(',')}`); +} + +const rrwebPackage = path.dirname(require.resolve('rrweb/package.json')); +const rrwebPlayerPackage = path.dirname(require.resolve('rrweb-player/package.json')); + +const browser = await chromium.launch({ headless: true }); +const page = await browser.newPage({ viewport: { width: 1000, height: 720 } }); +const errors = []; +page.on('console', (message) => { + if (message.type() === 'error') { + errors.push(message.text()); + } +}); +page.on('pageerror', (error) => errors.push(error.message)); + +await page.setContent('
'); +await page.addStyleTag({ path: path.join(rrwebPackage, 'dist', 'rrweb.min.css') }); +await page.addStyleTag({ path: path.join(rrwebPlayerPackage, 'dist', 'style.css') }); +await page.addScriptTag({ path: path.join(rrwebPackage, 'dist', 'rrweb-all.min.js') }); +await page.addScriptTag({ path: path.join(rrwebPlayerPackage, 'dist', 'index.js') }); + +await page.evaluate((replayEvents) => { + window.__hogflarePlayer = new window.rrwebPlayer({ + target: document.getElementById('player'), + props: { + events: replayEvents, + autoPlay: false, + showController: true, + width: 900, + height: 560, + }, + }); +}, playableEvents); + +await page.waitForSelector('.rr-player', { timeout: 10000 }); +await page.waitForTimeout(500); + +const duration = playableEvents[playableEvents.length - 1].timestamp - playableEvents[0].timestamp; +await page.evaluate((timeOffset) => { + window.__hogflarePlayer.goto(Math.max(0, timeOffset), false); +}, duration); +await page.waitForTimeout(500); + +const result = await page.evaluate(() => { + const iframe = document.querySelector('iframe'); + const replayText = iframe?.contentDocument?.body?.innerText || ''; + const playerBox = document.querySelector('.rr-player')?.getBoundingClientRect(); + + return { + replayText, + playerWidth: playerBox?.width || 0, + playerHeight: playerBox?.height || 0, + iframeCount: document.querySelectorAll('iframe').length, + }; +}); + +await browser.close(); + +if (errors.length > 0) { + throw new Error(`rrweb-player emitted errors: ${JSON.stringify(errors)}`); +} + +if (!result.replayText.includes('Replay Ready')) { + throw new Error(`Replay frame did not render the recorded DOM. text=${JSON.stringify(result.replayText)}`); +} + +if (result.playerWidth <= 0 || result.playerHeight <= 0 || result.iframeCount === 0) { + throw new Error(`rrweb-player did not render a visible replay surface: ${JSON.stringify(result)}`); +} + +console.log( + JSON.stringify({ + eventCount: playableEvents.length, + compressedEventCount: events.filter((event) => event?.cv === '2024-10').length, + eventTypes: [...eventTypes].sort(), + replayText: result.replayText, + playerWidth: result.playerWidth, + playerHeight: result.playerHeight, + }), +); diff --git a/tests/pipeline_e2e.rs b/tests/pipeline_e2e.rs index e9a0010..1fb25d1 100644 --- a/tests/pipeline_e2e.rs +++ b/tests/pipeline_e2e.rs @@ -117,10 +117,16 @@ async fn pipeline_handles_capture_batch_alias_and_session() -> Result<(), Box Result<(), Box String { let mut mac = Hmac::::new_from_slice(secret.as_bytes()).unwrap(); @@ -621,13 +622,60 @@ async fn posthog_compatibility_endpoints_forward_events() -> Result<(), Box Result<(), Box Result<(), Box> { + let (pipeline_endpoint, _pipeline_rx, pipeline_handle) = spawn_pipeline_stub().await?; + let (address, server_handle) = + spawn_app_with_options(pipeline_endpoint, None, None, None, None).await?; + + let base_url = format!("http://{}", address); + let client = Client::builder().timeout(Duration::from_secs(5)).build()?; + + let decide_response = client + .post(format!("{}/decide", base_url)) + .json(&json!({ "token": "phc_no_replay" })) + .send() + .await?; + assert!(decide_response.status().is_success()); + let decide_body: Value = decide_response.json().await?; + assert_eq!(decide_body["sessionRecording"], false); + + let config_response = client + .get(format!("{}/array/phc_no_replay/config", base_url)) + .send() + .await?; + assert!(config_response.status().is_success()); + let config_body: Value = config_response.json().await?; + assert_eq!(config_body["sessionRecording"], false); + + let flags_config_response = client + .post(format!("{}/flags?config=true", base_url)) + .json(&json!({ "token": "phc_no_replay" })) + .send() + .await?; + assert!(flags_config_response.status().is_success()); + let flags_config_body: Value = flags_config_response.json().await?; + assert_eq!(flags_config_body["sessionRecording"], false); + + cleanup(server_handle, pipeline_handle).await; + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn persons_pipeline_receives_anon_to_identified_snapshots( ) -> Result<(), Box> { diff --git a/tests/posthog_js.rs b/tests/posthog_js.rs index 02063cf..ab8bdce 100644 --- a/tests/posthog_js.rs +++ b/tests/posthog_js.rs @@ -1,7 +1,7 @@ #[path = "helpers/mod.rs"] mod helpers; -use std::time::Duration; +use std::{io::Write, time::Duration}; use helpers::{ cleanup, collect_events_until, spawn_app, spawn_app_with_options, @@ -11,6 +11,7 @@ use helpers::{ use hogflare::groups::GroupTypeMap; use reqwest::Client; use serde_json::Value; +use tempfile::NamedTempFile; use tokio::process::Command; #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -522,3 +523,89 @@ async fn posthog_js_multiple_events_forwarded_to_pipeline() -> Result<(), Box Result<(), Box> { + let api_key = "phc_replay_playback"; + let distinct_id = "js-replay-playback-user"; + let (pipeline_endpoint, mut pipeline_rx, pipeline_handle) = spawn_pipeline_stub().await?; + let (address, server_handle) = spawn_app_with_options( + pipeline_endpoint, + Some(api_key.to_string()), + Some("/s/".to_string()), + None, + None, + ) + .await?; + + let record_output = Command::new("bun") + .arg("run") + .arg("posthog_session_replay_playwright.js") + .current_dir("tests/js_client") + .env("HOGFLARE_HOST", format!("http://{}", address)) + .env("HOGFLARE_API_KEY", api_key) + .env("HOGFLARE_DISTINCT_ID", distinct_id) + .output() + .await?; + + if !record_output.status.success() { + return Err(format!( + "posthog replay script exited with status {:?}\nstdout:\n{}\nstderr:\n{}", + record_output.status, + String::from_utf8_lossy(&record_output.stdout), + String::from_utf8_lossy(&record_output.stderr) + ) + .into()); + } + + let events = collect_events_until(&mut pipeline_rx, 1, Duration::from_secs(15)).await?; + let replay_event = events + .iter() + .find(|event| event["event"] == "$snapshot_items") + .ok_or_else(|| format!("missing $snapshot_items event: {events:?}"))?; + + assert_eq!(replay_event["source"], "posthog"); + assert_eq!(replay_event["api_key"], api_key); + assert_eq!(replay_event["distinct_id"], distinct_id); + assert_eq!(replay_event["properties"]["distinct_id"], distinct_id); + assert_eq!(replay_event["properties"]["$lib"], "web"); + + let snapshot_items = replay_event["properties"]["$snapshot_items"] + .as_array() + .ok_or("expected $snapshot_items array")?; + assert!( + snapshot_items.len() >= 2, + "expected at least a full snapshot and one follow-up event, got {snapshot_items:?}" + ); + assert!( + snapshot_items.iter().any(|event| event["type"] == 2), + "expected replay output to contain an rrweb full snapshot: {snapshot_items:?}" + ); + + let mut replay_file = NamedTempFile::new()?; + replay_file.write_all(serde_json::to_string(snapshot_items)?.as_bytes())?; + replay_file.flush()?; + + let playback_output = Command::new("bun") + .arg("run") + .arg("verify_rrweb_playback.js") + .current_dir("tests/js_client") + .env("HOGFLARE_REPLAY_EVENTS_FILE", replay_file.path()) + .output() + .await?; + + if !playback_output.status.success() { + return Err(format!( + "rrweb playback script exited with status {:?}\nrecord stdout:\n{}\nrecord stderr:\n{}\nplayback stdout:\n{}\nplayback stderr:\n{}", + playback_output.status, + String::from_utf8_lossy(&record_output.stdout), + String::from_utf8_lossy(&record_output.stderr), + String::from_utf8_lossy(&playback_output.stdout), + String::from_utf8_lossy(&playback_output.stderr) + ) + .into()); + } + + cleanup(server_handle, pipeline_handle).await; + Ok(()) +}