Multi-Touch Attribution Without a $50K Platform

// Attribution model weight distribution
const attributionModels = {
  linear: (touchCount: number) => 1 / touchCount,
 
  uShaped: (position: number, total: number) => {
    if (position === 1) return 0.4; // First touch
    if (position === total) return 0.4; // Last touch
    return 0.2 / (total - 2); // Middle touches
  },
 
  wShaped: (position: number, total: number, isLeadCreation: boolean, isOppCreation: boolean) => {
    if (position === 1) return 0.3; // First touch
    if (isLeadCreation) return 0.3; // Lead creation
    if (isOppCreation) return 0.3; // Opportunity creation
    return 0.1 / (total - 3); // Middle touches
  },
 
  timeDecay: (daysFromConversion: number, halfLife: number = 7) => {
    return Math.pow(2, -daysFromConversion / halfLife);
  },
};

# NGINX configuration for first-party tracking
server {
    listen 443 ssl;
    server_name www.yourcompany.com;
 
    location /metrics/ {
        proxy_pass https://your-collector-url.com/;
        proxy_set_header Host your-collector-url.com;
        proxy_set_header X-Real-IP $remote_addr;
    }
}

// RudderStack SDK initialization
rudderanalytics.load("WRITE_KEY", "DATA_PLANE_URL", {
  // Enable cookie persistence
  storage: {
    cookie: {
      maxage: 31536000000, // 1 year in milliseconds
      path: "/",
      domain: ".yourcompany.com",
      samesite: "Lax",
      secure: true,
    },
  },
});
 
// Track page views with UTM parameters
rudderanalytics.page("Page Viewed", {
  url: window.location.href,
  referrer: document.referrer,
  utm_source: getQueryParam("utm_source"),
  utm_medium: getQueryParam("utm_medium"),
  utm_campaign: getQueryParam("utm_campaign"),
  utm_content: getQueryParam("utm_content"),
  utm_term: getQueryParam("utm_term"),
});
 
// Identify known users
rudderanalytics.identify(userId, {
  email: userEmail,
  company: companyName,
});

-- Identity resolution: Building the user map
-- dbt model: dim_identity_map.sql
 
WITH identify_events AS (
    SELECT
        anonymous_id,
        user_id,
        context_traits_email AS email,
        timestamp,
        ROW_NUMBER() OVER (
            PARTITION BY anonymous_id
            ORDER BY timestamp DESC
        ) AS rn
    FROM {{ source('rudderstack', 'identifies') }}
    WHERE user_id IS NOT NULL
),
 
-- Get the most recent identification for each anonymous_id
latest_identifies AS (
    SELECT
        anonymous_id,
        user_id,
        email
    FROM identify_events
    WHERE rn = 1
),
 
-- Build canonical user mapping
-- If multiple anonymous_ids map to the same user_id, group them
user_clusters AS (
    SELECT
        user_id AS canonical_user_id,
        anonymous_id,
        email
    FROM latest_identifies
)
 
SELECT
    anonymous_id,
    canonical_user_id,
    email
FROM user_clusters

-- Sessionization: Grouping events into sessions
-- dbt model: int_web_sessions.sql
 
WITH page_views AS (
    SELECT
        anonymous_id,
        timestamp,
        context_page_url AS url,
        context_page_referrer AS referrer,
        -- Extract UTMs
        context_campaign_source AS utm_source,
        context_campaign_medium AS utm_medium,
        context_campaign_name AS utm_campaign,
        context_campaign_content AS utm_content
    FROM {{ source('rudderstack', 'pages') }}
),
 
-- Flag new sessions (30-minute inactivity threshold)
session_flags AS (
    SELECT
        *,
        CASE
            WHEN TIMESTAMP_DIFF(
                timestamp,
                LAG(timestamp) OVER (PARTITION BY anonymous_id ORDER BY timestamp),
                MINUTE
            ) > 30 THEN 1
            ELSE 0
        END AS is_new_session
    FROM page_views
),
 
