Data Cleaning and Business Logic Edge Cases Questions

WITH ranked AS (
  SELECT
    *,
    ROW_NUMBER() OVER (
      PARTITION BY fuzzy_key
      ORDER BY
        (CASE WHEN source_system = 'crm' THEN 1 WHEN source_system = 'import' THEN 2 ELSE 3 END),
        completeness_score DESC,
        created_at DESC
    ) as rn
  FROM customers
),
canonical AS (
  SELECT * FROM ranked WHERE rn = 1
),
backfill AS (
  -- for each fuzzy_key get the first non-null value per attribute by precedence
  SELECT
    fuzzy_key,
    MAX(name) FILTER (WHERE rank_order = 1 AND name IS NOT NULL)       AS name_pref,
    MAX(email) FILTER (WHERE rank_order = 1 AND email IS NOT NULL)     AS email_pref
  FROM (
    SELECT
      fuzzy_key, name, email,
      ROW_NUMBER() OVER (
        PARTITION BY fuzzy_key
        ORDER BY
          (CASE WHEN source_system = 'crm' THEN 1 WHEN source_system = 'import' THEN 2 ELSE 3 END),
          completeness_score DESC,
          created_at DESC
      ) as rank_order
    FROM customers
  ) t
  GROUP BY fuzzy_key
)
SELECT
  c.customer_id AS canonical_customer_id,
  c.fuzzy_key,
  COALESCE(c.name, b.name_pref) AS name,
  COALESCE(c.email, b.email_pref) AS email,
  c.source_system,
  c.created_at,
  c.completeness_score
FROM canonical c
LEFT JOIN backfill b USING (fuzzy_key);

WITH base AS (
  SELECT
    event_id, user_id, ip_address, user_agent, event_time, event_type,
    DATE_TRUNC('minute', event_time) AS minute_bucket
  FROM events
  WHERE event_time >= now() - interval '7 days'
),
agg AS (
  SELECT
    user_id,
    ip_address,
    COUNT(*) AS events_last_7d,
    COUNT(DISTINCT minute_bucket) AS active_minutes,
    SUM(CASE WHEN user_agent IS NULL OR user_agent = '' THEN 1 ELSE 0 END) AS missing_ua,
    MAX(events_per_min) AS peak_events_per_min,
    COUNT(DISTINCT user_id) OVER (PARTITION BY ip_address) AS distinct_users_on_ip
  FROM (
    SELECT *, COUNT(*) OVER (PARTITION BY user_id, DATE_TRUNC('minute', event_time)) AS events_per_min
    FROM base
  ) t
  GROUP BY user_id, ip_address
)
SELECT *,
  (events_last_7d::numeric / NULLIF(active_minutes,1)) AS avg_events_per_min_active,
  (missing_ua::numeric / events_last_7d) AS frac_missing_ua
FROM agg
ORDER BY events_last_7d DESC
LIMIT 100;