diff --git a/.github/workflows/analytics.yml b/.github/workflows/analytics.yml new file mode 100644 index 00000000000..6a80f407b59 --- /dev/null +++ b/.github/workflows/analytics.yml @@ -0,0 +1,111 @@ +name: Analytics + +on: + pull_request: + paths: + - ".github/workflows/analytics.yml" + - "package.json" + - "pnpm-lock.yaml" + - "packages/local-docker/**" + - "packages/analytics/**" + - "packages/env/server.ts" + - "packages/web-backend/src/ProductAnalytics/**" + - "apps/web/actions/analytics/**" + - "apps/web/actions/organization/**" + - "apps/web/__tests__/unit/*analytics*.test.ts" + - "apps/web/__tests__/unit/developer-credits-webhook.test.ts" + - "apps/web/app/Layout/*Analytics*.tsx" + - "apps/web/app/api/events/**" + - "apps/web/app/api/invite/accept/route.ts" + - "apps/web/app/api/settings/billing/**" + - "apps/web/app/api/webhooks/stripe/route.ts" + - "apps/web/app/utils/analytics.ts" + - "apps/web/app/utils/product-analytics.ts" + - "apps/web/lib/analytics/**" + - "apps/web/lib/rate-limit.ts" + - "apps/desktop/src-tauri/src/lib.rs" + - "apps/desktop/src/utils/analytics.ts" + - "apps/desktop/src/utils/product-analytics.ts" + - "apps/desktop/src/utils/*analytics*.test.ts" + - "apps/desktop/src-tauri/src/posthog.rs" + - "scripts/analytics/**" + push: + branches: + - main + paths: + - ".github/workflows/analytics.yml" + - "package.json" + - "pnpm-lock.yaml" + - "packages/local-docker/**" + - "packages/analytics/**" + - "packages/env/server.ts" + - "packages/web-backend/src/ProductAnalytics/**" + - "apps/web/actions/analytics/**" + - "apps/web/actions/organization/**" + - "apps/web/__tests__/unit/*analytics*.test.ts" + - "apps/web/__tests__/unit/developer-credits-webhook.test.ts" + - "apps/web/app/Layout/*Analytics*.tsx" + - "apps/web/app/api/events/**" + - "apps/web/app/api/invite/accept/route.ts" + - "apps/web/app/api/settings/billing/**" + - "apps/web/app/api/webhooks/stripe/route.ts" + - "apps/web/app/utils/analytics.ts" + - "apps/web/app/utils/product-analytics.ts" + - "apps/web/lib/analytics/**" + - "apps/web/lib/rate-limit.ts" + - "apps/desktop/src-tauri/src/lib.rs" + - "apps/desktop/src/utils/analytics.ts" + - "apps/desktop/src/utils/product-analytics.ts" + - "apps/desktop/src/utils/*analytics*.test.ts" + - "apps/desktop/src-tauri/src/posthog.rs" + - "scripts/analytics/**" + workflow_dispatch: + inputs: + deploy: + description: Deploy the checked Tinybird project to production + required: true + default: false + type: boolean + +permissions: + contents: read + +jobs: + validate: + name: Validate and test analytics + runs-on: ubuntu-latest + timeout-minutes: 20 + steps: + - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 + - uses: ./.github/actions/setup-js + - name: Run analytics regression suites + run: pnpm analytics:test + - name: Validate Docker Compose + run: node scripts/analytics/analytics-cli.js compose-check + - name: Build and test with Tinybird Local + run: node scripts/analytics/analytics-cli.js local + - name: Stop Tinybird Local + if: always() + run: node scripts/analytics/analytics-cli.js local-stop + + deploy: + name: Deploy analytics + needs: validate + if: github.ref == 'refs/heads/main' && (github.event_name == 'push' || inputs.deploy == true) + runs-on: ubuntu-latest + timeout-minutes: 15 + concurrency: + group: analytics-production + cancel-in-progress: false + environment: production + env: + TINYBIRD_DEPLOY_TOKEN: ${{ secrets.TINYBIRD_DEPLOY_TOKEN }} + TINYBIRD_URL: ${{ secrets.TINYBIRD_URL }} + TINYBIRD_WORKSPACE_ID: ${{ secrets.TINYBIRD_WORKSPACE_ID }} + steps: + - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 + - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 + with: + node-version: 20 + - name: Check and deploy Tinybird project + run: node scripts/analytics/analytics-cli.js cloud-deploy diff --git a/README.md b/README.md index f8943348fba..b0bba710b23 100644 --- a/README.md +++ b/README.md @@ -176,14 +176,17 @@ The web API uses Effect and `@effect/platform` HTTP APIs. Desktop capture and ex ## Analytics -Cap uses [Tinybird](https://www.tinybird.co) for viewer telemetry dashboards. Set `TINYBIRD_ADMIN_TOKEN` or `TINYBIRD_TOKEN` before running analytics commands. +Cap uses [Tinybird](https://www.tinybird.co) for product and viewer analytics. The product event collector uses a separate append-only token, while agents use read-only query access. | Command | Purpose | | --- | --- | -| `pnpm analytics:setup` | Deploy Tinybird datasources and pipes from `scripts/analytics/tinybird` | +| `pnpm analytics:local` | Start the optional Tinybird Local Docker profile, build resources, run fixtures, and write `.env.analytics.local` | +| `pnpm analytics:test` | Run deterministic infrastructure tests and static validation | +| `pnpm analytics:deploy:check` | Validate a cloud deployment without promoting it | +| `pnpm analytics:deploy` | Safely deploy checked-in Tinybird resources | | `pnpm analytics:check` | Validate that the Tinybird workspace matches the app expectations | -`analytics:setup` can remove Tinybird resources outside the checked-in analytics configuration. Use it only against the workspace you intend to manage from this repo. +Routine analytics commands reject destructive operations. See [scripts/analytics/README.md](scripts/analytics/README.md) for local setup, scoped credentials, deployment, and agent access. ## Contributing diff --git a/apps/desktop/package.json b/apps/desktop/package.json index 3eb2586b91e..5b58754d170 100644 --- a/apps/desktop/package.json +++ b/apps/desktop/package.json @@ -15,6 +15,7 @@ }, "dependencies": { "@aerofoil/rive-solid-canvas": "^2.1.4", + "@cap/analytics": "workspace:*", "@cap/database": "workspace:*", "@cap/ui-solid": "workspace:*", "@cap/utils": "workspace:*", diff --git a/apps/desktop/src-tauri/src/lib.rs b/apps/desktop/src-tauri/src/lib.rs index 3f85c7c3b18..c917eb6d361 100644 --- a/apps/desktop/src-tauri/src/lib.rs +++ b/apps/desktop/src-tauri/src/lib.rs @@ -5184,6 +5184,7 @@ pub async fn run(recording_logging_handle: LoggingHandle, logs_dir: PathBuf) { specta_builder.mount_events(&app); hotkeys::init(&app); general_settings::init(&app); + posthog::init_product_session(&app); configure_camera_blur_recovery(&app, previous_termination); fake_window::init(&app); app.manage(target_select_overlay::WindowFocusManager::default()); diff --git a/apps/desktop/src-tauri/src/posthog.rs b/apps/desktop/src-tauri/src/posthog.rs index 54872b2b380..95d37fb182c 100644 --- a/apps/desktop/src-tauri/src/posthog.rs +++ b/apps/desktop/src-tauri/src/posthog.rs @@ -1,14 +1,31 @@ +use serde::Serialize; +use serde_json::{Map, Value}; use std::{ + future::Future, sync::{ OnceLock, PoisonError, RwLock, - atomic::{AtomicBool, Ordering}, + atomic::{AtomicBool, AtomicU64, Ordering}, }, time::Duration, }; use tauri::AppHandle; -use tracing::error; +use tauri_plugin_store::StoreExt; +use tokio::sync::mpsc; +use tracing::{debug, error, warn}; +use uuid::Uuid; -use crate::auth::AuthStore; +use crate::{ + auth::{AuthSecret, AuthStore}, + general_settings::GeneralSettingsStore, + web_api::ManagerExt, +}; + +const PRODUCT_EVENT_QUEUE_CAPACITY: usize = 100; +const PRODUCT_EVENT_BATCH_SIZE: usize = 20; +const PRODUCT_EVENT_BATCH_DELAY: Duration = Duration::from_millis(250); +const PRODUCT_EVENT_RETRY_DELAY: Duration = Duration::from_millis(500); +const PRODUCT_EVENT_REQUEST_TIMEOUT: Duration = Duration::from_secs(3); +const PRODUCT_EVENT_SESSION_STORE_KEY: &str = "product_analytics_session_id"; #[derive(Debug)] pub enum PostHogEvent { @@ -106,43 +123,58 @@ pub enum PostHogEvent { fn truncate_reason(mut s: String) -> String { const MAX_LEN: usize = 240; if s.len() > MAX_LEN { - s.truncate(MAX_LEN); + let mut end = MAX_LEN; + while !s.is_char_boundary(end) { + end = end.saturating_sub(1); + } + s.truncate(end); s.push('…'); } s } -fn posthog_event(event: PostHogEvent, distinct_id: Option<&str>) -> posthog_rs::Event { - fn make_event(name: &str, distinct_id: Option<&str>) -> posthog_rs::Event { - match distinct_id { - Some(id) => posthog_rs::Event::new(name, id), - None => posthog_rs::Event::new_anon(name), +#[derive(Clone, Debug)] +struct EventData { + name: &'static str, + properties: Map, +} + +impl EventData { + fn new(name: &'static str) -> Self { + Self { + name, + properties: Map::new(), } } - fn set(e: &mut posthog_rs::Event, key: &str, value: impl serde::Serialize) { - e.insert_prop(key, value) - .map_err(|err| error!("Error adding PostHog property {key}: {err:?}")) - .ok(); + fn set(&mut self, key: &str, value: impl Serialize) { + match serde_json::to_value(value) { + Ok(value) => { + self.properties.insert(key.to_string(), value); + } + Err(err) => error!("Error serializing analytics property {key}: {err:?}"), + } } +} +fn event_data(event: PostHogEvent) -> EventData { match event { PostHogEvent::MultipartUploadComplete { duration, length, size, } => { - let mut e = make_event("multipart_upload_complete", distinct_id); - set(&mut e, "duration", duration.as_secs()); - set(&mut e, "length", length.as_secs()); - set(&mut e, "size", size); - e + let mut data = EventData::new("multipart_upload_complete"); + data.set("duration", duration.as_secs()); + data.set("length", length.as_secs()); + data.set("size", size); + data } PostHogEvent::MultipartUploadFailed { duration, error } => { - let mut e = make_event("multipart_upload_failed", distinct_id); - set(&mut e, "duration", duration.as_secs()); - set(&mut e, "error", truncate_reason(error)); - e + let mut data = EventData::new("multipart_upload_failed"); + data.set("duration", duration.as_secs()); + data.set("error", truncate_reason(error)); + data } PostHogEvent::RecordingStarted { mode, @@ -156,18 +188,18 @@ fn posthog_event(event: PostHogEvent, distinct_id: Option<&str>) -> posthog_rs:: fragmented, custom_cursor_capture, } => { - let mut e = make_event("recording_started", distinct_id); - set(&mut e, "mode", mode); - set(&mut e, "target_kind", target_kind); - set(&mut e, "has_camera", has_camera); - set(&mut e, "has_mic", has_mic); - set(&mut e, "has_system_audio", has_system_audio); - set(&mut e, "target_fps", target_fps); - set(&mut e, "target_width", target_width); - set(&mut e, "target_height", target_height); - set(&mut e, "fragmented", fragmented); - set(&mut e, "custom_cursor_capture", custom_cursor_capture); - e + let mut data = EventData::new("recording_started"); + data.set("mode", mode); + data.set("target_kind", target_kind); + data.set("has_camera", has_camera); + data.set("has_mic", has_mic); + data.set("has_system_audio", has_system_audio); + data.set("target_fps", target_fps); + data.set("target_width", target_width); + data.set("target_height", target_height); + data.set("fragmented", fragmented); + data.set("custom_cursor_capture", custom_cursor_capture); + data } PostHogEvent::RecordingCompleted { mode, @@ -191,68 +223,58 @@ fn posthog_event(event: PostHogEvent, distinct_id: Option<&str>) -> posthog_rs:: audio_degraded_count, dropped_mic_messages, } => { - let mut e = make_event("recording_completed", distinct_id); - set(&mut e, "mode", mode); - set(&mut e, "status", status); - set(&mut e, "duration_secs", duration_secs); - set(&mut e, "segment_count", segment_count); - set(&mut e, "track_failure_count", track_failure_count); + let mut data = EventData::new("recording_completed"); + data.set("mode", mode); + data.set("status", status); + data.set("duration_secs", duration_secs); + data.set("segment_count", segment_count); + data.set("track_failure_count", track_failure_count); if let Some(ec) = error_class { - set(&mut e, "error_class", truncate_reason(ec)); + data.set("error_class", truncate_reason(ec)); } - set(&mut e, "video_frames_captured", video_frames_captured); - set(&mut e, "video_frames_dropped", video_frames_dropped); - set( - &mut e, - "drop_rate_pct", - (drop_rate_pct * 100.0).round() / 100.0, - ); - set(&mut e, "capture_stalls_count", capture_stalls_count); - set(&mut e, "capture_stalls_max_ms", capture_stalls_max_ms); - set(&mut e, "mixer_stalls_count", mixer_stalls_count); - set(&mut e, "mixer_stalls_max_ms", mixer_stalls_max_ms); - set(&mut e, "audio_gaps_count", audio_gaps_count); - set(&mut e, "audio_gaps_total_ms", audio_gaps_total_ms); - set( - &mut e, - "frame_drop_rate_high_count", - frame_drop_rate_high_count, - ); - set(&mut e, "source_restarts_count", source_restarts_count); - set(&mut e, "muxer_crash_count", muxer_crash_count); - set(&mut e, "audio_degraded_count", audio_degraded_count); - set(&mut e, "dropped_mic_messages", dropped_mic_messages); - e + data.set("video_frames_captured", video_frames_captured); + data.set("video_frames_dropped", video_frames_dropped); + data.set("drop_rate_pct", (drop_rate_pct * 100.0).round() / 100.0); + data.set("capture_stalls_count", capture_stalls_count); + data.set("capture_stalls_max_ms", capture_stalls_max_ms); + data.set("mixer_stalls_count", mixer_stalls_count); + data.set("mixer_stalls_max_ms", mixer_stalls_max_ms); + data.set("audio_gaps_count", audio_gaps_count); + data.set("audio_gaps_total_ms", audio_gaps_total_ms); + data.set("frame_drop_rate_high_count", frame_drop_rate_high_count); + data.set("source_restarts_count", source_restarts_count); + data.set("muxer_crash_count", muxer_crash_count); + data.set("audio_degraded_count", audio_degraded_count); + data.set("dropped_mic_messages", dropped_mic_messages); + data } PostHogEvent::RecordingMuxerCrashed { mode, reason, seconds_into_recording, } => { - let mut e = make_event("recording_muxer_crashed", distinct_id); - set(&mut e, "mode", mode); - set(&mut e, "reason", truncate_reason(reason)); - set( - &mut e, + let mut data = EventData::new("recording_muxer_crashed"); + data.set("mode", mode); + data.set("reason", truncate_reason(reason)); + data.set( "seconds_into_recording", (seconds_into_recording * 1000.0).round() / 1000.0, ); - e + data } PostHogEvent::RecordingAudioDegraded { mode, reason, seconds_into_recording, } => { - let mut e = make_event("recording_audio_degraded", distinct_id); - set(&mut e, "mode", mode); - set(&mut e, "reason", truncate_reason(reason)); - set( - &mut e, + let mut data = EventData::new("recording_audio_degraded"); + data.set("mode", mode); + data.set("reason", truncate_reason(reason)); + data.set( "seconds_into_recording", (seconds_into_recording * 1000.0).round() / 1000.0, ); - e + data } PostHogEvent::RecordingRecovered { trigger, @@ -260,70 +282,275 @@ fn posthog_event(event: PostHogEvent, distinct_id: Option<&str>) -> posthog_rs:: segments_recovered, validation_took_ms, } => { - let mut e = make_event("recording_recovered", distinct_id); - set(&mut e, "trigger", trigger); - set(&mut e, "recovered_duration_secs", recovered_duration_secs); - set(&mut e, "segments_recovered", segments_recovered); - set(&mut e, "validation_took_ms", validation_took_ms); - e + let mut data = EventData::new("recording_recovered"); + data.set("trigger", trigger); + data.set("recovered_duration_secs", recovered_duration_secs); + data.set("segments_recovered", segments_recovered); + data.set("validation_took_ms", validation_took_ms); + data } PostHogEvent::RecordingRecoveryFailed { trigger, reason } => { - let mut e = make_event("recording_recovery_failed", distinct_id); - set(&mut e, "trigger", trigger); - set(&mut e, "reason", truncate_reason(reason)); - e + let mut data = EventData::new("recording_recovery_failed"); + data.set("trigger", trigger); + data.set("reason", truncate_reason(reason)); + data } PostHogEvent::RecordingDiskSpaceLow { mode, bytes_remaining, } => { - let mut e = make_event("recording_disk_space_low", distinct_id); - set(&mut e, "mode", mode); - set(&mut e, "bytes_remaining", bytes_remaining); - e + let mut data = EventData::new("recording_disk_space_low"); + data.set("mode", mode); + data.set("bytes_remaining", bytes_remaining); + data } PostHogEvent::RecordingDiskSpaceExhausted { mode, bytes_remaining, } => { - let mut e = make_event("recording_disk_space_exhausted", distinct_id); - set(&mut e, "mode", mode); - set(&mut e, "bytes_remaining", bytes_remaining); - e + let mut data = EventData::new("recording_disk_space_exhausted"); + data.set("mode", mode); + data.set("bytes_remaining", bytes_remaining); + data } PostHogEvent::RecordingDeviceLost { mode, subsystem } => { - let mut e = make_event("recording_device_lost", distinct_id); - set(&mut e, "mode", mode); - set(&mut e, "subsystem", subsystem); - e + let mut data = EventData::new("recording_device_lost"); + data.set("mode", mode); + data.set("subsystem", subsystem); + data } PostHogEvent::RecordingEncoderRebuilt { mode, backend, attempt, } => { - let mut e = make_event("recording_encoder_rebuilt", distinct_id); - set(&mut e, "mode", mode); - set(&mut e, "backend", backend); - set(&mut e, "attempt", attempt); - e + let mut data = EventData::new("recording_encoder_rebuilt"); + data.set("mode", mode); + data.set("backend", backend); + data.set("attempt", attempt); + data } PostHogEvent::RecordingSourceAudioReset { mode, source, starvation_ms, } => { - let mut e = make_event("recording_source_audio_reset", distinct_id); - set(&mut e, "mode", mode); - set(&mut e, "source", source); - set(&mut e, "starvation_ms", starvation_ms); - e + let mut data = EventData::new("recording_source_audio_reset"); + data.set("mode", mode); + data.set("source", source); + data.set("starvation_ms", starvation_ms); + data } PostHogEvent::RecordingCaptureTargetLost { mode, target } => { - let mut e = make_event("recording_capture_target_lost", distinct_id); - set(&mut e, "mode", mode); - set(&mut e, "target", target); - e + let mut data = EventData::new("recording_capture_target_lost"); + data.set("mode", mode); + data.set("target", target); + data + } + } +} + +fn posthog_event( + data: &EventData, + distinct_id: &str, + process_person_profile: bool, +) -> posthog_rs::Event { + let mut event = posthog_rs::Event::new(data.name, distinct_id); + if !process_person_profile { + event + .insert_prop("$process_person_profile", false) + .map_err(|err| error!("Error disabling PostHog person profile: {err:?}")) + .ok(); + } + for (key, value) in &data.properties { + event + .insert_prop(key, value) + .map_err(|err| error!("Error adding PostHog property {key}: {err:?}")) + .ok(); + } + event +} + +fn is_core_product_event(name: &str) -> bool { + matches!( + name, + "recording_started" + | "recording_completed" + | "multipart_upload_complete" + | "multipart_upload_failed" + | "recording_recovery_failed" + ) +} + +#[derive(Clone, Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct ProductEvent { + event_id: String, + event_name: &'static str, + occurred_at: String, + anonymous_id: String, + session_id: String, + platform: &'static str, + app_version: &'static str, + properties: Map, +} + +#[derive(Serialize)] +struct ProductEventBatch<'a> { + events: &'a [ProductEvent], +} + +fn product_event(data: &EventData, anonymous_id: String) -> Option { + if !is_core_product_event(data.name) { + return None; + } + + Some(ProductEvent { + event_id: Uuid::new_v4().to_string(), + event_name: data.name, + occurred_at: chrono::Utc::now().to_rfc3339(), + anonymous_id, + session_id: PRODUCT_EVENT_SESSION_ID + .get_or_init(Uuid::new_v4) + .to_string(), + platform: "desktop", + app_version: env!("CARGO_PKG_VERSION"), + properties: product_event_properties(data), + }) +} + +fn product_event_properties(data: &EventData) -> Map { + data.properties + .iter() + .filter(|(key, value)| { + !matches!( + key.as_str(), + "error" | "error_message" | "file_name" | "file_path" | "raw_error" | "reason" + ) && (value.is_null() || value.is_boolean() || value.is_number() || value.is_string()) + }) + .map(|(key, value)| { + let value = match value { + Value::String(value) => Value::String(truncate_reason(value.clone())), + value => value.clone(), + }; + (key.clone(), value) + }) + .collect() +} + +fn live_telemetry_enabled(app: &AppHandle) -> bool { + let enabled = GeneralSettingsStore::get(app) + .ok() + .flatten() + .map(|settings| settings.enable_telemetry) + .unwrap_or_else(telemetry_enabled); + TELEMETRY_ENABLED.store(enabled, Ordering::Release); + enabled +} + +fn product_auth_token(app: &AppHandle) -> Option { + AuthStore::get(app) + .ok() + .flatten() + .map(|auth| match auth.secret { + AuthSecret::ApiKey { api_key } => api_key, + AuthSecret::Session { token, .. } => token, + }) +} + +async fn send_product_batch_once(app: &AppHandle, events: &[ProductEvent]) -> Result<(), String> { + if !live_telemetry_enabled(app) { + return Ok(()); + } + + let auth_token = product_auth_token(app); + let response = app + .api_request("/api/events", |client, url| { + let request = client + .post(url) + .timeout(PRODUCT_EVENT_REQUEST_TIMEOUT) + .json(&ProductEventBatch { events }); + match &auth_token { + Some(token) => request.bearer_auth(token), + None => request, + } + }) + .await + .map_err(|err| err.to_string())?; + + if response.status().is_success() || !should_retry_product_status(response.status().as_u16()) { + Ok(()) + } else { + Err(format!( + "product analytics endpoint returned {}", + response.status() + )) + } +} + +fn should_retry_product_status(status: u16) -> bool { + status == 429 || status >= 500 +} + +async fn retry_once(mut operation: F, retry_delay: Duration) -> Result<(), E> +where + F: FnMut() -> Fut, + Fut: Future>, +{ + if operation().await.is_ok() { + return Ok(()); + } + + tokio::time::sleep(retry_delay).await; + operation().await +} + +async fn run_product_event_worker(app: AppHandle, mut receiver: mpsc::Receiver) { + while let Some(first) = receiver.recv().await { + let mut events = Vec::with_capacity(PRODUCT_EVENT_BATCH_SIZE); + events.push(first); + let deadline = tokio::time::Instant::now() + PRODUCT_EVENT_BATCH_DELAY; + + while events.len() < PRODUCT_EVENT_BATCH_SIZE { + match tokio::time::timeout_at(deadline, receiver.recv()).await { + Ok(Some(event)) => events.push(event), + Ok(None) | Err(_) => break, + } + } + + if !live_telemetry_enabled(&app) { + continue; + } + + if let Err(err) = retry_once( + || send_product_batch_once(&app, &events), + PRODUCT_EVENT_RETRY_DELAY, + ) + .await + { + warn!( + event_count = events.len(), + "Dropping product analytics batch after one retry: {err}" + ); + } + } +} + +fn enqueue_product_event(app: &AppHandle, event: ProductEvent) { + let sender = PRODUCT_EVENT_SENDER.get_or_init(|| { + let (sender, receiver) = mpsc::channel(PRODUCT_EVENT_QUEUE_CAPACITY); + tokio::spawn(run_product_event_worker(app.clone(), receiver)); + sender + }); + + if let Err(err) = sender.try_send(event) { + let dropped = PRODUCT_EVENTS_DROPPED.fetch_add(1, Ordering::Relaxed) + 1; + if dropped.is_power_of_two() { + debug!( + dropped, + reason = %err, + "Product analytics queue is unavailable; event dropped" + ); } } } @@ -339,6 +566,21 @@ pub fn init() { } } +pub fn init_product_session(app: &AppHandle) { + let session_id = PRODUCT_EVENT_SESSION_ID + .get_or_init(Uuid::new_v4) + .to_string(); + match app.store("store") { + Ok(store) => { + store.set(PRODUCT_EVENT_SESSION_STORE_KEY, session_id); + if let Err(err) = store.save() { + warn!("Failed to persist product analytics session ID: {err}"); + } + } + Err(err) => warn!("Failed to access store for product analytics session: {err}"), + } +} + pub fn set_server_url(url: &str) { *API_SERVER_IS_CAP_CLOUD .get_or_init(Default::default) @@ -349,6 +591,9 @@ pub fn set_server_url(url: &str) { static API_SERVER_IS_CAP_CLOUD: OnceLock>> = OnceLock::new(); static TELEMETRY_ENABLED: AtomicBool = AtomicBool::new(true); +static PRODUCT_EVENT_SESSION_ID: OnceLock = OnceLock::new(); +static PRODUCT_EVENT_SENDER: OnceLock> = OnceLock::new(); +static PRODUCT_EVENTS_DROPPED: AtomicU64 = AtomicU64::new(0); pub fn set_telemetry_enabled(enabled: bool) { TELEMETRY_ENABLED.store(enabled, Ordering::Release); @@ -359,54 +604,244 @@ pub fn telemetry_enabled() -> bool { } pub fn async_capture_event(app: &AppHandle, event: PostHogEvent) { - if option_env!("VITE_POSTHOG_KEY").is_none() { + if !live_telemetry_enabled(app) { return; } - let live_enabled = crate::general_settings::GeneralSettingsStore::get(app) + let anonymous_id = GeneralSettingsStore::get(app) .ok() .flatten() - .map(|s| s.enable_telemetry) - .unwrap_or_else(telemetry_enabled); - TELEMETRY_ENABLED.store(live_enabled, Ordering::Release); - if !live_enabled { - return; - } - - let distinct_id = AuthStore::get(app) + .map(|settings| settings.instance_id.to_string()) + .unwrap_or_else(|| { + PRODUCT_EVENT_SESSION_ID + .get_or_init(Uuid::new_v4) + .to_string() + }); + let user_id = AuthStore::get(app) .ok() .flatten() .and_then(|auth| auth.user_id); + let process_person_profile = user_id.is_some(); + let distinct_id = user_id.unwrap_or_else(|| anonymous_id.clone()); + let mut data = event_data(event); + data.set("cap_version", env!("CARGO_PKG_VERSION")); + data.set( + "cap_backend", + match *API_SERVER_IS_CAP_CLOUD + .get_or_init(Default::default) + .read() + .unwrap_or_else(PoisonError::into_inner) + { + Some(true) => "cloud", + Some(false) => "self_hosted", + None => "unknown", + }, + ); + data.set("os", std::env::consts::OS); + data.set("arch", std::env::consts::ARCH); + + if let Some(event) = product_event(&data, anonymous_id) { + enqueue_product_event(app, event); + } + + if option_env!("VITE_POSTHOG_KEY").is_none() { + return; + } + + let app = app.clone(); tokio::spawn(async move { - let mut e = posthog_event(event, distinct_id.as_deref()); + if !live_telemetry_enabled(&app) { + return; + } - e.insert_prop("cap_version", env!("CARGO_PKG_VERSION")) - .map_err(|err| error!("Error adding PostHog property: {err:?}")) - .ok(); - e.insert_prop( - "cap_backend", - match *API_SERVER_IS_CAP_CLOUD - .get_or_init(Default::default) - .read() - .unwrap_or_else(PoisonError::into_inner) - { - Some(true) => "cloud", - Some(false) => "self_hosted", - None => "unknown", - }, - ) - .map_err(|err| error!("Error adding PostHog property: {err:?}")) - .ok(); - e.insert_prop("os", std::env::consts::OS) - .map_err(|err| error!("Error adding PostHog property: {err:?}")) - .ok(); - e.insert_prop("arch", std::env::consts::ARCH) - .map_err(|err| error!("Error adding PostHog property: {err:?}")) - .ok(); + let event = posthog_event(&data, &distinct_id, process_person_profile); - posthog_rs::capture(e) + posthog_rs::capture(event) .await .map_err(|err| error!("Error sending event to PostHog: {err:?}")) .ok(); }); } + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicUsize, Ordering}; + + use super::*; + + fn recording_started() -> PostHogEvent { + PostHogEvent::RecordingStarted { + mode: "studio", + target_kind: "screen", + has_camera: true, + has_mic: true, + has_system_audio: false, + target_fps: 60, + target_width: 1920, + target_height: 1080, + fragmented: true, + custom_cursor_capture: true, + } + } + + fn granular_health_event() -> PostHogEvent { + PostHogEvent::RecordingDeviceLost { + mode: "studio", + subsystem: "camera".to_string(), + } + } + + #[test] + fn core_product_event_catalog_is_intentionally_small() { + let included = [ + "recording_started", + "recording_completed", + "multipart_upload_complete", + "multipart_upload_failed", + "recording_recovery_failed", + ]; + let excluded = [ + "recording_recovered", + "recording_muxer_crashed", + "recording_audio_degraded", + "recording_disk_space_low", + "recording_disk_space_exhausted", + "recording_device_lost", + "recording_encoder_rebuilt", + "recording_source_audio_reset", + "recording_capture_target_lost", + ]; + + assert!(included.into_iter().all(is_core_product_event)); + assert!(!excluded.into_iter().any(is_core_product_event)); + } + + #[test] + fn recording_started_has_scalar_properties() { + let data = event_data(recording_started()); + + assert_eq!(data.name, "recording_started"); + assert_eq!(data.properties["mode"], "studio"); + assert_eq!(data.properties["target_fps"], 60); + assert_eq!(data.properties["has_camera"], true); + assert!(data.properties.values().all(|value| { + value.is_null() || value.is_boolean() || value.is_number() || value.is_string() + })); + } + + #[test] + fn product_event_reuses_install_id_and_process_session() { + let data = event_data(recording_started()); + let first = product_event(&data, "install-id".to_string()).unwrap(); + let second = product_event(&data, "install-id".to_string()).unwrap(); + + assert_eq!(first.anonymous_id, "install-id"); + assert_eq!(second.anonymous_id, "install-id"); + assert_eq!(first.session_id, second.session_id); + assert_ne!(first.event_id, second.event_id); + assert_eq!(first.platform, "desktop"); + } + + #[test] + fn stable_anonymous_posthog_event_remains_personless() { + let data = event_data(recording_started()); + let event = posthog_event(&data, "install-id", false); + let json = serde_json::to_value(event).unwrap(); + + assert_eq!(json["$distinct_id"], "install-id"); + assert_eq!(json["properties"]["$process_person_profile"], false); + } + + #[test] + fn granular_health_event_stays_out_of_product_analytics() { + let data = event_data(granular_health_event()); + + assert!(product_event(&data, "install-id".to_string()).is_none()); + } + + #[test] + fn product_events_remove_raw_error_details_before_networking() { + let data = event_data(PostHogEvent::MultipartUploadFailed { + duration: Duration::from_secs(2), + error: "/Users/private/recording.cap failed".to_string(), + }); + let event = product_event(&data, "install-id".to_string()).unwrap(); + + assert!(!event.properties.contains_key("error")); + assert_eq!(event.properties["duration"], 2); + } + + #[test] + fn truncation_is_safe_for_multibyte_text() { + let value = "🙂".repeat(100); + let truncated = truncate_reason(value); + + assert!(truncated.ends_with('…')); + assert!(truncated.len() <= 243); + assert!(truncated.is_char_boundary(truncated.len())); + } + + #[test] + fn batch_contract_uses_expected_camel_case_fields() { + let data = event_data(recording_started()); + let event = product_event(&data, "install-id".to_string()).unwrap(); + let json = serde_json::to_value(ProductEventBatch { events: &[event] }).unwrap(); + let serialized = &json["events"][0]; + + assert!(serialized.get("eventId").is_some()); + assert_eq!(serialized["eventName"], "recording_started"); + assert_eq!(serialized["anonymousId"], "install-id"); + assert_eq!(serialized["platform"], "desktop"); + assert!(serialized.get("occurredAt").is_some()); + assert!(serialized.get("sessionId").is_some()); + assert!(serialized.get("appVersion").is_some()); + } + + #[tokio::test] + async fn retry_once_stops_after_success() { + let attempts = AtomicUsize::new(0); + + let result = retry_once( + || { + let attempt = attempts.fetch_add(1, Ordering::Relaxed); + async move { + if attempt == 0 { + Err("temporary") + } else { + Ok(()) + } + } + }, + Duration::ZERO, + ) + .await; + + assert_eq!(result, Ok(())); + assert_eq!(attempts.load(Ordering::Relaxed), 2); + } + + #[tokio::test] + async fn retry_once_never_attempts_more_than_twice() { + let attempts = AtomicUsize::new(0); + + let result = retry_once( + || { + attempts.fetch_add(1, Ordering::Relaxed); + async { Err::<(), _>("offline") } + }, + Duration::ZERO, + ) + .await; + + assert_eq!(result, Err("offline")); + assert_eq!(attempts.load(Ordering::Relaxed), 2); + } + + #[test] + fn product_status_retry_policy_matches_the_browser() { + assert!(!should_retry_product_status(400)); + assert!(!should_retry_product_status(401)); + assert!(should_retry_product_status(429)); + assert!(should_retry_product_status(503)); + } +} diff --git a/apps/desktop/src/utils/analytics.test.ts b/apps/desktop/src/utils/analytics.test.ts new file mode 100644 index 00000000000..2f4587dcf3a --- /dev/null +++ b/apps/desktop/src/utils/analytics.test.ts @@ -0,0 +1,239 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => { + const state: { + enableTelemetry: boolean; + posthogConfig?: Record; + settingsListener?: (settings?: { enableTelemetry?: boolean }) => void; + } = { enableTelemetry: true }; + + return { + state, + fetch: vi.fn(async (_url: string, _request?: RequestInit) => ({ + ok: true, + status: 202, + })), + posthog: { + alias: vi.fn(), + capture: vi.fn(), + get_distinct_id: vi.fn(() => "install-id"), + has_opted_out_capturing: vi.fn(() => false), + identify: vi.fn(), + init: vi.fn((_key: string, config: Record) => { + state.posthogConfig = config; + }), + opt_in_capturing: vi.fn(), + opt_out_capturing: vi.fn(), + people: { set: vi.fn() }, + }, + }; +}); + +vi.mock("posthog-js", () => ({ default: mocks.posthog })); +vi.mock("@tauri-apps/api/app", () => ({ getVersion: async () => "0.5.6" })); +vi.mock("@tauri-apps/plugin-http", () => ({ fetch: mocks.fetch })); +vi.mock("@tauri-apps/plugin-store", () => ({ + Store: { + load: async () => ({ + get: async (key: string) => + key === "product_analytics_session_id" + ? "process-session-id" + : { enableTelemetry: mocks.state.enableTelemetry }, + onKeyChange: async ( + _key: string, + listener: (settings?: { enableTelemetry?: boolean }) => void, + ) => { + mocks.state.settingsListener = listener; + return () => {}; + }, + }), + }, +})); +vi.mock("~/store", () => ({ + generalSettingsStore: { + get: async () => ({ instanceId: "install-id" }), + }, +})); +vi.mock("./web-api", () => ({ + getConfiguredServerUrl: async () => "https://cap.so", + maybeProtectedHeaders: async () => ({ authorization: "Bearer token" }), +})); + +async function flushMicrotasks() { + for (let index = 0; index < 10; index++) await Promise.resolve(); +} + +async function loadAnalytics() { + const analytics = await import("./analytics"); + await flushMicrotasks(); + return analytics; +} + +function loadPostHog() { + const loaded = mocks.state.posthogConfig?.loaded; + if (typeof loaded !== "function") + throw new Error("PostHog was not initialized"); + loaded(); +} + +describe("desktop analytics", () => { + beforeEach(() => { + vi.resetModules(); + vi.useFakeTimers(); + vi.stubEnv("VITE_POSTHOG_KEY", "posthog-key"); + vi.stubEnv("VITE_POSTHOG_HOST", "https://posthog.example.com"); + mocks.state.enableTelemetry = true; + mocks.state.posthogConfig = undefined; + mocks.state.settingsListener = undefined; + mocks.posthog.alias.mockClear(); + mocks.posthog.capture.mockClear(); + mocks.posthog.get_distinct_id.mockClear(); + mocks.posthog.has_opted_out_capturing.mockClear(); + mocks.posthog.identify.mockClear(); + mocks.posthog.init.mockClear(); + mocks.posthog.opt_in_capturing.mockClear(); + mocks.posthog.opt_out_capturing.mockClear(); + mocks.posthog.people.set.mockClear(); + mocks.posthog.get_distinct_id.mockReturnValue("install-id"); + mocks.posthog.has_opted_out_capturing.mockReturnValue(false); + mocks.fetch.mockClear(); + }); + + afterEach(() => { + vi.useRealTimers(); + vi.unstubAllEnvs(); + }); + + it("disables every automatic PostHog collection surface", async () => { + await loadAnalytics(); + + expect(mocks.state.posthogConfig).toMatchObject({ + autocapture: false, + capture_pageleave: false, + capture_pageview: false, + disable_session_recording: true, + }); + }); + + it("sends a normalized core event through the first-party endpoint", async () => { + const { trackEvent } = await loadAnalytics(); + trackEvent("create_shareable_link_clicked", { + fps: 60, + has_existing_auth: true, + ignored: { nested: true }, + }); + await flushMicrotasks(); + await vi.advanceTimersByTimeAsync(250); + + expect(mocks.fetch).toHaveBeenCalledOnce(); + const [url, request] = mocks.fetch.mock.calls[0] ?? []; + expect(url).toBe("https://cap.so/api/events"); + expect(request).toMatchObject({ + method: "POST", + headers: { + authorization: "Bearer token", + "content-type": "application/json", + }, + }); + const body = JSON.parse(String(request?.body)); + expect(body.events).toHaveLength(1); + expect(body.events[0]).toMatchObject({ + eventName: "create_shareable_link_clicked", + anonymousId: "install-id", + sessionId: "process-session-id", + platform: "desktop", + appVersion: "0.5.6", + properties: { fps: 60, has_existing_auth: true }, + }); + }); + + it("keeps non-core events in explicit PostHog rollback only", async () => { + const { trackEvent } = await loadAnalytics(); + loadPostHog(); + trackEvent("camera_selected", { source: "dropdown" }); + await flushMicrotasks(); + await vi.runAllTimersAsync(); + + expect(mocks.fetch).not.toHaveBeenCalled(); + expect(mocks.posthog.capture).toHaveBeenCalledWith("camera_selected", { + source: "dropdown", + platform: "desktop", + }); + }); + + it("never accepts a client-authored revenue event", async () => { + const { trackEvent } = await loadAnalytics(); + loadPostHog(); + trackEvent("purchase_completed", { quantity: 10 }); + await flushMicrotasks(); + await vi.runAllTimersAsync(); + + expect(mocks.fetch).not.toHaveBeenCalled(); + expect(mocks.posthog.capture).toHaveBeenCalledWith("purchase_completed", { + quantity: 10, + platform: "desktop", + }); + }); + + it("drops permanent collector errors without retrying", async () => { + mocks.fetch.mockResolvedValue({ ok: false, status: 400 }); + const { trackEvent } = await loadAnalytics(); + trackEvent("export_button_clicked"); + await flushMicrotasks(); + await vi.advanceTimersByTimeAsync(1_000); + + expect(mocks.fetch).toHaveBeenCalledOnce(); + }); + + it("retries transient collector errors once", async () => { + mocks.fetch.mockResolvedValue({ ok: false, status: 503 }); + const { trackEvent } = await loadAnalytics(); + trackEvent("export_button_clicked"); + await flushMicrotasks(); + await vi.advanceTimersByTimeAsync(250); + await vi.advanceTimersByTimeAsync(500); + + expect(mocks.fetch).toHaveBeenCalledTimes(2); + }); + + it("bounds the pre-initialization PostHog queue without retry timers", async () => { + const { trackEvent } = await loadAnalytics(); + for (let index = 0; index < 105; index++) { + trackEvent(`legacy_event_${index}`); + } + await flushMicrotasks(); + await vi.advanceTimersByTimeAsync(5_000); + + expect(mocks.posthog.capture).not.toHaveBeenCalled(); + loadPostHog(); + await flushMicrotasks(); + + expect(mocks.posthog.capture).toHaveBeenCalledTimes(100); + expect(mocks.posthog.capture.mock.calls[0]?.[0]).toBe("legacy_event_5"); + }); + + it("suppresses and clears both providers as soon as telemetry is disabled", async () => { + const { trackEvent } = await loadAnalytics(); + trackEvent("export_button_clicked"); + await flushMicrotasks(); + mocks.state.settingsListener?.({ enableTelemetry: false }); + await vi.runAllTimersAsync(); + loadPostHog(); + await flushMicrotasks(); + + expect(mocks.fetch).not.toHaveBeenCalled(); + expect(mocks.posthog.capture).not.toHaveBeenCalled(); + expect(mocks.posthog.opt_out_capturing).toHaveBeenCalledOnce(); + }); + + it("does not initialize either provider when telemetry starts disabled", async () => { + mocks.state.enableTelemetry = false; + const { trackEvent } = await loadAnalytics(); + trackEvent("export_button_clicked"); + await vi.runAllTimersAsync(); + + expect(mocks.posthog.init).not.toHaveBeenCalled(); + expect(mocks.posthog.capture).not.toHaveBeenCalled(); + expect(mocks.fetch).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/desktop/src/utils/analytics.ts b/apps/desktop/src/utils/analytics.ts index d50316d3eb0..08ac149d3cc 100644 --- a/apps/desktop/src/utils/analytics.ts +++ b/apps/desktop/src/utils/analytics.ts @@ -1,128 +1,323 @@ +import { + isCoreEventName, + isServerOnlyEventName, + normalizeProductEventProperties, + type ProductEventInput, +} from "@cap/analytics"; +import { getVersion } from "@tauri-apps/api/app"; +import { fetch } from "@tauri-apps/plugin-http"; import { Store } from "@tauri-apps/plugin-store"; import posthog from "posthog-js"; import { v4 as uuid } from "uuid"; +import { generalSettingsStore } from "~/store"; +import { ProductAnalyticsQueue } from "./product-analytics"; +import { getConfiguredServerUrl, maybeProtectedHeaders } from "./web-api"; + const key = import.meta.env.VITE_POSTHOG_KEY as string; const host = import.meta.env.VITE_POSTHOG_HOST as string; +const POSTHOG_QUEUE_CAPACITY = 100; +const PRODUCT_ANALYTICS_REQUEST_TIMEOUT_MS = 3000; -let isPostHogInitialized = false; +type PendingPostHogEvent = { + eventName: string; + properties: Record; +}; +let isPostHogInitialized = false; +let isPostHogInitializationStarted = false; let telemetryEnabledCache = true; +let telemetryStateReady = false; +let telemetryStatePromise: Promise | undefined; +let anonymousIdPromise: Promise | undefined; +let appVersionPromise: Promise | undefined; +let productSessionIdPromise: Promise | undefined; +let fallbackAnonymousIdValue: string | undefined; +let pendingPostHogEvents: PendingPostHogEvent[] = []; +let activeProductRequest: AbortController | undefined; + +const productAnalyticsQueue = new ProductAnalyticsQueue({ + sendBatch: sendProductEventBatch, + isEnabled: isTelemetryEnabled, +}); + +function applyTelemetryState(enabled: boolean) { + telemetryEnabledCache = enabled; + if (!enabled) { + pendingPostHogEvents = []; + productAnalyticsQueue.clear(); + activeProductRequest?.abort(); + if (isPostHogInitializationStarted) posthog.opt_out_capturing(); + return; + } + + initializePostHog(); + if (isPostHogInitializationStarted && posthog.has_opted_out_capturing()) { + posthog.opt_in_capturing({ captureEventName: false }); + } +} + +async function initializeTelemetryState() { + if (telemetryStatePromise) return telemetryStatePromise; + + telemetryStatePromise = (async () => { + try { + const store = await Store.load("store"); + const settings = await store.get<{ enableTelemetry?: boolean }>( + "general_settings", + ); + applyTelemetryState(settings?.enableTelemetry !== false); + await store.onKeyChange<{ enableTelemetry?: boolean }>( + "general_settings", + (settings) => applyTelemetryState(settings?.enableTelemetry !== false), + ); + } catch { + applyTelemetryState(telemetryEnabledCache); + } finally { + telemetryStateReady = true; + } + })(); + + return telemetryStatePromise; +} + +async function isTelemetryEnabled() { + if (!telemetryStateReady) await initializeTelemetryState(); + return telemetryEnabledCache; +} + +function fallbackAnonymousId() { + if (fallbackAnonymousIdValue) return fallbackAnonymousIdValue; + try { + const storage = getAnalyticsStorage(); + const existing = storage?.getItem("anonymous_id"); + if (existing) { + fallbackAnonymousIdValue = existing; + return existing; + } + } catch {} + + fallbackAnonymousIdValue = uuid(); + try { + getAnalyticsStorage()?.setItem("anonymous_id", fallbackAnonymousIdValue); + } catch {} + return fallbackAnonymousIdValue; +} -async function isTelemetryEnabled(): Promise { +function getAnalyticsStorage() { try { - const store = await Store.load("store"); - const settings = - (await store.get<{ enableTelemetry?: boolean }>("general_settings")) ?? - null; - telemetryEnabledCache = settings?.enableTelemetry !== false; + return typeof window === "undefined" ? undefined : window.localStorage; } catch { - // fall back to cached value; defaults to enabled + return undefined; } - return telemetryEnabledCache; } -if (key && host) { +async function getAnonymousId() { + if (!anonymousIdPromise) { + anonymousIdPromise = generalSettingsStore + .get() + .then((settings) => settings?.instanceId ?? fallbackAnonymousId()) + .then((anonymousId) => { + getAnalyticsStorage()?.setItem("anonymous_id", anonymousId); + return anonymousId; + }) + .catch(fallbackAnonymousId); + } + return anonymousIdPromise; +} + +async function getAppVersion() { + if (!appVersionPromise) { + appVersionPromise = getVersion().catch(() => undefined); + } + return appVersionPromise; +} + +async function getProductSessionId() { + if (!productSessionIdPromise) { + productSessionIdPromise = Store.load("store") + .then((store) => store.get("product_analytics_session_id")) + .then((stored) => stored ?? uuid()) + .catch(uuid); + } + return productSessionIdPromise; +} + +async function sendProductEventBatch(events: ProductEventInput[]) { + if (!(await isTelemetryEnabled())) return; + + const controller = new AbortController(); + activeProductRequest = controller; + const timeout = setTimeout( + () => controller.abort(), + PRODUCT_ANALYTICS_REQUEST_TIMEOUT_MS, + ); + + try { + const { authorization } = await maybeProtectedHeaders(); + const headers: Record = { + "content-type": "application/json", + }; + if (authorization) headers.authorization = authorization; + + const response = await fetch( + new URL("/api/events", await getConfiguredServerUrl()).toString(), + { + method: "POST", + headers, + body: JSON.stringify({ events }), + signal: controller.signal, + }, + ); + if (!response.ok) { + if (response.status === 429 || response.status >= 500) { + throw new Error(`Product analytics returned ${response.status}`); + } + } + } finally { + clearTimeout(timeout); + if (activeProductRequest === controller) activeProductRequest = undefined; + } +} + +async function enqueueProductEvent( + eventId: string, + eventName: string, + occurredAt: string, + properties?: Record, +) { + if (!isCoreEventName(eventName) || isServerOnlyEventName(eventName)) return; + + const [anonymousId, appVersion, productSessionId] = await Promise.all([ + getAnonymousId(), + getAppVersion(), + getProductSessionId(), + ]); + if (!telemetryEnabledCache) return; + + const normalizedProperties = normalizeProductEventProperties(properties); + productAnalyticsQueue.enqueue({ + eventId, + eventName, + occurredAt, + anonymousId, + sessionId: productSessionId, + platform: "desktop", + ...(appVersion ? { appVersion } : {}), + ...(normalizedProperties ? { properties: normalizedProperties } : {}), + }); +} + +function capturePostHogEvent(event: PendingPostHogEvent) { + try { + posthog.capture(event.eventName, event.properties); + } catch (error) { + console.error(`Error capturing event ${event.eventName}:`, error); + } +} + +function enqueuePostHogEvent(event: PendingPostHogEvent) { + if (pendingPostHogEvents.length >= POSTHOG_QUEUE_CAPACITY) { + pendingPostHogEvents.shift(); + } + pendingPostHogEvents.push(event); +} + +async function flushPendingPostHogEvents() { + if (!(await isTelemetryEnabled())) { + pendingPostHogEvents = []; + return; + } + + const events = pendingPostHogEvents; + pendingPostHogEvents = []; + for (const event of events) capturePostHogEvent(event); +} + +function initializePostHog() { + if (isPostHogInitializationStarted || !key || !host) return; + isPostHogInitializationStarted = true; + try { posthog.init(key, { api_host: host, + autocapture: false, + capture_pageleave: false, capture_pageview: false, - loaded: (_posthogInstance) => { + disable_session_recording: true, + loaded: () => { isPostHogInitialized = true; + void flushPendingPostHogEvents(); }, }); - console.log("PostHog initialization started"); } catch (error) { console.error("Failed to initialize PostHog:", error); } } export function initAnonymousUser() { - if (!key || !host) { - console.warn("Cannot initialize anonymous user - missing key or host"); - return; - } + if (!key || !host) return; - try { - const anonymousId = localStorage.getItem("anonymous_id") ?? uuid(); - localStorage.setItem("anonymous_id", anonymousId); - posthog.identify(anonymousId); - console.log("Anonymous user identified:", anonymousId); - } catch (error) { - console.error("Error initializing anonymous user:", error); - } + void Promise.all([isTelemetryEnabled(), getAnonymousId()]) + .then(([enabled, anonymousId]) => { + if (enabled) posthog.identify(anonymousId); + }) + .catch((error) => + console.error("Error initializing anonymous user:", error), + ); } export function identifyUser( userId: string, properties?: Record, ) { - if (!key || !host) { - console.warn("Cannot identify user - missing key or host"); - return; - } + if (!key || !host) return; - try { - const currentId = posthog.get_distinct_id(); - const anonymousId = localStorage.getItem("anonymous_id"); + void isTelemetryEnabled().then((enabled) => { + if (!enabled) return; - if (currentId !== userId) { + try { + const currentId = posthog.get_distinct_id(); + const storage = getAnalyticsStorage(); + const anonymousId = storage?.getItem("anonymous_id"); + + if (currentId === userId) return; if (anonymousId && currentId === anonymousId) { - console.log(`Aliasing user ${userId} from anonymous ID ${anonymousId}`); posthog.alias(userId, anonymousId); } posthog.identify(userId); - if (properties) { - posthog.people.set(properties); - } - localStorage.removeItem("anonymous_id"); - console.log(`User identified: ${userId}`); - } else { - console.log(`User already identified as ${userId}`); + if (properties) posthog.people.set(properties); + storage?.removeItem("anonymous_id"); + } catch (error) { + console.error("Error identifying user:", error); } - } catch (error) { - console.error("Error identifying user:", error); - } + }); } export function trackEvent( eventName: string, properties?: Record, ) { - if (!key || !host) { - console.warn( - "PostHog event not captured - missing key or host:", - eventName, - ); - return; - } - - if (!telemetryEnabledCache) { - return; - } + const eventId = uuid(); + const occurredAt = new Date().toISOString(); void isTelemetryEnabled().then((enabled) => { if (!enabled) return; - try { - if (!isPostHogInitialized) { - console.warn( - `PostHog not initialized yet, queuing event: ${eventName}`, - ); - setTimeout(() => { - console.log(`Retrying event ${eventName} after delay`); - trackEvent(eventName, properties); - }, 1000); - return; - } + void enqueueProductEvent(eventId, eventName, occurredAt, properties); - const eventProperties = { ...properties, platform: "desktop" }; - console.log(`Capturing event ${eventName}:`, eventProperties); - posthog.capture(eventName, eventProperties); - } catch (error) { - console.error(`Error capturing event ${eventName}:`, error); - } + if (!key || !host) return; + const event = { + eventName, + properties: { ...properties, platform: "desktop" }, + }; + if (isPostHogInitialized) capturePostHogEvent(event); + else enqueuePostHogEvent(event); }); } -void isTelemetryEnabled(); +if (typeof window !== "undefined") { + window.addEventListener("pagehide", () => void productAnalyticsQueue.flush()); +} + +void initializeTelemetryState(); diff --git a/apps/desktop/src/utils/product-analytics.test.ts b/apps/desktop/src/utils/product-analytics.test.ts new file mode 100644 index 00000000000..209e2b46cf8 --- /dev/null +++ b/apps/desktop/src/utils/product-analytics.test.ts @@ -0,0 +1,248 @@ +import type { ProductEventInput } from "@cap/analytics"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { ProductAnalyticsQueue } from "./product-analytics"; + +function productEvent(index: number): ProductEventInput { + return { + eventId: `event-${index}`, + eventName: "recording_started", + occurredAt: "2026-07-12T12:00:00.000Z", + anonymousId: "install-id", + sessionId: "session-id", + platform: "desktop", + appVersion: "0.5.6", + properties: { index }, + }; +} + +function sendBatchMock() { + return vi.fn(async (_events: ProductEventInput[]) => {}); +} + +describe("ProductAnalyticsQueue", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("batches events after the short collection window", async () => { + const sendBatch = sendBatchMock(); + const queue = new ProductAnalyticsQueue({ + sendBatch, + isEnabled: () => true, + batchDelayMs: 25, + }); + + queue.enqueue(productEvent(1)); + queue.enqueue(productEvent(2)); + expect(sendBatch).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(25); + + expect(sendBatch).toHaveBeenCalledOnce(); + expect(sendBatch.mock.calls[0]?.[0].map((event) => event.eventId)).toEqual([ + "event-1", + "event-2", + ]); + }); + + it("flushes immediately at the batch limit", async () => { + const sendBatch = sendBatchMock(); + const queue = new ProductAnalyticsQueue({ + sendBatch, + isEnabled: () => true, + batchSize: 3, + }); + + queue.enqueue(productEvent(1)); + queue.enqueue(productEvent(2)); + queue.enqueue(productEvent(3)); + await Promise.resolve(); + await Promise.resolve(); + + expect(sendBatch).toHaveBeenCalledOnce(); + expect(sendBatch.mock.calls[0]?.[0]).toHaveLength(3); + }); + + it("never sends more than the configured batch size", async () => { + const sizes: number[] = []; + const queue = new ProductAnalyticsQueue({ + sendBatch: async (events) => { + sizes.push(events.length); + }, + isEnabled: () => true, + batchSize: 5, + batchDelayMs: 10, + }); + + for (let index = 0; index < 13; index++) { + queue.enqueue(productEvent(index)); + } + await vi.runAllTimersAsync(); + + expect(sizes.reduce((total, size) => total + size, 0)).toBe(13); + expect(sizes.every((size) => size <= 5)).toBe(true); + }); + + it("keeps serialized requests under the byte limit", async () => { + const requestSizes: number[] = []; + const queue = new ProductAnalyticsQueue({ + sendBatch: async (events) => { + requestSizes.push( + new TextEncoder().encode(JSON.stringify({ events })).byteLength, + ); + }, + isEnabled: () => true, + batchDelayMs: 10, + maxBatchBytes: 1500, + }); + + for (let index = 0; index < 10; index++) { + queue.enqueue({ + ...productEvent(index), + properties: { value: "x".repeat(500) }, + }); + } + await vi.runAllTimersAsync(); + + expect(requestSizes).toHaveLength(5); + expect(requestSizes.every((size) => size <= 1500)).toBe(true); + }); + + it("bounds memory and drops the oldest queued event", async () => { + const sendBatch = sendBatchMock(); + const queue = new ProductAnalyticsQueue({ + sendBatch, + isEnabled: () => true, + batchSize: 10, + capacity: 3, + batchDelayMs: 10, + }); + + for (let index = 0; index < 4; index++) { + queue.enqueue(productEvent(index)); + } + await vi.advanceTimersByTimeAsync(10); + + expect(queue.dropped).toBe(1); + expect(sendBatch.mock.calls[0]?.[0].map((event) => event.eventId)).toEqual([ + "event-1", + "event-2", + "event-3", + ]); + }); + + it("does not send when telemetry is disabled", async () => { + const sendBatch = sendBatchMock(); + const queue = new ProductAnalyticsQueue({ + sendBatch, + isEnabled: () => false, + batchDelayMs: 10, + }); + + queue.enqueue(productEvent(1)); + await vi.advanceTimersByTimeAsync(10); + + expect(sendBatch).not.toHaveBeenCalled(); + expect(queue.size).toBe(0); + expect(queue.dropped).toBe(1); + }); + + it("retries a failed event once and then drops it", async () => { + const sendBatch = vi.fn(async (_events: ProductEventInput[]) => { + throw new Error("offline"); + }); + const queue = new ProductAnalyticsQueue({ + sendBatch, + isEnabled: () => true, + batchDelayMs: 10, + retryDelayMs: 20, + }); + + queue.enqueue(productEvent(1)); + await vi.advanceTimersByTimeAsync(10); + await vi.advanceTimersByTimeAsync(20); + + expect(sendBatch).toHaveBeenCalledTimes(2); + expect(queue.size).toBe(0); + expect(queue.dropped).toBe(1); + }); + + it("does not retry an in-flight request after telemetry is disabled", async () => { + let enabled = true; + let failRequest: (() => void) | undefined; + const sendBatch = vi.fn( + (_events: ProductEventInput[]) => + new Promise((_resolve, reject) => { + failRequest = () => reject(new Error("aborted")); + }), + ); + const queue = new ProductAnalyticsQueue({ + sendBatch, + isEnabled: () => enabled, + batchSize: 1, + retryDelayMs: 10, + }); + + queue.enqueue(productEvent(1)); + await Promise.resolve(); + enabled = false; + failRequest?.(); + await vi.runAllTimersAsync(); + + expect(sendBatch).toHaveBeenCalledOnce(); + expect(queue.size).toBe(0); + expect(queue.dropped).toBe(1); + }); + + it("keeps at most one request in flight", async () => { + let completeFirst: (() => void) | undefined; + let calls = 0; + const sendBatch = vi.fn((_events: ProductEventInput[]) => { + calls += 1; + if (calls > 1) return Promise.resolve(); + return new Promise((resolve) => { + completeFirst = resolve; + }); + }); + const queue = new ProductAnalyticsQueue({ + sendBatch, + isEnabled: () => true, + batchSize: 1, + batchDelayMs: 10, + }); + + queue.enqueue(productEvent(1)); + await Promise.resolve(); + queue.enqueue(productEvent(2)); + await queue.flush(); + + expect(sendBatch).toHaveBeenCalledOnce(); + completeFirst?.(); + await vi.advanceTimersByTimeAsync(10); + + expect(sendBatch).toHaveBeenCalledTimes(2); + }); + + it("can synchronously clear queued events on opt-out", async () => { + const sendBatch = sendBatchMock(); + const queue = new ProductAnalyticsQueue({ + sendBatch, + isEnabled: () => true, + batchDelayMs: 10, + }); + + queue.enqueue(productEvent(1)); + queue.enqueue(productEvent(2)); + queue.clear(); + await vi.runAllTimersAsync(); + + expect(queue.size).toBe(0); + expect(queue.dropped).toBe(2); + expect(sendBatch).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/desktop/src/utils/product-analytics.ts b/apps/desktop/src/utils/product-analytics.ts new file mode 100644 index 00000000000..e6c0294fb5e --- /dev/null +++ b/apps/desktop/src/utils/product-analytics.ts @@ -0,0 +1,171 @@ +import { + PRODUCT_ANALYTICS_LIMITS, + type ProductEventInput, +} from "@cap/analytics"; + +export const PRODUCT_ANALYTICS_BATCH_SIZE = PRODUCT_ANALYTICS_LIMITS.batchSize; +export const PRODUCT_ANALYTICS_QUEUE_CAPACITY = + PRODUCT_ANALYTICS_LIMITS.queueSize; +export const PRODUCT_ANALYTICS_BATCH_DELAY_MS = 250; +export const PRODUCT_ANALYTICS_RETRY_DELAY_MS = 500; + +type QueuedEvent = { + event: ProductEventInput; + attempts: number; +}; + +type ProductAnalyticsQueueOptions = { + sendBatch: (events: ProductEventInput[]) => Promise; + isEnabled: () => boolean | Promise; + batchSize?: number; + capacity?: number; + batchDelayMs?: number; + retryDelayMs?: number; + maxBatchBytes?: number; + onDrop?: (count: number) => void; +}; + +export class ProductAnalyticsQueue { + readonly #sendBatch: ProductAnalyticsQueueOptions["sendBatch"]; + readonly #isEnabled: ProductAnalyticsQueueOptions["isEnabled"]; + readonly #batchSize: number; + readonly #capacity: number; + readonly #batchDelayMs: number; + readonly #retryDelayMs: number; + readonly #maxBatchBytes: number; + readonly #onDrop: ProductAnalyticsQueueOptions["onDrop"]; + #queue: QueuedEvent[] = []; + #timer: ReturnType | undefined; + #inFlight = false; + #dropped = 0; + + constructor(options: ProductAnalyticsQueueOptions) { + this.#sendBatch = options.sendBatch; + this.#isEnabled = options.isEnabled; + this.#batchSize = options.batchSize ?? PRODUCT_ANALYTICS_BATCH_SIZE; + this.#capacity = options.capacity ?? PRODUCT_ANALYTICS_QUEUE_CAPACITY; + this.#batchDelayMs = + options.batchDelayMs ?? PRODUCT_ANALYTICS_BATCH_DELAY_MS; + this.#retryDelayMs = + options.retryDelayMs ?? PRODUCT_ANALYTICS_RETRY_DELAY_MS; + this.#maxBatchBytes = + options.maxBatchBytes ?? PRODUCT_ANALYTICS_LIMITS.requestBytes; + this.#onDrop = options.onDrop; + } + + get size() { + return this.#queue.length; + } + + get dropped() { + return this.#dropped; + } + + enqueue(event: ProductEventInput) { + if (this.#queue.length >= this.#capacity) { + this.#queue.shift(); + this.#recordDrop(1); + } + + this.#queue.push({ event, attempts: 0 }); + if (this.#queue.length >= this.#batchSize) { + void this.flush(); + } else { + this.#schedule(this.#batchDelayMs); + } + } + + clear() { + if (this.#queue.length > 0) { + this.#recordDrop(this.#queue.length); + this.#queue = []; + } + this.#cancelTimer(); + } + + async flush() { + if (this.#inFlight || this.#queue.length === 0) return; + + this.#inFlight = true; + this.#cancelTimer(); + + try { + if (!(await this.#isEnabled())) { + this.clear(); + return; + } + + const batch = this.#takeBatch(); + if (batch.length === 0) return; + try { + await this.#sendBatch(batch.map(({ event }) => event)); + } catch { + if (!(await this.#isEnabled())) { + this.#recordDrop(batch.length); + return; + } + const retryable = batch + .filter(({ attempts }) => attempts === 0) + .map(({ event, attempts }) => ({ event, attempts: attempts + 1 })); + this.#recordDrop(batch.length - retryable.length); + const queued = [...retryable, ...this.#queue]; + const overflow = Math.max(0, queued.length - this.#capacity); + this.#recordDrop(overflow); + this.#queue = queued.slice(overflow); + if (this.#queue.length > 0) this.#schedule(this.#retryDelayMs); + return; + } + } finally { + this.#inFlight = false; + } + + if (this.#queue.length > 0) { + this.#schedule(this.#batchDelayMs); + } + } + + #schedule(delayMs: number) { + if (this.#timer !== undefined) return; + this.#timer = setTimeout(() => { + this.#timer = undefined; + void this.flush(); + }, delayMs); + } + + #cancelTimer() { + if (this.#timer === undefined) return; + clearTimeout(this.#timer); + this.#timer = undefined; + } + + #recordDrop(count: number) { + if (count === 0) return; + this.#dropped += count; + this.#onDrop?.(this.#dropped); + } + + #takeBatch() { + const batch: QueuedEvent[] = []; + + while (batch.length < this.#batchSize && this.#queue.length > 0) { + const next = this.#queue[0]; + if (!next) break; + const candidate = [...batch, next]; + const bytes = new TextEncoder().encode( + JSON.stringify({ events: candidate.map(({ event }) => event) }), + ).byteLength; + + if (bytes > this.#maxBatchBytes) { + if (batch.length > 0) break; + this.#queue.shift(); + this.#recordDrop(1); + continue; + } + + batch.push(next); + this.#queue.shift(); + } + + return batch; + } +} diff --git a/apps/web/__tests__/unit/guest-checkout-analytics.test.ts b/apps/web/__tests__/unit/guest-checkout-analytics.test.ts new file mode 100644 index 00000000000..8782675c934 --- /dev/null +++ b/apps/web/__tests__/unit/guest-checkout-analytics.test.ts @@ -0,0 +1,79 @@ +import { NextRequest } from "next/server"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + createSession: vi.fn(), + product: vi.fn(), + posthog: vi.fn(), + readAnonymousId: vi.fn(), +})); + +vi.mock("@cap/env", () => ({ + serverEnv: () => ({ WEB_URL: "https://cap.so" }), +})); +vi.mock("@cap/utils", () => ({ + stripe: () => ({ + checkout: { sessions: { create: mocks.createSession } }, + }), +})); +vi.mock("@/lib/analytics/server", () => ({ + readAnalyticsAnonymousId: mocks.readAnonymousId, + scheduleLegacyPostHogEvent: mocks.posthog, + scheduleServerProductEvent: mocks.product, +})); + +describe("guest checkout analytics", () => { + let POST: typeof import("@/app/api/settings/billing/guest-checkout/route").POST; + + beforeEach(async () => { + vi.clearAllMocks(); + mocks.createSession.mockResolvedValue({ + id: "cs_guest_1", + url: "https://checkout.stripe.com/session", + }); + POST = (await import("@/app/api/settings/billing/guest-checkout/route")) + .POST; + }); + + it("uses one stable fallback identity through checkout and purchase metadata", async () => { + mocks.readAnonymousId.mockReturnValue(undefined); + const response = await POST( + new NextRequest("https://cap.so/api/settings/billing/guest-checkout", { + method: "POST", + body: JSON.stringify({ priceId: "price_team", quantity: 3 }), + }), + ); + expect(response.status).toBe(200); + + const metadata = mocks.createSession.mock.calls[0]?.[0].metadata; + expect(metadata.analyticsAnonymousId).toMatch(/^guest:/); + expect(metadata.analyticsIsFirstPurchase).toBe("true"); + expect(mocks.product).toHaveBeenCalledWith( + expect.objectContaining({ + eventId: "checkout:cs_guest_1", + anonymousId: metadata.analyticsAnonymousId, + }), + ); + expect(mocks.posthog).toHaveBeenCalledWith( + expect.objectContaining({ + distinctId: metadata.analyticsAnonymousId, + properties: expect.objectContaining({ + $insert_id: "checkout:cs_guest_1", + }), + }), + ); + }); + + it("preserves an existing browser identity", async () => { + mocks.readAnonymousId.mockReturnValue("anonymous-browser-1"); + await POST( + new NextRequest("https://cap.so/api/settings/billing/guest-checkout", { + method: "POST", + body: JSON.stringify({ priceId: "price_team", quantity: 1 }), + }), + ); + expect( + mocks.createSession.mock.calls[0]?.[0].metadata.analyticsAnonymousId, + ).toBe("anonymous-browser-1"); + }); +}); diff --git a/apps/web/__tests__/unit/product-analytics-browser-token.test.ts b/apps/web/__tests__/unit/product-analytics-browser-token.test.ts new file mode 100644 index 00000000000..941cbf7fb96 --- /dev/null +++ b/apps/web/__tests__/unit/product-analytics-browser-token.test.ts @@ -0,0 +1,75 @@ +import { describe, expect, it } from "vitest"; +import { + createProductAnalyticsBrowserToken, + PRODUCT_ANALYTICS_BROWSER_TOKEN_COOKIE, + PRODUCT_ANALYTICS_BROWSER_TOKEN_TTL_SECONDS, + readProductAnalyticsBrowserToken, + readProductAnalyticsBrowserTokenClaims, + verifyProductAnalyticsBrowserToken, +} from "@/lib/analytics/browser-token"; + +const secret = "analytics-browser-token-test-secret"; +const now = Date.parse("2026-07-12T12:00:00.000Z"); + +describe("product analytics browser token", () => { + it("accepts an untampered token inside its bounded lifetime", () => { + const token = createProductAnalyticsBrowserToken( + secret, + "anonymous-1", + now, + ); + expect(verifyProductAnalyticsBrowserToken(token, secret, now)).toBe(true); + expect(readProductAnalyticsBrowserTokenClaims(token, secret, now)).toEqual({ + anonymousId: "anonymous-1", + }); + expect( + verifyProductAnalyticsBrowserToken( + token, + secret, + now + PRODUCT_ANALYTICS_BROWSER_TOKEN_TTL_SECONDS * 1000, + ), + ).toBe(true); + }); + + it("rejects expired, future, tampered, and malformed tokens", () => { + const token = createProductAnalyticsBrowserToken( + secret, + "anonymous-1", + now, + ); + expect( + verifyProductAnalyticsBrowserToken( + token, + secret, + now + (PRODUCT_ANALYTICS_BROWSER_TOKEN_TTL_SECONDS + 1) * 1000, + ), + ).toBe(false); + expect( + verifyProductAnalyticsBrowserToken( + createProductAnalyticsBrowserToken(secret, "anonymous-1", now + 61_000), + secret, + now, + ), + ).toBe(false); + expect(verifyProductAnalyticsBrowserToken(`${token}x`, secret, now)).toBe( + false, + ); + expect(verifyProductAnalyticsBrowserToken("invalid", secret, now)).toBe( + false, + ); + }); + + it("reads only the analytics token cookie", () => { + const token = createProductAnalyticsBrowserToken( + secret, + "anonymous-1", + now, + ); + expect( + readProductAnalyticsBrowserToken( + `other=value; ${PRODUCT_ANALYTICS_BROWSER_TOKEN_COOKIE}=${token}`, + ), + ).toBe(token); + expect(readProductAnalyticsBrowserToken("other=value")).toBeUndefined(); + }); +}); diff --git a/apps/web/__tests__/unit/product-analytics-queue.test.ts b/apps/web/__tests__/unit/product-analytics-queue.test.ts new file mode 100644 index 00000000000..f79f2c6264b --- /dev/null +++ b/apps/web/__tests__/unit/product-analytics-queue.test.ts @@ -0,0 +1,377 @@ +import { + PRODUCT_ANALYTICS_LIMITS, + type ProductEventInput, +} from "@cap/analytics"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + createProductEventId, + getOrCreateBrowserAnonymousId, + getOrCreateStorageId, + ProductAnalyticsQueue, + type ProductAnalyticsTransport, + readFirstTouchAttribution, + sendBrowserProductAnalytics, + shouldCaptureProductPageView, +} from "@/app/utils/product-analytics"; + +const makeEvent = (index: number): ProductEventInput => ({ + eventId: `event-${index}`, + eventName: "page_view", + occurredAt: "2026-07-12T12:00:00.000Z", + anonymousId: "anonymous-1", + sessionId: "session-1", + platform: "web", +}); + +describe("ProductAnalyticsQueue", () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it("does not perform network work on enqueue", () => { + const transport = vi.fn(); + const queue = new ProductAnalyticsQueue(transport); + queue.enqueue(makeEvent(1)); + expect(transport).not.toHaveBeenCalled(); + }); + + it("flushes one batch after the interval", async () => { + const transport = vi + .fn() + .mockResolvedValue("success"); + const queue = new ProductAnalyticsQueue(transport); + queue.enqueue(makeEvent(1)); + queue.enqueue(makeEvent(2)); + + await vi.advanceTimersByTimeAsync(5_000); + expect(transport).toHaveBeenCalledTimes(1); + expect(transport.mock.calls[0]?.[0]).toHaveLength(2); + }); + + it("flushes immediately when a full batch is queued", async () => { + const transport = vi + .fn() + .mockResolvedValue("success"); + const queue = new ProductAnalyticsQueue(transport); + for (let i = 0; i < PRODUCT_ANALYTICS_LIMITS.batchSize; i += 1) { + queue.enqueue(makeEvent(i)); + } + await queue.flush(); + expect(transport).toHaveBeenCalledTimes(1); + expect(transport.mock.calls[0]?.[0]).toHaveLength( + PRODUCT_ANALYTICS_LIMITS.batchSize, + ); + }); + + it("allows only one request in flight", async () => { + let resolveTransport: ((value: "success") => void) | undefined; + const transport = vi.fn().mockImplementation( + () => + new Promise((resolve) => { + resolveTransport = resolve; + }), + ); + const queue = new ProductAnalyticsQueue(transport); + queue.enqueue(makeEvent(1)); + const first = queue.flush(); + const second = queue.flush(); + expect(first).toBe(second); + expect(transport).toHaveBeenCalledTimes(1); + resolveTransport?.("success"); + await first; + }); + + it("retries a failed batch once", async () => { + const transport = vi + .fn() + .mockResolvedValueOnce("retry") + .mockResolvedValueOnce("retry"); + const queue = new ProductAnalyticsQueue(transport); + queue.enqueue(makeEvent(1)); + await queue.flush(); + await vi.advanceTimersByTimeAsync(3_000); + expect(transport).toHaveBeenCalledTimes(2); + expect(queue.size).toBe(0); + }); + + it("honors retry backoff for a full failed batch", async () => { + const transport = vi + .fn() + .mockResolvedValueOnce("retry") + .mockResolvedValueOnce("success"); + const queue = new ProductAnalyticsQueue(transport); + for (let i = 0; i < PRODUCT_ANALYTICS_LIMITS.batchSize; i += 1) { + queue.enqueue(makeEvent(i)); + } + await Promise.resolve(); + await Promise.resolve(); + + expect(transport).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(1_999); + expect(transport).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(1); + expect(transport).toHaveBeenCalledTimes(2); + }); + + it("does not retry a rejected batch", async () => { + const transport = vi + .fn() + .mockResolvedValue("drop"); + const queue = new ProductAnalyticsQueue(transport); + queue.enqueue(makeEvent(1)); + await queue.flush(); + await vi.runAllTimersAsync(); + expect(transport).toHaveBeenCalledTimes(1); + }); + + it("bounds memory and drops the oldest queued events", async () => { + let resolveFirst: ((value: "success") => void) | undefined; + const transport = vi + .fn() + .mockImplementationOnce( + () => + new Promise((resolve) => { + resolveFirst = resolve; + }), + ) + .mockResolvedValue("success"); + const queue = new ProductAnalyticsQueue(transport); + for (let i = 0; i < PRODUCT_ANALYTICS_LIMITS.queueSize + 30; i += 1) { + queue.enqueue(makeEvent(i)); + } + expect(queue.size).toBe(PRODUCT_ANALYTICS_LIMITS.queueSize); + expect(transport.mock.calls[0]?.[0][0]?.eventId).toBe("event-0"); + resolveFirst?.("success"); + await vi.waitFor(() => + expect(transport.mock.calls.length).toBeGreaterThanOrEqual(2), + ); + expect(transport.mock.calls[1]?.[0][0]?.eventId).toBe("event-30"); + }); + + it("keeps every request under the body size limit", async () => { + const requestSizes: number[] = []; + const transport = vi.fn(async (events) => { + requestSizes.push( + new TextEncoder().encode(JSON.stringify({ events })).byteLength, + ); + return "success"; + }); + const queue = new ProductAnalyticsQueue(transport); + for (let i = 0; i < 10; i += 1) { + queue.enqueue({ + ...makeEvent(i), + properties: { value: "x".repeat(20_000) }, + }); + } + + await vi.runAllTimersAsync(); + expect( + requestSizes.every( + (size) => size <= PRODUCT_ANALYTICS_LIMITS.requestBytes, + ), + ).toBe(true); + expect( + transport.mock.calls.reduce( + (count, [events]) => count + events.length, + 0, + ), + ).toBe(10); + }); + + it("drops a single event larger than the request limit", async () => { + const transport = vi.fn(); + const queue = new ProductAnalyticsQueue(transport); + queue.enqueue({ + ...makeEvent(1), + properties: { + value: "x".repeat(PRODUCT_ANALYTICS_LIMITS.requestBytes), + }, + }); + + await vi.runAllTimersAsync(); + expect(transport).not.toHaveBeenCalled(); + expect(queue.size).toBe(0); + }); +}); + +describe("browser analytics identity", () => { + it("falls back when secure UUID generation is unavailable", () => { + const randomValues = (values: Uint32Array) => { + values.set([123, 456]); + return values; + }; + expect(createProductEventId(null, 1_000, randomValues)).toBe( + "fallback-rs-3f-co", + ); + expect( + createProductEventId( + () => { + throw new Error("blocked"); + }, + 1_000, + randomValues, + ), + ).toBe("fallback-rs-3f-co"); + expect(createProductEventId(null, 1_000, null)).toMatch( + /^fallback-rs-counter-[a-z0-9]+$/, + ); + }); + + it("reuses a persisted identifier", () => { + const storage = { + getItem: vi.fn(() => "existing-id"), + setItem: vi.fn(), + }; + expect(getOrCreateStorageId(storage, "key", () => "new-id")).toBe( + "existing-id", + ); + expect(storage.setItem).not.toHaveBeenCalled(); + }); + + it("creates and persists an identifier once", () => { + const storage = { getItem: vi.fn(() => null), setItem: vi.fn() }; + expect(getOrCreateStorageId(storage, "key", () => "new-id")).toBe("new-id"); + expect(storage.setItem).toHaveBeenCalledWith("key", "new-id"); + }); + + it("uses the server-issued cookie identity", () => { + const storage = { getItem: vi.fn(() => "stale-id"), setItem: vi.fn() }; + expect( + getOrCreateBrowserAnonymousId(storage, "signed-id", () => "new-id"), + ).toBe("signed-id"); + expect(storage.setItem).toHaveBeenCalledWith( + "cap_analytics_anonymous_id_v1", + "signed-id", + ); + }); + + it("falls back when storage is unavailable", () => { + const storage = { + getItem: vi.fn(() => { + throw new Error("blocked"); + }), + setItem: vi.fn(), + }; + expect(getOrCreateStorageId(storage, "key", () => "memory-id")).toBe( + "memory-id", + ); + }); + + it("keeps one generated id when persistence is unavailable", () => { + const createId = vi.fn(() => "memory-id"); + const storage = { + getItem: vi.fn(() => null), + setItem: vi.fn(() => { + throw new Error("blocked"); + }), + }; + expect(getOrCreateStorageId(storage, "key", createId)).toBe("memory-id"); + expect(createId).toHaveBeenCalledOnce(); + }); +}); + +describe("first-touch attribution", () => { + it("stores only allowlisted attribution fields", () => { + const storage = { getItem: vi.fn(() => null), setItem: vi.fn() }; + const result = readFirstTouchAttribution( + "?utm_source=google&utm_campaign=launch&email=private%40example.com", + storage, + ); + expect(result).toEqual({ utm_source: "google", utm_campaign: "launch" }); + expect(storage.setItem).toHaveBeenCalledOnce(); + }); + + it("does not overwrite existing attribution", () => { + const storage = { + getItem: vi.fn(() => '{"utm_source":"original"}'), + setItem: vi.fn(), + }; + expect(readFirstTouchAttribution("?utm_source=new", storage)).toEqual({ + utm_source: "original", + }); + expect(storage.setItem).not.toHaveBeenCalled(); + }); +}); + +describe("product page views", () => { + it.each(["/", "/pricing", "/dashboard", "/dashboard/settings"])( + "captures %s", + (pathname) => { + expect(shouldCaptureProductPageView(pathname)).toBe(true); + }, + ); + + it.each(["/s/video-id", "/c/comment-id", "/embed/video-id"])( + "excludes high-volume viewer route %s", + (pathname) => { + expect(shouldCaptureProductPageView(pathname)).toBe(false); + }, + ); +}); + +describe("browser product analytics transport", () => { + it("uses a beacon during unload without also fetching", async () => { + const fetchImpl = vi.fn(); + const sendBeacon = vi.fn(() => true); + await expect( + sendBrowserProductAnalytics([makeEvent(1)], "unload", { + fetchImpl, + sendBeacon, + }), + ).resolves.toBe("success"); + expect(sendBeacon).toHaveBeenCalledOnce(); + expect(fetchImpl).not.toHaveBeenCalled(); + }); + + it("falls back to keepalive fetch when a beacon is rejected", async () => { + const fetchImpl = vi + .fn() + .mockResolvedValue(new Response(null, { status: 202 })); + await expect( + sendBrowserProductAnalytics([makeEvent(1)], "unload", { + fetchImpl, + sendBeacon: () => false, + }), + ).resolves.toBe("success"); + expect(fetchImpl.mock.calls[0]?.[1]).toMatchObject({ keepalive: true }); + }); + + it.each([ + [429, "retry"], + [503, "retry"], + [400, "drop"], + ] as const)("maps HTTP %s to %s", async (status, result) => { + const fetchImpl = vi + .fn() + .mockResolvedValue(new Response(null, { status })); + await expect( + sendBrowserProductAnalytics([makeEvent(1)], "normal", { fetchImpl }), + ).resolves.toBe(result); + }); + + it("retries transport failures", async () => { + const fetchImpl = vi + .fn() + .mockRejectedValue(new Error("offline")); + await expect( + sendBrowserProductAnalytics([makeEvent(1)], "normal", { fetchImpl }), + ).resolves.toBe("retry"); + }); + + it("times out a stalled request", async () => { + vi.useFakeTimers(); + const fetchImpl = vi.fn( + (_url, init) => + new Promise((_resolve, reject) => { + init?.signal?.addEventListener("abort", () => + reject(new DOMException("Aborted", "AbortError")), + ); + }), + ); + const result = sendBrowserProductAnalytics([makeEvent(1)], "normal", { + fetchImpl, + }); + await vi.advanceTimersByTimeAsync(3_000); + await expect(result).resolves.toBe("retry"); + vi.useRealTimers(); + }); +}); diff --git a/apps/web/__tests__/unit/product-analytics-request.test.ts b/apps/web/__tests__/unit/product-analytics-request.test.ts new file mode 100644 index 00000000000..59ba91ba862 --- /dev/null +++ b/apps/web/__tests__/unit/product-analytics-request.test.ts @@ -0,0 +1,239 @@ +import { + PRODUCT_ANALYTICS_LIMITS, + type ProductEventInput, +} from "@cap/analytics"; +import { describe, expect, it } from "vitest"; +import { + getProductAnalyticsRateLimitKey, + hasExpectedBrowserAnalyticsMetadata, + isAllowedAnonymousBrowserProductEvent, + isAuthenticatedAnalyticsRequestCandidate, + normalizeGeoHeader, + normalizeProductEventBatch, + ProductAnalyticsRateLimiter, +} from "@/lib/analytics/request"; + +const allowedOrigins = ["https://cap.so", "tauri://localhost"]; +const event: ProductEventInput = { + eventId: "event-1", + eventName: "page_view", + occurredAt: "2026-07-12T12:00:00.000Z", + anonymousId: "anonymous-1", + sessionId: "session-1", + platform: "web", +}; +const now = Date.parse("2026-07-12T12:00:01.000Z"); + +describe("hasExpectedBrowserAnalyticsMetadata", () => { + it.each([ + [ + "same-origin browser", + { origin: "https://cap.so", secFetchSite: "same-origin" }, + ], + [ + "same-site browser", + { origin: "https://cap.so", secFetchSite: "same-site" }, + ], + ])("accepts %s", (_label, headers) => { + expect(hasExpectedBrowserAnalyticsMetadata(headers, allowedOrigins)).toBe( + true, + ); + }); + + it("rejects cross-site browser requests", () => { + expect( + hasExpectedBrowserAnalyticsMetadata( + { origin: "https://attacker.example", secFetchSite: "cross-site" }, + allowedOrigins, + ), + ).toBe(false); + }); + + it("rejects requests without browser metadata", () => { + expect(hasExpectedBrowserAnalyticsMetadata({}, allowedOrigins)).toBe(false); + expect( + hasExpectedBrowserAnalyticsMetadata( + { origin: "https://cap.so" }, + allowedOrigins, + ), + ).toBe(false); + expect( + hasExpectedBrowserAnalyticsMetadata( + { origin: "https://cap.so", secFetchSite: "none" }, + allowedOrigins, + ), + ).toBe(false); + }); + + it("allows API-key requests to attempt actor resolution", () => { + expect( + isAuthenticatedAnalyticsRequestCandidate({ + authorization: `Bearer ${"a".repeat(36)}`, + }), + ).toBe(true); + expect( + isAuthenticatedAnalyticsRequestCandidate({ + authorization: "Bearer invalid", + }), + ).toBe(false); + expect( + isAuthenticatedAnalyticsRequestCandidate({ + authorization: `Bearer ${"a".repeat(36)} extra`, + }), + ).toBe(false); + expect( + isAuthenticatedAnalyticsRequestCandidate({ + authorization: `Bearer ${"a".repeat(36)}`, + origin: "https://attacker.example", + }), + ).toBe(true); + }); + + it("allows only bounded top-of-funnel web events without an actor", () => { + expect(isAllowedAnonymousBrowserProductEvent(event, "anonymous-1")).toBe( + true, + ); + expect( + isAllowedAnonymousBrowserProductEvent( + { ...event, eventName: "recording_started" }, + "anonymous-1", + ), + ).toBe(false); + expect( + isAllowedAnonymousBrowserProductEvent( + { ...event, anonymousId: "attacker-chosen" }, + "anonymous-1", + ), + ).toBe(false); + expect( + isAllowedAnonymousBrowserProductEvent( + { ...event, platform: "desktop" }, + "anonymous-1", + ), + ).toBe(false); + }); + + it("rejects oversized declared bodies", () => { + expect( + hasExpectedBrowserAnalyticsMetadata( + { contentLength: String(PRODUCT_ANALYTICS_LIMITS.requestBytes + 1) }, + allowedOrigins, + ), + ).toBe(false); + }); + + it.each(["invalid", "-1", "1.5"])( + "rejects malformed content length %s", + (contentLength) => { + expect( + hasExpectedBrowserAnalyticsMetadata({ contentLength }, allowedOrigins), + ).toBe(false); + }, + ); +}); + +describe("normalizeProductEventBatch", () => { + it("accepts a bounded valid batch", () => { + expect(normalizeProductEventBatch([event], now)).toEqual([event]); + }); + + it("rejects an empty batch", () => { + expect(normalizeProductEventBatch([], now)).toBeNull(); + }); + + it("rejects a batch above the cap", () => { + expect( + normalizeProductEventBatch( + Array.from( + { length: PRODUCT_ANALYTICS_LIMITS.batchSize + 1 }, + () => event, + ), + now, + ), + ).toBeNull(); + }); + + it("rejects the whole batch when one event is invalid", () => { + expect( + normalizeProductEventBatch( + [event, { ...event, eventName: "$autocapture" }], + now, + ), + ).toBeNull(); + }); + + it("rejects an undeclared oversized body", () => { + expect( + normalizeProductEventBatch( + [ + { + ...event, + properties: { + value: "x".repeat(PRODUCT_ANALYTICS_LIMITS.requestBytes), + }, + }, + ], + now, + ), + ).toBeNull(); + }); + + it.each([ + "user_signed_up", + "checkout_started", + "guest_checkout_started", + "purchase_completed", + ] as const)("rejects client-authored %s", (eventName) => { + expect( + normalizeProductEventBatch([{ ...event, eventName }], now), + ).toBeNull(); + }); +}); + +describe("ProductAnalyticsRateLimiter", () => { + it("enforces per-key and process-wide fallback limits", () => { + const limiter = new ProductAnalyticsRateLimiter({ + perKeyLimit: 2, + globalLimit: 4, + windowMs: 1_000, + }); + expect(limiter.isRateLimited("a", 0)).toBe(false); + expect(limiter.isRateLimited("a", 0)).toBe(false); + expect(limiter.isRateLimited("a", 0)).toBe(true); + expect(limiter.isRateLimited("b", 0)).toBe(false); + expect(limiter.isRateLimited("c", 0)).toBe(true); + expect(limiter.isRateLimited("a", 1_000)).toBe(false); + }); + + it("uses only a platform-owned Vercel request identity", () => { + expect( + getProductAnalyticsRateLimitKey({ + trustedVercelProxy: true, + xVercelForwardedFor: "203.0.113.10, 10.0.0.1", + }), + ).toBe("203.0.113.10"); + expect( + getProductAnalyticsRateLimitKey({ + trustedVercelProxy: false, + xVercelForwardedFor: "attacker-controlled", + }), + ).toBe("self-hosted"); + expect(getProductAnalyticsRateLimitKey({ trustedVercelProxy: true })).toBe( + "vercel-unknown", + ); + }); +}); + +describe("normalizeGeoHeader", () => { + it("decodes and bounds a city header", () => { + expect(normalizeGeoHeader("Nicosia%20Centre", true)).toBe("Nicosia Centre"); + }); + + it("rejects malformed encoded data", () => { + expect(normalizeGeoHeader("%E0%A4%A", true)).toBeUndefined(); + }); + + it("removes unknown values", () => { + expect(normalizeGeoHeader("unknown")).toBeUndefined(); + }); +}); diff --git a/apps/web/__tests__/unit/product-analytics-scheduler.test.ts b/apps/web/__tests__/unit/product-analytics-scheduler.test.ts new file mode 100644 index 00000000000..3ed5183580f --- /dev/null +++ b/apps/web/__tests__/unit/product-analytics-scheduler.test.ts @@ -0,0 +1,47 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + after: vi.fn(() => { + throw new Error("after unavailable"); + }), +})); + +vi.mock("next/server", () => ({ after: mocks.after })); +vi.mock("@cap/env", () => ({ + buildEnv: { + NEXT_PUBLIC_POSTHOG_KEY: "", + NEXT_PUBLIC_POSTHOG_HOST: "", + }, + serverEnv: () => ({ + PRODUCT_ANALYTICS_TINYBIRD_HOST: undefined, + PRODUCT_ANALYTICS_TINYBIRD_TOKEN: undefined, + }), +})); +vi.mock("posthog-node", () => ({ PostHog: vi.fn() })); + +describe("analytics scheduling", () => { + afterEach(() => vi.restoreAllMocks()); + + it("cannot make a business route fail when after is unavailable", async () => { + vi.spyOn(console, "error").mockImplementation(() => {}); + const { scheduleLegacyPostHogEvent, scheduleServerProductEvent } = + await import("@/lib/analytics/server"); + + expect(() => + scheduleServerProductEvent({ + eventId: "checkout:cs_1", + eventName: "checkout_started", + anonymousId: "anonymous-1", + platform: "web", + }), + ).not.toThrow(); + expect(() => + scheduleLegacyPostHogEvent({ + distinctId: "anonymous-1", + eventName: "checkout_started", + }), + ).not.toThrow(); + await Promise.resolve(); + expect(mocks.after).toHaveBeenCalledTimes(2); + }); +}); diff --git a/apps/web/__tests__/unit/product-analytics-server.test.ts b/apps/web/__tests__/unit/product-analytics-server.test.ts new file mode 100644 index 00000000000..f1dda108db9 --- /dev/null +++ b/apps/web/__tests__/unit/product-analytics-server.test.ts @@ -0,0 +1,52 @@ +import { describe, expect, it } from "vitest"; +import { createServerProductEventRows } from "@/lib/analytics/server-event"; + +describe("server product analytics", () => { + it("builds a deterministic trusted server event", () => { + const [row] = createServerProductEventRows({ + eventId: "stripe:evt_123:purchase_completed", + eventName: "purchase_completed", + occurredAt: "2026-07-12T12:00:00.000Z", + anonymousId: "anonymous-1", + platform: "web", + userId: "user-1", + organizationId: "org-1", + properties: { + quantity: 3, + email: "private@example.com", + nested: { private: true }, + }, + }); + + expect(row).toMatchObject({ + event_id: "stripe:evt_123:purchase_completed", + event_name: "purchase_completed", + source: "server", + platform: "web", + anonymous_id: "anonymous-1", + user_id: "user-1", + organization_id: "org-1", + properties: '{"quantity":3}', + }); + }); + + it("uses an authenticated fallback identity", () => { + const [row] = createServerProductEventRows({ + eventId: "signup:user-1", + eventName: "user_signed_up", + platform: "server", + userId: "user-1", + }); + expect(row?.anonymous_id).toBe("user:user-1"); + }); + + it("drops an event without any stable identity", () => { + expect( + createServerProductEventRows({ + eventId: "event-1", + eventName: "page_view", + platform: "server", + }), + ).toEqual([]); + }); +}); diff --git a/apps/web/__tests__/unit/product-analytics-transport.test.ts b/apps/web/__tests__/unit/product-analytics-transport.test.ts new file mode 100644 index 00000000000..2fa585cbe32 --- /dev/null +++ b/apps/web/__tests__/unit/product-analytics-transport.test.ts @@ -0,0 +1,134 @@ +import { + createProductEventRows, + type ProductAnalyticsError, + sendProductAnalyticsRows, +} from "@cap/analytics"; +import { hasAnalyticsSessionCookie } from "@cap/web-backend"; +import { describe, expect, it, vi } from "vitest"; + +const rows = createProductEventRows( + [ + { + eventId: "event-1", + eventName: "page_view", + occurredAt: "2026-07-12T12:00:00.000Z", + anonymousId: "anonymous-1", + platform: "web", + }, + ], + { + receivedAt: "2026-07-12T12:00:01.000Z", + source: "client", + }, +); + +describe("Tinybird product event transport", () => { + it("skips session resolution for anonymous requests", () => { + expect(hasAnalyticsSessionCookie()).toBe(false); + expect(hasAnalyticsSessionCookie("theme=dark; visitor=123")).toBe(false); + expect( + hasAnalyticsSessionCookie( + "theme=dark; next-auth.session-token=token; visitor=123", + ), + ).toBe(true); + expect(hasAnalyticsSessionCookie("next-auth.session-token.0=chunk")).toBe( + true, + ); + }); + + it("posts NDJSON with append-only credentials", async () => { + const fetchImpl = vi + .fn() + .mockResolvedValue(new Response(null, { status: 202 })); + + await sendProductAnalyticsRows({ + host: "https://api.tinybird.co", + token: "append-token", + rows, + fetchImpl, + }); + + const [url, request] = fetchImpl.mock.calls[0] ?? []; + expect(String(url)).toBe( + "https://api.tinybird.co/v0/events?name=product_events_v1&format=ndjson", + ); + expect(request).toMatchObject({ + method: "POST", + headers: { + Authorization: "Bearer append-token", + "Content-Type": "application/x-ndjson", + }, + body: JSON.stringify(rows[0]), + }); + }); + + it("retries a transient response once", async () => { + const fetchImpl = vi + .fn() + .mockResolvedValueOnce(new Response("busy", { status: 503 })) + .mockResolvedValueOnce(new Response(null, { status: 202 })); + + await sendProductAnalyticsRows({ + host: "https://api.tinybird.co", + token: "append-token", + rows, + fetchImpl, + }); + expect(fetchImpl).toHaveBeenCalledTimes(2); + }); + + it("does not retry a permanent response", async () => { + const fetchImpl = vi + .fn() + .mockResolvedValue(new Response("invalid", { status: 400 })); + + await expect( + sendProductAnalyticsRows({ + host: "https://api.tinybird.co", + token: "append-token", + rows, + fetchImpl, + }), + ).rejects.toMatchObject({ + _tag: "ProductAnalyticsError", + retryable: false, + status: 400, + } satisfies Partial); + expect(fetchImpl).toHaveBeenCalledOnce(); + }); + + it("supports a single-attempt collector path", async () => { + const fetchImpl = vi + .fn() + .mockRejectedValue(new Error("offline")); + await expect( + sendProductAnalyticsRows({ + host: "https://api.tinybird.co", + token: "append-token", + rows, + maxAttempts: 1, + fetchImpl, + }), + ).rejects.toMatchObject({ retryable: true }); + expect(fetchImpl).toHaveBeenCalledOnce(); + }); + + it("fails after two network attempts", async () => { + const fetchImpl = vi + .fn() + .mockRejectedValue(new Error("offline")); + + await expect( + sendProductAnalyticsRows({ + host: "https://api.tinybird.co", + token: "append-token", + rows, + fetchImpl, + }), + ).rejects.toMatchObject({ + _tag: "ProductAnalyticsError", + retryable: true, + }); + expect(fetchImpl).toHaveBeenCalledTimes(2); + }); +}); diff --git a/apps/web/__tests__/unit/subscription-analytics-webhook.test.ts b/apps/web/__tests__/unit/subscription-analytics-webhook.test.ts new file mode 100644 index 00000000000..c0e530ffede --- /dev/null +++ b/apps/web/__tests__/unit/subscription-analytics-webhook.test.ts @@ -0,0 +1,246 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + product: vi.fn(), + posthog: vi.fn(), + constructEvent: vi.fn(), + retrieveCustomer: vi.fn(), + retrieveSubscription: vi.fn(), +})); + +const dbChain = { + select: vi.fn(), + from: vi.fn(), + where: vi.fn(), + limit: vi.fn(), + update: vi.fn(), + set: vi.fn(), +}; + +vi.mock("@/lib/analytics/server", () => ({ + scheduleLegacyPostHogEvent: mocks.posthog, + scheduleServerProductEvent: mocks.product, +})); +vi.mock("@/lib/developer-credits", () => ({ addCreditsToAccount: vi.fn() })); +vi.mock("@cap/database", () => ({ db: () => dbChain })); +vi.mock("@cap/database/helpers", () => ({ nanoId: () => "new-user" })); +vi.mock("@cap/database/schema", () => ({ + developerCreditTransactions: {}, + users: { + id: "id", + email: "email", + }, +})); +vi.mock("@cap/env", () => ({ + serverEnv: () => ({ STRIPE_WEBHOOK_SECRET: "whsec_test" }), +})); +vi.mock("@cap/utils", () => ({ + stripe: () => ({ + webhooks: { constructEvent: mocks.constructEvent }, + customers: { + retrieve: mocks.retrieveCustomer, + update: vi.fn(), + }, + subscriptions: { + retrieve: mocks.retrieveSubscription, + list: vi.fn(), + }, + }), +})); +vi.mock("@cap/web-domain", () => ({ + Organisation: { OrganisationId: { make: (value: string) => value } }, + User: { UserId: { make: (value: string) => value } }, +})); +vi.mock("drizzle-orm", () => ({ + and: (...args: unknown[]) => args, + eq: (left: unknown, right: unknown) => ({ left, right }), +})); + +const dbUser = { + id: "user-1", + email: "user@example.com", + activeOrganizationId: "org-1", + stripeSubscriptionId: null, + name: "User", +}; + +const customer = { + id: "cus_1", + deleted: false, + email: "user@example.com", + metadata: { userId: "user-1" }, +}; + +const subscription = { + id: "sub_1", + status: "active", + items: { + data: [ + { + quantity: 3, + price: { + id: "price_team", + unit_amount: 900, + recurring: { interval: "month", interval_count: 1 }, + }, + }, + ], + }, +}; + +function session(overrides: Record = {}) { + return { + id: "cs_1", + customer: "cus_1", + subscription: "sub_1", + payment_status: "paid", + amount_total: 2700, + amount_subtotal: 3000, + currency: "usd", + total_details: { amount_discount: 300 }, + metadata: { + platform: "web", + analyticsAnonymousId: "anonymous-1", + analyticsIsFirstPurchase: "true", + }, + ...overrides, + }; +} + +function request() { + return new Request("https://cap.so/api/webhooks/stripe", { + method: "POST", + headers: { "Stripe-Signature": "signature" }, + body: "{}", + }); +} + +function event(type: string, checkoutSession: ReturnType) { + return { + id: `evt_${type}`, + created: 1_752_537_600, + type, + data: { object: checkoutSession }, + }; +} + +describe("Stripe subscription analytics", () => { + let POST: typeof import("@/app/api/webhooks/stripe/route").POST; + + beforeEach(async () => { + vi.clearAllMocks(); + dbChain.select.mockReturnValue(dbChain); + dbChain.from.mockReturnValue(dbChain); + dbChain.where.mockReturnValue(dbChain); + dbChain.limit.mockResolvedValue([dbUser]); + dbChain.update.mockReturnValue(dbChain); + dbChain.set.mockReturnValue(dbChain); + mocks.retrieveCustomer.mockResolvedValue(customer); + mocks.retrieveSubscription.mockResolvedValue(subscription); + POST = (await import("@/app/api/webhooks/stripe/route")).POST; + }); + + it("emits a paid purchase with revenue dimensions and deterministic IDs", async () => { + mocks.constructEvent.mockReturnValue( + event("checkout.session.completed", session()), + ); + expect((await POST(request())).status).toBe(200); + + expect(mocks.product).toHaveBeenCalledWith( + expect.objectContaining({ + eventId: "stripe:evt_checkout.session.completed:purchase_completed", + eventName: "purchase_completed", + occurredAt: "2025-07-15T00:00:00.000Z", + anonymousId: "anonymous-1", + userId: "user-1", + organizationId: "org-1", + properties: expect.objectContaining({ + payment_status: "paid", + amount_total_minor: 2700, + currency: "usd", + unit_amount_minor: 900, + billing_interval: "month", + quantity: 3, + }), + }), + ); + expect(mocks.posthog).toHaveBeenCalledWith( + expect.objectContaining({ + properties: expect.objectContaining({ + $insert_id: + "stripe:evt_checkout.session.completed:purchase_completed", + }), + }), + ); + }); + + it("keeps first-purchase attribution stable on duplicate delivery", async () => { + dbChain.limit.mockResolvedValue([ + { ...dbUser, stripeSubscriptionId: "sub_1" }, + ]); + mocks.constructEvent.mockReturnValue( + event("checkout.session.completed", session()), + ); + + expect((await POST(request())).status).toBe(200); + expect(mocks.product).toHaveBeenCalledWith( + expect.objectContaining({ + properties: expect.objectContaining({ + is_first_purchase: true, + }), + }), + ); + }); + + it("does not count an unpaid checkout as a purchase", async () => { + mocks.constructEvent.mockReturnValue( + event( + "checkout.session.completed", + session({ payment_status: "unpaid" }), + ), + ); + expect((await POST(request())).status).toBe(200); + expect(mocks.product).not.toHaveBeenCalled(); + expect(mocks.posthog).not.toHaveBeenCalled(); + }); + + it("emits when an asynchronous subscription payment settles", async () => { + mocks.constructEvent.mockReturnValue( + event("checkout.session.async_payment_succeeded", session()), + ); + expect((await POST(request())).status).toBe(200); + expect(mocks.product).toHaveBeenCalledWith( + expect.objectContaining({ + eventId: + "stripe:evt_checkout.session.async_payment_succeeded:purchase_completed", + userId: "user-1", + }), + ); + }); + + it("counts a no-payment trial while exposing its zero revenue", async () => { + mocks.retrieveSubscription.mockResolvedValue({ + ...subscription, + status: "trialing", + }); + mocks.constructEvent.mockReturnValue( + event( + "checkout.session.completed", + session({ + payment_status: "no_payment_required", + amount_total: 0, + }), + ), + ); + expect((await POST(request())).status).toBe(200); + expect(mocks.product).toHaveBeenCalledWith( + expect.objectContaining({ + properties: expect.objectContaining({ + payment_status: "no_payment_required", + amount_total_minor: 0, + subscription_status: "trialing", + }), + }), + ); + }); +}); diff --git a/apps/web/actions/analytics/track-user-signed-up.ts b/apps/web/actions/analytics/track-user-signed-up.ts index 2ffb33c6f0a..ce0af3d7b3c 100644 --- a/apps/web/actions/analytics/track-user-signed-up.ts +++ b/apps/web/actions/analytics/track-user-signed-up.ts @@ -1,9 +1,16 @@ "use server"; +import { PRODUCT_ANALYTICS_ANONYMOUS_ID_COOKIE } from "@cap/analytics"; import { db } from "@cap/database"; import { getCurrentUser } from "@cap/database/auth/session"; import { users } from "@cap/database/schema"; import { sql } from "drizzle-orm"; +import { cookies } from "next/headers"; +import { + captureServerProductEvent, + scheduleAfterResponse, +} from "@/lib/analytics/server"; +import { normalizeServerIdentifier } from "@/lib/analytics/server-event"; const SIGNUP_TRACKING_WINDOW_MS = 7 * 24 * 60 * 60 * 1000; @@ -15,6 +22,7 @@ type UserPreferences = { pauseReactions: boolean; }; trackedEvents?: { + product_user_signed_up?: boolean; user_signed_up?: boolean; }; } | null; @@ -53,10 +61,9 @@ export async function checkAndMarkUserSignedUpTracked(): Promise<{ try { const prefs = currentUser.preferences as UserPreferences; const alreadyTracked = Boolean(prefs?.trackedEvents?.user_signed_up); - - if (alreadyTracked) { - return { shouldTrack: false }; - } + const productAlreadyTracked = Boolean( + prefs?.trackedEvents?.product_user_signed_up, + ); const createdAtTime = getCreatedAtTime(currentUser.created_at); @@ -67,6 +74,41 @@ export async function checkAndMarkUserSignedUpTracked(): Promise<{ return { shouldTrack: false }; } + if (!productAlreadyTracked) { + const analyticsAnonymousId = normalizeServerIdentifier( + (await cookies()).get(PRODUCT_ANALYTICS_ANONYMOUS_ID_COOKIE)?.value, + ); + scheduleAfterResponse(async () => { + try { + const captured = await captureServerProductEvent({ + eventId: `signup:${currentUser.id}`, + eventName: "user_signed_up", + occurredAt: new Date(createdAtTime).toISOString(), + anonymousId: analyticsAnonymousId, + platform: "web", + userId: currentUser.id, + organizationId: currentUser.activeOrganizationId, + }); + if (!captured) return; + + await db() + .update(users) + .set({ + preferences: sql`JSON_SET(COALESCE(${users.preferences}, JSON_OBJECT()), '$.trackedEvents.product_user_signed_up', true)`, + }) + .where( + sql`(${users.id} = ${currentUser.id}) AND JSON_CONTAINS(COALESCE(${users.preferences}, JSON_OBJECT()), CAST(true AS JSON), '$.trackedEvents.product_user_signed_up') = 0`, + ); + } catch (error) { + console.error("Failed to capture user_signed_up", error); + } + }); + } + + if (alreadyTracked) { + return { shouldTrack: false }; + } + const result = await db() .update(users) .set({ diff --git a/apps/web/actions/organization/send-invites.ts b/apps/web/actions/organization/send-invites.ts index ad3642d4e3b..b46f8e5193b 100644 --- a/apps/web/actions/organization/send-invites.ts +++ b/apps/web/actions/organization/send-invites.ts @@ -15,6 +15,7 @@ import { serverEnv } from "@cap/env"; import type { Organisation } from "@cap/web-domain"; import { and, eq, inArray } from "drizzle-orm"; import { revalidatePath } from "next/cache"; +import { scheduleServerProductEvent } from "@/lib/analytics/server"; import { provisionOrganizationInvitee } from "@/lib/organization-provisioning"; import { type AssignableOrganizationRole, @@ -211,6 +212,31 @@ export async function sendOrganizationInvites( } } + const failedInviteIds = new Set(failedInvites.map((invite) => invite.id)); + const successfulInvites = inviteRecords.filter( + (invite) => !failedInviteIds.has(invite.id), + ); + const firstSuccessfulInvite = successfulInvites[0]; + if (firstSuccessfulInvite) { + scheduleServerProductEvent({ + eventId: `organization_invites:${firstSuccessfulInvite.id}`, + eventName: "organization_invite_sent", + platform: "web", + userId: user.id, + organizationId, + properties: { + invite_count: successfulInvites.length, + admin_count: successfulInvites.filter( + (invite) => invite.role === "admin", + ).length, + member_count: successfulInvites.filter( + (invite) => invite.role === "member", + ).length, + delivery: "email", + }, + }); + } + revalidatePath("/dashboard/settings/organization"); return { success: true, failedEmails }; diff --git a/apps/web/actions/organization/update-seat-quantity.ts b/apps/web/actions/organization/update-seat-quantity.ts index 942bbb0bf08..89f365161a1 100644 --- a/apps/web/actions/organization/update-seat-quantity.ts +++ b/apps/web/actions/organization/update-seat-quantity.ts @@ -11,6 +11,7 @@ import { stripe } from "@cap/utils"; import type { Organisation } from "@cap/web-domain"; import { eq } from "drizzle-orm"; import { revalidatePath } from "next/cache"; +import { scheduleServerProductEvent } from "@/lib/analytics/server"; import { calculateProSeats } from "@/utils/organization"; async function getOwnerSubscription( @@ -205,6 +206,27 @@ export async function updateSeatQuantity( } revalidatePath("/dashboard/settings/organization"); + const latestInvoice = + typeof updatedSubscription.latest_invoice === "string" + ? updatedSubscription.latest_invoice + : updatedSubscription.latest_invoice?.id; + scheduleServerProductEvent({ + eventId: `seat_quantity:${subscription.id}:${latestInvoice ?? updatedSubscription.current_period_start}:${newQuantity}`, + eventName: "seat_quantity_changed", + platform: "web", + userId: user.id, + organizationId, + properties: { + previous_quantity: currentQuantity, + new_quantity: newQuantity, + quantity_delta: newQuantity - currentQuantity, + direction: isSeatIncrease ? "increase" : "decrease", + price_id: subscriptionItem.price.id, + unit_amount_minor: subscriptionItem.price.unit_amount, + currency: subscriptionItem.price.currency, + billing_interval: subscriptionItem.price.recurring?.interval, + }, + }); return { success: true, newQuantity }; } diff --git a/apps/web/app/Layout/PosthogIdentify.tsx b/apps/web/app/Layout/PosthogIdentify.tsx index 851143c9a4d..ef074fa5f06 100644 --- a/apps/web/app/Layout/PosthogIdentify.tsx +++ b/apps/web/app/Layout/PosthogIdentify.tsx @@ -5,7 +5,7 @@ import { checkAndMarkUserSignedUpTracked } from "@/actions/analytics/track-user- import { identifyUser, initAnonymousUser, - trackEvent, + trackExternalEvent, } from "../utils/analytics"; import { useCurrentUser } from "./AuthContext"; @@ -30,9 +30,8 @@ function Inner() { (async () => { const { shouldTrack } = await checkAndMarkUserSignedUpTracked(); if (shouldTrack) { - trackEvent("user_signed_up"); + trackExternalEvent("user_signed_up"); } - trackEvent("user_signed_in"); })(); } }, [user]); diff --git a/apps/web/app/Layout/PosthogPageView.tsx b/apps/web/app/Layout/PosthogPageView.tsx index 41996db03bb..0838ced327a 100644 --- a/apps/web/app/Layout/PosthogPageView.tsx +++ b/apps/web/app/Layout/PosthogPageView.tsx @@ -1,28 +1,23 @@ -// app/PostHogPageView.tsx "use client"; -import { usePathname, useSearchParams } from "next/navigation"; +import { usePathname } from "next/navigation"; import { usePostHog } from "posthog-js/react"; -import { Suspense, useEffect, useMemo } from "react"; +import { useEffect } from "react"; +import { shouldCaptureProductPageView } from "../utils/product-analytics"; let lastTrackedUrl: string | null = null; function PostHogPageView(): null { const pathname = usePathname(); - const searchParams = useSearchParams(); const posthog = usePostHog(); - const search = useMemo(() => searchParams?.toString() ?? "", [searchParams]); useEffect(() => { - if (!pathname || !posthog) { + if (!pathname || !posthog || !shouldCaptureProductPageView(pathname)) { return; } try { - let url = window.location.origin + pathname; - if (search) { - url = `${url}?${search}`; - } + const url = window.location.origin + pathname; if (lastTrackedUrl === url) { return; @@ -33,18 +28,9 @@ function PostHogPageView(): null { } catch (error) { console.error("Error capturing pageview:", error); } - }, [pathname, search, posthog]); + }, [pathname, posthog]); return null; } -// Wrap this in Suspense to avoid the `useSearchParams` usage above -// from de-opting the whole app into client-side rendering -// See: https://nextjs.org/docs/messages/deopted-into-client-rendering -export default function SuspendedPostHogPageView() { - return ( - - - - ); -} +export default PostHogPageView; diff --git a/apps/web/app/Layout/ProductAnalyticsPageView.tsx b/apps/web/app/Layout/ProductAnalyticsPageView.tsx new file mode 100644 index 00000000000..5bccfd96b9d --- /dev/null +++ b/apps/web/app/Layout/ProductAnalyticsPageView.tsx @@ -0,0 +1,29 @@ +"use client"; + +import { usePathname } from "next/navigation"; +import { useEffect } from "react"; +import { + captureProductPageView, + shouldCaptureProductPageView, +} from "../utils/product-analytics"; + +let lastCapturedPathname: string | undefined; + +export function ProductAnalyticsPageView() { + const pathname = usePathname(); + + useEffect(() => { + if ( + !pathname || + pathname === lastCapturedPathname || + !shouldCaptureProductPageView(pathname) + ) { + return; + } + + lastCapturedPathname = pathname; + captureProductPageView(); + }, [pathname]); + + return null; +} diff --git a/apps/web/app/Layout/providers.tsx b/apps/web/app/Layout/providers.tsx index 79ff4a0675e..8076fe1bae8 100644 --- a/apps/web/app/Layout/providers.tsx +++ b/apps/web/app/Layout/providers.tsx @@ -44,17 +44,15 @@ export function PostHogProvider({ if (!host) return undefined; const base: CapPostHogConfig = { api_host: host, + autocapture: false, capture_pageview: false, - capture_pageleave: true, + capture_pageleave: false, + disable_session_recording: true, bootstrap: initialBootstrap.current?.distinctID ? initialBootstrap.current : undefined, }; - if (process.env.NEXT_PUBLIC_POSTHOG_DISABLE_SESSION_RECORDING === "true") { - base.disable_session_recording = true; - } - return base; }, [host]); diff --git a/apps/web/app/api/desktop/[...route]/root.ts b/apps/web/app/api/desktop/[...route]/root.ts index 694119771b9..b6f4e798242 100644 --- a/apps/web/app/api/desktop/[...route]/root.ts +++ b/apps/web/app/api/desktop/[...route]/root.ts @@ -8,7 +8,7 @@ import { organizations, users, } from "@cap/database/schema"; -import { buildEnv, serverEnv } from "@cap/env"; +import { serverEnv } from "@cap/env"; import { stripe, userIsPro } from "@cap/utils"; import { OrganizationBrandingPatchBody } from "@cap/web-api-contract"; import { ImageUploads } from "@cap/web-backend"; @@ -17,9 +17,12 @@ import { zValidator } from "@hono/zod-validator"; import { and, eq, isNull } from "drizzle-orm"; import { Effect, Option } from "effect"; import { type Context, Hono } from "hono"; -import { PostHog } from "posthog-node"; import type Stripe from "stripe"; import { z } from "zod"; +import { + scheduleLegacyPostHogEvent, + scheduleServerProductEvent, +} from "@/lib/analytics/server"; import { runPromise } from "@/lib/server"; import { withAuth, withOptionalAuth } from "../../utils"; import { @@ -756,31 +759,37 @@ app.post( success_url: `${serverEnv().WEB_URL}/dashboard/caps?upgrade=true`, cancel_url: `${serverEnv().WEB_URL}/pricing`, allow_promotion_codes: true, - metadata: { platform: "desktop", dubCustomerId: user.id }, + metadata: { + platform: "desktop", + dubCustomerId: user.id, + analyticsIsFirstPurchase: user.stripeSubscriptionId ? "false" : "true", + }, }); if (checkoutSession.url) { console.log("[POST] Checkout session created successfully"); + scheduleServerProductEvent({ + eventId: `checkout:${checkoutSession.id}`, + eventName: "checkout_started", + platform: "desktop", + userId: user.id, + organizationId: user.activeOrganizationId, + properties: { + price_id: priceId, + quantity: 1, + }, + }); - try { - const ph = new PostHog(buildEnv.NEXT_PUBLIC_POSTHOG_KEY || "", { - host: buildEnv.NEXT_PUBLIC_POSTHOG_HOST || "", - }); - - ph.capture({ - distinctId: user.id, - event: "checkout_started", - properties: { - price_id: priceId, - quantity: 1, - platform: "desktop", - }, - }); - - await ph.shutdown(); - } catch (e) { - console.error("Failed to capture checkout_started in PostHog", e); - } + scheduleLegacyPostHogEvent({ + distinctId: user.id, + eventName: "checkout_started", + properties: { + $insert_id: `checkout:${checkoutSession.id}`, + price_id: priceId, + quantity: 1, + platform: "desktop", + }, + }); return c.json({ url: checkoutSession.url }); } diff --git a/apps/web/app/api/events/route.ts b/apps/web/app/api/events/route.ts new file mode 100644 index 00000000000..7e73d92e60c --- /dev/null +++ b/apps/web/app/api/events/route.ts @@ -0,0 +1,181 @@ +import { + createProductEventRows, + PRODUCT_ANALYTICS_LIMITS, +} from "@cap/analytics"; +import { serverEnv } from "@cap/env"; +import { + ProductAnalytics, + resolveProductAnalyticsActor, +} from "@cap/web-backend"; +import { + HttpApi, + HttpApiBuilder, + HttpApiEndpoint, + HttpApiError, + HttpApiGroup, + HttpApiSchema, + HttpServerRequest, +} from "@effect/platform"; +import { Effect, Layer, Schema } from "effect"; +import { + readProductAnalyticsBrowserToken, + readProductAnalyticsBrowserTokenClaims, +} from "@/lib/analytics/browser-token"; +import { + getProductAnalyticsRateLimitKey, + hasExpectedBrowserAnalyticsMetadata, + isAllowedAnonymousBrowserProductEvent, + isAuthenticatedAnalyticsRequestCandidate, + normalizeGeoHeader, + normalizeProductEventBatch, + ProductAnalyticsRateLimiter, +} from "@/lib/analytics/request"; +import { isRateLimited, RATE_LIMIT_IDS } from "@/lib/rate-limit"; +import { apiToHandler } from "@/lib/server"; +import { allowedOrigins } from "@/utils/cors"; + +class RateLimited extends Schema.TaggedError()( + "RateLimited", + {}, + HttpApiSchema.annotations({ status: 429 }), +) {} + +class Api extends HttpApi.make("ProductAnalyticsApi").add( + HttpApiGroup.make("events").add( + HttpApiEndpoint.post("capture", "/api/events") + .setPayload( + Schema.Struct({ + events: Schema.Array(Schema.Unknown).pipe( + Schema.minItems(1), + Schema.maxItems(PRODUCT_ANALYTICS_LIMITS.batchSize), + ), + }), + ) + .addSuccess(Schema.Struct({ accepted: Schema.Number })) + .addError(HttpApiError.BadRequest) + .addError(HttpApiError.ServiceUnavailable) + .addError(RateLimited), + ), +) {} + +const RequestHeaders = Schema.Struct({ + authorization: Schema.optional(Schema.String), + "content-length": Schema.optional(Schema.String), + cookie: Schema.optional(Schema.String), + "sec-fetch-site": Schema.optional(Schema.String), + origin: Schema.optional(Schema.String), + "x-vercel-ip-country": Schema.optional(Schema.String), + "x-vercel-ip-country-region": Schema.optional(Schema.String), + "x-vercel-ip-city": Schema.optional(Schema.String), + "x-vercel-forwarded-for": Schema.optional(Schema.String), +}); + +const fallbackRateLimiter = new ProductAnalyticsRateLimiter(); + +const ApiLive = HttpApiBuilder.api(Api).pipe( + Layer.provide( + HttpApiBuilder.group(Api, "events", (handlers) => + Effect.gen(function* () { + const analytics = yield* ProductAnalytics; + + return handlers.handle("capture", ({ payload }) => + Effect.gen(function* () { + const headers = yield* HttpServerRequest.schemaHeaders( + RequestHeaders, + ).pipe(Effect.mapError(() => new HttpApiError.BadRequest())); + const requestMetadata = { + authorization: headers.authorization, + contentLength: headers["content-length"], + origin: headers.origin, + secFetchSite: headers["sec-fetch-site"], + }; + const browserClaims = + hasExpectedBrowserAnalyticsMetadata( + requestMetadata, + allowedOrigins, + ) && + readProductAnalyticsBrowserTokenClaims( + readProductAnalyticsBrowserToken(headers.cookie), + serverEnv().NEXTAUTH_SECRET, + ); + if ( + !browserClaims && + !isAuthenticatedAnalyticsRequestCandidate(requestMetadata) + ) { + return yield* Effect.fail(new HttpApiError.BadRequest()); + } + + if ( + fallbackRateLimiter.isRateLimited( + getProductAnalyticsRateLimitKey({ + trustedVercelProxy: process.env.VERCEL === "1", + xVercelForwardedFor: headers["x-vercel-forwarded-for"], + }), + ) + ) { + return yield* Effect.fail(new RateLimited()); + } + + if ( + yield* Effect.promise(() => + isRateLimited(RATE_LIMIT_IDS.PRODUCT_ANALYTICS_EVENTS), + ) + ) { + return yield* Effect.fail(new RateLimited()); + } + + const events = normalizeProductEventBatch(payload.events); + if (!events) { + return yield* Effect.fail(new HttpApiError.BadRequest()); + } + + const actor = yield* resolveProductAnalyticsActor; + if ( + !actor && + (!browserClaims || + !events.every((event) => + isAllowedAnonymousBrowserProductEvent( + event, + browserClaims.anonymousId, + ), + )) + ) { + return yield* Effect.fail(new HttpApiError.BadRequest()); + } + const rows = createProductEventRows(events, { + receivedAt: new Date().toISOString(), + source: "client", + userId: actor?.userId, + organizationId: actor?.organizationId, + country: normalizeGeoHeader(headers["x-vercel-ip-country"]), + region: normalizeGeoHeader(headers["x-vercel-ip-country-region"]), + city: normalizeGeoHeader(headers["x-vercel-ip-city"], true), + }); + + yield* analytics + .append(rows) + .pipe( + Effect.catchTag("ProductAnalyticsError", (error) => + Effect.logWarning( + "Product analytics ingestion failed", + error, + ).pipe( + Effect.andThen( + Effect.fail(new HttpApiError.ServiceUnavailable()), + ), + ), + ), + ); + + return { accepted: rows.length }; + }), + ); + }), + ), + ), +); + +const handler = apiToHandler(ApiLive); + +export const POST = handler; +export const OPTIONS = handler; diff --git a/apps/web/app/api/invite/accept/route.ts b/apps/web/app/api/invite/accept/route.ts index 4f99e2022a8..1352bd25390 100644 --- a/apps/web/app/api/invite/accept/route.ts +++ b/apps/web/app/api/invite/accept/route.ts @@ -9,6 +9,10 @@ import { } from "@cap/database/schema"; import { and, eq } from "drizzle-orm"; import { type NextRequest, NextResponse } from "next/server"; +import { + readAnalyticsAnonymousId, + scheduleServerProductEvent, +} from "@/lib/analytics/server"; import { normalizeAssignableOrganizationRole } from "@/lib/permissions/roles"; import { calculateProSeats } from "@/utils/organization"; @@ -33,6 +37,10 @@ export async function POST(request: NextRequest) { } try { + let joinedMemberId: string | undefined; + let joinedOrganizationId: string | undefined; + let joinedRole: string | undefined; + let assignedProSeat = false; await db().transaction(async (tx) => { const [invite] = await tx .select() @@ -72,6 +80,9 @@ export async function POST(request: NextRequest) { role, }); memberId = newId; + joinedMemberId = newId; + joinedOrganizationId = invite.organizationId; + joinedRole = role; } const [org] = await tx @@ -111,6 +122,7 @@ export async function POST(request: NextRequest) { }); if (proSeatsRemaining > 0) { + assignedProSeat = true; await tx .update(organizationMembers) .set({ hasProSeat: true }) @@ -148,6 +160,21 @@ export async function POST(request: NextRequest) { .where(eq(organizationInvites.id, inviteId)); }); + if (joinedMemberId && joinedOrganizationId) { + scheduleServerProductEvent({ + eventId: `organization_member:${joinedMemberId}:joined`, + eventName: "organization_member_joined", + anonymousId: readAnalyticsAnonymousId(request), + platform: "web", + userId: user.id, + organizationId: joinedOrganizationId, + properties: { + role: joinedRole, + assigned_pro_seat: assignedProSeat, + }, + }); + } + return NextResponse.json({ success: true }); } catch (error) { if (error instanceof Error) { diff --git a/apps/web/app/api/settings/billing/guest-checkout/route.ts b/apps/web/app/api/settings/billing/guest-checkout/route.ts index a8e5dc63f30..775225e834a 100644 --- a/apps/web/app/api/settings/billing/guest-checkout/route.ts +++ b/apps/web/app/api/settings/billing/guest-checkout/route.ts @@ -1,11 +1,18 @@ -import { buildEnv, serverEnv } from "@cap/env"; +import { randomUUID } from "node:crypto"; +import { serverEnv } from "@cap/env"; import { stripe } from "@cap/utils"; import type { NextRequest } from "next/server"; -import { PostHog } from "posthog-node"; +import { + readAnalyticsAnonymousId, + scheduleLegacyPostHogEvent, + scheduleServerProductEvent, +} from "@/lib/analytics/server"; export async function POST(request: NextRequest) { console.log("Starting guest checkout process"); const { priceId, quantity } = await request.json(); + const analyticsAnonymousId = readAnalyticsAnonymousId(request); + const checkoutAnonymousId = analyticsAnonymousId ?? `guest:${randomUUID()}`; console.log("Received guest checkout request:", { priceId, quantity }); @@ -25,32 +32,35 @@ export async function POST(request: NextRequest) { metadata: { platform: "web", guestCheckout: "true", + analyticsIsFirstPurchase: "true", + analyticsAnonymousId: checkoutAnonymousId, }, }); if (checkoutSession.url) { console.log("Successfully created guest checkout session"); + scheduleServerProductEvent({ + eventId: `checkout:${checkoutSession.id}`, + eventName: "guest_checkout_started", + anonymousId: checkoutAnonymousId, + platform: "web", + properties: { + price_id: priceId, + quantity: quantity || 1, + }, + }); - try { - const ph = new PostHog(buildEnv.NEXT_PUBLIC_POSTHOG_KEY || "", { - host: buildEnv.NEXT_PUBLIC_POSTHOG_HOST || "", - }); - - ph.capture({ - distinctId: `guest-${checkoutSession.id}`, - event: "guest_checkout_started", - properties: { - price_id: priceId, - quantity: quantity || 1, - platform: "web", - session_id: checkoutSession.id, - }, - }); - - await ph.shutdown(); - } catch (e) { - console.error("Failed to capture guest_checkout_started in PostHog", e); - } + scheduleLegacyPostHogEvent({ + distinctId: checkoutAnonymousId, + eventName: "guest_checkout_started", + properties: { + $insert_id: `checkout:${checkoutSession.id}`, + price_id: priceId, + quantity: quantity || 1, + platform: "web", + session_id: checkoutSession.id, + }, + }); return Response.json({ url: checkoutSession.url }, { status: 200 }); } diff --git a/apps/web/app/api/settings/billing/subscribe/route.ts b/apps/web/app/api/settings/billing/subscribe/route.ts index ebb9b816bad..c702471c621 100644 --- a/apps/web/app/api/settings/billing/subscribe/route.ts +++ b/apps/web/app/api/settings/billing/subscribe/route.ts @@ -1,17 +1,22 @@ import { db } from "@cap/database"; import { getCurrentUser } from "@cap/database/auth/session"; import { users } from "@cap/database/schema"; -import { buildEnv, serverEnv } from "@cap/env"; +import { serverEnv } from "@cap/env"; import { stripe, userIsPro } from "@cap/utils"; import { eq } from "drizzle-orm"; import type { NextRequest } from "next/server"; -import { PostHog } from "posthog-node"; import type Stripe from "stripe"; +import { + readAnalyticsAnonymousId, + scheduleLegacyPostHogEvent, + scheduleServerProductEvent, +} from "@/lib/analytics/server"; export async function POST(request: NextRequest) { const user = await getCurrentUser(); let customerId = user?.stripeCustomerId; const { priceId, quantity, isOnBoarding } = await request.json(); + const analyticsAnonymousId = readAnalyticsAnonymousId(request); if (!priceId) { console.error("Price ID not found"); @@ -78,29 +83,36 @@ export async function POST(request: NextRequest) { platform: "web", dubCustomerId: user.id, isOnBoarding: isOnBoarding ? "true" : "false", + analyticsIsFirstPurchase: user.stripeSubscriptionId ? "false" : "true", + ...(analyticsAnonymousId ? { analyticsAnonymousId } : {}), }, }); if (checkoutSession.url) { - try { - const ph = new PostHog(buildEnv.NEXT_PUBLIC_POSTHOG_KEY || "", { - host: buildEnv.NEXT_PUBLIC_POSTHOG_HOST || "", - }); - - ph.capture({ - distinctId: user.id, - event: "checkout_started", - properties: { - price_id: priceId, - quantity: quantity, - platform: "web", - }, - }); + scheduleServerProductEvent({ + eventId: `checkout:${checkoutSession.id}`, + eventName: "checkout_started", + anonymousId: analyticsAnonymousId, + platform: "web", + userId: user.id, + organizationId: user.activeOrganizationId, + properties: { + price_id: priceId, + quantity: quantity ?? 1, + is_onboarding: Boolean(isOnBoarding), + }, + }); - await ph.shutdown(); - } catch (e) { - console.error("Failed to capture checkout_started in PostHog", e); - } + scheduleLegacyPostHogEvent({ + distinctId: user.id, + eventName: "checkout_started", + properties: { + $insert_id: `checkout:${checkoutSession.id}`, + price_id: priceId, + quantity, + platform: "web", + }, + }); return Response.json({ url: checkoutSession.url }, { status: 200 }); } diff --git a/apps/web/app/api/webhooks/stripe/route.ts b/apps/web/app/api/webhooks/stripe/route.ts index e0f3dff4f06..ce7ee372ada 100644 --- a/apps/web/app/api/webhooks/stripe/route.ts +++ b/apps/web/app/api/webhooks/stripe/route.ts @@ -1,13 +1,16 @@ import { db } from "@cap/database"; import { nanoId } from "@cap/database/helpers"; import { developerCreditTransactions, users } from "@cap/database/schema"; -import { buildEnv, serverEnv } from "@cap/env"; +import { serverEnv } from "@cap/env"; import { stripe } from "@cap/utils"; import { Organisation, User } from "@cap/web-domain"; import { and, eq } from "drizzle-orm"; import { NextResponse } from "next/server"; -import { PostHog } from "posthog-node"; import type Stripe from "stripe"; +import { + scheduleLegacyPostHogEvent, + scheduleServerProductEvent, +} from "@/lib/analytics/server"; import { addCreditsToAccount } from "@/lib/developer-credits"; const relevantEvents = new Set([ @@ -17,6 +20,98 @@ const relevantEvents = new Set([ "customer.subscription.deleted", ]); +type PurchaseAnalyticsUser = Pick< + typeof users.$inferSelect, + "id" | "activeOrganizationId" +>; + +function isSettledSubscriptionPurchase( + session: Stripe.Checkout.Session, + subscription: Stripe.Subscription, +) { + return ( + (session.payment_status === "paid" || + session.payment_status === "no_payment_required") && + (subscription.status === "active" || subscription.status === "trialing") + ); +} + +function scheduleSubscriptionPurchaseEvents({ + eventId, + occurredAt, + session, + subscription, + inviteQuota, + user, + isFirstPurchase, +}: { + eventId: string; + occurredAt: string; + session: Stripe.Checkout.Session; + subscription: Stripe.Subscription; + inviteQuota: number; + user?: PurchaseAnalyticsUser; + isFirstPurchase: boolean; +}) { + const isGuestCheckout = session.metadata?.guestCheckout === "true"; + const platform = + session.metadata?.platform === "desktop" + ? "desktop" + : session.metadata?.platform === "web" + ? "web" + : "server"; + const anonymousId = session.metadata?.analyticsAnonymousId; + const distinctId = user?.id ?? anonymousId ?? `stripe:${session.id}`; + const price = subscription.items.data[0]?.price; + const revenueProperties = { + payment_status: session.payment_status, + subscription_status: subscription.status, + amount_total_minor: session.amount_total, + amount_subtotal_minor: session.amount_subtotal, + discount_amount_minor: session.total_details?.amount_discount, + currency: session.currency, + unit_amount_minor: price?.unit_amount, + billing_interval: price?.recurring?.interval, + billing_interval_count: price?.recurring?.interval_count, + }; + + scheduleServerProductEvent({ + eventId: `stripe:${eventId}:purchase_completed`, + eventName: "purchase_completed", + occurredAt, + anonymousId, + platform, + userId: user?.id, + organizationId: user?.activeOrganizationId, + properties: { + ...revenueProperties, + invite_quota: inviteQuota, + price_id: price?.id, + quantity: inviteQuota, + is_onboarding: session.metadata?.isOnBoarding === "true", + is_first_purchase: isFirstPurchase, + is_guest_checkout: isGuestCheckout, + }, + }); + + scheduleLegacyPostHogEvent({ + distinctId, + eventName: "purchase_completed", + properties: { + $insert_id: `stripe:${eventId}:purchase_completed`, + subscription_id: subscription.id, + ...revenueProperties, + invite_quota: inviteQuota, + price_id: price?.id, + quantity: inviteQuota, + is_onboarding: session.metadata?.isOnBoarding === "true", + platform: platform === "server" ? "unknown" : platform, + is_first_purchase: isFirstPurchase, + is_guest_checkout: isGuestCheckout, + }, + }); +} + async function grantDeveloperCredits( session: Stripe.Checkout.Session, ): Promise { @@ -329,39 +424,17 @@ export const POST = async (req: Request) => { console.log("Successfully updated user in database"); - try { - const serverPostHog = new PostHog( - buildEnv.NEXT_PUBLIC_POSTHOG_KEY || "", - { host: buildEnv.NEXT_PUBLIC_POSTHOG_HOST || "" }, - ); - - const isFirstPurchase = !dbUser.stripeSubscriptionId; - const isGuestCheckout = session.metadata?.guestCheckout === "true"; - serverPostHog.capture({ - distinctId: dbUser.id, - event: "purchase_completed", - properties: { - subscription_id: subscription.id, - subscription_status: subscription.status, - invite_quota: inviteQuota, - price_id: subscription.items.data[0]?.price.id, - quantity: inviteQuota, - is_onboarding: session.metadata?.isOnBoarding === "true", - platform: - session.metadata?.platform === "desktop" - ? "desktop" - : session.metadata?.platform === "web" - ? "web" - : "unknown", - is_first_purchase: isFirstPurchase, - is_guest_checkout: isGuestCheckout, - }, + if (isSettledSubscriptionPurchase(session, subscription)) { + scheduleSubscriptionPurchaseEvents({ + eventId: event.id, + occurredAt: new Date(event.created * 1000).toISOString(), + session, + subscription, + inviteQuota, + user: dbUser, + isFirstPurchase: + session.metadata?.analyticsIsFirstPurchase === "true", }); - - await serverPostHog.shutdown(); - console.log("Successfully tracked purchase event in PostHog"); - } catch (error) { - console.error("Error tracking purchase in PostHog:", error); } } @@ -374,6 +447,45 @@ export const POST = async (req: Request) => { if (session.metadata?.type === "developer_credits") { return await grantDeveloperCredits(session); } + + if (typeof session.subscription === "string") { + const subscription = await stripe().subscriptions.retrieve( + session.subscription, + ); + if (isSettledSubscriptionPurchase(session, subscription)) { + const inviteQuota = subscription.items.data.reduce( + (total, item) => total + (item.quantity || 1), + 0, + ); + let dbUser: typeof users.$inferSelect | null = null; + if (typeof session.customer === "string") { + const customer = await stripe().customers.retrieve( + session.customer, + ); + if (!customer.deleted) { + const userId = customer.metadata.userId + ? User.UserId.make(customer.metadata.userId) + : undefined; + dbUser = await findUserWithRetry( + customer.email ?? "", + userId, + 1, + ); + } + } + + scheduleSubscriptionPurchaseEvents({ + eventId: event.id, + occurredAt: new Date(event.created * 1000).toISOString(), + session, + subscription, + inviteQuota, + ...(dbUser ? { user: dbUser } : {}), + isFirstPurchase: + session.metadata?.analyticsIsFirstPurchase === "true", + }); + } + } } if (event.type === "customer.subscription.updated") { diff --git a/apps/web/app/layout.tsx b/apps/web/app/layout.tsx index f8c3bbcff49..0a446ced264 100644 --- a/apps/web/app/layout.tsx +++ b/apps/web/app/layout.tsx @@ -3,6 +3,7 @@ import type { Metadata } from "next"; import localFont from "next/font/local"; import Script from "next/script"; import type { PropsWithChildren } from "react"; +import { ProductAnalyticsPageView } from "./Layout/ProductAnalyticsPageView"; const defaultFont = localFont({ src: [ @@ -84,6 +85,7 @@ export default function RootLayout({ children }: PropsWithChildren) {