{% macro cents_to_dollars(column_name, scale=2) %}
  round({{ column_name }} / 100.0, {{ scale }})
{% endmacro %}


-- Usage in any model:
select
  order_id,
  {{ cents_to_dollars('amount_cents') }}          as amount,
  {{ cents_to_dollars('tax_cents', scale=4) }}  as tax

-- Compiles to:
select
  order_id,
  round(amount_cents / 100.0, 2) as amount,
  round(tax_cents    / 100.0, 4) as tax

{% macro pivot_status_counts(statuses) %}

  {% for status in statuses %}

    countif(status = '{{ status }}') as {{ status | lower }}_count

    {% if not loop.last %},{% endif %}

  {% endfor %}

{% endmacro %}


-- Usage:
select
  customer_id,
  {{ pivot_status_counts([
      'PENDING',
      'COMPLETED',
      'CANCELLED',
      'REFUNDED'
  ]) }}
from {{ ref('stg_orders') }}
group by 1

{% macro get_column_names(model) %}

  {% set relation = ref(model) %}
  {% set columns  = adapter.get_columns_in_relation(relation) %}

  {{ log("Columns in " ~ model ~ ":", info=true) }}
  {% for col in columns %}
    {{ log(col.name ~ " (" ~ col.dtype ~ ")", info=true) }}
  {% endfor %}

{% endmacro %}


-- A model that auto-selects all non-PII columns:
{% macro select_non_pii(model, pii_columns) %}

  {% set relation = ref(model) %}
  {% set all_cols  = adapter.get_columns_in_relation(relation) %}

  select
  {% for col in all_cols %}
    {% if col.name not in pii_columns %}
      {{ col.name }}{% if not loop.last %},{% endif %}
    {% endif %}
  {% endfor %}
  from {{ relation }}

{% endmacro %}

{% macro get_max_date(model, column) %}

  {% set query %}
    select max({{ column }}) as max_date
    from {{ ref(model) }}
  {% endset %}

  {% set result = run_query(query) %}

  {% if execute %}
    {% set max_date = result.columns[0].values()[0] %}
    {{ return(max_date) }}
  {% endif %}

{% endmacro %}

-- Usage: dynamically backfill from last loaded date
{% set cutoff = get_max_date('stg_orders', 'created_at') %}

models:
  - name: stg_orders
    columns:
      - name: order_id
        tests:
          - unique                # SELECT count(*) WHERE count > 1
          - not_null             # SELECT count(*) WHERE col IS NULL

      - name: status
        tests:
          - accepted_values:     # SELECT ... WHERE NOT IN (values)
              values: [PENDING, COMPLETED, CANCELLED, REFUNDED]
              quote: false

      - name: customer_id
        tests:
          - relationships:       # referential integrity
              to: ref('stg_customers')
              field: customer_id

      - name: amount_dollars
        tests:
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 0
              max_value: 100000
              strictly: true

-- ANY ROW RETURNED = TEST FAILS.
-- dbt expects zero rows from a passing test.
-- This is pure SQL — use everything you know.

with revenue_check as (

  select
    order_date,
    sum(net_revenue)  as daily_revenue,
    count(*)          as order_count

  from {{ ref('mart_revenue') }}
  group by 1

)

-- Return rows where revenue is IMPOSSIBLE
select *
from revenue_check
where
  daily_revenue < 0                    -- negative revenue
  or order_count > 1000000            -- astronomical count
  or daily_revenue / order_count > 50000 -- avg order > $50k

{% test not_future_date(model, column_name, tolerance_days=1) %}

  -- Reusable: call from schema.yml on any date column.
  -- tolerance_days handles timezone skew.

  select
    {{ column_name }}  as bad_date,
    current_timestamp() as checked_at
  from {{ model }}
  where
    {{ column_name }} >
    current_timestamp() + interval '{{ tolerance_days }} days'

{% endtest %}

-- Usage in schema.yml:
-- tests:
--   - not_future_date:
--       column_name: created_at
--       tolerance_days: 2

{% snapshot customers_snapshot %}

{{
  config(
    target_schema  = 'snapshots',
    unique_key     = 'customer_id',
    strategy       = 'timestamp',
    updated_at     = 'updated_at'   -- change detector
  )
}}

select *
from {{ source('postgres', 'customers') }}

{% endsnapshot %}

