diff --git a/Cargo.toml b/Cargo.toml index e1f7168..1ed65e3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -35,7 +35,7 @@ regex = "1" axum = { version = "0.7", default-features = false, features = ["json", "query", "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"] } +tokio = { version = "1.37", features = ["macros", "rt-multi-thread", "process", "net", "time", "io-util", "sync"] } tracing-subscriber = { version = "0.3", features = ["env-filter"] } [dev-dependencies] diff --git a/README.md b/README.md index 3b56612..e79bd6b 100644 --- a/README.md +++ b/README.md @@ -71,7 +71,7 @@ flowchart TB EventsPipeline --> EventsR2["R2 Data Catalog
events table"] PersonsPipeline --> PersonsR2["R2 Data Catalog
persons table"] - ReplayUI["Replay Explorer"] -->|"R2 SQL"| EventsR2 + WebUI["Hogflare UI"] -->|"R2 SQL"| EventsR2 Models["Semantic Models
models/*.yml"] --> EventsR2 Models --> PersonsR2 Consumers["DuckDB / R2 SQL / BI"] --> Models diff --git a/docs/product-analytics.md b/docs/product-analytics.md new file mode 100644 index 0000000..2b9034f --- /dev/null +++ b/docs/product-analytics.md @@ -0,0 +1,28 @@ +# Product Analytics + +Hogflare serves product analytics from `/analytics`. The dashboard uses the repo's Sidemantic `events`, `pageviews`, and `sessions` models through the native Sidemantic/DuckDB Iceberg bridge. + +## API Config + +Product analytics shares the same R2 Data Catalog warehouse credentials as replay, but uses analytics-specific names when provided: + +| Setting | Notes | +| --- | --- | +| `HOGFLARE_ANALYTICS_ACCOUNT_ID` | Cloudflare account id for the R2 Data Catalog warehouse. Falls back to `HOGFLARE_REPLAY_ACCOUNT_ID`. | +| `HOGFLARE_ANALYTICS_BUCKET` | R2 bucket name backing the warehouse. Falls back to `HOGFLARE_REPLAY_BUCKET`. | +| `HOGFLARE_ANALYTICS_R2_SQL_TOKEN` | R2 SQL/Data Catalog token. Store as a secret. Falls back to `HOGFLARE_REPLAY_R2_SQL_TOKEN`. | +| `HOGFLARE_ANALYTICS_EVENTS_TABLE` | Iceberg events table. Falls back to `HOGFLARE_REPLAY_EVENTS_TABLE`. | +| `HOGFLARE_ANALYTICS_PERSONS_TABLE` | Optional Iceberg persons table. Defaults are inferred from the events table. | +| `HOGFLARE_ANALYTICS_MODEL_DIR` | Optional Sidemantic model directory. Defaults to `models`. | +| `HOGFLARE_ANALYTICS_SIDEMANTIC_SCRIPT` | Optional override for the native analytics worker script. | +| `HOGFLARE_ANALYTICS_PREAGG` | Optional Sidemantic pre-aggregation switch. Defaults to enabled. Set to `0` to disable. | +| `HOGFLARE_ANALYTICS_PREAGG_SCHEMA` | Optional DuckDB schema for Sidemantic rollup tables. Defaults to `sidemantic_preagg`. | + +## Routes + +- `/` serves the Hogflare app and opens product analytics by default. +- `/analytics` serves the product analytics view. +- `/analytics/api/charts` returns overview metrics, a focused trend, and semantic breakdowns including domains, referrers, browser, country, region, and city leaderboards. +- `/replay` serves the replay feature. + +At worker startup, Sidemantic materializes known count-based chart shapes into daily pre-aggregation rollups and automatically routes eligible queries through those tables. `metric`, `dimension`, and `granularity` choose the focused chart. `semantic_filters` carries clickable cross-filter state as a JSON object of semantic dimensions to values. Analytics leaderboards use a fixed top-10 row cap plus an Others row. diff --git a/docs/session-replay.md b/docs/session-replay.md index 3259f1c..d617571 100644 --- a/docs/session-replay.md +++ b/docs/session-replay.md @@ -2,7 +2,7 @@ ![Hogflare replay explorer](assets/replay-explorer.jpg) -Hogflare stores replay uploads in the same events table as analytics events, then serves a read-only replay explorer from `/replay`. The UI is for product analytics workflows: browse recent sessions, search events, inspect funnel drop-offs, review computed friction signals, and follow a person journey. +Hogflare stores replay uploads in the same events table as product events, then serves a read-only replay explorer from `/replay`. The replay feature is for browsing recent sessions, searching events, inspecting funnel drop-offs, reviewing computed friction signals, and following a person journey. ## Ingestion @@ -27,14 +27,16 @@ Replay APIs require: The token stays server-side in the Worker. The browser only calls Hogflare's replay API. +Product analytics has a separate first-class route and API. See [product analytics](product-analytics.md). + ## Routes - `/replay` serves the explorer UI. - `/replay/api/sessions` lists replay sessions by reading `$snapshot_items` and legacy `$snapshot` rows from Iceberg through R2 SQL. -- `/replay/api/events` searches analytics events while excluding replay recording rows. +- `/replay/api/events` searches product events while excluding replay recording rows. - `/replay/api/funnels` classifies sessions as converted, stuck, or dropped for an ordered `steps` list of event names. - `/replay/api/friction` computes replay-derived signals such as rage clicks, dead clicks, form thrash, long idle gaps, repeated navigation, and deep scroll without follow-up. -- `/replay/api/person` joins a distinct ID's replay sessions and analytics events into one journey timeline. +- `/replay/api/person` joins a distinct ID's replay sessions and product events into one journey timeline. - `/replay/api/sessions/:session_id` returns normalized rrweb events plus an activity timeline for one session. ## Filters diff --git a/models/events.yml b/models/events.yml index a2b0587..090cbdc 100644 --- a/models/events.yml +++ b/models/events.yml @@ -54,7 +54,6 @@ models: group3, group4, group_properties, - api_key, extra, coalesce( json_extract_string(properties, '$.$session_id'), @@ -87,6 +86,14 @@ models: json_extract_string(properties, '$.$referrer'), json_extract_string(properties, '$.referrer') ) as referrer, + nullif(regexp_extract( + coalesce( + json_extract_string(properties, '$.$referrer'), + json_extract_string(properties, '$.referrer') + ), + '^(?:[a-zA-Z][a-zA-Z0-9+.-]*://)?([^/?#]+)', + 1 + ), '') as referrer_domain, coalesce( json_extract_string(properties, '$.$utm_source'), json_extract_string(properties, '$.utm_source'), @@ -188,11 +195,11 @@ models: select *, coalesce(session_id, actor_id || ':' || strftime(event_time, '%Y-%m-%d')) as session_key, - event_type not in ('$identify', '$groupidentify', '$create_alias', '$engage', '$snapshot') as is_capture_event, + event_type not in ('$identify', '$groupidentify', '$create_alias', '$engage', '$snapshot', '$snapshot_items') as is_capture_event, event_type in ('$pageview', 'page_view', '$screen', 'screen') as is_pageview_event, event_type in ('$identify', '$engage') as is_person_mutation_event, event_type = '$groupidentify' as is_group_event, - event_type = '$snapshot' as is_session_recording_event, + event_type in ('$snapshot', '$snapshot_items') as is_session_recording_event, resolved_person_id is not null as has_person, group0 is not null or group1 is not null or group2 is not null or group3 is not null or group4 is not null as has_group, session_id is not null as has_session_id @@ -240,9 +247,6 @@ models: - name: team_id type: numeric description: Optional PostHog team id assigned by the Worker. - - name: api_key - type: categorical - description: PostHog project API key. - name: distinct_id type: categorical description: Original PostHog distinct id. @@ -285,6 +289,9 @@ models: - name: referrer type: categorical description: Browser referrer. + - name: referrer_domain + type: categorical + description: Domain extracted from the browser referrer. - name: utm_source type: categorical description: UTM source. @@ -326,15 +333,19 @@ models: description: Country code from Cloudflare enrichment or event properties. - name: geo_region type: categorical + parent: geo_country_code description: Region/subdivision from Cloudflare enrichment or event properties. - name: geo_city type: categorical + parent: geo_region description: City from Cloudflare enrichment or event properties. - name: geo_timezone type: categorical + parent: geo_country_code description: Timezone from Cloudflare enrichment. - name: cf_colo type: categorical + parent: geo_country_code description: Cloudflare colo. - name: cf_asn type: numeric @@ -479,8 +490,8 @@ models: - name: snapshot_events agg: count filters: - - "event_type = '$snapshot'" - description: Session recording snapshot events. + - "event_type in ('$snapshot', '$snapshot_items')" + description: Session recording snapshot and normalized snapshot-item events. - name: grouped_events agg: count filters: @@ -500,10 +511,6 @@ models: agg: count_distinct sql: coalesce(group0, group1, group2, group3, group4) description: Distinct group keys across populated group slots. - - name: unique_api_keys - agg: count_distinct - sql: api_key - description: Distinct PostHog project API keys. - name: power_users type: cohort entity: actor_id @@ -524,6 +531,64 @@ models: agg: count description: Count of actors with at least two sessions in the query scope. + pre_aggregations: + - name: analytics_hourly + measures: + - event_count + - capture_events + - pageviews + - snapshot_events + dimensions: + - event_type + - pathname + - host + - referrer_domain + - referrer + - browser + - os + - device_type + - geo_country_code + - geo_region + - geo_city + - geo_timezone + - cf_asn + - utm_source + - utm_campaign + time_dimension: event_time + granularity: hour + partition_granularity: day + refresh_key: + every: 1 hour + incremental: false + - name: analytics_daily + measures: + - event_count + - capture_events + - pageviews + - snapshot_events + dimensions: + - event_type + - pathname + - host + - referrer_domain + - referrer + - browser + - os + - device_type + - geo_country_code + - geo_region + - geo_city + - geo_timezone + - cf_asn + - utm_source + - utm_campaign + time_dimension: event_time + granularity: day + partition_granularity: month + refresh_key: + every: 1 hour + incremental: false + segments: - name: capture_events sql: is_capture_event @@ -546,4 +611,3 @@ models: - name: session_recordings sql: is_session_recording_event description: Session recording snapshot events. - diff --git a/models/pageviews.yml b/models/pageviews.yml index aa12af8..6355e39 100644 --- a/models/pageviews.yml +++ b/models/pageviews.yml @@ -48,20 +48,29 @@ models: group2, group3, group4, - api_key, coalesce(json_extract_string(properties, '$.$session_id'), json_extract_string(properties, '$.session_id')) as session_id, coalesce(json_extract_string(properties, '$.$current_url'), json_extract_string(properties, '$.current_url'), json_extract_string(properties, '$.url')) as current_url, coalesce(json_extract_string(properties, '$.$pathname'), json_extract_string(properties, '$.pathname'), json_extract_string(properties, '$.path')) as pathname, coalesce(json_extract_string(properties, '$.$host'), json_extract_string(properties, '$.host')) as host, coalesce(json_extract_string(properties, '$.$title'), json_extract_string(properties, '$.title')) as page_title, coalesce(json_extract_string(properties, '$.$referrer'), json_extract_string(properties, '$.referrer')) as referrer, + nullif(regexp_extract( + coalesce(json_extract_string(properties, '$.$referrer'), json_extract_string(properties, '$.referrer')), + '^(?:[a-zA-Z][a-zA-Z0-9+.-]*://)?([^/?#]+)', + 1 + ), '') as referrer_domain, coalesce(json_extract_string(properties, '$.$utm_source'), json_extract_string(properties, '$.utm_source'), json_extract_string(properties, '$.$initial_utm_source')) as utm_source, coalesce(json_extract_string(properties, '$.$utm_medium'), json_extract_string(properties, '$.utm_medium'), json_extract_string(properties, '$.$initial_utm_medium')) as utm_medium, coalesce(json_extract_string(properties, '$.$utm_campaign'), json_extract_string(properties, '$.utm_campaign'), json_extract_string(properties, '$.$initial_utm_campaign')) as utm_campaign, coalesce(json_extract_string(properties, '$.$browser'), json_extract_string(properties, '$.browser')) as browser, coalesce(json_extract_string(properties, '$.$os'), json_extract_string(properties, '$.os')) as os, coalesce(json_extract_string(properties, '$.$device_type'), json_extract_string(properties, '$.device_type')) as device_type, - coalesce(json_extract_string(properties, '$.$geoip_country_code'), json_extract_string(properties, '$.country')) as geo_country_code + coalesce(json_extract_string(properties, '$.$geoip_country_code'), json_extract_string(properties, '$.country')) as geo_country_code, + coalesce(json_extract_string(properties, '$.$geoip_subdivision_1_code'), json_extract_string(properties, '$.region')) as geo_region, + coalesce(json_extract_string(properties, '$.$geoip_city_name'), json_extract_string(properties, '$.city')) as geo_city, + json_extract_string(properties, '$.$geoip_time_zone') as geo_timezone, + json_extract_string(properties, '$.cf_colo') as cf_colo, + try_cast(json_extract_string(properties, '$.cf_asn') as bigint) as cf_asn from {{ events_table }} left join identity_map on distinct_id = identity_map.linked_distinct_id ) @@ -105,6 +114,8 @@ models: type: categorical - name: referrer type: categorical + - name: referrer_domain + type: categorical - name: utm_source type: categorical - name: utm_medium @@ -119,6 +130,20 @@ models: type: categorical - name: geo_country_code type: categorical + - name: geo_region + type: categorical + parent: geo_country_code + - name: geo_city + type: categorical + parent: geo_region + - name: geo_timezone + type: categorical + parent: geo_country_code + - name: cf_colo + type: categorical + parent: geo_country_code + - name: cf_asn + type: numeric - name: event_time type: time granularity: day @@ -140,3 +165,54 @@ models: sql: resolved_person_id description: Distinct resolved persons with pageviews. + pre_aggregations: + - name: analytics_hourly + measures: + - pageviews + dimensions: + - event_type + - pathname + - host + - referrer_domain + - referrer + - browser + - os + - device_type + - geo_country_code + - geo_region + - geo_city + - geo_timezone + - cf_asn + - utm_source + - utm_campaign + time_dimension: event_time + granularity: hour + partition_granularity: day + refresh_key: + every: 1 hour + incremental: false + - name: analytics_daily + measures: + - pageviews + dimensions: + - event_type + - pathname + - host + - referrer_domain + - referrer + - browser + - os + - device_type + - geo_country_code + - geo_region + - geo_city + - geo_timezone + - cf_asn + - utm_source + - utm_campaign + time_dimension: event_time + granularity: day + partition_granularity: month + refresh_key: + every: 1 hour + incremental: false diff --git a/models/sessions.yml b/models/sessions.yml index 5a03937..1ed7eb5 100644 --- a/models/sessions.yml +++ b/models/sessions.yml @@ -39,18 +39,27 @@ models: identity_map.canonical_distinct_id, coalesce(timestamp, created_at) as event_time, properties, - api_key, coalesce(json_extract_string(properties, '$.$session_id'), json_extract_string(properties, '$.session_id')) as session_id, coalesce(json_extract_string(properties, '$.$pathname'), json_extract_string(properties, '$.pathname'), json_extract_string(properties, '$.path')) as pathname, coalesce(json_extract_string(properties, '$.$current_url'), json_extract_string(properties, '$.current_url'), json_extract_string(properties, '$.url')) as current_url, coalesce(json_extract_string(properties, '$.$referrer'), json_extract_string(properties, '$.referrer')) as referrer, + nullif(regexp_extract( + coalesce(json_extract_string(properties, '$.$referrer'), json_extract_string(properties, '$.referrer')), + '^(?:[a-zA-Z][a-zA-Z0-9+.-]*://)?([^/?#]+)', + 1 + ), '') as referrer_domain, coalesce(json_extract_string(properties, '$.$utm_source'), json_extract_string(properties, '$.utm_source'), json_extract_string(properties, '$.$initial_utm_source')) as utm_source, coalesce(json_extract_string(properties, '$.$utm_medium'), json_extract_string(properties, '$.utm_medium'), json_extract_string(properties, '$.$initial_utm_medium')) as utm_medium, coalesce(json_extract_string(properties, '$.$utm_campaign'), json_extract_string(properties, '$.utm_campaign'), json_extract_string(properties, '$.$initial_utm_campaign')) as utm_campaign, coalesce(json_extract_string(properties, '$.$browser'), json_extract_string(properties, '$.browser')) as browser, coalesce(json_extract_string(properties, '$.$os'), json_extract_string(properties, '$.os')) as os, coalesce(json_extract_string(properties, '$.$device_type'), json_extract_string(properties, '$.device_type')) as device_type, - coalesce(json_extract_string(properties, '$.$geoip_country_code'), json_extract_string(properties, '$.country')) as geo_country_code + coalesce(json_extract_string(properties, '$.$geoip_country_code'), json_extract_string(properties, '$.country')) as geo_country_code, + coalesce(json_extract_string(properties, '$.$geoip_subdivision_1_code'), json_extract_string(properties, '$.region')) as geo_region, + coalesce(json_extract_string(properties, '$.$geoip_city_name'), json_extract_string(properties, '$.city')) as geo_city, + json_extract_string(properties, '$.$geoip_time_zone') as geo_timezone, + json_extract_string(properties, '$.cf_colo') as cf_colo, + try_cast(json_extract_string(properties, '$.cf_asn') as bigint) as cf_asn from {{ events_table }} left join identity_map on distinct_id = identity_map.linked_distinct_id where coalesce(timestamp, created_at) is not null @@ -68,7 +77,6 @@ models: first(resolved_person_id order by event_time asc) as person_id, first(canonical_distinct_id order by event_time asc) as canonical_distinct_id, true as has_session_id, - first(api_key order by event_time asc) as api_key, min(event_time) as session_start_at, max(event_time) as session_end_at, date_diff('second', min(event_time), max(event_time)) as duration_seconds, @@ -80,13 +88,19 @@ models: first(current_url order by event_time asc) as landing_url, last(current_url order by event_time asc) as exit_url, first(referrer order by event_time asc) as referrer, + first(referrer_domain order by event_time asc) as referrer_domain, first(utm_source order by event_time asc) as utm_source, first(utm_medium order by event_time asc) as utm_medium, first(utm_campaign order by event_time asc) as utm_campaign, first(browser order by event_time asc) as browser, first(os order by event_time asc) as os, first(device_type order by event_time asc) as device_type, - first(geo_country_code order by event_time asc) as geo_country_code + first(geo_country_code order by event_time asc) as geo_country_code, + first(geo_region order by event_time asc) as geo_region, + first(geo_city order by event_time asc) as geo_city, + first(geo_timezone order by event_time asc) as geo_timezone, + first(cf_colo order by event_time asc) as cf_colo, + first(cf_asn order by event_time asc) as cf_asn from sessionized group by session_id description: Real PostHog SDK session rollup keyed only by captured $session_id. @@ -113,8 +127,6 @@ models: type: categorical - name: has_session_id type: boolean - - name: api_key - type: categorical - name: landing_path type: categorical - name: exit_path @@ -125,6 +137,8 @@ models: type: categorical - name: referrer type: categorical + - name: referrer_domain + type: categorical - name: utm_source type: categorical - name: utm_medium @@ -139,6 +153,20 @@ models: type: categorical - name: geo_country_code type: categorical + - name: geo_region + type: categorical + parent: geo_country_code + - name: geo_city + type: categorical + parent: geo_region + - name: geo_timezone + type: categorical + parent: geo_country_code + - name: cf_colo + type: categorical + parent: geo_country_code + - name: cf_asn + type: numeric - name: duration_seconds type: numeric - name: event_count @@ -190,3 +218,48 @@ models: sql: pageview_count description: Average pageviews per session. + pre_aggregations: + - name: analytics_hourly + measures: + - session_count + dimensions: + - landing_path + - browser + - os + - device_type + - referrer_domain + - referrer + - geo_country_code + - geo_region + - geo_city + - geo_timezone + - cf_asn + - utm_source + time_dimension: session_start_at + granularity: hour + partition_granularity: day + refresh_key: + every: 1 hour + incremental: false + - name: analytics_daily + measures: + - session_count + dimensions: + - landing_path + - browser + - os + - device_type + - referrer_domain + - referrer + - geo_country_code + - geo_region + - geo_city + - geo_timezone + - cf_asn + - utm_source + time_dimension: session_start_at + granularity: day + partition_granularity: month + refresh_key: + every: 1 hour + incremental: false diff --git a/scripts/product_analytics_sidemantic.py b/scripts/product_analytics_sidemantic.py new file mode 100644 index 0000000..a57cf00 --- /dev/null +++ b/scripts/product_analytics_sidemantic.py @@ -0,0 +1,1362 @@ +# /// script +# requires-python = ">=3.11" +# dependencies = [ +# "duckdb>=1.1", +# "sidemantic @ git+https://github.com/sidequery/sidemantic", +# ] +# /// + +from __future__ import annotations + +import json +import os +import re +import sys +import time +from datetime import date, datetime +from pathlib import Path +from typing import Any + +from sidemantic import SemanticLayer, load_from_directory +from sidemantic.sql.generator import SQLGenerator + + +SNAPSHOT_EVENT = "$snapshot" +SNAPSHOT_ITEMS_EVENT = "$snapshot_items" +IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9_]+(?:\.[A-Za-z0-9_]+)*$") +NUMERIC_LITERAL_RE = re.compile(r"^-?\d+(?:\.\d+)?$") +GRANULARITIES = {"hour", "day", "week", "month"} +ANALYTICS_BREAKDOWN_LIMIT = 10 +LEADERBOARD_PANELS = [ + "top_events", + "top_pages", + "domains", + "referring_domains", + "referrers", + "browsers", + "operating_systems", + "devices", + "countries", + "regions", + "cities", + "timezones", + "asns", + "utm_sources", + "utm_campaigns", +] +CHART_PANELS = {"summary", "series", "focus_breakdown", *LEADERBOARD_PANELS} +TIME_DIMENSIONS = { + "events": "event_time", + "pageviews": "event_time", + "sessions": "session_start_at", +} +DEFAULT_DIMENSIONS = { + "events": "events.event_type", + "pageviews": "pageviews.pathname", + "sessions": "sessions.landing_path", +} +BUILTIN_BREAKDOWN_DIMENSIONS = { + "events.event_type", + "events.pathname", + "pageviews.pathname", + "pageviews.event_type", + "pageviews.host", + "pageviews.referrer_domain", + "pageviews.referrer", + "events.browser", + "events.geo_country_code", + "events.geo_region", + "events.geo_city", +} +PANEL_DIMENSIONS = { + "top_events": { + "events": "events.event_type", + "pageviews": "pageviews.event_type", + }, + "top_pages": { + "events": "events.pathname", + "pageviews": "pageviews.pathname", + "sessions": "sessions.landing_path", + }, + "domains": { + "events": "events.host", + "pageviews": "pageviews.host", + }, + "referring_domains": { + "events": "events.referrer_domain", + "pageviews": "pageviews.referrer_domain", + "sessions": "sessions.referrer_domain", + }, + "referrers": { + "events": "events.referrer", + "pageviews": "pageviews.referrer", + "sessions": "sessions.referrer", + }, + "browsers": { + "events": "events.browser", + "pageviews": "pageviews.browser", + "sessions": "sessions.browser", + }, + "operating_systems": { + "events": "events.os", + "pageviews": "pageviews.os", + "sessions": "sessions.os", + }, + "devices": { + "events": "events.device_type", + "pageviews": "pageviews.device_type", + "sessions": "sessions.device_type", + }, + "countries": { + "events": "events.geo_country_code", + "pageviews": "pageviews.geo_country_code", + "sessions": "sessions.geo_country_code", + }, + "regions": { + "events": "events.geo_region", + "pageviews": "pageviews.geo_region", + "sessions": "sessions.geo_region", + }, + "cities": { + "events": "events.geo_city", + "pageviews": "pageviews.geo_city", + "sessions": "sessions.geo_city", + }, + "timezones": { + "events": "events.geo_timezone", + "pageviews": "pageviews.geo_timezone", + "sessions": "sessions.geo_timezone", + }, + "asns": { + "events": "events.cf_asn", + "pageviews": "pageviews.cf_asn", + "sessions": "sessions.cf_asn", + }, + "utm_sources": { + "events": "events.utm_source", + "pageviews": "pageviews.utm_source", + "sessions": "sessions.utm_source", + }, + "utm_campaigns": { + "events": "events.utm_campaign", + "pageviews": "pageviews.utm_campaign", + }, +} +METRICS = { + "events.capture_events": { + "model": "events", + "metric": "capture_events", + "label": "Events", + "display": "count", + "context_key": "event_count", + }, + "pageviews.pageviews": { + "model": "pageviews", + "metric": "pageviews", + "label": "Pageviews", + "display": "count", + "context_key": "pageviews", + }, + "events.unique_users": { + "model": "events", + "metric": "unique_users", + "label": "Users", + "display": "count", + }, + "sessions.session_count": { + "model": "sessions", + "metric": "session_count", + "label": "SDK sessions", + "display": "count", + "context_key": "session_count", + }, + "sessions.average_session_seconds": { + "model": "sessions", + "metric": "average_session_seconds", + "label": "Avg session", + "display": "seconds", + }, + "events.snapshot_events": { + "model": "events", + "metric": "snapshot_events", + "label": "Replay chunks", + "display": "count", + "context_key": "recordings", + }, +} +QUERY_METRICS = { + "events.capture_events", + "events.unique_users", + "pageviews.pageviews", + "sessions.session_count", + "sessions.average_session_seconds", +} +METRIC_ALIASES = { + "events.event_count": "events.capture_events", +} +DIMENSIONS = { + "events.event_type": { + "model": "events", + "dimension": "event_type", + "label": "Event", + "title": "Events", + }, + "events.pathname": { + "model": "events", + "dimension": "pathname", + "label": "Page", + "title": "Pages", + }, + "events.browser": { + "model": "events", + "dimension": "browser", + "label": "Browser", + "title": "Browsers", + }, + "events.os": { + "model": "events", + "dimension": "os", + "label": "OS", + "title": "Operating systems", + }, + "events.device_type": { + "model": "events", + "dimension": "device_type", + "label": "Device type", + "title": "Devices", + }, + "events.host": { + "model": "events", + "dimension": "host", + "label": "Host", + "title": "Domains", + }, + "events.referrer_domain": { + "model": "events", + "dimension": "referrer_domain", + "label": "Referring domain", + "title": "Referring domains", + }, + "events.referrer": { + "model": "events", + "dimension": "referrer", + "label": "Referrer", + "title": "Referrers", + }, + "events.geo_country_code": { + "model": "events", + "dimension": "geo_country_code", + "label": "Country", + "title": "Countries", + }, + "events.geo_region": { + "model": "events", + "dimension": "geo_region", + "label": "Region", + "title": "Regions", + }, + "events.geo_city": { + "model": "events", + "dimension": "geo_city", + "label": "City", + "title": "Cities", + }, + "events.geo_timezone": { + "model": "events", + "dimension": "geo_timezone", + "label": "Timezone", + "title": "Timezones", + }, + "events.cf_colo": { + "model": "events", + "dimension": "cf_colo", + "label": "Cloudflare colo", + "title": "Cloudflare colos", + }, + "events.cf_asn": { + "model": "events", + "dimension": "cf_asn", + "label": "ASN", + "title": "ASNs", + "type": "numeric", + }, + "events.utm_source": { + "model": "events", + "dimension": "utm_source", + "label": "UTM source", + "title": "UTM sources", + }, + "events.utm_campaign": { + "model": "events", + "dimension": "utm_campaign", + "label": "UTM campaign", + "title": "UTM campaigns", + }, + "pageviews.pathname": { + "model": "pageviews", + "dimension": "pathname", + "label": "Page", + "title": "Pages", + }, + "pageviews.event_type": { + "model": "pageviews", + "dimension": "event_type", + "label": "Event", + "title": "Events", + }, + "pageviews.host": { + "model": "pageviews", + "dimension": "host", + "label": "Host", + "title": "Domains", + }, + "pageviews.referrer_domain": { + "model": "pageviews", + "dimension": "referrer_domain", + "label": "Referring domain", + "title": "Referring domains", + }, + "pageviews.referrer": { + "model": "pageviews", + "dimension": "referrer", + "label": "Referrer", + "title": "Referrers", + }, + "pageviews.browser": { + "model": "pageviews", + "dimension": "browser", + "label": "Browser", + "title": "Browsers", + }, + "pageviews.os": { + "model": "pageviews", + "dimension": "os", + "label": "OS", + "title": "Operating systems", + }, + "pageviews.device_type": { + "model": "pageviews", + "dimension": "device_type", + "label": "Device type", + "title": "Devices", + }, + "pageviews.geo_country_code": { + "model": "pageviews", + "dimension": "geo_country_code", + "label": "Country", + "title": "Countries", + }, + "pageviews.geo_region": { + "model": "pageviews", + "dimension": "geo_region", + "label": "Region", + "title": "Regions", + }, + "pageviews.geo_city": { + "model": "pageviews", + "dimension": "geo_city", + "label": "City", + "title": "Cities", + }, + "pageviews.geo_timezone": { + "model": "pageviews", + "dimension": "geo_timezone", + "label": "Timezone", + "title": "Timezones", + }, + "pageviews.cf_colo": { + "model": "pageviews", + "dimension": "cf_colo", + "label": "Cloudflare colo", + "title": "Cloudflare colos", + }, + "pageviews.cf_asn": { + "model": "pageviews", + "dimension": "cf_asn", + "label": "ASN", + "title": "ASNs", + "type": "numeric", + }, + "pageviews.utm_source": { + "model": "pageviews", + "dimension": "utm_source", + "label": "UTM source", + "title": "UTM sources", + }, + "pageviews.utm_campaign": { + "model": "pageviews", + "dimension": "utm_campaign", + "label": "UTM campaign", + "title": "UTM campaigns", + }, + "sessions.landing_path": { + "model": "sessions", + "dimension": "landing_path", + "label": "Landing path", + "title": "Landing paths", + }, + "sessions.browser": { + "model": "sessions", + "dimension": "browser", + "label": "Browser", + "title": "Browsers", + }, + "sessions.os": { + "model": "sessions", + "dimension": "os", + "label": "OS", + "title": "Operating systems", + }, + "sessions.device_type": { + "model": "sessions", + "dimension": "device_type", + "label": "Device type", + "title": "Devices", + }, + "sessions.referrer_domain": { + "model": "sessions", + "dimension": "referrer_domain", + "label": "Referring domain", + "title": "Referring domains", + }, + "sessions.referrer": { + "model": "sessions", + "dimension": "referrer", + "label": "Referrer", + "title": "Referrers", + }, + "sessions.geo_country_code": { + "model": "sessions", + "dimension": "geo_country_code", + "label": "Country", + "title": "Countries", + }, + "sessions.geo_region": { + "model": "sessions", + "dimension": "geo_region", + "label": "Region", + "title": "Regions", + }, + "sessions.geo_city": { + "model": "sessions", + "dimension": "geo_city", + "label": "City", + "title": "Cities", + }, + "sessions.geo_timezone": { + "model": "sessions", + "dimension": "geo_timezone", + "label": "Timezone", + "title": "Timezones", + }, + "sessions.cf_colo": { + "model": "sessions", + "dimension": "cf_colo", + "label": "Cloudflare colo", + "title": "Cloudflare colos", + }, + "sessions.cf_asn": { + "model": "sessions", + "dimension": "cf_asn", + "label": "ASN", + "title": "ASNs", + "type": "numeric", + }, + "sessions.utm_source": { + "model": "sessions", + "dimension": "utm_source", + "label": "UTM source", + "title": "UTM sources", + }, +} + + +def main() -> None: + if len(sys.argv) > 1 and sys.argv[1] == "--serve": + serve() + return + + query = json.loads(sys.argv[1]) if len(sys.argv) > 1 else json.load(sys.stdin) + config = AnalyticsConfig.from_env() + runner = SidemanticAnalytics(config) + print(json.dumps(runner.run(query), separators=(",", ":"))) + + +def serve() -> None: + config = AnalyticsConfig.from_env() + runner = SidemanticAnalytics(config) + for line in sys.stdin: + line = line.strip() + if not line: + continue + request_id = None + try: + query = json.loads(line) + request_id = query.get("_request_id") + response = {"ok": True, "request_id": request_id, "result": runner.run(query)} + except Exception as exc: + response = {"ok": False, "request_id": request_id, "error": str(exc)} + print(json.dumps(response, separators=(",", ":")), flush=True) + + +class AnalyticsConfig: + def __init__( + self, + *, + account_id: str, + bucket: str, + token: str, + events_table: str, + persons_table: str, + model_dir: Path, + preagg_enabled: bool, + preagg_refresh: bool, + preagg_schema: str, + ) -> None: + self.account_id = account_id + self.bucket = bucket + self.token = token + self.events_table = events_table + self.persons_table = persons_table + self.model_dir = model_dir + self.preagg_enabled = preagg_enabled + self.preagg_refresh = preagg_refresh + self.preagg_schema = preagg_schema + + @classmethod + def from_env(cls) -> "AnalyticsConfig": + events_table = env_required_any("HOGFLARE_ANALYTICS_EVENTS_TABLE", "HOGFLARE_REPLAY_EVENTS_TABLE") + persons_table = ( + os.environ.get("HOGFLARE_ANALYTICS_PERSONS_TABLE") + or infer_persons_table(events_table) + ) + for name, value in { + "HOGFLARE_ANALYTICS_EVENTS_TABLE": events_table, + "HOGFLARE_ANALYTICS_PERSONS_TABLE": persons_table, + }.items(): + if not IDENTIFIER_RE.match(value): + raise ValueError(f"{name} must be a dotted SQL identifier") + preagg_schema = ( + os.environ.get("HOGFLARE_ANALYTICS_PREAGG_SCHEMA") + or "sidemantic_preagg" + ) + if not IDENTIFIER_RE.match(preagg_schema): + raise ValueError("HOGFLARE_ANALYTICS_PREAGG_SCHEMA must be a SQL identifier") + + return cls( + account_id=env_required_any("HOGFLARE_ANALYTICS_ACCOUNT_ID", "HOGFLARE_REPLAY_ACCOUNT_ID"), + bucket=env_required_any("HOGFLARE_ANALYTICS_BUCKET", "HOGFLARE_REPLAY_BUCKET"), + token=env_required_any( + "HOGFLARE_ANALYTICS_R2_SQL_TOKEN", + "HOGFLARE_REPLAY_R2_SQL_TOKEN", + ), + events_table=events_table, + persons_table=persons_table, + model_dir=Path( + os.environ.get("HOGFLARE_ANALYTICS_MODEL_DIR") + or "models" + ), + preagg_enabled=env_flag("HOGFLARE_ANALYTICS_PREAGG", default=True), + preagg_refresh=env_flag("HOGFLARE_ANALYTICS_PREAGG_REFRESH", default=False), + preagg_schema=preagg_schema, + ) + + @property + def warehouse_name(self) -> str: + return f"{self.account_id}_{self.bucket}" + + @property + def catalog_endpoint(self) -> str: + return f"https://catalog.cloudflarestorage.com/{self.account_id}/{self.bucket}" + + @property + def attached_events_table(self) -> str: + return f"iceberg_catalog.{self.events_table}" + + @property + def attached_persons_table(self) -> str: + return f"iceberg_catalog.{self.persons_table}" + + +class SidemanticAnalytics: + def __init__(self, config: AnalyticsConfig) -> None: + self.config = config + self.layer = SemanticLayer( + connection="duckdb:///:memory:", + auto_register=False, + use_preaggregations=config.preagg_enabled, + preagg_schema=config.preagg_schema, + ) + self._connect_iceberg() + load_from_directory(self.layer, str(config.model_dir)) + self._bind_model_tables() + self._materialize_preaggregations() + self.generator = SQLGenerator( + self.layer.graph, + dialect="duckdb", + preagg_schema=config.preagg_schema, + ) + + def run(self, query: dict[str, Any]) -> dict[str, Any]: + top_limit = ANALYTICS_BREAKDOWN_LIMIT + metric = metric_def(query.get("metric")) + dimension = dimension_def(query.get("dimension"), metric["model"]) + granularity = granularity_for(query.get("granularity")) + events_filters = self._filters_for("events", query) + pageviews_filters = self._filters_for("pageviews", query) + sessions_filters = self._filters_for("sessions", query) + focus_filters = self._filters_for(metric["model"], query) + panel = clean(query.get("panel")) + if panel in CHART_PANELS: + return self._run_panel( + panel=panel, + query=query, + top_limit=top_limit, + metric=metric, + dimension=dimension, + granularity=granularity, + events_filters=events_filters, + pageviews_filters=pageviews_filters, + sessions_filters=sessions_filters, + focus_filters=focus_filters, + ) + + events_summary = self._one( + metrics=["events.capture_events", "events.unique_users"], + filters=events_filters, + skip_default_time_dimensions=True, + ) + pageviews_summary = self._one( + metrics=["pageviews.pageviews"], + filters=pageviews_filters, + skip_default_time_dimensions=True, + ) + sessions_summary = self._one( + metrics=["sessions.session_count", "sessions.average_session_seconds"], + filters=sessions_filters, + skip_default_time_dimensions=True, + ) + focus_time_dimension = TIME_DIMENSIONS[metric["model"]] + focus_bucket_ref = f"{metric['model']}.{focus_time_dimension}__{granularity}" + context_series_rows = self._rows( + metrics=[ + "events.capture_events", + "events.pageviews", + "events.unique_sessions", + "events.snapshot_events", + ], + dimensions=[f"events.event_time__{granularity}"], + filters=events_filters, + order_by=[f"events.event_time__{granularity}"], + ) + focus_series_rows = self._rows( + metrics=[metric["ref"]], + dimensions=[focus_bucket_ref], + filters=focus_filters, + order_by=[focus_bucket_ref], + ) + focus_breakdown_filters = [ + *self._filters_for(metric["model"], query, exclude_dimension_ref=dimension["ref"]), + f"{dimension['ref']} is not null", + ] + focus_breakdown_rows = self._rows( + metrics=[metric["ref"]], + dimensions=[dimension["ref"]], + filters=focus_breakdown_filters, + order_by=[f"{metric['ref']} DESC"], + limit=top_limit, + skip_default_time_dimensions=True, + ) + focus_breakdown_total = self._metric_total( + metric["ref"], + metric["metric"], + focus_breakdown_filters, + ) + + event_count = to_int(events_summary.get("capture_events")) + pageviews = to_int(pageviews_summary.get("pageviews")) + unique_users = to_int(events_summary.get("unique_users")) + session_count = to_int(sessions_summary.get("session_count")) + average_session_seconds = to_float(sessions_summary.get("average_session_seconds")) + + breakdowns = [ + breakdown( + dimension["title"], + dimension["model"], + dimension["dimension"], + metric["ref"], + focus_breakdown_rows, + dimension["dimension"], + metric["metric"], + focus_breakdown_total, + ) + ] + for panel in panel_keys_for_metric(metric): + candidate = self._leaderboard_panel(panel, query, top_limit, metric) + if not candidate: + continue + if not any(same_breakdown(candidate, existing) for existing in breakdowns): + breakdowns.append(candidate) + + return { + "focus": { + "metric": metric["ref"], + "metric_label": metric["label"], + "dimension": dimension["ref"], + "dimension_label": dimension["label"], + "granularity": granularity, + }, + "summary": [ + count_metric("Events", event_count, "events", "capture_events"), + count_metric("Pageviews", pageviews, "pageviews", "pageviews"), + count_metric("Users", unique_users, "events", "unique_users"), + count_metric("SDK sessions", session_count, "sessions", "session_count"), + seconds_metric( + "Avg session", + average_session_seconds, + "sessions", + "average_session_seconds", + ), + ], + "series": series_points( + context_series_rows, + focus_series_rows, + focus_time_dimension, + granularity, + metric["metric"], + ), + "breakdowns": breakdowns, + } + + def _run_panel( + self, + *, + panel: str, + query: dict[str, Any], + top_limit: int, + metric: dict[str, Any], + dimension: dict[str, Any], + granularity: str, + events_filters: list[str], + pageviews_filters: list[str], + sessions_filters: list[str], + focus_filters: list[str], + ) -> dict[str, Any]: + response = self._empty_response(metric, dimension, granularity) + + if panel == "summary": + events_summary = self._one( + metrics=["events.capture_events", "events.unique_users"], + filters=events_filters, + skip_default_time_dimensions=True, + ) + pageviews_summary = self._one( + metrics=["pageviews.pageviews"], + filters=pageviews_filters, + skip_default_time_dimensions=True, + ) + sessions_summary = self._one( + metrics=["sessions.session_count", "sessions.average_session_seconds"], + filters=sessions_filters, + skip_default_time_dimensions=True, + ) + response["summary"] = [ + count_metric("Events", to_int(events_summary.get("capture_events")), "events", "capture_events"), + count_metric("Pageviews", to_int(pageviews_summary.get("pageviews")), "pageviews", "pageviews"), + count_metric("Users", to_int(events_summary.get("unique_users")), "events", "unique_users"), + count_metric("SDK sessions", to_int(sessions_summary.get("session_count")), "sessions", "session_count"), + seconds_metric( + "Avg session", + to_float(sessions_summary.get("average_session_seconds")), + "sessions", + "average_session_seconds", + ), + ] + return response + + if panel == "series": + focus_time_dimension = TIME_DIMENSIONS[metric["model"]] + focus_bucket_ref = f"{metric['model']}.{focus_time_dimension}__{granularity}" + context_series_rows = self._rows( + metrics=[ + "events.capture_events", + "events.pageviews", + "events.unique_sessions", + "events.snapshot_events", + ], + dimensions=[f"events.event_time__{granularity}"], + filters=events_filters, + order_by=[f"events.event_time__{granularity}"], + ) + focus_series_rows = self._rows( + metrics=[metric["ref"]], + dimensions=[focus_bucket_ref], + filters=focus_filters, + order_by=[focus_bucket_ref], + ) + response["series"] = series_points( + context_series_rows, + focus_series_rows, + focus_time_dimension, + granularity, + metric["metric"], + ) + return response + + if panel == "focus_breakdown": + focus_breakdown_filters = [ + *self._filters_for(metric["model"], query, exclude_dimension_ref=dimension["ref"]), + f"{dimension['ref']} is not null", + ] + focus_breakdown_rows = self._rows( + metrics=[metric["ref"]], + dimensions=[dimension["ref"]], + filters=focus_breakdown_filters, + order_by=[f"{metric['ref']} DESC"], + limit=top_limit, + skip_default_time_dimensions=True, + ) + focus_breakdown_total = self._metric_total( + metric["ref"], + metric["metric"], + focus_breakdown_filters, + ) + response["breakdowns"] = [ + breakdown( + dimension["title"], + dimension["model"], + dimension["dimension"], + metric["ref"], + focus_breakdown_rows, + dimension["dimension"], + metric["metric"], + focus_breakdown_total, + ) + ] + return response + + leaderboard = self._leaderboard_panel(panel, query, top_limit, metric) + if leaderboard: + response["breakdowns"] = [leaderboard] + return response + + def _leaderboard_panel( + self, + panel: str, + query: dict[str, Any], + top_limit: int, + metric: dict[str, Any], + ) -> dict[str, Any] | None: + dimension_ref = PANEL_DIMENSIONS.get(panel, {}).get(metric["model"]) + if not dimension_ref: + return None + dimension_defn = DIMENSIONS[dimension_ref] + filters = [ + *self._filters_for(metric["model"], query, exclude_dimension_ref=dimension_ref), + f"{dimension_ref} is not null", + ] + rows = self._rows( + metrics=[metric["ref"]], + dimensions=[dimension_ref], + filters=filters, + order_by=[f"{metric['ref']} DESC"], + limit=top_limit, + skip_default_time_dimensions=True, + ) + total = self._metric_total(metric["ref"], metric["metric"], filters) + return breakdown( + dimension_defn["title"], + dimension_defn["model"], + dimension_defn["dimension"], + metric["ref"], + rows, + dimension_defn["dimension"], + metric["metric"], + total, + ) + + @staticmethod + def _empty_response(metric: dict[str, Any], dimension: dict[str, Any], granularity: str) -> dict[str, Any]: + return { + "focus": { + "metric": metric["ref"], + "metric_label": metric["label"], + "dimension": dimension["ref"], + "dimension_label": dimension["label"], + "granularity": granularity, + }, + "summary": [], + "series": [], + "breakdowns": [], + } + + def _connect_iceberg(self) -> None: + con = self.layer.adapter.raw_connection + con.execute("INSTALL httpfs") + con.execute("INSTALL iceberg") + con.execute("LOAD httpfs") + con.execute("LOAD iceberg") + con.execute("CREATE OR REPLACE SECRET r2_catalog_secret (TYPE ICEBERG, TOKEN ?)", [self.config.token]) + con.execute( + f"ATTACH '{self.config.warehouse_name}' AS iceberg_catalog " + f"(TYPE ICEBERG, ENDPOINT '{self.config.catalog_endpoint}')" + ) + + def _bind_model_tables(self) -> None: + for model in self.layer.graph.models.values(): + if model.sql: + model.sql = ( + model.sql.replace("{{ events_table }}", self.config.attached_events_table) + .replace("{{ persons_table }}", self.config.attached_persons_table) + ) + + def _rows( + self, + *, + metrics: list[str], + dimensions: list[str] | None = None, + filters: list[str] | None = None, + order_by: list[str] | None = None, + limit: int | None = None, + skip_default_time_dimensions: bool = False, + ) -> list[dict[str, Any]]: + use_preaggregations = self.config.preagg_enabled + sql = self.generator.generate( + metrics=metrics, + dimensions=dimensions or [], + filters=filters or [], + order_by=order_by, + limit=limit, + skip_default_time_dimensions=skip_default_time_dimensions, + use_preaggregations=use_preaggregations, + ) + try: + relation = self.layer.adapter.execute(sql) + except Exception: + if not use_preaggregations: + raise + fallback_sql = self.generator.generate( + metrics=metrics, + dimensions=dimensions or [], + filters=filters or [], + order_by=order_by, + limit=limit, + skip_default_time_dimensions=skip_default_time_dimensions, + use_preaggregations=False, + ) + relation = self.layer.adapter.execute(fallback_sql) + columns = [column[0] for column in relation.description] + return [dict(zip(columns, row, strict=False)) for row in relation.fetchall()] + + def _materialize_preaggregations(self) -> None: + if not self.config.preagg_enabled or not self.config.preagg_refresh: + return + con = self.layer.adapter.raw_connection + con.execute(f"CREATE SCHEMA IF NOT EXISTS {self.config.preagg_schema}") + for model_name, model in self.layer.graph.models.items(): + for preagg in model.pre_aggregations: + table_name = preagg.get_table_name(model_name, schema=self.config.preagg_schema) + source_sql = preagg.generate_materialization_sql(model) + started = time.perf_counter() + print( + f"sidemantic preagg refresh start {model_name}.{preagg.name} -> {table_name}", + file=sys.stderr, + flush=True, + ) + result = preagg.refresh( + connection=con, + source_sql=source_sql, + table_name=table_name, + mode="full", + ) + con.execute(f"ANALYZE {table_name}") + elapsed = time.perf_counter() - started + print( + "sidemantic preagg refresh done " + f"{model_name}.{preagg.name} rows={result.rows_inserted} " + f"seconds={elapsed:.2f}", + file=sys.stderr, + flush=True, + ) + + def _one(self, **kwargs: Any) -> dict[str, Any]: + rows = self._rows(**kwargs) + return rows[0] if rows else {} + + def _metric_total(self, metric_ref: str, value_key: str, filters: list[str]) -> float: + return to_float( + self._one( + metrics=[metric_ref], + filters=filters, + skip_default_time_dimensions=True, + ).get(value_key) + ) + + def _filters_for( + self, + model: str, + query: dict[str, Any], + *, + exclude_dimension_ref: str | None = None, + ) -> list[str]: + fields = { + "events": { + "distinct_id": "distinct_id", + "session_id": "session_id", + "event_name": "event_type", + "url": "current_url", + "time": "event_time", + }, + "pageviews": { + "distinct_id": "actor_id", + "session_id": "session_id", + "event_name": "event_type", + "url": "current_url", + "time": "event_time", + }, + "sessions": { + "distinct_id": "actor_id", + "session_id": "session_id", + "url": "landing_url", + "time": "session_start_at", + }, + }[model] + filters: list[str] = [] + + for query_key in ("distinct_id", "session_id", "event_name"): + value = clean(query.get(query_key)) + field = fields.get(query_key) + if value and field: + filters.append(f"{model}.{field} = {sql_string(value)}") + + url = clean(query.get("url")) + if url and fields.get("url"): + needle = sql_like(url.lower()) + if model == "sessions": + filters.append( + f"(lower({model}.landing_url) like {needle} or lower({model}.exit_url) like {needle})" + ) + else: + filters.append(f"lower({model}.{fields['url']}) like {needle}") + + date_from = clean(query.get("date_from")) + if date_from: + filters.append(f"{model}.{fields['time']} >= {sql_string(date_from)}") + date_to = clean(query.get("date_to")) + if date_to: + filters.append(f"{model}.{fields['time']} <= {sql_string(date_to)}") + + filters.extend(semantic_filters_for(model, query, exclude_dimension_ref=exclude_dimension_ref)) + return filters + + +def metric_def(value: Any) -> dict[str, Any]: + ref = clean(value) or "events.capture_events" + ref = METRIC_ALIASES.get(ref, ref) + if ref not in QUERY_METRICS: + raise ValueError(f"Unsupported analytics metric: {ref}") + metric = dict(METRICS[ref]) + metric["ref"] = ref + return metric + + +def dimension_def(value: Any, model: str) -> dict[str, Any]: + ref = clean(value) or DEFAULT_DIMENSIONS[model] + if ref not in DIMENSIONS or DIMENSIONS[ref]["model"] != model: + ref = DEFAULT_DIMENSIONS[model] + dimension = dict(DIMENSIONS[ref]) + dimension["ref"] = ref + return dimension + + +def granularity_for(value: Any) -> str: + granularity = clean(value) or "day" + return granularity if granularity in GRANULARITIES else "day" + + +def panel_keys_for_metric(metric: dict[str, Any]) -> list[str]: + return [ + panel + for panel in LEADERBOARD_PANELS + if metric["model"] in PANEL_DIMENSIONS.get(panel, {}) + ] + + +def semantic_filters_for( + model: str, + query: dict[str, Any], + *, + exclude_dimension_ref: str | None = None, +) -> list[str]: + filters: list[str] = [] + for source_ref, values in semantic_filters_from_query(query).items(): + if source_ref == exclude_dimension_ref: + continue + target_ref = dimension_ref_for_model(source_ref, model) + if not target_ref: + continue + if target_ref == exclude_dimension_ref: + continue + target = DIMENSIONS[target_ref] + sql_values = [] + for value in values: + sql_value = value_sql(target, value) + if sql_value is not None: + sql_values.append(sql_value) + if not sql_values: + continue + if len(sql_values) == 1: + filters.append(f"{target_ref} = {sql_values[0]}") + else: + filters.append(f"{target_ref} in ({', '.join(sql_values)})") + return filters + + +def semantic_filters_from_query(query: dict[str, Any]) -> dict[str, list[str]]: + raw = clean(query.get("semantic_filters")) + if not raw: + return {} + try: + parsed = json.loads(raw) + except json.JSONDecodeError: + return {} + if not isinstance(parsed, dict): + return {} + + filters: dict[str, list[str]] = {} + for ref, values in parsed.items(): + ref = clean(ref) + if ref not in DIMENSIONS: + continue + value_list = values if isinstance(values, list) else [values] + cleaned = [value for value in (clean(value) for value in value_list) if value is not None] + if cleaned: + filters[ref] = cleaned[:20] + return filters + + +def dimension_ref_for_model(source_ref: str, model: str) -> str | None: + source = DIMENSIONS.get(source_ref) + if not source: + return None + if source["model"] == model: + return source_ref + source_dimension = source["dimension"] + for ref, candidate in DIMENSIONS.items(): + if candidate["model"] == model and candidate["dimension"] == source_dimension: + return ref + return None + + +def series_points( + context_rows: list[dict[str, Any]], + focus_rows: list[dict[str, Any]], + focus_time_dimension: str, + granularity: str, + focus_value_key: str, +) -> list[dict[str, Any]]: + context_bucket_key = f"event_time__{granularity}" + focus_bucket_key = f"{focus_time_dimension}__{granularity}" + context_by_bucket = { + date_label(row.get(context_bucket_key), granularity): row + for row in context_rows + if row.get(context_bucket_key) is not None + } + focus_by_bucket = { + date_label(row.get(focus_bucket_key), granularity): row + for row in focus_rows + if row.get(focus_bucket_key) is not None + } + buckets = sorted(set(context_by_bucket) | set(focus_by_bucket)) + return [ + { + "bucket": bucket, + "event_count": to_int(context_by_bucket.get(bucket, {}).get("capture_events")), + "pageviews": to_int(context_by_bucket.get(bucket, {}).get("pageviews")), + "session_count": to_int(context_by_bucket.get(bucket, {}).get("unique_sessions")), + "recordings": to_int(context_by_bucket.get(bucket, {}).get("snapshot_events")), + "focused_value": to_float(focus_by_bucket.get(bucket, {}).get(focus_value_key)), + } + for bucket in buckets + ] + + +def same_breakdown(left: dict[str, Any], right: dict[str, Any]) -> bool: + return ( + left.get("model") == right.get("model") + and left.get("dimension") == right.get("dimension") + and left.get("metric") == right.get("metric") + ) + + +def use_geo_hierarchy_breakdown(dimension: dict[str, Any]) -> bool: + return dimension.get("dimension") in {"geo_country_code", "geo_region", "geo_city"} + + +def use_builtin_breakdown(dimension: dict[str, Any]) -> bool: + return dimension.get("ref") in BUILTIN_BREAKDOWN_DIMENSIONS or use_geo_hierarchy_breakdown(dimension) + + +def breakdown( + title: str, + model: str, + dimension: str, + metric: str, + rows: list[dict[str, Any]], + label_key: str, + value_key: str, + total: float, + *, + denominator: str = "max", +) -> dict[str, Any]: + row_values = [ + (str(row.get(label_key)), to_float(row.get(value_key))) + for row in rows + if row.get(label_key) is not None + ] + values = [value for _, value in row_values] + percent_base = max(values, default=0.0) if denominator == "max" else total + payload_rows = [ + { + "label": label, + "value": value, + "percent": percent(value, percent_base), + } + for label, value in row_values + ] + other_value = max(0.0, total - sum(values)) + if other_value > 0.0001: + payload_rows.append( + { + "label": "Others", + "value": other_value, + "percent": percent(other_value, percent_base), + "is_other": True, + } + ) + return { + "title": title, + "model": model, + "dimension": dimension, + "metric": metric, + "rows": payload_rows, + } + + +def count_metric(label: str, value: int, model: str, metric: str) -> dict[str, Any]: + return { + "label": label, + "value": float(value), + "display_value": format_count(value), + "model": model, + "metric": metric, + "semantic_ref": f"{model}.{metric}", + } + + +def seconds_metric(label: str, value: float, model: str, metric: str) -> dict[str, Any]: + return { + "label": label, + "value": value, + "display_value": f"{value:.1f}s", + "model": model, + "metric": metric, + "semantic_ref": f"{model}.{metric}", + } + + +def infer_persons_table(events_table: str) -> str: + replacements = [ + ("hogflare_events_v3", "hogflare_persons_v2"), + ("hogflare_events", "hogflare_persons"), + ("events", "persons"), + ] + for needle, replacement in replacements: + if needle in events_table: + return events_table.replace(needle, replacement) + return "default.hogflare_persons_v2" + + +def clean(value: Any) -> str | None: + if value is None: + return None + value = str(value).strip() + return value or None + + +def sql_string(value: str) -> str: + return "'" + value.replace("'", "''") + "'" + + +def value_sql(dimension: dict[str, Any], value: str) -> str | None: + normalized = clean(value) + if normalized is None: + return None + if dimension.get("type") == "numeric": + return normalized if NUMERIC_LITERAL_RE.match(normalized) else None + return sql_string(normalized) + + +def sql_like(value: str) -> str: + escaped = value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_").replace("'", "''") + return f"'%{escaped}%'" + + +def to_int(value: Any) -> int: + return int(value or 0) + + +def to_float(value: Any) -> float: + return float(value or 0) + + +def percent(value: float, total: float) -> float: + return 0.0 if total == 0 else (value / total) * 100.0 + + +def format_count(value: int) -> str: + return f"{value:,}" + + +def date_label(value: Any, granularity: str = "day") -> str: + if isinstance(value, datetime): + if granularity == "hour": + return value.replace(minute=0, second=0, microsecond=0).isoformat() + return value.date().isoformat() + if isinstance(value, date): + return value.isoformat() + return str(value) + + +def clamp_int(value: Any, *, default: int, low: int, high: int) -> int: + try: + parsed = int(value) + except (TypeError, ValueError): + parsed = default + return min(high, max(low, parsed)) + + +def env_required(name: str) -> str: + value = os.environ.get(name) + if not value: + raise RuntimeError(f"{name} is required") + return value + + +def env_required_any(*names: str) -> str: + for name in names: + value = os.environ.get(name) + if value: + return value + raise RuntimeError(f"{names[0]} is required") + + +def env_flag(name: str, *, default: bool) -> bool: + value = clean(os.environ.get(name)) + if value is None: + return default + return value.lower() not in {"0", "false", "no", "off"} + + +if __name__ == "__main__": + main() diff --git a/src/app_ui.html b/src/app_ui.html new file mode 100644 index 0000000..ab186cd --- /dev/null +++ b/src/app_ui.html @@ -0,0 +1,4500 @@ + + + + + + Hogflare + + + + + +
+ + +
+
+
+ No session selected + Search replay sessions or choose an event. +
+
+ + +
+
+
+
+
+ Ready + Load sessions, then select a recording to replay. +
+
+
+
+ + +
+ + + + diff --git a/src/config.rs b/src/config.rs index ce7f4cd..e605d67 100644 --- a/src/config.rs +++ b/src/config.rs @@ -7,6 +7,7 @@ use thiserror::Error; use url::Url; use crate::feature_flags::FeatureFlagStore; +use crate::product_analytics::ProductAnalyticsConfig; use crate::replay::{ReplayConfig, ReplayConfigError}; #[derive(Debug, Clone)] @@ -22,6 +23,7 @@ pub struct Config { pub posthog_project_api_key: Option, pub session_recording_endpoint: Option, pub replay: Option, + pub analytics: Option, pub posthog_signing_secret: Option, pub person_debug_token: Option, pub feature_flags: FeatureFlagStore, @@ -136,6 +138,7 @@ impl Config { .ok() .map(|v| v.to_string()); let replay = replay_config_from_worker_env(env)?; + let analytics = analytics_config_from_worker_env(env)?; let posthog_signing_secret = env .secret("POSTHOG_SIGNING_SECRET") .ok() @@ -174,6 +177,7 @@ impl Config { posthog_project_api_key, session_recording_endpoint, replay, + analytics, posthog_signing_secret, person_debug_token, feature_flags, @@ -269,6 +273,7 @@ impl Config { let posthog_project_api_key = env::var("POSTHOG_API_KEY").ok(); let session_recording_endpoint = env::var("POSTHOG_SESSION_RECORDING_ENDPOINT").ok(); let replay = replay_config_from_env()?; + let analytics = analytics_config_from_env()?; let posthog_signing_secret = env::var("POSTHOG_SIGNING_SECRET").ok(); let person_debug_token = env::var("PERSON_DEBUG_TOKEN").ok(); let feature_flags = match env::var("HOGFLARE_FEATURE_FLAGS") { @@ -294,6 +299,7 @@ impl Config { posthog_project_api_key, session_recording_endpoint, replay, + analytics, posthog_signing_secret, person_debug_token, feature_flags, @@ -354,6 +360,50 @@ fn replay_config_from_worker_env(env: &worker::Env) -> Result Result, ConfigError> { + let account_id = worker_var_any( + env, + &[ + "HOGFLARE_ANALYTICS_ACCOUNT_ID", + "HOGFLARE_REPLAY_ACCOUNT_ID", + ], + ); + let bucket_name = worker_var_any( + env, + &["HOGFLARE_ANALYTICS_BUCKET", "HOGFLARE_REPLAY_BUCKET"], + ); + let auth_token = worker_secret_or_var_any( + env, + &[ + "HOGFLARE_ANALYTICS_R2_SQL_TOKEN", + "HOGFLARE_REPLAY_R2_SQL_TOKEN", + ], + ); + + let configured = account_id.is_some() || bucket_name.is_some() || auth_token.is_some(); + if !configured { + return Ok(None); + } + + let events_table = worker_var_any( + env, + &[ + "HOGFLARE_ANALYTICS_EVENTS_TABLE", + "HOGFLARE_REPLAY_EVENTS_TABLE", + ], + ); + + Ok(Some(ProductAnalyticsConfig::new( + account_id.ok_or(ConfigError::MissingVar("HOGFLARE_ANALYTICS_ACCOUNT_ID"))?, + bucket_name.ok_or(ConfigError::MissingVar("HOGFLARE_ANALYTICS_BUCKET"))?, + auth_token.ok_or(ConfigError::MissingVar("HOGFLARE_ANALYTICS_R2_SQL_TOKEN"))?, + events_table, + ))) +} + #[cfg(not(target_arch = "wasm32"))] fn replay_config_from_env() -> Result, ConfigError> { let account_id = env::var("HOGFLARE_REPLAY_ACCOUNT_ID").ok(); @@ -384,6 +434,36 @@ fn replay_config_from_env() -> Result, ConfigError> { .map_err(ConfigError::from) } +#[cfg(not(target_arch = "wasm32"))] +fn analytics_config_from_env() -> Result, ConfigError> { + let account_id = env_any(&[ + "HOGFLARE_ANALYTICS_ACCOUNT_ID", + "HOGFLARE_REPLAY_ACCOUNT_ID", + ]); + let bucket_name = env_any(&["HOGFLARE_ANALYTICS_BUCKET", "HOGFLARE_REPLAY_BUCKET"]); + let auth_token = env_any(&[ + "HOGFLARE_ANALYTICS_R2_SQL_TOKEN", + "HOGFLARE_REPLAY_R2_SQL_TOKEN", + ]); + + let configured = account_id.is_some() || bucket_name.is_some() || auth_token.is_some(); + if !configured { + return Ok(None); + } + + let events_table = env_any(&[ + "HOGFLARE_ANALYTICS_EVENTS_TABLE", + "HOGFLARE_REPLAY_EVENTS_TABLE", + ]); + + Ok(Some(ProductAnalyticsConfig::new( + account_id.ok_or(ConfigError::MissingVar("HOGFLARE_ANALYTICS_ACCOUNT_ID"))?, + bucket_name.ok_or(ConfigError::MissingVar("HOGFLARE_ANALYTICS_BUCKET"))?, + auth_token.ok_or(ConfigError::MissingVar("HOGFLARE_ANALYTICS_R2_SQL_TOKEN"))?, + events_table, + ))) +} + fn parse_replay_query_limit(value: Option) -> Result, ConfigError> { value .map(|value| { @@ -412,3 +492,25 @@ fn parse_optional_url( }) .transpose() } + +#[cfg(not(target_arch = "wasm32"))] +fn env_any(names: &[&str]) -> Option { + names.iter().find_map(|name| env::var(name).ok()) +} + +#[cfg(target_arch = "wasm32")] +fn worker_var_any(env: &worker::Env, names: &[&str]) -> Option { + names + .iter() + .find_map(|name| env.var(name).ok().map(|value| value.to_string())) +} + +#[cfg(target_arch = "wasm32")] +fn worker_secret_or_var_any(env: &worker::Env, names: &[&str]) -> Option { + names.iter().find_map(|name| { + env.secret(name) + .ok() + .map(|secret| secret.to_string()) + .or_else(|| env.var(name).ok().map(|value| value.to_string())) + }) +} diff --git a/src/extractors.rs b/src/extractors.rs index 8e141b8..0c9887a 100644 --- a/src/extractors.rs +++ b/src/extractors.rs @@ -910,6 +910,7 @@ mod tests { pipeline: Arc::new(pipeline), persons_pipeline: None, replay: None, + analytics: None, posthog_team_id: None, decide_api_token: None, session_recording_endpoint: None, diff --git a/src/lib.rs b/src/lib.rs index 63334b5..5882f55 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -7,7 +7,9 @@ pub mod importer; pub mod models; pub mod persons; pub mod pipeline; +pub mod product_analytics; pub mod replay; +pub mod ui; use std::{collections::HashMap, sync::Arc}; @@ -34,6 +36,7 @@ use persons::{ NoopPersonStore, PersonAlias, PersonError, PersonStore, PersonUpdate, }; use pipeline::{PersonPipelineRecord, PipelineClient, PipelineError, PipelineEvent}; +use product_analytics::{ProductAnalyticsClient, ProductAnalyticsError, ProductAnalyticsQuery}; use replay::{ ReplayClient, ReplayError, ReplayEventsQuery, ReplayFrictionQuery, ReplayFunnelQuery, ReplayPersonQuery, ReplaySessionEventsQuery, ReplaySessionsQuery, @@ -63,6 +66,7 @@ pub(crate) struct AppState { pub(crate) pipeline: Arc, pub(crate) persons_pipeline: Option>, pub(crate) replay: Option>, + pub(crate) analytics: Option>, pub(crate) posthog_team_id: Option, pub(crate) decide_api_token: Option, pub(crate) session_recording_endpoint: Option, @@ -180,6 +184,11 @@ pub async fn run_with_config(config: Config) -> Result<(), RunError> { .map(ReplayClient::new) .transpose()? .map(Arc::new); + let analytics = config + .analytics + .clone() + .map(ProductAnalyticsClient::new) + .map(Arc::new); info!( endpoint = %config.pipeline_endpoint, @@ -199,19 +208,23 @@ pub async fn run_with_config(config: Config) -> Result<(), RunError> { let listener = TcpListener::bind(config.address).await?; info!(address = %config.address, "listening for requests"); - serve_with_person_pipeline_and_replay( + serve_with_state( listener, - Arc::new(pipeline), - persons_pipeline, - replay, - config.posthog_team_id, - Arc::new(NoopGroupStore), - GroupTypeMap::new(config.posthog_group_types.clone()), - config.posthog_project_api_key.clone(), - config.session_recording_endpoint.clone(), - config.posthog_signing_secret.clone(), - config.person_debug_token.clone(), - Arc::new(config.feature_flags), + build_state( + Arc::new(pipeline), + persons_pipeline, + replay, + analytics, + config.posthog_team_id, + Arc::new(NoopGroupStore), + GroupTypeMap::new(config.posthog_group_types.clone()), + config.posthog_project_api_key.clone(), + config.session_recording_endpoint.clone(), + config.posthog_signing_secret.clone(), + config.person_debug_token.clone(), + Arc::new(config.feature_flags), + Arc::new(persons::MemoryPersonStore::new(config.posthog_team_id)), + ), ) .await } @@ -286,11 +299,17 @@ pub async fn fetch( }, None => None, }; + let analytics = config + .analytics + .clone() + .map(ProductAnalyticsClient::new) + .map(Arc::new); - let mut router = build_router_with_person_pipeline_and_replay( + let mut router = router(build_state( Arc::new(pipeline), persons_pipeline, replay, + analytics, config.posthog_team_id, group_store, group_type_map, @@ -300,7 +319,7 @@ pub async fn fetch( config.person_debug_token.clone(), feature_flags, person_store, - ); + )); Ok(router.call(req).await?) } @@ -394,6 +413,7 @@ pub fn build_router_with_person_pipeline_and_replay( pipeline, persons_pipeline, replay, + None, posthog_team_id, group_store, group_type_map, @@ -415,6 +435,7 @@ pub async fn serve(listener: TcpListener, pipeline: Arc) -> Resu None, None, None, + None, Arc::new(NoopGroupStore), GroupTypeMap::default(), None, @@ -507,6 +528,7 @@ pub async fn serve_with_person_pipeline_and_replay( pipeline, persons_pipeline, replay, + None, posthog_team_id, group_store, group_type_map, @@ -522,6 +544,7 @@ pub async fn serve_with_person_pipeline_and_replay( fn router(state: AppState) -> Router { let router = Router::new() + .route("/", get(app_ui)) .route("/capture", post(capture)) .route("/e", post(browser_capture)) .route("/e/", post(browser_capture)) @@ -538,8 +561,11 @@ fn router(state: AppState) -> Router { .route("/array/:token/config.js", get(array_config_js)) .route("/s", post(session_recording)) .route("/s/", post(session_recording)) - .route("/replay", get(replay_ui)) - .route("/replay/", get(replay_ui)) + .route("/analytics", get(app_ui)) + .route("/analytics/", get(app_ui)) + .route("/analytics/api/charts", get(product_analytics)) + .route("/replay", get(app_ui)) + .route("/replay/", get(app_ui)) .route("/replay/api/sessions", get(replay_sessions)) .route("/replay/api/events", get(replay_events)) .route("/replay/api/funnels", get(replay_funnels)) @@ -567,6 +593,7 @@ fn build_state( pipeline: Arc, persons_pipeline: Option>, replay: Option>, + analytics: Option>, posthog_team_id: Option, group_store: Arc, group_type_map: GroupTypeMap, @@ -581,6 +608,7 @@ fn build_state( pipeline, persons_pipeline, replay, + analytics, posthog_team_id, group_store, group_type_map, @@ -1727,8 +1755,8 @@ async fn health() -> impl IntoResponse { } #[cfg_attr(target_arch = "wasm32", worker::send)] -async fn replay_ui() -> impl IntoResponse { - Html(replay::replay_ui_html()) +async fn app_ui() -> impl IntoResponse { + Html(ui::app_html()) } #[cfg_attr(target_arch = "wasm32", worker::send)] @@ -1746,6 +1774,21 @@ async fn replay_sessions( } } +#[cfg_attr(target_arch = "wasm32", worker::send)] +async fn product_analytics( + State(state): State, + axum::extract::Query(query): axum::extract::Query, +) -> impl IntoResponse { + let Some(client) = state.analytics.as_ref() else { + return product_analytics_error_response(ProductAnalyticsError::NotConfigured); + }; + + match client.query(query).await { + Ok(response) => Json(response).into_response(), + Err(err) => product_analytics_error_response(err), + } +} + #[cfg_attr(target_arch = "wasm32", worker::send)] async fn replay_events( State(state): State, @@ -1853,6 +1896,33 @@ fn replay_error_response(err: ReplayError) -> axum::response::Response { .into_response() } +fn product_analytics_error_response(err: ProductAnalyticsError) -> axum::response::Response { + let status = match &err { + ProductAnalyticsError::NotConfigured => StatusCode::SERVICE_UNAVAILABLE, + ProductAnalyticsError::Serialize(_) => StatusCode::INTERNAL_SERVER_ERROR, + ProductAnalyticsError::Failed(_) | ProductAnalyticsError::InvalidResponse(_) => { + StatusCode::BAD_GATEWAY + } + #[cfg(not(target_arch = "wasm32"))] + ProductAnalyticsError::Process(_) => StatusCode::BAD_GATEWAY, + #[cfg(target_arch = "wasm32")] + ProductAnalyticsError::Unavailable => StatusCode::SERVICE_UNAVAILABLE, + }; + + if status.is_server_error() { + error!(error = %err, "analytics request failed"); + } + + ( + status, + Json(ErrorResponse { + status: 0, + error: err.to_string(), + }), + ) + .into_response() +} + #[cfg_attr(target_arch = "wasm32", worker::send)] async fn debug_person( State(state): State, diff --git a/src/product_analytics.rs b/src/product_analytics.rs new file mode 100644 index 0000000..83fbd8f --- /dev/null +++ b/src/product_analytics.rs @@ -0,0 +1,417 @@ +#[cfg(not(target_arch = "wasm32"))] +use std::path::PathBuf; +#[cfg(not(target_arch = "wasm32"))] +use std::process::Stdio; +#[cfg(not(target_arch = "wasm32"))] +use std::sync::atomic::{AtomicU64, Ordering}; + +use serde::{Deserialize, Serialize}; +#[cfg(not(target_arch = "wasm32"))] +use serde_json::{json, Value}; +use thiserror::Error; + +#[cfg(not(target_arch = "wasm32"))] +use tokio::{ + io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Lines}, + process::{Child, ChildStdin, ChildStdout, Command}, + sync::Mutex, +}; + +#[derive(Debug, Clone)] +#[cfg_attr(target_arch = "wasm32", allow(dead_code))] +pub struct ProductAnalyticsConfig { + account_id: String, + bucket_name: String, + auth_token: String, + events_table: String, +} + +impl ProductAnalyticsConfig { + pub fn new( + account_id: String, + bucket_name: String, + auth_token: String, + events_table: Option, + ) -> Self { + Self { + account_id, + bucket_name, + auth_token, + events_table: events_table.unwrap_or_else(|| "default.hogflare_events".to_string()), + } + } +} + +#[derive(Clone)] +#[cfg_attr(target_arch = "wasm32", allow(dead_code))] +pub struct ProductAnalyticsClient { + config: ProductAnalyticsConfig, + #[cfg(not(target_arch = "wasm32"))] + worker: std::sync::Arc>>, + #[cfg(not(target_arch = "wasm32"))] + request_id: std::sync::Arc, +} + +impl ProductAnalyticsClient { + pub fn new(config: ProductAnalyticsConfig) -> Self { + Self { + config, + #[cfg(not(target_arch = "wasm32"))] + worker: std::sync::Arc::new(Mutex::new(None)), + #[cfg(not(target_arch = "wasm32"))] + request_id: std::sync::Arc::new(AtomicU64::new(0)), + } + } + + #[cfg(not(target_arch = "wasm32"))] + pub async fn query( + &self, + query: ProductAnalyticsQuery, + ) -> Result { + let request_id = self.request_id.fetch_add(1, Ordering::Relaxed) + 1; + let mut payload_value = + serde_json::to_value(query).map_err(ProductAnalyticsError::Serialize)?; + if let Value::Object(fields) = &mut payload_value { + fields.insert("_request_id".to_string(), json!(request_id)); + } + let payload = + serde_json::to_string(&payload_value).map_err(ProductAnalyticsError::Serialize)?; + let runtime = SidemanticRuntimeConfig::from_config(&self.config); + let mut worker = self.worker.lock().await; + if worker + .as_ref() + .map(|existing| existing.runtime != runtime) + .unwrap_or(true) + { + *worker = Some(SidemanticWorker::start(runtime.clone(), &self.config).await?); + } + + let active = worker + .as_mut() + .expect("sidemantic worker must be initialized before querying"); + match active.query(&payload, request_id).await { + Ok(response) => Ok(response), + Err(err) if err.should_restart_worker() => { + drop(worker.take()); + let mut restarted = SidemanticWorker::start(runtime, &self.config).await?; + let response = restarted + .query(&payload, request_id) + .await + .map_err(SidemanticWorkerError::into_product_analytics_error)?; + *worker = Some(restarted); + Ok(response) + } + Err(err) => Err(err.into_product_analytics_error()), + } + } + + #[cfg(target_arch = "wasm32")] + pub async fn query( + &self, + _query: ProductAnalyticsQuery, + ) -> Result { + Err(ProductAnalyticsError::Unavailable) + } +} + +#[derive(Debug, Error)] +pub enum ProductAnalyticsError { + #[error("product analytics is not configured")] + NotConfigured, + #[error("failed to serialize sidemantic analytics query: {0}")] + Serialize(#[source] serde_json::Error), + #[error("failed to run sidemantic analytics: {0}")] + #[cfg(not(target_arch = "wasm32"))] + Process(#[source] std::io::Error), + #[error("sidemantic analytics failed: {0}")] + Failed(String), + #[error("sidemantic analytics requires the native runtime")] + #[cfg(target_arch = "wasm32")] + Unavailable, + #[error("invalid sidemantic analytics response: {0}")] + InvalidResponse(#[source] serde_json::Error), +} + +#[derive(Debug, Deserialize, Serialize)] +pub struct ProductAnalyticsQuery { + #[serde(default)] + pub api_key: Option, + #[serde(default)] + pub distinct_id: Option, + #[serde(default)] + pub session_id: Option, + #[serde(default)] + pub event_name: Option, + #[serde(default)] + pub url: Option, + #[serde(default)] + pub date_from: Option, + #[serde(default)] + pub date_to: Option, + #[serde(default)] + pub limit: Option, + #[serde(default)] + pub metric: Option, + #[serde(default)] + pub dimension: Option, + #[serde(default)] + pub granularity: Option, + #[serde(default)] + pub semantic_filters: Option, + #[serde(default)] + pub panel: Option, +} + +#[derive(Debug, Deserialize, Serialize, Clone, PartialEq)] +pub struct ProductAnalyticsResponse { + pub focus: ProductAnalyticsFocus, + pub summary: Vec, + pub series: Vec, + pub breakdowns: Vec, +} + +#[derive(Debug, Deserialize, Serialize, Clone, PartialEq)] +pub struct ProductAnalyticsFocus { + pub metric: String, + pub metric_label: String, + pub dimension: String, + pub dimension_label: String, + pub granularity: String, +} + +#[derive(Debug, Deserialize, Serialize, Clone, PartialEq)] +pub struct ProductAnalyticsMetric { + pub label: String, + pub value: f64, + pub display_value: String, + pub model: String, + pub metric: String, + pub semantic_ref: String, +} + +#[derive(Debug, Deserialize, Serialize, Clone, PartialEq)] +pub struct ProductAnalyticsSeriesPoint { + pub bucket: String, + pub event_count: usize, + pub pageviews: usize, + pub session_count: usize, + pub recordings: usize, + pub focused_value: f64, +} + +#[derive(Debug, Deserialize, Serialize, Clone, PartialEq)] +pub struct ProductAnalyticsBreakdown { + pub title: String, + pub model: String, + pub dimension: String, + pub metric: String, + pub rows: Vec, +} + +#[derive(Debug, Deserialize, Serialize, Clone, PartialEq)] +pub struct ProductAnalyticsBreakdownRow { + pub label: String, + pub value: f64, + pub percent: f64, + #[serde(default, skip_serializing_if = "is_false")] + pub is_other: bool, +} + +#[cfg(not(target_arch = "wasm32"))] +#[derive(Clone, PartialEq, Eq)] +struct SidemanticRuntimeConfig { + manifest_dir: PathBuf, + script_path: PathBuf, + model_dir: String, + persons_table: String, +} + +#[cfg(not(target_arch = "wasm32"))] +impl SidemanticRuntimeConfig { + fn from_config(config: &ProductAnalyticsConfig) -> Self { + let manifest_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR")); + let script_path = std::env::var("HOGFLARE_ANALYTICS_SIDEMANTIC_SCRIPT") + .map(PathBuf::from) + .unwrap_or_else(|_| manifest_dir.join("scripts/product_analytics_sidemantic.py")); + let model_dir = std::env::var("HOGFLARE_ANALYTICS_MODEL_DIR") + .unwrap_or_else(|_| manifest_dir.join("models").display().to_string()); + let persons_table = std::env::var("HOGFLARE_ANALYTICS_PERSONS_TABLE") + .unwrap_or_else(|_| infer_persons_table(&config.events_table)); + + Self { + manifest_dir, + script_path, + model_dir, + persons_table, + } + } +} + +#[cfg(not(target_arch = "wasm32"))] +struct SidemanticWorker { + runtime: SidemanticRuntimeConfig, + child: Child, + stdin: ChildStdin, + stdout: Lines>, +} + +#[cfg(not(target_arch = "wasm32"))] +impl SidemanticWorker { + async fn start( + runtime: SidemanticRuntimeConfig, + config: &ProductAnalyticsConfig, + ) -> Result { + let mut child = Command::new("uv") + .arg("run") + .arg("--script") + .arg(&runtime.script_path) + .arg("--serve") + .current_dir(&runtime.manifest_dir) + .env("HOGFLARE_ANALYTICS_ACCOUNT_ID", &config.account_id) + .env("HOGFLARE_ANALYTICS_BUCKET", &config.bucket_name) + .env("HOGFLARE_ANALYTICS_R2_SQL_TOKEN", &config.auth_token) + .env("HOGFLARE_ANALYTICS_EVENTS_TABLE", &config.events_table) + .env("HOGFLARE_ANALYTICS_PERSONS_TABLE", &runtime.persons_table) + .env("HOGFLARE_ANALYTICS_MODEL_DIR", &runtime.model_dir) + .env("UV_NO_PROGRESS", "1") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .spawn() + .map_err(ProductAnalyticsError::Process)?; + + let stdin = child.stdin.take().ok_or_else(|| { + ProductAnalyticsError::Process(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "sidemantic worker stdin was not available", + )) + })?; + let stdout = child.stdout.take().ok_or_else(|| { + ProductAnalyticsError::Process(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "sidemantic worker stdout was not available", + )) + })?; + + Ok(Self { + runtime, + child, + stdin, + stdout: BufReader::new(stdout).lines(), + }) + } + + async fn query( + &mut self, + payload: &str, + request_id: u64, + ) -> Result { + self.stdin + .write_all(payload.as_bytes()) + .await + .map_err(SidemanticWorkerError::Process)?; + self.stdin + .write_all(b"\n") + .await + .map_err(SidemanticWorkerError::Process)?; + self.stdin + .flush() + .await + .map_err(SidemanticWorkerError::Process)?; + + loop { + let Some(line) = self + .stdout + .next_line() + .await + .map_err(SidemanticWorkerError::Process)? + else { + return Err(SidemanticWorkerError::Eof); + }; + if line.trim().is_empty() { + continue; + } + let envelope: SidemanticWorkerEnvelope = + serde_json::from_str(&line).map_err(SidemanticWorkerError::InvalidResponse)?; + if envelope.request_id != Some(request_id) { + continue; + } + if envelope.ok { + return envelope.result.ok_or_else(|| { + SidemanticWorkerError::Analytics( + "sidemantic worker returned ok without a result".to_string(), + ) + }); + } + return Err(SidemanticWorkerError::Analytics( + envelope + .error + .unwrap_or_else(|| "sidemantic worker returned an analytics error".to_string()), + )); + } + } +} + +#[cfg(not(target_arch = "wasm32"))] +impl Drop for SidemanticWorker { + fn drop(&mut self) { + let _ = self.child.start_kill(); + } +} + +#[cfg(not(target_arch = "wasm32"))] +#[derive(Deserialize)] +struct SidemanticWorkerEnvelope { + ok: bool, + request_id: Option, + result: Option, + error: Option, +} + +#[cfg(not(target_arch = "wasm32"))] +enum SidemanticWorkerError { + Process(std::io::Error), + Eof, + InvalidResponse(serde_json::Error), + Analytics(String), +} + +#[cfg(not(target_arch = "wasm32"))] +impl SidemanticWorkerError { + fn should_restart_worker(&self) -> bool { + matches!( + self, + Self::Process(_) | Self::Eof | Self::InvalidResponse(_) + ) + } + + fn into_product_analytics_error(self) -> ProductAnalyticsError { + match self { + Self::Process(err) => ProductAnalyticsError::Process(err), + Self::Eof => ProductAnalyticsError::Process(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "sidemantic worker exited before returning analytics", + )), + Self::InvalidResponse(err) => ProductAnalyticsError::InvalidResponse(err), + Self::Analytics(message) => ProductAnalyticsError::Failed(message), + } + } +} + +fn is_false(value: &bool) -> bool { + !*value +} + +#[cfg(not(target_arch = "wasm32"))] +fn infer_persons_table(events_table: &str) -> String { + for (needle, replacement) in [ + ("hogflare_events_v3", "hogflare_persons_v2"), + ("hogflare_events", "hogflare_persons"), + ("events", "persons"), + ] { + if events_table.contains(needle) { + return events_table.replace(needle, replacement); + } + } + "default.hogflare_persons_v2".to_string() +} diff --git a/src/replay.rs b/src/replay.rs index 83b44b8..b124223 100644 --- a/src/replay.rs +++ b/src/replay.rs @@ -81,7 +81,7 @@ pub enum ReplayConfigError { InvalidQueryLimit { value: String, message: String }, } -#[derive(Debug, Clone)] +#[derive(Clone)] pub struct ReplayClient { config: ReplayConfig, #[cfg(not(target_arch = "wasm32"))] @@ -696,7 +696,7 @@ pub struct ReplayActivityItem { #[derive(Debug, Serialize)] pub struct ReplayEventsResponse { - pub events: Vec, + pub events: Vec, } #[derive(Debug, Serialize)] @@ -768,7 +768,7 @@ pub struct ReplayPersonJourneyResponse { #[serde(skip_serializing_if = "Option::is_none")] pub distinct_id: Option, pub sessions: Vec, - pub events: Vec, + pub events: Vec, pub timeline: Vec, } @@ -786,7 +786,7 @@ pub struct ReplayPersonTimelineItem { } #[derive(Debug, Serialize, Clone, PartialEq)] -pub struct ReplayAnalyticsEvent { +pub struct ReplayEventSummary { pub uuid: String, pub event: String, pub distinct_id: String, @@ -1043,7 +1043,7 @@ pub fn build_session_events_response( pub fn build_events_response(rows: Vec, limit: usize) -> ReplayEventsResponse { let mut events = rows .into_iter() - .map(ReplayAnalyticsEvent::from) + .map(ReplayEventSummary::from) .collect::>(); events.sort_by(|a, b| b.created_at.cmp(&a.created_at)); events.truncate(limit); @@ -1062,9 +1062,9 @@ pub fn build_funnel_response( }; } - let mut grouped: HashMap> = HashMap::new(); + let mut grouped: HashMap> = HashMap::new(); for row in rows { - let event = ReplayAnalyticsEvent::from(row); + let event = ReplayEventSummary::from(row); let key = event .session_id .clone() @@ -1133,7 +1133,7 @@ pub fn build_friction_response( pub fn build_person_journey_response( distinct_id: Option, sessions: Vec, - events: Vec, + events: Vec, limit: usize, ) -> ReplayPersonJourneyResponse { let mut timeline = Vec::new(); @@ -1204,11 +1204,7 @@ pub fn rows_from_r2_sql_response(value: Value) -> Result, ReplayError Err(ReplayError::MissingRows) } -pub fn replay_ui_html() -> &'static str { - include_str!("replay_ui.html") -} - -impl From for ReplayAnalyticsEvent { +impl From for ReplayEventSummary { fn from(row: ReplayEventRow) -> Self { let session_id = session_id_from_value(&row.properties); let url = event_url_from_properties(&row.properties); @@ -1229,7 +1225,7 @@ impl From for ReplayAnalyticsEvent { } fn funnel_session_from_events( - events: &[ReplayAnalyticsEvent], + events: &[ReplayEventSummary], steps: &[String], ) -> Option { let first = events.first()?; @@ -1633,7 +1629,7 @@ fn current_url_at(events: &[Value], timestamp: i64) -> Option { .last() } -fn event_property_line(event: &ReplayAnalyticsEvent) -> String { +fn event_property_line(event: &ReplayEventSummary) -> String { if event.properties.is_empty() { return event .url @@ -2302,7 +2298,7 @@ mod tests { } } - fn analytics_row( + fn event_row( uuid: &str, event: &str, session_id: &str, @@ -2475,7 +2471,7 @@ mod tests { } #[test] - fn builds_analytics_event_response_without_raw_properties() { + fn builds_event_response_without_raw_properties() { let response = build_events_response( vec![ReplayEventRow { uuid: "event-1".to_string(), @@ -2513,13 +2509,13 @@ mod tests { fn classifies_funnel_sessions() { let response = build_funnel_response( vec![ - analytics_row("a1", "Viewed Pricing", "converted", "user-a", 1_000), - analytics_row("a2", "Checkout Started", "converted", "user-a", 2_000), - analytics_row("a3", "Paid", "converted", "user-a", 3_000), - analytics_row("b1", "Viewed Pricing", "stuck", "user-b", 4_000), - analytics_row("b2", "Viewed Pricing", "stuck", "user-b", 5_000), - analytics_row("c1", "Viewed Pricing", "dropped", "user-c", 6_000), - analytics_row("d1", "Signed Up", "ignored", "user-d", 7_000), + event_row("a1", "Viewed Pricing", "converted", "user-a", 1_000), + event_row("a2", "Checkout Started", "converted", "user-a", 2_000), + event_row("a3", "Paid", "converted", "user-a", 3_000), + event_row("b1", "Viewed Pricing", "stuck", "user-b", 4_000), + event_row("b2", "Viewed Pricing", "stuck", "user-b", 5_000), + event_row("c1", "Viewed Pricing", "dropped", "user-c", 6_000), + event_row("d1", "Signed Up", "ignored", "user-d", 7_000), ], vec![ "Viewed Pricing".to_string(), diff --git a/src/replay_ui.html b/src/replay_ui.html deleted file mode 100644 index 4b67e82..0000000 --- a/src/replay_ui.html +++ /dev/null @@ -1,2235 +0,0 @@ - - - - - - Hogflare Replay Explorer - - - - - -
- - -
-
-
- No session selected - Search replay sessions or choose an analytics event. -
-
- - -
-
-
-
-
- Ready - Load sessions, then select a recording to replay. -
-
-
-
- - -
- - - - diff --git a/src/ui.rs b/src/ui.rs new file mode 100644 index 0000000..1161ad0 --- /dev/null +++ b/src/ui.rs @@ -0,0 +1,3 @@ +pub fn app_html() -> &'static str { + include_str!("app_ui.html") +} diff --git a/tests/posthog_endpoints.rs b/tests/posthog_endpoints.rs index 71d407b..0509308 100644 --- a/tests/posthog_endpoints.rs +++ b/tests/posthog_endpoints.rs @@ -932,13 +932,17 @@ async fn replay_ui_lists_and_loads_session_events() -> Result<(), BoxHogflare")); assert!(ui.contains("aria-label=\"Hogflare\"")); assert!(ui.contains("id=\"status\" hidden")); assert!(ui.contains("Activity timeline")); assert!(ui.contains("Date range")); assert!(ui.contains("type=\"datetime-local\"")); assert!(ui.contains("data-date-preset=\"24h\"")); + assert!(ui.contains("data-date-preset=\"30d\" aria-pressed=\"true\"")); + assert!(ui.contains("")); + assert!(ui.contains("id=\"clear\" type=\"button\" data-section=\"replays\"")); + assert!(!ui.contains(">Reset<")); assert!(ui.contains("aria-label=\"Hide filters\"")); assert!(ui.contains("aria-label=\"Hide context\"")); assert!(ui.contains(">Search<"));