-- Generate session IDs
session_ids AS (
    SELECT
        *,
        CONCAT(
            anonymous_id,
            '-',
            SUM(is_new_session) OVER (
                PARTITION BY anonymous_id
                ORDER BY timestamp
            )
        ) AS session_id
    FROM session_flags
)
 
SELECT
    session_id,
    anonymous_id,
    MIN(timestamp) AS session_start_at,
    MAX(timestamp) AS session_end_at,
    COUNT(*) AS page_views,
    -- Attribution: First non-null UTM wins for the session
    FIRST_VALUE(utm_source IGNORE NULLS) OVER (
        PARTITION BY session_id ORDER BY timestamp
    ) AS source,
    FIRST_VALUE(utm_medium IGNORE NULLS) OVER (
        PARTITION BY session_id ORDER BY timestamp
    ) AS medium,
    FIRST_VALUE(utm_campaign IGNORE NULLS) OVER (
        PARTITION BY session_id ORDER BY timestamp
    ) AS campaign
FROM session_ids
GROUP BY session_id, anonymous_id

-- Attribution spine: Connecting touchpoints to opportunities
-- dbt model: fct_attribution_spine.sql
 
WITH opportunities AS (
    SELECT
        id AS opportunity_id,
        account_id,
        amount,
        created_date,
        close_date,
        is_won
    FROM {{ ref('stg_salesforce_opportunities') }}
    WHERE is_won = TRUE
),
 
contacts AS (
    SELECT
        id AS contact_id,
        account_id,
        email
    FROM {{ ref('stg_salesforce_contacts') }}
),
 
identity_map AS (
    SELECT
        anonymous_id,
        email
    FROM {{ ref('dim_identity_map') }}
),
 
sessions AS (
    SELECT * FROM {{ ref('int_web_sessions') }}
)
 
SELECT
    o.opportunity_id,
    o.amount AS conversion_value,
    o.created_date AS conversion_date,
    s.session_id,
    s.session_start_at,
    s.source,
    s.medium,
    s.campaign,
 
    -- Calculate touchpoint position
    ROW_NUMBER() OVER (
        PARTITION BY o.opportunity_id
        ORDER BY s.session_start_at ASC
    ) AS touch_position,
 
    COUNT(*) OVER (
        PARTITION BY o.opportunity_id
    ) AS total_touches
 
FROM opportunities o
-- Join opportunity to account contacts (Account-Based Logic)
JOIN contacts c ON o.account_id = c.account_id
-- Join contact to identity map via email
JOIN identity_map i ON LOWER(c.email) = LOWER(i.email)
-- Join identity to sessions
JOIN sessions s ON i.anonymous_id = s.anonymous_id
 
WHERE
    -- Lookback window: 90 days before opportunity creation
    s.session_start_at BETWEEN
        DATE_SUB(o.created_date, INTERVAL 90 DAY)
        AND o.created_date
    -- Exclude sessions with no source
    AND s.source IS NOT NULL

-- Attribution model: Applying weights
-- dbt model: fct_attribution_model.sql
 
WITH spine AS (
    SELECT * FROM {{ ref('fct_attribution_spine') }}
)
 
SELECT
    opportunity_id,
    session_id,
    source,
    medium,
    campaign,
    conversion_value,
    touch_position,
    total_touches,
 
    -- Linear Attribution
    (1.0 / total_touches) * conversion_value AS linear_revenue,
 
    -- U-Shaped Attribution
    CASE
        WHEN touch_position = 1 THEN 0.4
        WHEN touch_position = total_touches THEN 0.4
        ELSE 0.2 / NULLIF(total_touches - 2, 0)
    END * conversion_value AS u_shaped_revenue,
 
    -- Time-Decay Attribution (7-day half-life)
    POWER(2, -TIMESTAMP_DIFF(conversion_date, session_start_at, DAY) / 7.0)
        / SUM(POWER(2, -TIMESTAMP_DIFF(conversion_date, session_start_at, DAY) / 7.0))
            OVER (PARTITION BY opportunity_id)
        * conversion_value AS time_decay_revenue
 
