cureate saber; deprecate legacy swaps (#838)

* cureate saber; deprecate legacy swaps

* remove dev ref

models/gold/defi/defi__fact_swaps.sql

@@ -22,37 +22,7 @@
     {% endif %}
 {% endif %}
 
-    -- Separate cte since the silver.swaps model has an existing _log_id
-with swaps_general as (
-SELECT
-    block_timestamp,
-    block_id,
-    tx_id,
-    succeeded,
-    swapper,
-    from_amt AS swap_from_amount,
-    from_mint AS swap_from_mint,
-    to_amt AS swap_to_amount,
-    to_mint AS swap_to_mint,
-    program_id,
-    _log_id,
-    swaps_id AS fact_swaps_id,
-    inserted_timestamp,
-    modified_timestamp
-FROM
-    {{ ref('silver__swaps') }}
-WHERE
-    program_id not in ('JUP4Fb2cqiRUcaTHdrPC8h2gNsA2ETXiPDD33WcGuJB','9W959DqEETiGZocYWCQPaJ6sBmUzgfxXfqGeTEdp3aQP', 'DjVE6JNiYqPL2QXyCUUh8rNjHrbz9hXHNYt99MQ59qw1', 'whirLbMiicVdio4qvUfM5KAg6Ct8VwpYzGff3uctyCc')
-{% if is_incremental() %}
-AND
-    modified_timestamp >= '{{ max_modified_timestamp }}'
-{% else %}
-AND
-    modified_timestamp::date < '{{ backfill_to_date }}'
-{% endif %}
-)
-,
-swaps_individual as (
+WITH swaps_individual as (
 SELECT
     block_timestamp,
     block_id,
@@ -349,8 +319,54 @@ FROM
 {% if is_incremental() %}
 WHERE modified_timestamp >= '{{ max_modified_timestamp }}'
 {% endif %}
+UNION ALL
+SELECT
+    block_timestamp,
+    block_id,
+    tx_id,
+    succeeded,
+    swapper,
+    swap_from_amount,
+    swap_from_mint,
+    swap_to_amount,
+    swap_to_mint,
+    program_id,
+    swap_index,
+    swaps_intermediate_saber_id as fact_swaps_id,
+    inserted_timestamp,
+    modified_timestamp
+FROM
+    {{ ref('silver__swaps_intermediate_saber') }}
+{% if is_incremental() %}
+WHERE modified_timestamp >= '{{ max_modified_timestamp }}'
+{% endif %}
 )
 
+{% if not is_incremental() %}
+-- Separate cte since the silver.swaps model has an existing _log_id
+-- Only select from the deprecated model during the initial FR
+,   
+swaps_general as (
+SELECT
+    block_timestamp,
+    block_id,
+    tx_id,
+    succeeded,
+    swapper,
+    from_amt AS swap_from_amount,
+    from_mint AS swap_from_mint,
+    to_amt AS swap_to_amount,
+    to_mint AS swap_to_mint,
+    program_id,
+    _log_id,
+    swaps_id AS fact_swaps_id,
+    inserted_timestamp,
+    modified_timestamp
+FROM
+    {{ ref('silver__swaps_view') }}
+WHERE
+    program_id not in ('JUP4Fb2cqiRUcaTHdrPC8h2gNsA2ETXiPDD33WcGuJB','9W959DqEETiGZocYWCQPaJ6sBmUzgfxXfqGeTEdp3aQP', 'DjVE6JNiYqPL2QXyCUUh8rNjHrbz9hXHNYt99MQ59qw1', 'whirLbMiicVdio4qvUfM5KAg6Ct8VwpYzGff3uctyCc','SSwpkEEcbUqx4vtoEByFjSkhKdCT862DNVb52nZg1UZ','Crt7UoUR6QgrFrN7j8rmSQpUTNWNSitSwWvsWGf1qZ5t')
+)
 select 
     block_timestamp,
     block_id,
@@ -374,6 +390,8 @@ FROM
     l
     ON s.program_id = l.address
 union all
+{% endif %}
+
 select 
     block_timestamp,
     block_id,

models/gold/defi/defi__fact_swaps_jupiter_summary.sql

@@ -121,7 +121,7 @@ SELECT
     inserted_timestamp,
     modified_timestamp
 FROM
-    {{ ref('silver__swaps') }}
+    {{ ref('silver__swaps_view') }}
 WHERE
     program_id = 'JUP4Fb2cqiRUcaTHdrPC8h2gNsA2ETXiPDD33WcGuJB'
     and block_timestamp::date <= '2023-10-31'

models/silver/swaps/saber/silver__swaps_intermediate_saber.sql

@@ -0,0 +1,166 @@
+-- depends_on: {{ ref('silver__decoded_instructions_combined') }}
+
+{{ config(
+    materialized = 'incremental',
+    unique_key = "swaps_intermediate_saber_id",
+    incremental_predicates = ["dynamic_range_predicate", "block_timestamp::date"],
+    merge_exclude_columns = ["inserted_timestamp"],
+    cluster_by = ['block_timestamp::DATE','modified_timestamp::DATE'],
+    tags = ['scheduled_non_core','scheduled_non_core_hourly'],
+) }}
+
+{% if execute %}
+    {% set base_query %}
+    CREATE OR REPLACE TEMPORARY TABLE silver.swaps_intermediate_saber__intermediate_tmp AS
+    SELECT
+        block_timestamp,
+        block_id,
+        tx_id,
+        succeeded,
+        INDEX,
+        inner_index,
+        program_id,
+        event_type,
+        decoded_instruction,
+        _inserted_timestamp
+    FROM
+        {{ ref('silver__decoded_instructions_combined') }}
+    WHERE
+        program_id = 'SSwpkEEcbUqx4vtoEByFjSkhKdCT862DNVb52nZg1UZ'
+        AND event_type = 'swap'
+        AND succeeded
+
+    {% if is_incremental() %}
+    AND _inserted_timestamp >= (
+        SELECT
+            MAX(_inserted_timestamp) - INTERVAL '1 hour'
+        FROM
+            {{ this }}
+    )
+    {% else %}
+        AND _inserted_timestamp::DATE between '2025-04-30' and '2025-05-03'
+    {% endif %}
+    {% endset %}
+    
+    {% do run_query(base_query) %}
+    {% set between_stmts = fsc_utils.dynamic_range_predicate(
+        "silver.swaps_intermediate_saber__intermediate_tmp",
+        "block_timestamp::date"
+    ) %}
+{% endif %}
+
+with base AS (
+    SELECT
+        *
+    FROM
+        silver.swaps_intermediate_saber__intermediate_tmp
+),
+decoded AS (
+    SELECT
+        block_timestamp,
+        block_id,
+        tx_id,
+        INDEX,
+        inner_index,
+        COALESCE(LEAD(inner_index) OVER (PARTITION BY tx_id, index
+            ORDER BY inner_index) -1, 999999
+        ) AS inner_index_end,
+        program_id,
+        silver.udf_get_account_pubkey_by_name('userAuthority', decoded_instruction:accounts) as swapper,
+        silver.udf_get_account_pubkey_by_name('inputUser', decoded_instruction:accounts) as source_token_account,
+        null as source_mint,
+        null as destination_mint,
+        silver.udf_get_account_pubkey_by_name('outputUserTokenUser', decoded_instruction:accounts) as destination_token_account,  
+        silver.udf_get_account_pubkey_by_name('outputUserTokenReserve', decoded_instruction:accounts) as program_destination_token_account, 
+        silver.udf_get_account_pubkey_by_name('inputReserve', decoded_instruction:accounts) as program_source_token_account,
+        _inserted_timestamp
+    FROM
+        base
+)
+,
+transfers AS (
+    SELECT
+        A.*,
+        COALESCE(SPLIT_PART(INDEX :: text, '.', 1) :: INT, INDEX :: INT) AS index_1,
+        NULLIF(SPLIT_PART(INDEX :: text, '.', 2), '') :: INT AS inner_index_1
+    FROM
+        {{ ref('silver__transfers') }} A
+        INNER JOIN (
+            SELECT
+                DISTINCT tx_id,
+                    block_timestamp::DATE AS block_date
+            FROM
+                decoded
+        ) d
+        ON d.block_date = A.block_timestamp::DATE
+        AND d.tx_id = A.tx_id
+    WHERE
+        A.succeeded
+        and {{ between_stmts }}
+),
+pre_final AS (
+    SELECT
+        A.block_id,
+        A.block_timestamp,
+        A.program_id,
+        A.tx_id,
+        A.index,
+        A.inner_index,
+        A.inner_index_end,
+        C.succeeded,
+        A.swapper,
+        b.amount AS from_amt,
+        b.mint AS from_mint,
+        C.amount AS to_amt,
+        C.mint AS to_mint,
+        A._inserted_timestamp
+    FROM
+        decoded A
+        LEFT JOIN transfers b
+        ON A.tx_id = b.tx_id
+        AND A.source_token_account = b.source_token_account
+        AND A.program_source_token_account = b.dest_token_account
+        AND A.index = b.index_1
+        AND (
+            (b.inner_index_1 BETWEEN A.inner_index AND A.inner_index_end)
+            OR A.inner_index IS NULL
+        )
+        LEFT JOIN transfers C
+        ON A.tx_id = C.tx_id
+        AND A.destination_token_account = C.dest_token_account
+        AND A.program_destination_token_account = C.source_token_account
+        AND A.index = C.index_1
+        AND (
+            (C.inner_index_1 BETWEEN A.inner_index AND A.inner_index_end)
+            OR A.inner_index IS NULL
+        ) 
+    QUALIFY ROW_NUMBER() over (PARTITION BY A.tx_id, A.index, A.inner_INDEX ORDER BY inner_index) = 1
+)
+SELECT
+    block_id,
+    block_timestamp,
+    program_id,
+    tx_id,
+    succeeded,
+    ROW_NUMBER() over (
+        PARTITION BY tx_id
+        ORDER BY
+            INDEX,
+            inner_index
+    ) AS swap_index,
+    index,
+    inner_index,
+    swapper,
+    from_amt as swap_from_amount,
+    from_mint as swap_from_mint,
+    to_amt as swap_to_amount,
+    to_mint as swap_to_mint,
+    _inserted_timestamp,
+    {{ dbt_utils.generate_surrogate_key(['tx_id','index','inner_index']) }} AS swaps_intermediate_saber_id,
+    SYSDATE() AS inserted_timestamp,
+    SYSDATE() AS modified_timestamp,
+    '{{ invocation_id }}' AS _invocation_id
+FROM
+    pre_final
+WHERE
+    (swap_to_mint is not null and swap_to_amount is not null)

models/silver/swaps/saber/silver__swaps_intermediate_saber.yml

@@ -0,0 +1,83 @@
+version: 2
+models:
+  - name: silver__swaps_intermediate_saber
+    tests:
+      - dbt_utils.unique_combination_of_columns:
+          combination_of_columns:
+            - TX_ID
+            - INDEX
+            - INNER_INDEX
+          where: block_timestamp::date > current_date - 7
+    recent_date_filter: &recent_date_filter
+      config:
+        where: _inserted_timestamp >= current_date - 7
+    columns:
+      - name: BLOCK_TIMESTAMP
+        description: "{{ doc('block_timestamp') }}"
+        tests:
+          - not_null: *recent_date_filter
+          - dbt_expectations.expect_row_values_to_have_recent_data:
+              datepart: day
+              interval: 2
+      - name: BLOCK_ID
+        description: "{{ doc('block_id') }}"
+        tests:
+          - not_null: *recent_date_filter
+      - name: TX_ID
+        description: "{{ doc('tx_id') }}"
+        tests:
+          - not_null: *recent_date_filter
+      - name: INDEX
+        description: "{{ doc('event_index') }}"
+        tests: 
+          - not_null: *recent_date_filter
+      - name: INNER_INDEX
+        description: "{{ doc('inner_index') }}. This is the inner index of the log event listing the inner swap"
+      - name: SUCCEEDED
+        description: "{{ doc('tx_succeeded') }}"
+        tests: 
+          - not_null: *recent_date_filter
+      - name: PROGRAM_ID
+        description: "{{ doc('program_id') }}"
+        tests: 
+          - not_null: *recent_date_filter
+      - name: SWAPPER
+        description: "{{ doc('swaps_swapper') }}"
+        tests: 
+          - not_null: *recent_date_filter
+      - name: SWAP_FROM_AMOUNT
+        description:  "{{ doc('swaps_from_amt') }}"
+        tests: 
+          - not_null: *recent_date_filter
+      - name: SWAP_FROM_MINT
+        description:  "{{ doc('swaps_from_mint') }}"
+        tests: 
+          - not_null: *recent_date_filter
+      - name: SWAP_TO_AMOUNT
+        description:  "{{ doc('swaps_to_amt') }}"
+        tests: 
+          - not_null: *recent_date_filter
+      - name: SWAP_TO_MINT
+        description:  "{{ doc('swaps_to_mint') }}"
+        tests: 
+          - not_null: *recent_date_filter 
+      - name: _INSERTED_TIMESTAMP
+        description: "{{ doc('_inserted_timestamp') }}"
+      - name: SWAPS_INTERMEDIATE_SABER_ID
+        description: '{{ doc("pk") }}'   
+        tests: 
+          - unique: *recent_date_filter
+      - name: INSERTED_TIMESTAMP
+        description: '{{ doc("inserted_timestamp") }}'   
+        tests: 
+          - not_null: *recent_date_filter
+      - name: MODIFIED_TIMESTAMP
+        description: '{{ doc("modified_timestamp") }}' 
+        tests: 
+          - not_null: *recent_date_filter
+      - name: _INVOCATION_ID
+        description: '{{ doc("_invocation_id") }}' 
+        tests: 
+          - not_null: 
+              name: test_silver__not_null_swaps_intermediate_saber__invocation_id
+              <<: *recent_date_filter

models/silver/swaps/silver__swaps.sql

@@ -4,7 +4,8 @@
     incremental_strategy = 'merge',
     cluster_by = ['block_timestamp::DATE','modified_timestamp::DATE'],
     merge_exclude_columns = ["inserted_timestamp"],
-    tags = ['scheduled_non_core','scheduled_non_core_hourly']
+    full_refresh = false,
+    enabled = false,
 ) }}
 
 WITH base AS (

models/silver/swaps/silver__swaps_view.sql

@@ -0,0 +1,26 @@
+{{ config(
+  materialized = 'view'
+) }}
+
+SELECT
+    block_id,
+    block_timestamp,
+    program_id,
+    tx_id,
+    succeeded,
+    swapper,
+    from_amt,
+    from_mint,
+    to_amt,
+    to_mint,
+    _log_id,
+    _inserted_timestamp,
+    swaps_id,
+    inserted_timestamp,
+    modified_timestamp,
+    _invocation_id
+FROM
+  {{ source(
+    'solana_silver',
+    'swaps'
+  ) }}

models/sources.yml

@@ -119,6 +119,7 @@ sources:
       - name: nft_sales_solsniper
       - name: nft_sales_solanart
       - name: swaps_intermediate_orca
+      - name: swaps
   - name: solana_streamline
     database: solana
     schema: streamline