-- models/vault/hub_customer.sql
{{
    config(
        materialized='incremental',
        unique_key='hub_customer_hashkey',
        incremental_strategy='merge',
        tags=['vault', 'hub']
    )
}}

{%- set source_models = ['stg_crm_customers', 'stg_erp_customers'] -%}

with source_union as (
    {%- for source in source_models %}
    select
        customer_id as business_key,
        'source_{{ source }}' as record_source,
        _loaded_at as load_date
    from {{ ref(source) }}
    {%- if not loop.last %}
    union all
    {%- endif %}
    {%- endfor %}
),

deduplicated as (
    select
        {{ dbt_utils.generate_surrogate_key(['business_key']) }} as hub_customer_hashkey,
        business_key,
        min(load_date) as load_date,
        min(record_source) as record_source
    from source_union
    group by business_key
)

select
    hub_customer_hashkey,
    business_key,
    load_date,
    record_source
from deduplicated

{% if is_incremental() %}
where hub_customer_hashkey not in (
    select hub_customer_hashkey from {{ this }}
)
{% endif %}

-- models/vault/link_order_customer.sql
{{
    config(
        materialized='incremental',
        unique_key='link_order_customer_hashkey',
        incremental_strategy='merge',
        tags=['vault', 'link']
    )
}}

with source_data as (
    select
        order_id as order_business_key,
        customer_id as customer_business_key,
        'stg_orders' as record_source,
        _loaded_at as load_date
    from {{ ref('stg_orders') }}
),

hash_keys as (
    select
        {{ dbt_utils.generate_surrogate_key(['order_business_key']) }} as hub_order_hashkey,
        {{ dbt_utils.generate_surrogate_key(['customer_business_key']) }} as hub_customer_hashkey,
        {{ dbt_utils.generate_surrogate_key(['order_business_key', 'customer_business_key']) }} as link_order_customer_hashkey,
        record_source,
        load_date
    from source_data
)

select distinct
    link_order_customer_hashkey,
    hub_order_hashkey,
    hub_customer_hashkey,
    load_date,
    record_source
from hash_keys

{% if is_incremental() %}
where link_order_customer_hashkey not in (
    select link_order_customer_hashkey from {{ this }}
)
{% endif %}

-- models/vault/sat_order_details.sql
{{
    config(
        materialized='incremental',
        unique_key=['hub_order_hashkey', 'load_date'],
        incremental_strategy='merge',
        tags=['vault', 'satellite']
    )
}}

with source_data as (
    select
        order_id as business_key,
        status,
        amount,
        currency,
        order_date,
        _loaded_at as load_date,
        'stg_orders' as record_source
    from {{ ref('stg_orders') }}
),

hash_keys as (
    select
        {{ dbt_utils.generate_surrogate_key(['business_key']) }} as hub_order_hashkey,
        status,
        amount,
        currency,
        order_date,
        load_date,
        record_source,
        {{ dbt_utils.generate_surrogate_key(['status', 'amount', 'currency', 'order_date']) }} as hashdiff
    from source_data
),

{% if is_incremental() %}
previous_hashes as (
    select hub_order_hashkey, hashdiff
    from {{ this }}
),

new_records as (
    select
        h.hub_order_hashkey,
        h.status,
        h.amount,
        h.currency,
        h.order_date,
        h.load_date,
        h.record_source,
        h.hashdiff
    from hash_keys h
    left join previous_hashes p on h.hub_order_hashkey = p.hub_order_hashkey
    where p.hub_order_hashkey is null
       or h.hashdiff != p.hashdiff
)
select * from new_records
{% else %}
select * from hash_keys
{% endif %}

-- macros/generate_hashkey.sql
{% macro generate_hashkey(columns, algorithm='sha256') %}
    {% if target.type == 'bigquery' %}
        {{ algorithm }}(concat({% for col in columns %}coalesce(cast({{ col }} as string), '^^')
        {%- if not loop.last %}, '||'{% endif %}{% endfor %}))
    {% elif target.type == 'snowflake' %}
        hash({% for col in columns %}coalesce(cast({{ col }} as varchar), '^^')
        {%- if not loop.last %}, '||'{% endif %}{% endfor %}, '{{ algorithm }}')
    {% else %}
        md5({% for col in columns %}coalesce(cast({{ col }} as varchar), '^^')
        {%- if not loop.last %}, '||'{% endif %}{% endfor %})
    {% endif %}
{% endmacro %}

-- models/vault/pit_order.sql
{{
    config(
        materialized='incremental',
        unique_key='hub_order_hashkey',
        tags=['vault', 'pit']
    )
}}

with dates as (
    select * from {{ ref('dim_date') }}
    where date_value >= '2024-01-01'
),

satellites as (
    select
        hub_order_hashkey,
        load_date,
        status,
        amount
    from {{ ref('sat_order_details') }}
),

pit_records as (
    select
        d.date_value as pit_date,
        s.hub_order_hashkey,
        s.status,
        s.amount,
        s.load_date as sat_load_date
    from dates d
    cross join (
        select distinct hub_order_hashkey from satellites
    ) s
    left join satellites s
        on s.hub_order_hashkey = s.hub_order_hashkey
        and s.load_date = (
            select max(load_date)
            from satellites
            where hub_order_hashkey = s.hub_order_hashkey
            and load_date <= d.date_value
        )
)

select * from pit_records