Enrich Every Event in Real Time with Streaming JOINs

CREATE SOURCE touchpoints WITH (
  connector = 'kafka',
  topic = 'marketing.touchpoints',
  properties.bootstrap.server = 'broker:9092'
) FORMAT PLAIN ENCODE JSON;

CREATE MATERIALIZED VIEW user_journeys AS
WITH conversions AS (
  SELECT
    user_id,
    ts AS conversion_ts,
    revenue
  FROM touchpoints
  WHERE revenue > 0
),
journeys AS (
  SELECT
    c.user_id,
    c.conversion_ts,
    c.revenue,
    t.channel,
    t.campaign,
    t.action,
    t.ts AS touch_ts,
    ROW_NUMBER() OVER (
      PARTITION BY c.user_id, c.conversion_ts
      ORDER BY t.ts
    ) AS touch_position,
    COUNT(*) OVER (
      PARTITION BY c.user_id, c.conversion_ts
    ) AS journey_length
  FROM conversions c
  JOIN touchpoints t
    ON c.user_id = t.user_id
    AND t.ts <= c.conversion_ts
    AND t.ts >= c.conversion_ts - INTERVAL '30 DAYS'
)
SELECT
  user_id,
  conversion_ts,
  revenue,
  channel,
  campaign,
  touch_position,
  journey_length,
  CASE
    WHEN journey_length = 1 THEN 1.0
    WHEN touch_position = 1 THEN 0.30
    WHEN touch_position = journey_length THEN 0.30
    ELSE 0.40 / (journey_length - 2)
  END AS position_weight
FROM journeys;