Skip to content

Commit 8037fac

Browse files
authored
Merge pull request #12 from sidequery/session-replay-support
2 parents e2f1620 + e7a6e08 commit 8037fac

13 files changed

Lines changed: 1098 additions & 84 deletions

.github/workflows/ci.yml

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,10 @@ jobs:
4949
- name: Install JS test dependencies
5050
run: bun install --cwd tests/js_client --frozen-lockfile
5151

52+
- name: Install Playwright browser
53+
working-directory: tests/js_client
54+
run: bunx playwright install --with-deps chromium
55+
5256
- name: Verify Docker Compose
5357
run: docker compose version
5458

README.md

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -6,9 +6,9 @@ Hogflare is a Cloudflare Workers ingestion layer for PostHog SDKs. It supports P
66

77
#### What works today
88

9-
- Ingestion endpoints: `/capture`, `/identify`, `/alias`, `/batch`, `/e`, `/engage`, `/groups`
9+
- Ingestion endpoints: `/capture`, `/identify`, `/alias`, `/batch`, `/e`, `/engage`, `/groups`, `/s`
1010
- Persons and groups: `$set`, `$set_once`, `$unset`, aliasing, and group properties
11-
- Feature flags: `/flags` and `/decide` are evaluated in the Worker (used by PostHog SDKs)
11+
- SDK config and feature flags: `/array/:token/config`, `/flags`, and `/decide` are evaluated in the Worker
1212
- Request enrichment: Cloudflare IP/geo fields added when missing
1313
- Queryable people: append-only person snapshots can be written to a separate Iceberg table
1414

@@ -137,6 +137,7 @@ CLOUDFLARE_PIPELINE_TIMEOUT_SECS = "10"
137137
# POSTHOG_GROUP_TYPE_2 = "project"
138138
# POSTHOG_GROUP_TYPE_3 = "org"
139139
# POSTHOG_GROUP_TYPE_4 = "workspace"
140+
# POSTHOG_SESSION_RECORDING_ENDPOINT = "/s/"
140141

141142
[[durable_objects.bindings]]
142143
name = "PERSONS"
@@ -532,6 +533,7 @@ Use `--skip-person-output` when resuming an event import after person rows were
532533
- `/e` (event payloads)
533534
- `/engage`
534535
- `/groups`
536+
- `/s` (session replay payloads)
535537

536538
### Persons
537539

@@ -551,11 +553,14 @@ The Durable Object is the source of truth for the current person record. When `C
551553

552554
### Session replay
553555

554-
- `/s` stores raw session recording chunks only.
556+
- 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.
557+
- `/s` accepts PostHog replay payloads, including gzip/gzip-js compressed browser SDK requests.
558+
- Modern `$snapshot` payloads are normalized to `$snapshot_items` rows before they are sent through Cloudflare Pipelines into R2.
559+
- Legacy raw chunk payloads are still accepted as `$snapshot` rows.
555560

556561
### Feature flags
557562

558-
Feature flags are evaluated in the Worker and exposed via `/decide` and `/flags`.
563+
Feature flags and SDK remote config are evaluated in the Worker and exposed via `/array/:token/config`, `/decide`, and `/flags`.
559564

560565
Configuration is a JSON blob in `HOGFLARE_FEATURE_FLAGS`. It can be either:
561566

src/extractors.rs

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,11 @@ pub struct PostHogBatchPayload {
4141
pub batch: BatchRequest,
4242
}
4343

44+
pub struct PostHogRawPayload {
45+
pub items: Vec<Value>,
46+
pub sent_at: Option<DateTime<Utc>>,
47+
}
48+
4449
pub struct RequestEnrichment {
4550
properties: Map<String, Value>,
4651
}
@@ -291,6 +296,52 @@ impl FromRequest<AppState, Body> for PostHogBatchPayload {
291296
}
292297
}
293298

299+
#[async_trait]
300+
impl FromRequest<AppState, Body> for PostHogRawPayload {
301+
type Rejection = PayloadExtractorError;
302+
303+
async fn from_request(
304+
request: Request<Body>,
305+
state: &AppState,
306+
) -> Result<Self, Self::Rejection> {
307+
let (parts, body) = request.into_parts();
308+
let headers = parts.headers;
309+
let query = parts.uri.query().map(str::to_string);
310+
let bytes = to_bytes(body, usize::MAX)
311+
.await
312+
.map_err(PayloadExtractorError::BodyRead)?;
313+
314+
verify_signature(&headers, &bytes, state.signing_secret.as_deref())?;
315+
let decoded = decode_content_encoding(&headers, &bytes)?;
316+
let query_params = parse_query_params(query.as_deref())?;
317+
let query_compression = query_param(&query_params, "compression")
318+
.or_else(|| query_param(&query_params, "compression_method"));
319+
let mut payloads =
320+
parse_posthog_body_with_compression::<Value>(&headers, &decoded, query_compression)?;
321+
322+
if let Some(api_key) = header_api_key(&headers) {
323+
apply_api_key_to_values(&mut payloads, &api_key);
324+
}
325+
326+
Ok(PostHogRawPayload {
327+
items: payloads,
328+
sent_at: header_sent_at(&headers),
329+
})
330+
}
331+
}
332+
333+
fn apply_api_key_to_values(items: &mut [Value], api_key: &str) {
334+
for item in items {
335+
let Value::Object(map) = item else {
336+
continue;
337+
};
338+
if map.get("api_key").is_none() && map.get("token").is_none() && map.get("$token").is_none()
339+
{
340+
map.insert("api_key".to_string(), Value::String(api_key.to_string()));
341+
}
342+
}
343+
}
344+
294345
fn decode_content_encoding(
295346
headers: &HeaderMap,
296347
body: &[u8],

0 commit comments

Comments
 (0)