{%- set incremental_predicate = partition_filter_daily('DBT_INTERNAL_DEST.p_date') -%}
{{
    config(
        materialized='incremental',
        incremental_strategy='delete+insert',
        incremental_predicates=[incremental_predicate],
        unique_key=['user_id', 'p_date'],
        cluster_by=['p_date'],
        snowflake_warehouse='FACT_PLAY_BACKFILL_SMALL',
        on_schema_change='fail',
    )
}}

-- Read from snapshot to get point-in-time follower/following relationships
-- Only read records that could be active during the partition window
WITH snapshot AS (
    SELECT
        id,
        follower_user_id,
        following_user_id,
        created_at,
        dbt_valid_from::date as dbt_valid_from_date,
        dbt_valid_to::date as dbt_valid_to_date
    FROM {{ ref('rds_profile_follow__snapshot') }}
    WHERE 1=1
        AND (
            -- Either (1) the follow happened during the partition window,
            ({{ partition_filter_daily("dbt_valid_from_date") }})
            -- or (2) the follow ended during the partition window,
            OR (dbt_valid_to_date IS NOT NULL AND {{ partition_filter_daily("dbt_valid_to_date") }})
            -- or (3) the follow started before partition and extends into/past partition,
            OR (dbt_valid_from_date < '{{ var("partition_start_date", "1900-01-01") }}'
                AND (dbt_valid_to_date IS NULL OR dbt_valid_to_date >= '{{ var("partition_start_date", "1900-01-01") }}'))
    )
),

-- Generate date spine for the range we need based on dagster partition
date_spine AS (
    SELECT date_day AS p_date
    FROM {{ ref('dim_date') }}
    WHERE {{ partition_filter_daily('date_day') }}
),

-- Get all unique users from the snapshot
unique_users AS (
    SELECT DISTINCT following_user_id AS user_id FROM snapshot
    UNION
    SELECT DISTINCT follower_user_id AS user_id FROM snapshot
),

-- Expand snapshot to all dates it was active
active_relationships AS (
    SELECT
        date_spine.p_date,
        snapshot.follower_user_id,
        snapshot.following_user_id
    FROM date_spine
    INNER JOIN snapshot
        ON snapshot.dbt_valid_from_date <= date_spine.p_date
        AND (snapshot.dbt_valid_to_date > date_spine.p_date OR snapshot.dbt_valid_to_date IS NULL)
),

-- Count followers and following for each user-date
follower_counts AS (
    SELECT
        p_date,
        user_id,
        COALESCE(SUM(follower_cnt), 0) AS total_follower_cnt,
        COALESCE(SUM(following_cnt), 0) AS total_following_cnt
    FROM (
        -- Count people following this user
        SELECT
            p_date,
            following_user_id AS user_id,
            COUNT(DISTINCT follower_user_id) AS follower_cnt,
            0 AS following_cnt
        FROM active_relationships
        GROUP BY p_date, following_user_id

        UNION ALL

        -- Count people this user is following
        SELECT
            p_date,
            follower_user_id AS user_id,
            0 AS follower_cnt,
            COUNT(DISTINCT following_user_id) AS following_cnt
        FROM active_relationships
        GROUP BY p_date, follower_user_id
    )
    GROUP BY p_date, user_id
)

SELECT
    user_id,
    total_following_cnt,
    total_follower_cnt,
    p_date
FROM follower_counts
WHERE 1=1
    AND {{ partition_filter_daily('p_date') }}
    AND (total_follower_cnt > 0 OR total_following_cnt > 0)  -- Only include users with actual relationships
