SQL for Growth Analytics Questions

WITH first_purchase AS (
  SELECT
    u.user_id,
    DATE_TRUNC('week', u.signup_ts) AS cohort_week,
    MIN(p.purchase_ts) AS first_purchase_ts
  FROM users u
  LEFT JOIN purchases p
    ON p.user_id = u.user_id
    AND p.purchase_ts >= u.signup_ts
  GROUP BY 1,2
),
deltas AS (
  SELECT
    cohort_week,
    EXTRACT(EPOCH FROM (first_purchase_ts - signup_ts))/3600.0 AS hours_to_first_purchase
  FROM first_purchase f
  JOIN users u ON u.user_id = f.user_id
  WHERE first_purchase_ts IS NOT NULL
)
SELECT
  cohort_week,
  percentile_disc(0.25) WITHIN GROUP (ORDER BY hours_to_first_purchase) AS p25_hours,
  percentile_disc(0.50) WITHIN GROUP (ORDER BY hours_to_first_purchase) AS p50_hours,
  percentile_disc(0.75) WITHIN GROUP (ORDER BY hours_to_first_purchase) AS p75_hours
FROM deltas
GROUP BY cohort_week
ORDER BY cohort_week;

WITH first_purchase AS (
  SELECT user_id, DATE_TRUNC('week', signup_ts) AS cohort_week, MIN(purchase_ts) AS first_purchase_ts
  FROM users u
  LEFT JOIN purchases p USING (user_id)
  WHERE p.purchase_ts >= u.signup_ts
  GROUP BY user_id, cohort_week
),
deltas AS (
  SELECT cohort_week,
         EXTRACT(EPOCH FROM (first_purchase_ts - signup_ts))/3600.0 AS hours_to_first_purchase
  FROM first_purchase f JOIN users u USING (user_id)
  WHERE first_purchase_ts IS NOT NULL
)
SELECT
  cohort_week,
  approx_percentile(hours_to_first_purchase, ARRAY[0.25,0.5,0.75]) AS percentiles_hours
FROM deltas
GROUP BY cohort_week
ORDER BY cohort_week;

WITH user_cohorts AS (
  SELECT
    user_id,
    DATE_TRUNC('month', created_at)::date AS cohort_month
  FROM users
),

purchases_enriched AS (
  SELECT
    u.cohort_month,
    p.user_id,
    p.amount,
    -- month_index: 0 for purchases in same calendar month as created_at
    DATE_PART('year', p.occurred_at)::int * 12 + DATE_PART('month', p.occurred_at)::int
      - (DATE_PART('year', u.cohort_month)::int * 12 + DATE_PART('month', u.cohort_month)::int)
      AS month_index
  FROM purchases p
  JOIN user_cohorts u ON p.user_id = u.user_id
),

discounted_by_month AS (
  -- keep only months 0..11 and compute PV
  SELECT
    cohort_month,
    user_id,
    month_index,
    amount / POWER(1.05, month_index) AS discounted_amount
  FROM purchases_enriched
  WHERE month_index BETWEEN 0 AND 11
),

cohort_revenue AS (
  -- sum discounted revenue per cohort
  SELECT
    cohort_month,
    SUM(discounted_amount) AS total_discounted_revenue
  FROM discounted_by_month
  GROUP BY cohort_month
),

cohort_size AS (
  -- count users per cohort (include users with no purchases)
  SELECT
    cohort_month,
    COUNT(*)::numeric AS cohort_size
  FROM user_cohorts
  GROUP BY cohort_month
)

SELECT
  c.cohort_month,
  cs.cohort_size,
  COALESCE(cr.total_discounted_revenue, 0) / cs.cohort_size AS avg_discounted_ltv_12m
FROM cohort_size cs
LEFT JOIN cohort_revenue cr ON cs.cohort_month = cr.cohort_month
ORDER BY c.cohort_month;

-- deterministic: email_hash or login_user_id
WITH events AS (
  SELECT event_id, device_id, email_hash, login_user_id, ts
  FROM raw_events
),
det_matches AS (
  SELECT e.*, COALESCE(login_user_id, email_hash) AS deterministic_id
  FROM events e
  WHERE login_user_id IS NOT NULL OR email_hash IS NOT NULL
)
SELECT * FROM det_matches;

WITH unmatched AS (
  SELECT e.* FROM raw_events e
  LEFT JOIN det_matches d ON e.event_id = d.event_id
  WHERE d.event_id IS NULL
),
candidates AS (
  SELECT u.event_id, a.user_id AS candidate_user,
    (CASE WHEN u.device_fingerprint = a.device_fingerprint THEN 0.5 ELSE 0 END) +
    (CASE WHEN inet_nton(u.ip) >> 16 = inet_nton(a.ip) >> 16 THEN 0.2 ELSE 0 END) +
    (CASE WHEN regexp_replace(u.user_agent,'[^a-z]','','i') = regexp_replace(a.user_agent,'[^a-z]','','i') THEN 0.2 ELSE 0 END) +
    (CASE WHEN ABS(EXTRACT(EPOCH FROM u.ts - a.ts)) < 3600 THEN 0.1 ELSE 0 END) AS score
  FROM unmatched u
  JOIN known_id_events a ON TRUE -- restrict by time window, geo, etc.
),
best AS (
  SELECT event_id, candidate_user, score,
    ROW_NUMBER() OVER (PARTITION BY event_id ORDER BY score DESC) rn
  FROM candidates
)
SELECT event_id, candidate_user AS probabilistic_user
FROM best
WHERE rn = 1 AND score >= 0.6;

