ez intents - need a couple tweaks

models/bronze/prices/bronze__complete_token_prices.sql

@@ -27,9 +27,3 @@ FROM
         'crosschain_silver',
         'complete_token_prices'
     ) }}
-WHERE
-    blockchain IN (
-        'near',
-        'near protocol',
-        'ethereum'
-    )

models/gold/defi/defi__ez_intents.sql

@@ -9,9 +9,51 @@
     tags = ['intents','curated','scheduled_non_core']
 ) }}
 
-with
-intents as (
-    select
+{% if execute %}
+
+{% if is_incremental() %}
+{% set query %}
+
+SELECT
+    MIN(DATE_TRUNC('day', block_timestamp)) AS block_timestamp_day
+FROM
+    {{ ref('defi__fact_intents') }}
+WHERE
+    modified_timestamp >= (
+        SELECT
+            MAX(modified_timestamp)
+        FROM
+            {{ this }}
+    ) {% endset %}
+    {% set min_block_timestamp_day = run_query(query).columns [0].values() [0] %}
+    {% elif var('MANUAL_FIX') %}
+    {% set query %}
+SELECT
+    MIN(DATE_TRUNC('day', block_timestamp)) AS block_timestamp_day
+FROM
+    {{ this }}
+WHERE
+    FLOOR(
+        block_id,
+        -3
+    ) = {{ var('RANGE_START') }}
+
+    {% endset %}
+    {% set min_block_timestamp_day = run_query(query).columns [0].values() [0] %}
+{% endif %}
+
+{% if not min_block_timestamp_day or min_block_timestamp_day == 'None' %}
+    {% set min_block_timestamp_day = '2024-11-01' %}
+{% endif %}
+
+{{ log(
+    "min_block_timestamp_day: " ~ min_block_timestamp_day,
+    info = True
+) }}
+{% endif %}
+
+WITH intents AS (
+    SELECT
         block_timestamp,
         block_id,
         tx_hash,
@@ -28,33 +70,153 @@ intents as (
         amount_index,
         amount_raw,
         token_id,
+        SPLIT(
+            token_id,
+            'nep141:'
+        ) [1] AS contract_address_raw,
         referral,
         dip4_version,
         gas_burnt,
-        receipt_succeeded
-    from
+        receipt_succeeded,
+        fact_intents_id,
+        FLOOR(
+            block_id,
+            -3
+        ) AS _partition_by_block_number
+    FROM
         {{ ref('defi__fact_intents') }}
-    {% if is_incremental() %}
-    where modified_timestamp >= (
-        select coalesce(max(modified_timestamp),'1970-01-01' :: timestamp)
-        from {{ this }}
-    )
-    {% endif %}
+
+        {% if var("MANUAL_FIX") %}
+        WHERE
+            {{ partition_load_manual('no_buffer') }}
+        {% else %}
+
+        {% if is_incremental() %}
+            WHERE
+                GREATEST(
+                    modified_timestamp,
+                    '2000-01-01'
+                ) >= DATEADD(
+                    'minute',
+                    -5,(
+                        SELECT
+                            MAX(
+                                modified_timestamp
+                            )
+                        FROM
+                            {{ this }}
+                    )
+                )
+        {% endif %}
+        {% endif %}
 ),
-native_labels as (
-    select
+labels AS (
+    SELECT
+        near_token_id AS contract_address_raw,
+        SPLIT(
+            defuse_asset_identifier,
+            ':'
+        ) [0] :: STRING AS ecosystem,
+        SPLIT(
+            defuse_asset_identifier,
+            ':'
+        ) [1] :: STRING AS chain_id,
+        SPLIT(
+            defuse_asset_identifier,
+            ':'
+        ) [2] :: STRING AS contract_address,
+        asset_name AS symbol,
+        decimals
+    FROM
+        {{ ref('silver__defuse_tokens_metadata') }}
+    UNION ALL
+    SELECT
+        contract_address AS contract_address_raw,
+        'near' AS ecosystem,
+        '397' AS chain_id,
         contract_address,
-        name,
         symbol,
         decimals
-    from 
+    FROM
         {{ ref('silver__ft_contract_metadata') }}
 ),
-defuse_labels as (
-    select
-        defuse_asset_identifier,
-        asset_name as name,
-        decimals
-    from 
-        {{ ref('silver__defuse_tokens_metadata') }}
-)
+prices AS (
+    SELECT
+        token_address AS contract_address,
+        price,
+        HOUR
+    FROM
+        {{ ref('price__ez_prices_hourly') }}
+
+{% if is_incremental() or var('MANUAL_FIX') %}
+WHERE
+    DATE_TRUNC(
+        'day',
+        HOUR
+    ) >= '{{ min_block_timestamp_day }}'
+{% endif %}
+
+qualify(ROW_NUMBER() over (PARTITION BY token_address, HOUR
+ORDER BY
+    HOUR DESC) = 1)
+),
+FINAL AS (
+    SELECT
+        block_timestamp,
+        block_id,
+        tx_hash,
+        receipt_id,
+        receiver_id,
+        predecessor_id,
+        log_event,
+        log_index,
+        log_event_index,
+        owner_id,
+        old_owner_id,
+        new_owner_id,
+        memo,
+        amount_index,
+        amount_raw,
+        token_id,
+        i.contract_address_raw,
+        referral,
+        dip4_version,
+        gas_burnt,
+        receipt_succeeded,
+        fact_intents_id AS ez_intents_id,
+        COALESCE(
+            dl.short_name,
+            l.ecosystem
+        ) AS blockchain,
+        l.contract_address,
+        l.symbol,
+        l.decimals,
+        amount_raw / pow(
+            10,
+            l.decimals
+        ) AS amount_adj,
+        p.price,
+        amount_raw / pow(
+            10,
+            l.decimals
+        ) * p.price AS amount_usd
+    FROM
+        intents i
+        LEFT JOIN labels l
+        ON i.contract_address_raw = l.contract_address_raw
+        LEFT JOIN EXTERNAL.defillama.dim_chains dl
+        ON l.chain_id = dl.chain_id 
+        ASOF JOIN prices p match_condition (
+            i.block_timestamp >= p.hour
+        )
+        ON (
+            l.contract_address = p.contract_address
+        )
+)
+SELECT
+    *,
+    SYSDATE() AS inserted_timestamp,
+    SYSDATE() AS modified_timestamp,
+    '{{ invocation_id }}' AS _invocation_id
+FROM
+    FINAL

models/silver/labels/DEPRECATING/livequery__request_pagoda_nft_metadata.sql

@@ -1,8 +1,7 @@
 {{ config(
     materialized = 'incremental',
     unique_key = 'metadata_id',
-    incremental_strategy = 'delete+insert',
-    tags = ['streamline_non_core']
+    incremental_strategy = 'delete+insert'
 ) }}
 -- DEPRECATED
 -- DELETE ALONGSIDE DEFI__DIM_NFT_SERIES_METADATA