{{ config(
    materialized='incremental',
    unique_key='permit_id',
    schema='silver'
) }}

WITH cdc_raw AS (
    SELECT
        CAST(permit_id AS VARCHAR) AS permit_id,
        TRIM(applicant_name) AS applicant_name,
        LOWER(permit_type) AS permit_type,
        CAST(status AS VARCHAR) AS permit_status,
        CAST(issued_date AS TIMESTAMP) AS issued_date,
        CAST(updated_at AS TIMESTAMP) AS updated_at,
        _sling_op,
        CAST(_sling_ts AS TIMESTAMP) AS _sling_ts,
        ROW_NUMBER() OVER (
            PARTITION BY permit_id 
            ORDER BY _sling_ts DESC
        ) AS row_num
    FROM {{ source('bronze_lakehouse', 'environmental_permits_cdc') }}
    {% if is_incremental() %}
        WHERE _sling_ts > (SELECT MAX(_sling_ts) FROM {{ this }})
    {% endif %}
)

SELECT
    permit_id,
    applicant_name,
    permit_type,
    permit_status,
    issued_date,
    updated_at,
    _sling_ts,
    CURRENT_TIMESTAMP AS ingested_at_silver
FROM cdc_raw
WHERE row_num = 1 
  AND _sling_op != 'D'  -- Buang record yang di-DELETE