WITH trials AS (
  -- extract each user's trial_start with the country at that time
  SELECT
    user_id,
    DATE(occurred_at) AS trial_date,
    country AS trial_country,
    MIN(occurred_at) FILTER (WHERE event_name = 'trial_start') OVER (PARTITION BY user_id) AS first_trial_ts
  FROM events
  WHERE event_name = 'trial_start'
),
distinct_trials AS (
  -- one row per user per trial (use the first trial_start if multiple)
  SELECT DISTINCT ON (user_id) user_id, first_trial_ts AS trial_ts, DATE(first_trial_ts) AS trial_date, trial_country
  FROM trials
  ORDER BY user_id, first_trial_ts
),
pays AS (
  -- find first pay after trial for each user (if any)
  SELECT
    t.user_id,
    t.trial_date,
    t.trial_country,
    MIN(e.occurred_at) AS pay_ts
  FROM distinct_trials t
  LEFT JOIN events e
    ON e.user_id = t.user_id
    AND e.event_name = 'pay'
    AND e.occurred_at >= t.trial_ts
  GROUP BY t.user_id, t.trial_date, t.trial_country
),
daily_calendar AS (
  -- generate calendar of days for reporting
  SELECT generate_series(min(trial_date), max(trial_date), interval '1 day')::date AS day
  FROM distinct_trials
),
cohort_counts AS (
  -- trials per day+country
  SELECT trial_date AS day, trial_country AS country, COUNT(*) AS trials
  FROM distinct_trials
  GROUP BY 1,2
),
cohort_pays AS (
  -- pays attributed to trial_date+country
  SELECT trial_date AS day, trial_country AS country, COUNT(*) AS pays
  FROM pays
  WHERE pay_ts IS NOT NULL
  GROUP BY 1,2
)
SELECT
  d.day,
  c.country,
  COALESCE(SUM(c.trials) OVER (PARTITION BY c.country ORDER BY d.day
       ROWS BETWEEN 6 PRECEDING AND CURRENT ROW),0) AS trials_7d,
  COALESCE(SUM(cp.pays) OVER (PARTITION BY c.country ORDER BY d.day
       ROWS BETWEEN 6 PRECEDING AND CURRENT ROW),0) AS pays_7d,
  CASE WHEN COALESCE(SUM(c.trials) OVER (PARTITION BY c.country ORDER BY d.day
       ROWS BETWEEN 6 PRECEDING AND CURRENT ROW),0) = 0
       THEN NULL
       ELSE ROUND(
         COALESCE(SUM(cp.pays) OVER (PARTITION BY c.country ORDER BY d.day
           ROWS BETWEEN 6 PRECEDING AND CURRENT ROW)::numeric
         / SUM(c.trials) OVER (PARTITION BY c.country ORDER BY d.day
           ROWS BETWEEN 6 PRECEDING AND CURRENT ROW), 4)
  END AS conversion_rate_7d
FROM
  (SELECT day FROM daily_calendar) d
CROSS JOIN
  (SELECT DISTINCT trial_country AS country FROM distinct_trials) countries
LEFT JOIN cohort_counts c ON c.day = d.day AND c.country = countries.country
LEFT JOIN cohort_pays cp ON cp.day = d.day AND cp.country = countries.country
ORDER BY d.day, countries.country;

-- 1. Build staging aggregates for window [run_date-6 .. run_date]
WITH events_window AS (
  SELECT *
  FROM stg_events
  WHERE event_date BETWEEN DATE_SUB(@run_date, INTERVAL 6 DAY) AND @run_date
),

-- deterministic dedupe: keep latest ingestion or canonical row per raw_event_id
deduped AS (
  SELECT *
  FROM (
    SELECT e.*,
           ROW_NUMBER() OVER (PARTITION BY raw_event_id ORDER BY ingestion_ts DESC) rn
    FROM events_window e
  ) t
  WHERE rn = 1
),

-- compute per-cohort aggregates and checksum
stg_cohort_agg AS (
  SELECT
    DATE(event_time) AS cohort_date,
    user_id,
    metric_id,
    COUNT(*) AS events_count,
    SUM(value) AS value_sum,
    -- deterministic checksum of inputs for idempotency
    MD5(CONCAT(user_id, '|', metric_id, '|', DATE(event_time), '|', CAST(COUNT(*) AS STRING), '|', CAST(SUM(value) AS STRING))) AS agg_checksum
  FROM deduped
  GROUP BY 1,2,3
)

-- 2. MERGE into partitioned target table
MERGE INTO analytics.cohort_agg T
USING stg_cohort_agg S
  ON T.cohort_date = S.cohort_date
 AND T.user_id = S.user_id
 AND T.metric_id = S.metric_id
WHEN MATCHED AND T.agg_checksum != S.agg_checksum THEN
  UPDATE SET
    events_count = S.events_count,
    value_sum = S.value_sum,
    agg_checksum = S.agg_checksum,
    updated_at = CURRENT_TIMESTAMP()
WHEN NOT MATCHED THEN
  INSERT (cohort_date, user_id, metric_id, events_count, value_sum, agg_checksum, created_at, updated_at)
  VALUES (S.cohort_date, S.user_id, S.metric_id, S.events_count, S.value_sum, S.agg_checksum, CURRENT_TIMESTAMP(), CURRENT_TIMESTAMP());