-- dbt adds these columns automatically:
-- dbt_scd_id     UUID for each version row
-- dbt_updated_at timestamp this version was captured
-- dbt_valid_from when this version became active
-- dbt_valid_to   NULL = current version, else end timestamp

{% snapshot products_snapshot %}

{{
  config(
    target_schema = 'snapshots',
    unique_key    = 'product_id',
    strategy      = 'check',
    check_cols    = ['price', 'category', 'is_active']
    -- dbt hashes these cols and detects any change
    -- check_cols = 'all' hashes every column
  )
}}

select *
from {{ source('postgres', 'products') }}

{% endsnapshot %}

# Python models must define a model(dbt, session) function.
# dbt.ref() and dbt.source() return DataFrames.

import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import LabelEncoder

def model(dbt, session):

    # Config lives in the function body
    dbt.config(
        materialized='table',
        packages=["scikit-learn", "pandas"]
    )

    # ref() returns a Snowpark / Spark DataFrame
    features_df = dbt.ref('int_churn_features')
    labels_df   = dbt.ref('mart_churn_labels')

    # Convert to pandas (Snowpark / BQ Dataproc)
    features = features_df.to_pandas()
    labels   = labels_df.to_pandas()

    X_train = features[['days_since_login', 'order_count_30d',
                         'support_tickets', 'plan_tier']]
    y_train = labels['churned']

    clf = RandomForestClassifier(n_estimators=100, random_state=42)
    clf.fit(X_train, y_train)

    # Score production data
    prod_df = dbt.ref('int_current_customers').to_pandas()
    prod_df['churn_probability'] = clf.predict_proba(
        prod_df[['days_since_login', 'order_count_30d',
                 'support_tickets', 'plan_tier']]
    )[:, 1]

    # Return a DataFrame — dbt persists it as a table
    return prod_df[['customer_id', 'churn_probability']]

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StringType

def model(dbt, spark):

    dbt.config(
        materialized='incremental',
        incremental_strategy='merge',
        unique_key='event_id',
        file_format='delta'         # Delta Lake!
    )

    # dbt.ref() returns a Spark DataFrame
    raw = dbt.source('kafka', 'raw_avro_events')

    parsed = (
        raw
        .withColumn('payload', F.from_json(
            F.col('value'),
            'struct<event_id:string, user_id:string, ts:long>'
        ))
        .select(
            F.col('payload.event_id').alias('event_id'),
            F.col('payload.user_id').alias('user_id'),
            F.to_timestamp(F.col('payload.ts') / 1000).alias('event_ts'),
            F.col('partition').alias('kafka_partition'),
            F.col('offset').alias('kafka_offset')
        )
        .filter(F.col('event_id').isNotNull())
    )

    if dbt.is_incremental:
        max_ts = spark.sql(f"SELECT max(event_ts) FROM {dbt.this}")
        cutoff = max_ts.collect()[0][0]
        parsed = parsed.filter(F.col('event_ts') > cutoff)

    return parsed

{% macro generate_schema_name(custom_schema_name, node) %}

  {% set default_schema = target.schema %}

  {% if target.name == 'prod' %}

    -- In prod: use EXACTLY the custom_schema_name you set.
    -- stg_orders → analytics.staging (not analytics_staging)
    {% if custom_schema_name is not none %}
      {{ custom_schema_name | trim }}
    {% else %}
      {{ default_schema | trim }}
    {% endif %}

  {% else %}

    -- In dev: prefix with user target schema.
    -- avoids dev engineers stomping on each other.
    {% if custom_schema_name is not none %}
      {{ default_schema }}_{{ custom_schema_name | trim }}
    {% else %}
      {{ default_schema | trim }}
    {% endif %}

  {% endif %}

{% endmacro %}

{{
  config(
    materialized = 'table',
    pre_hook     = [
      "CALL analytics.lock_table('{{ this }}')",
      "{{ log('Starting mart_revenue build', info=True) }}"
    ],
    post_hook    = [
      -- Grant access to BI roles after build
      "GRANT SELECT ON {{ this }} TO ROLE REPORTER",
      -- Analyze table for query planner
      "ANALYZE {{ this }}",
      -- Custom audit log
      """
        INSERT INTO analytics.build_audit
        SELECT '{{ this }}', current_timestamp, '{{ invocation_id }}'
      """
    ]
  )
}}

select * from {{ ref('int_orders_enriched') }}