Fix tests and update README

Author: nicosuave

.gitignore

@@ -8,4 +8,5 @@ tests/js_client/node_modules/
 dist/
 wrangler.toml
 scripts/setup-cloudflare.sh
+scripts/pipeline-schema.json
 .claude/

Cargo.lock

@@ -4,7 +4,7 @@ version = "0.1.0"
 edition = "2021"
 
 [lib]
-crate-type = ["cdylib"]
+crate-type = ["cdylib", "rlib"]
 
 [dependencies]
 axum = { version = "0.7", default-features = false, features = ["json"] }
@@ -29,6 +29,7 @@ worker = { version = "0.7", features = ["http"] }
 tower-service = "0.3.3"
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
+axum = { version = "0.7", default-features = false, features = ["json", "tokio", "http1"] }
 dotenvy = "0.15"
 reqwest = { version = "0.12", features = ["json", "rustls-tls"] }
 tokio = { version = "1.37", features = ["macros", "rt-multi-thread", "process", "net", "time"] }

Cargo.toml

@@ -6,7 +6,7 @@ Hogflare is a Cloudflare Workers service that accepts PostHog ingestion requests
 
 ## Why?
 
-PostHog is a nice-to-use web & product analytics platform. However, self-hosting PostHog is prohibitively complex so most users seem to rely on the cloud offering. This is an alternative for cost-conscious data folks & businesses interested in a low maintenance way to ingest web & product analytics directly into a managed data lake. 
+PostHog is a nice-to-use web & product analytics platform. However, self-hosting PostHog is prohibitively complex so most users seem to rely on the cloud offering. This is an alternative for cost-conscious data folks & businesses interested in a low maintenance way to ingest web & product analytics directly into a managed data lake.
 
 A [hobby deployment of PostHog](https://github.com/PostHog/posthog/blob/master/docker-compose.hobby.yml) includes: postgres, redis, redis7, clickhouse, zookeeper, kafka, worker, web, plugins, proxy, objectstorage, seaweedfs, asyncmigrationscheck, temporal, elasticsearch, temporal-admin-tools, temporal-ui, temporal-django-worker, cyclotron-janitor, capture, replay-capture, property-defs-rs, livestream, feature-flags, cymbal
 
@@ -16,7 +16,7 @@ Admittedly, PostHog does a *lot* more than this package, but some folks really j
 
 1) Create a Pipeline stream and sink in the Cloudflare dashboard or via `wrangler pipelines setup`.
 2) Use the schema below for the stream.
-3) Create a local `wrangler.toml` (gitignored) and set variables.
+3) Copy `wrangler.toml.example` to `wrangler.toml` and set variables.
 4) Set Wrangler secrets.
 5) Deploy the Worker.
 
@@ -38,11 +38,17 @@ Admittedly, PostHog does a *lot* more than this package, but some folks really j
 }
 ```
 
-### Local `wrangler.toml` (gitignored)
+### Wrangler config
+
+Copy the example (gitignored) and fill in your stream endpoint:
+
+```bash
+cp wrangler.toml.example wrangler.toml
+```
 
 ```toml
 name = "hogflare"
-main = "build/index.js"
+main = "build/index.js" # generated entrypoint from worker-build for the Rust worker
 compatibility_date = "2025-01-09"
 
 [vars]
@@ -78,6 +84,35 @@ curl -X POST https://<your-worker>.workers.dev/capture \
   ]'
 ```
 
+## PostHog SDK config (posthog-js)
+
+```js
+import posthog from "posthog-js";
+
+posthog.init("<project_api_key>", {
+  api_host: "https://<your-worker>.workers.dev",
+  capture_pageview: true,
+});
+```
+
+## Local development (fake pipeline)
+
+The repo includes a lightweight fake pipeline (FastAPI + DuckDB) used by tests.
+
+```bash
+docker compose up --build -d fake-pipeline
+```
+
+```bash
+# .env.local (not committed)
+CLOUDFLARE_PIPELINE_ENDPOINT=http://127.0.0.1:8088/
+CLOUDFLARE_PIPELINE_TIMEOUT_SECS=5
+```
+
+```bash
+cargo run
+```
+
 ## Query data (DuckDB)
 
 ```sql
@@ -86,14 +121,6 @@ INSTALL iceberg;
 LOAD httpfs;
 LOAD iceberg;
 
