Merge pull request #11 from sidequery/posthog-import-idempotency

Add idempotent PostHog import

Author: nicosuave

Cargo.lock

@@ -27,7 +27,7 @@ tower-http = { version = "0.5", features = ["trace"] }
 url = "2.5"
 worker = { version = "0.7", features = ["http"] }
 tower-service = "0.3.3"
-uuid = { version = "1.7", features = ["v4", "serde", "js"] }
+uuid = { version = "1.7", features = ["v4", "v5", "serde", "js"] }
 getrandom = { version = "0.2", features = ["js"] }
 regex = "1"
 

Cargo.toml

@@ -409,6 +409,118 @@ catalog.drop_table(("default", "<table-name>"), purge_requested=True)
 PY
 ```
 
+## Import existing PostHog data
+
+Hogflare includes a host-side importer for backfilling an existing PostHog project into the same Cloudflare Pipeline sink used by the Worker. It reads PostHog's private API with a personal API key, then writes normalized rows to the pipeline:
+
+- persons as `$identify` rows
+- groups as `$groupidentify` rows
+- historical events from HogQL with original `timestamp`, `created_at`, and PostHog event `uuid` when available
+
+The importer writes historical rows directly to the pipeline. It does not mutate Worker Durable Object state.
+
+Required inputs:
+
+```bash
+export POSTHOG_PROJECT_ID="<project_id>"
+export POSTHOG_PERSONAL_API_KEY="phx_..."
+export CLOUDFLARE_PIPELINE_ENDPOINT="https://<stream-id>.ingest.cloudflare.com"
+export CLOUDFLARE_PIPELINE_AUTH_TOKEN="<pipeline token>" # if your stream requires it
+```
+
+Optional inputs:
+
+```bash
+export POSTHOG_HOST="https://us.posthog.com" # or https://eu.posthog.com / self-hosted URL
+export POSTHOG_ENVIRONMENT_ID="<environment_id>" # recommended for current PostHog persons/groups APIs
+export HOGFLARE_API_KEY="phc_..."
+export POSTHOG_TEAM_ID="1"
+export POSTHOG_GROUP_TYPE_0="company"
+export IMPORT_FROM="2025-01-01"
+export IMPORT_TO="2025-02-01"
+export IMPORT_BATCH_SIZE="500"
+export IMPORT_PERSONS_OFFSET="0" # resume guardrails
+export IMPORT_EVENTS_OFFSET="0"
+export IMPORT_EVENTS_AFTER_TIMESTAMP="2024-09-21T03:24:11Z"
+export IMPORT_EVENTS_AFTER_UUID="0192129b-c354-77b4-b496-9be7ec571fb4"
+export IMPORT_EVENT_UUIDS_FILE="/tmp/missing-event-uuids.txt"
+export IMPORT_EVENT_WINDOW_DAYS="7"
+export IMPORT_EVENT_WINDOW_HOURS="6" # use days or hours, not both
+export IMPORT_MAX_PERSONS="1000" # optional guardrails for smoke tests
+export IMPORT_MAX_GROUPS="1000"
+export IMPORT_MAX_EVENTS="1000"
+export IMPORT_STATE_FILE=".hogflare-import-state.jsonl"
+export IMPORT_TARGET_ACCOUNT_ID="<cloudflare_account_id>"
+export IMPORT_TARGET_BUCKET="<r2_bucket>"
+export IMPORT_TARGET_TABLE="default.hogflare_events_v3"
+export WRANGLER_R2_SQL_AUTH_TOKEN="<r2 sql token>"
+export IMPORT_CLOUDFLARE_API_TOKEN="<token with Pipelines read>" # optional auto flush discovery
+export IMPORT_PIPELINE_FLUSH_SECS="300" # fallback if Pipelines read is unavailable
+```
+
+Production imports require R2 SQL target checks by default. The importer uses stable import
+keys, queries the target before each batch, and skips rows that are already present.
+Cloudflare Pipeline/R2 is append-only and does not enforce uniqueness by itself. Passing
+`--no-target-check` or `IMPORT_TARGET_CHECKS=false` opts out and should only be used for
+local tests.
+
+Retry behavior is intentionally conservative. Import sends are not blindly retried after a
+transport or response error because the pipeline may have accepted the batch even if the
+client did not receive the response. The importer aligns its wait window to the Cloudflare
+Pipeline sink rolling policy when `IMPORT_CLOUDFLARE_API_TOKEN` can read Pipelines. Without
+that API access, it uses `IMPORT_PIPELINE_FLUSH_SECS`, defaulting conservatively to 300
+seconds. The wait is `max(60s, 2 * flush + 30s)`, unless `IMPORT_TARGET_WAIT_SECS` is set
+explicitly.
+
+The local state file makes normal same-machine resumes cheap, but it is not a substitute for
+target checks if the state file is lost, multiple importers run concurrently, or a send has an
+unknown commit state.
+
+Run a dry run first:
+
+```bash
+cargo run --bin import_posthog -- --dry-run
+```
+
+Run the import:
+
+```bash
+cargo run --bin import_posthog
+```
+
+You can also pass flags instead of env vars:
+
+```bash
+cargo run --bin import_posthog -- \
+  --posthog-host https://us.posthog.com \
+  --project-id 12345 \
+  --environment-id 67890 \
+  --personal-api-key "$POSTHOG_PERSONAL_API_KEY" \
+  --pipeline-endpoint "$CLOUDFLARE_PIPELINE_ENDPOINT" \
+  --pipeline-auth-token "$CLOUDFLARE_PIPELINE_AUTH_TOKEN" \
+  --hogflare-api-key phc_example \
+  --from 2025-01-01 \
+  --to 2025-02-01 \
+  --persons-offset 0 \
+  --events-offset 0 \
+  --events-after-timestamp 2024-09-21T03:24:11Z \
+  --events-after-uuid 0192129b-c354-77b4-b496-9be7ec571fb4 \
+  --event-uuids-file /tmp/missing-event-uuids.txt \
+  --event-window-hours 6 \
+  --max-persons 1000 \
+  --max-groups 1000 \
+  --max-events 1000 \
+  --import-state-file .hogflare-import-state.jsonl \
+  --target-account-id "$CLOUDFLARE_ACCOUNT_ID" \
+  --target-bucket hogflare \
+  --target-table default.hogflare_events_v3 \
+  --target-auth-token "$WRANGLER_R2_SQL_AUTH_TOKEN" \
+  --cloudflare-api-token "$CLOUDFLARE_API_TOKEN"
+```
+
+Use `--skip-persons`, `--skip-groups`, or `--skip-events` to import only part of the project.
+Use `--skip-person-output` when resuming an event import after person rows were already written; it still loads people for event hydration.
+
 ## PostHog compatibility
 
 ### Ingestion endpoints

README.md

@@ -0,0 +1,33 @@
+use hogflare::importer::{run_import, ImportConfig, ImportError};
+
+#[tokio::main]
+async fn main() {
+    dotenvy::from_filename(".env.local")
+        .or_else(|_| dotenvy::dotenv())
+        .ok();
+
+    match run().await {
+        Ok(()) => {}
+        Err(ImportError::Usage(message)) => {
+            println!("{message}");
+        }
+        Err(err) => {
+            eprintln!("posthog import failed: {err}");
+            std::process::exit(1);
+        }
+    }
+}
+
+async fn run() -> Result<(), ImportError> {
+    let config = ImportConfig::from_env_and_args(std::env::args().skip(1))?;
+    let dry_run = config.dry_run;
+    let summary = run_import(config).await?;
+
+    let mode = if dry_run { "dry run" } else { "import" };
+    println!(
+        "PostHog {mode} complete: persons={}, groups={}, events={}, skipped={}, pipeline_batches={}",
+        summary.persons, summary.groups, summary.events, summary.skipped, summary.pipeline_batches
+    );
+
+    Ok(())
+}