FROM spine

-- Channel performance report
-- dbt model: rpt_channel_attribution.sql
 
WITH attributed AS (
    SELECT * FROM {{ ref('fct_attribution_model') }}
),
 
ad_spend AS (
    -- Pull from Airbyte-synced ad platform data
    SELECT
        date,
        source,
        medium,
        campaign,
        SUM(spend) AS spend
    FROM {{ ref('stg_ad_spend_unified') }}
    GROUP BY 1, 2, 3, 4
)
 
SELECT
    a.source,
    a.medium,
    a.campaign,
 
    -- Attributed revenue by model
    SUM(a.linear_revenue) AS linear_revenue,
    SUM(a.u_shaped_revenue) AS u_shaped_revenue,
    SUM(a.time_decay_revenue) AS time_decay_revenue,
 
    -- Touchpoint counts
    COUNT(DISTINCT a.opportunity_id) AS opportunities_influenced,
    COUNT(*) AS total_touches,
 
    -- Efficiency metrics (join with spend data)
    COALESCE(SUM(s.spend), 0) AS total_spend,
    CASE
        WHEN SUM(s.spend) > 0
        THEN SUM(a.u_shaped_revenue) / SUM(s.spend)
        ELSE NULL
    END AS roas_u_shaped
 
FROM attributed a
LEFT JOIN ad_spend s
    ON a.source = s.source
    AND a.medium = s.medium
    AND a.campaign = s.campaign
    AND DATE(a.session_start_at) = s.date
GROUP BY 1, 2, 3
ORDER BY u_shaped_revenue DESC

-- Campaign persistence for direct traffic
WITH sessions_with_lag AS (
    SELECT
        *,
        LAG(source) OVER (PARTITION BY anonymous_id ORDER BY session_start_at) AS prev_source,
        LAG(medium) OVER (PARTITION BY anonymous_id ORDER BY session_start_at) AS prev_medium,
        LAG(campaign) OVER (PARTITION BY anonymous_id ORDER BY session_start_at) AS prev_campaign,
        LAG(session_start_at) OVER (PARTITION BY anonymous_id ORDER BY session_start_at) AS prev_session_at
    FROM {{ ref('int_web_sessions') }}
)
 
SELECT
    session_id,
    anonymous_id,
    session_start_at,
    -- Inherit UTMs from previous session if current is direct/null
    -- and previous session was within 30 days
    CASE
        WHEN source IS NULL
            AND TIMESTAMP_DIFF(session_start_at, prev_session_at, DAY) <= 30
        THEN prev_source
        ELSE source
    END AS source,
    CASE
        WHEN medium IS NULL
            AND TIMESTAMP_DIFF(session_start_at, prev_session_at, DAY) <= 30
        THEN prev_medium
        ELSE medium
    END AS medium
FROM sessions_with_lag

-- Offline touchpoints from Salesforce
WITH offline_touches AS (
    SELECT
        cm.contact_id,
        c.campaign_name AS campaign,
        'offline' AS source,
        c.type AS medium, -- 'Webinar', 'Conference', 'Dinner'
        cm.created_date AS touch_date
    FROM {{ ref('stg_salesforce_campaign_members') }} cm
    JOIN {{ ref('stg_salesforce_campaigns') }} c
        ON cm.campaign_id = c.id
),
 
-- Union with digital sessions
all_touchpoints AS (
    SELECT * FROM {{ ref('int_web_sessions') }}
    UNION ALL
    SELECT * FROM offline_touches
)
 
SELECT * FROM all_touchpoints

// Reverse ETL configuration (Census example)
const conversionSync = {
  source: "snowflake.fct_attribution_model",
  destination: "facebook_conversions_api",
  mapping: {
    email: "email_hash", // SHA-256 hashed
    event_name: "Purchase",
    event_time: "conversion_date",
    value: "conversion_value",
    currency: "USD",
  },
  schedule: "daily",
};