-CREATE SECRET r2_secret (
-  TYPE S3,
-  KEY_ID '<R2_ACCESS_KEY_ID>',
-  SECRET '<R2_SECRET_ACCESS_KEY>',
-  ENDPOINT '<ACCOUNT_ID>.r2.cloudflarestorage.com',
-  REGION 'auto'
-);
-
 CREATE SECRET r2_catalog_secret (
   TYPE ICEBERG,
   TOKEN '<CLOUDFLARE_API_TOKEN>'
@@ -132,6 +159,5 @@ Hogflare adds Cloudflare request data into `properties` when those keys are not
 
 ## Limitations
 
-- This repo does not include a local Pipelines emulator. You need a real Cloudflare Pipeline endpoint.
 - `/decide` returns placeholders, not evaluated flags.
 - `/s` stores raw session recording chunks only.

README.md

@@ -769,7 +769,7 @@ mod tests {
     use chrono::TimeZone;
     use flate2::write::{GzEncoder, ZlibEncoder};
     use flate2::Compression;
-    use serde_json::json;
+    use serde_json::{json, Value};
     use std::{io::Write, sync::Arc, time::Duration};
 
     use crate::{models::CaptureRequest, pipeline::PipelineClient, AppState};
@@ -1039,4 +1039,42 @@ mod tests {
         let expected = chrono::Utc.with_ymd_and_hms(2025, 2, 2, 0, 0, 0).unwrap();
         assert_eq!(payload.batch.sent_at, Some(expected));
     }
+
+    #[test]
+    fn enrichment_reads_cloudflare_headers() {
+        let request = Request::builder()
+            .uri("/capture")
+            .header("cf-connecting-ip", "203.0.113.10")
+            .header("cf-ray", "ray-123")
+            .body(Body::empty())
+            .unwrap();
+        let (parts, _) = request.into_parts();
+
+        let props = build_enrichment_properties(&parts);
+
+        assert_eq!(
+            props.get("$ip"),
+            Some(&Value::String("203.0.113.10".to_string()))
+        );
+        assert_eq!(
+            props.get("cf_ray"),
+            Some(&Value::String("ray-123".to_string()))
+        );
+    }
+
+    #[test]
+    fn enrichment_ignores_empty_header_values() {
+        let request = Request::builder()
+            .uri("/capture")
+            .header("cf-connecting-ip", "   ")
+            .header("cf-ray", "")
+            .body(Body::empty())
+            .unwrap();
+        let (parts, _) = request.into_parts();
+
+        let props = build_enrichment_properties(&parts);
+
+        assert!(props.get("$ip").is_none());
+        assert!(props.get("cf_ray").is_none());
+    }
 }

src/extractors.rs

@@ -334,7 +334,7 @@ pub enum PipelineError {
 mod tests {
     use super::*;
     use chrono::TimeZone;
-    use serde_json::json;
+    use serde_json::{json, Value};
 
     #[test]
     fn converts_group_identify_payload() {
@@ -430,4 +430,46 @@ mod tests {
         assert_eq!(event.api_key.as_deref(), Some("phc_session"));
         assert_eq!(event.properties, Some(payload));
     }
+
+    #[test]
+    fn merges_enrichment_without_overwrite() {
+        let payload = CaptureRequest {
+            api_key: Some("phc_capture".to_string()),
+            event: "test-event".to_string(),
+            distinct_id: "user-1".to_string(),
+            properties: Some(json!({
+                "$ip": "203.0.113.1",
+                "existing": true
+            })),
+            timestamp: None,
+            context: None,
+            extra: std::collections::HashMap::new(),
+        };
+
+        let mut enrichment = Map::new();
+        enrichment.insert(
+            "$ip".to_string(),
+            Value::String("198.51.100.2".to_string()),
+        );
+        enrichment.insert(
+            "cf_ray".to_string(),
+            Value::String("ray-xyz".to_string()),
+        );
+
+        let event = PipelineEvent::from_capture(payload).with_enrichment(&enrichment);
+        let props = match event.properties {
+            Some(Value::Object(props)) => props,
+            other => panic!("expected properties object, got {other:?}"),
+        };
+
+        assert_eq!(
+            props.get("$ip"),
+            Some(&Value::String("203.0.113.1".to_string()))
+        );
+        assert_eq!(
+            props.get("cf_ray"),
+            Some(&Value::String("ray-xyz".to_string()))
+        );
+        assert_eq!(props.get("existing"), Some(&Value::Bool(true)));
+    }
 }

src/pipeline.rs

@@ -0,0 +1,7 @@
+name = "hogflare"
+main = "build/index.js" # generated entrypoint from worker-build for the Rust worker
+compatibility_date = "2025-01-09"
+
+[vars]
+CLOUDFLARE_PIPELINE_ENDPOINT = "https://<stream-id>.ingest.cloudflare.com"
+CLOUDFLARE_PIPELINE_TIMEOUT_SECS = "10"