{{
  config(
    materialized='incremental',
    unique_key='id',
    cluster_by=['p_date'],
    on_schema_change='sync_all_columns',
  )
}}

WITH

-- Get Stripe credit topups
stripe_credit_topups AS (
    SELECT
        -- Primary key and user identifiers
        {{ md5_to_uuid("TO_VARCHAR('Stripe' || COALESCE(du.user_id, 0) || i.customer_id || i.id)") }} AS id
        , du.user_id
        , du.user_uid

        -- Billing provider information
        , 'Stripe' AS billing_provider
        , 'web' AS purchased_on_platform
        , i.customer_id AS provider_customer_id
        , i.id AS provider_transaction_id

        -- Credit topup information
        , CASE
            WHEN pd.name = 'Suno Pro - 500 Credits' THEN 500
            WHEN pd.name = 'Suno Pro - 1000 Credits' THEN 1000
            WHEN pd.name = 'Suno Pro - 2000 Credits' THEN 2000
            WHEN pd.name = 'Suno Pro - 4000 Credits' THEN 4000
            WHEN REGEXP_LIKE(pd.name, 'Suno Pro - ([0-9]+) Credits')
                THEN CAST(REGEXP_SUBSTR(pd.name, '([0-9]+)', 1, 1) AS NUMBER(38,0))
            ELSE NULL
        END AS n_credits
        , CAST(i.date AS TIMESTAMP_NTZ(9)) AS purchased_ts_utc
        , pd.id AS provider_product_id
        , pd.name AS provider_product_name
        -- p.unit_amount_decimal is in local currency by smallest denomination (with some special cases)
        , CAST(
            p.unit_amount_decimal /
            CASE
                WHEN p.currency IN ('ugx', 'isk') THEN 100.0 -- Zero-decimal currencies with 100 subunits for backwards compatibility
                WHEN p.currency IN ('bif', 'clp', 'djf', 'gnf', 'jpy', 'kmf', 'krw', 'mga', 'pyg', 'rwf', 'ugx', 'vnd', 'vuv', 'xaf', 'xof', 'xpf') THEN 1.0 -- Currencies with no subunits https://docs.stripe.com/currencies#zero-decimal
                ELSE 100.0 -- Default to standard 100 subunits
            END
            / CAST(JSON_EXTRACT_PATH_TEXT(er.buy_currency_exchange_rates, i.currency) AS DOUBLE)
        AS NUMBER(38,2)) AS amount_usd
        , CAST(
            p.unit_amount_decimal /
            CASE
                WHEN p.currency IN ('ugx', 'isk') THEN 100.0 -- Zero-decimal currencies with 100 subunits for backwards compatibility
                WHEN p.currency IN ('bif', 'clp', 'djf', 'gnf', 'jpy', 'kmf', 'krw', 'mga', 'pyg', 'rwf', 'ugx', 'vnd', 'vuv', 'xaf', 'xof', 'xpf') THEN 1.0 -- Currencies with no subunits https://docs.stripe.com/currencies#zero-decimal
                ELSE 100.0 -- Default to standard 100 subunits
            END
        AS NUMBER(38,2)) AS amount_local
        , CAST(UPPER(p.currency) AS VARCHAR) AS currency_code
        , DATE(i.date) AS p_date
    FROM
        {{ source('stripe', 'invoices') }} i
    LEFT JOIN
        {{ source('stripe', 'invoice_line_items') }} ili 
        ON ili.invoice_id = i.id
    LEFT JOIN
        {{ source('stripe', 'prices') }} p
        ON p.id = ili.price_id
    LEFT JOIN
        {{ source('stripe', 'products') }} pd
        ON pd.id = p.product_id
    LEFT JOIN
        {{ source('stripe', 'exchange_rates_from_usd') }} er
        ON er.sell_currency = 'usd'
        AND {{ partition_filter_daily('er.date') }}
    LEFT JOIN
        {{ source('snowflake', 'dim_user') }} du
        ON du.stripe_customer_id = i.customer_id
    WHERE 1=1
        AND {{ partition_filter_daily('DATE(i.date)') }}
        AND i.billing_reason IN ('manual', 'subscription_update') -- Topups show up in both `manual` and `subscription_update` billing reasons
        AND REGEXP_LIKE(LOWER(pd.name), '^Suno.* - \\d+ Credits$')
        AND i.paid
)

, revcat_credit_topups AS (
    SELECT
        id
        , user_id
        , user_uid
        , billing_provider
        , purchased_on_platform
        , provider_customer_id
        , provider_transaction_id
        , n_credits
        , purchased_ts_utc
        , provider_product_id
        , provider_product_name
        , amount_usd
        , amount_local
        , currency_code
        , p_date
    FROM (
        SELECT
            -- Primary key and user identifiers
            {{ md5_to_uuid("TO_VARCHAR('RevenueCat' || COALESCE(du.user_id, 0) || rtd.rc_original_app_user_id || rtd.store_transaction_id)") }} AS id
            , du.user_id
            , du.user_uid

            -- Billing provider information
            , 'RevenueCat' AS billing_provider
            , LOWER(rtd.platform) AS purchased_on_platform
            , rtd.rc_original_app_user_id AS provider_customer_id
            , rtd.store_transaction_id AS provider_transaction_id

            -- Credit topup information
            , CASE
                WHEN rtd.product_identifier = '500_credits' THEN 500
                WHEN rtd.product_identifier = '1000_credits' THEN 1000
                WHEN rtd.product_identifier = '2000_credits' THEN 2000
                WHEN rtd.product_identifier = '4000_credits' THEN 4000
                WHEN REGEXP_LIKE(rtd.product_identifier, '([0-9]+)_credits')
                    THEN CAST(REGEXP_SUBSTR(rtd.product_identifier, '([0-9]+)', 1, 1) AS NUMBER(38,0))
                ELSE NULL
            END AS n_credits
            , CAST(rtd.start_time AS TIMESTAMP_NTZ(9)) AS purchased_ts_utc
            , NULL AS provider_product_id
            , rtd.product_identifier AS provider_product_name
            , rtd.price_in_usd AS amount_usd
            , rtd.price_in_purchased_currency AS amount_local
            , rtd.purchased_currency AS currency_code
            , DATE(rtd.start_time) AS p_date
            , ROW_NUMBER() OVER (PARTITION BY rtd.store_transaction_id ORDER BY rtd.start_time DESC) AS rn
        FROM
            {{ source('revenuecat', 'revenuecat_transaction_data') }} rtd
        LEFT JOIN
            {{ source('snowflake', 'dim_user') }} du
            ON du.user_uid = rtd.rc_last_seen_app_user_id_alias
        WHERE 1=1
            AND {{ partition_filter_daily('DATE(rtd.start_time)') }}
            AND (rtd.presented_offering = 'credits' OR rtd.presented_offering IS NULL) -- Some edge cases where the presented_offering is null
            AND REGEXP_LIKE(rtd.product_identifier, '.*credit.*') -- Double check that the product identifier contains 'credit'
    )
    WHERE rn = 1
)

SELECT * FROM stripe_credit_topups
UNION ALL
SELECT * FROM revcat_credit_topups
