An 3943/get compressed nft addresses (#439)

* external funcs (wip)

* call compressed nft mint parser with external func

* ignore bronze_api models

* move to match standard location

* fix missing mints due to inconsistent ordering

* add a tag to bronze_api model, exclude accordingly

* update silver table to use new bronze api data

* wip

* handle non-mint events and multi mints per tx_id

* add regression test

.github/workflows/dbt_run_macro_get_compressed_nft.yml

@@ -42,5 +42,4 @@ jobs:
       - name: Run DBT Jobs
         run: |
           dbt run -s models/silver/nfts/silver__nft_compressed_mints_onchain.sql
-          dbt run-operation get_helius_compressed_nft
           dbt run -s models/silver/nfts/silver__nft_compressed_mints.sql

data/testing__nft_compressed_mints.csv

@@ -0,0 +1,10 @@
+tx_id,mint,leaf_owner,collection_mint,purchaser,instruction_name
+43nKZtMkpcKAJF5Sj9sekrVqM3StnTzuQkBqghzdgvWNDxoG3hTtQ8H2qHLaiYy4ogjErYxrAGShpfbKashFtqG9,3bXzBaxodhNdwW5HKeW2rKbj7rXdPpYNA5anPEVvptax,ErNMP3aJZWdb7YKGXZfdeVAAm9QH1XqNb4w6DceWmshf,EXGU3FiZDB4XY6WfJadvanjYxPJxJQTi6CMogM4wCZFd,noKiefqP19U1pe3Az1NSDJTWbaMNjPY584LVikxVkac,mintToCollectionV1
+43nKZtMkpcKAJF5Sj9sekrVqM3StnTzuQkBqghzdgvWNDxoG3hTtQ8H2qHLaiYy4ogjErYxrAGShpfbKashFtqG9,2SWSntiJpVhk4QvDwzdjS8wS5VumUSQsSBxJd8KVqWrG,5hXEWkdyXymkYiatScrM4GR1krAs4s9Zhs7Z9NDFSEyS,EXGU3FiZDB4XY6WfJadvanjYxPJxJQTi6CMogM4wCZFd,noKiefqP19U1pe3Az1NSDJTWbaMNjPY584LVikxVkac,mintToCollectionV1
+43cY1oaPwhJeQGbVCbhN8Pw9JsfN8CjHJ8xoPstEPebDVaWX9ZrvUDNfNXrdM1HxyWfkkMXogFm1VdmjPmhU4NqE,9LRXLLok4GSUqNXHWssVfwaqBsGZgabtWhaQXtyfd5sZ,8b5tQbMG2ReswMYZ7fNHM6KHjGoxXbdV3taJZoAbQdMj,E4UoNp3xrE2VYRe4huepMqVGjQD4SaJ8MvsRG4Dd49r7,7tYXi34ZR3u9x4opMPoUzyyagSCepPaYBXBLZDYnsa1u,mintToCollectionV1
+4pvJwVszQfX8JWpNMvv1hmvduTigCVnZabcs9KbqwusbPFjTWxLJzHCvzHZVCrg7svHgzuqaACp7uxMNMqemRc4,Bok26qz6UL5KthJWhT2rmz2iDG2qoxFoafKNveEwmakN,FiKfji18bGgRgurFtQHbNerchWTZHqhBgcFUUMd83urM,E4UoNp3xrE2VYRe4huepMqVGjQD4SaJ8MvsRG4Dd49r7,7tYXi34ZR3u9x4opMPoUzyyagSCepPaYBXBLZDYnsa1u,mintToCollectionV1
+3hnRu4Sp6s3pPKZ2pRCkrM1zQkC6HxrruKhDfBmX3qNbKn4mu4UL2M5W8JonnEn7zQKT3fpsSjyGu4qb5JbhXNZf,GrLmkGFiDC46LL1hk2agDdrXnf8yfh2FbW4YkVsmfcza,9XYtuY83nGZsQLW98UkCtts2o9iSfwukvFbSD1oc2kcC,E4UoNp3xrE2VYRe4huepMqVGjQD4SaJ8MvsRG4Dd49r7,7tYXi34ZR3u9x4opMPoUzyyagSCepPaYBXBLZDYnsa1u,mintToCollectionV1
+2mHjoyQ1fCnjeJk51ZiTU3RWXejjSsbzEgrEbHkgeRjQX4b5Y36FzLQzsKxYXpP8RcLeYPbuyMWXj4qfVGqa8cFD,3kKPEG3VsnJAoy3edVipfAhFg6huuycyQwBaNYeaFoiB,BMY299EkPXrf6g3ZggxqREGzDU3DWxAzAmrY9jrv3AXY,EZuLJQoRAQUWgyVFbuWK1exGGeAHDewnTRQwMiL2D1iQ,3T1fQuahAAwNGLoVdbLLviJyMVetsLt1KqShWSnFEtS4,mintToCollectionV1
+2mHjoyQ1fCnjeJk51ZiTU3RWXejjSsbzEgrEbHkgeRjQX4b5Y36FzLQzsKxYXpP8RcLeYPbuyMWXj4qfVGqa8cFD,5VTEwjc3SLKk3MHg6xv5Djk8s96Hx2iah2V2aAJZFykz,BMY14t4R1Kd3JbaXnVcfd35YvC45YftZXVYXr3nxfAd5,EZuLJQoRAQUWgyVFbuWK1exGGeAHDewnTRQwMiL2D1iQ,3T1fQuahAAwNGLoVdbLLviJyMVetsLt1KqShWSnFEtS4,mintToCollectionV1
+2mHjoyQ1fCnjeJk51ZiTU3RWXejjSsbzEgrEbHkgeRjQX4b5Y36FzLQzsKxYXpP8RcLeYPbuyMWXj4qfVGqa8cFD,FKeq26oCK5RccSPbcWt8ru6WC4WbWqoUEMfsRbqDjkNy,BMXzvpLFBBufyio9GcwfyfmB1muNSohiHZVkeYV4ZFrA,EZuLJQoRAQUWgyVFbuWK1exGGeAHDewnTRQwMiL2D1iQ,3T1fQuahAAwNGLoVdbLLviJyMVetsLt1KqShWSnFEtS4,mintToCollectionV1
+rQ4y9cvCKYKYUZPcxn6MJMpxPo3LYDSJ85Y4jpN59k7KPb1niZrWoFJ6XqShfDdWrxCFccRJuXMqwV3N3aNr7W2,6ELqXaV2uJRoc4jk4g7VcizvJMB3EvP6CVBoSPAsexPq,6RLj2uiJJiYqUsRksjdkHosggpboPZ2qrARHPq1PMwKw,DRiP3vrQ7LuvQtmwL74RcejviWmQ1HuuKFcJdUpYG6Di,DRiPPP2LytGjNZ5fVpdZS7Xi1oANSY3Df1gSxvUKpzny,mintToCollectionV1

macros/bronze_api/get_helius_compressed_nft.sql

@@ -1,113 +0,0 @@
-{% macro get_helius_compressed_nft() %}
-  {% set final_calls_query %}
-  CREATE temporary TABLE final_calls AS
-SELECT
-  ARRAY_AGG(
-    { 'id': 'my-id',
-    'jsonrpc': '2.0',
-    'method': 'searchAssets',
-    'params':
-      { 'compressed': TRUE,
-      'grouping': [ 'collection', collection_mint],
-      'page': page,
-      'limit': LIMIT, 
-      'sortBy':{ 'sortBy': 'created', 'sortDirection': 'desc' }} 
-    }
-  ) calls, 
-  collection_mint AS nft_collection_mint,
-  page,
-  LIMIT, 
-  -- FLOOR((ROW_NUMBER() OVER(ORDER BY collection_mint, page) - 1) / 2) as batch_id
-  (ROW_NUMBER() over(ORDER BY collection_mint, page) - 1) AS batch_id
-FROM
-  (
-    WITH counted_items AS (
-      SELECT
-        collection_mint,
-        COUNT(*) AS item_count,
-        CEIL((COUNT(*)) / 500) AS total_pages
-      FROM
-        {{ target.database }}.silver.nft_compressed_mints_onchain
-      WHERE
-        _inserted_timestamp >= (
-          SELECT
-            MAX(_inserted_timestamp)
-          FROM
-            {{ target.database }}.silver.nft_compressed_mints
-        )
-      GROUP BY
-        collection_mint
-    ),
-    temp AS (
-      SELECT
-        collection_mint,
-        CASE
-          WHEN item_count > 500
-          AND page < total_pages THEN 500
-          WHEN page = total_pages THEN 500
-          ELSE item_count
-        END AS LIMIT, 
-        page
-      FROM
-        counted_items
-        CROSS JOIN (
-          SELECT
-            ROW_NUMBER() over (
-              ORDER BY
-                SEQ4()
-            ) AS page
-          FROM
-            TABLE(GENERATOR(rowcount => 10000))
-        ) page_nums
-      WHERE
-        page_nums.page <= counted_items.total_pages
-        OR (
-          item_count > 500
-          AND page_nums.page = counted_items.total_pages
-        )
-      ORDER BY
-        collection_mint,
-        page
-    )
-    SELECT
-      *
-    FROM
-      temp
-  )
-GROUP BY collection_mint, LIMIT, page
-ORDER BY batch_id ASC;
-{% endset %}
-  {% do run_query(final_calls_query) %}
-  {% for batch_id in range(0,5000) %}
-    {% set results_query %}
-  INSERT INTO
-    {{ target.database }}.bronze_API.helius_compressed_nfts WITH results AS (
-      SELECT
-        ethereum.streamline.udf_json_rpc_call(
-          'https://rpc.helius.xyz/?api-key=' || (
-            SELECT
-              api_key
-            FROM
-              crosschain.silver.apis_keys
-            WHERE
-              api_name = 'helius'
-          ),{},
-          calls
-        ) AS DATA,
-        nft_collection_mint,
-        page,
-        LIMIT, 
-        SYSDATE() AS _inserted_timestamp
-      FROM
-        final_calls
-      WHERE
-        batch_id = {{ batch_id }}
-    )
-  SELECT
-    *
-  FROM
-    results;
-{% endset %}
-    {% do run_query(results_query) %}
-  {% endfor %}
-{% endmacro %}

models/gold/nft/nft__fact_nft_mints.sql

@@ -46,7 +46,7 @@ SELECT
     COALESCE (
         nft_compressed_mints_id,
         {{ dbt_utils.generate_surrogate_key(
-            ['tx_id']
+            ['tx_id','mint']
         ) }}
     ) AS fact_nft_mints_id,
     COALESCE(

models/silver/nfts/silver__nft_compressed_mints.sql

@@ -8,88 +8,121 @@
 WITH offchain AS (
 
     SELECT
-        items.value,
-        items.value ['id'] :: STRING AS mint,
+        r.value :tx_id :: STRING AS tx_id,
+        r.value :index :: INTEGER AS mint_index,
+        r.value :inner_index :: INTEGER AS mint_inner_index,
+        COALESCE(
+            r.value :mint :: STRING,
+            ''
+        ) AS mint,
         0.000005 AS mint_price,
         'So11111111111111111111111111111111111111111' AS mint_currency,
         'BGUMAp9Gq7iTEuizy4pqaxsTyUCBK68MDfK752saRPUY' AS program_id,
-        page,
-        _inserted_timestamp,
-        nft_collection_mint,
-        items.seq AS seq
+        start_inserted_timestamp AS _inserted_timestamp
     FROM
-        {{ target.database }}.bronze_api.helius_compressed_nfts,
-        LATERAL FLATTEN(
-            input => DATA :data [0] :result :items
-        ) AS items
-
-{% if is_incremental() %}
-WHERE _inserted_timestamp >= (
-    SELECT
-        MAX(_inserted_timestamp)
-    FROM
-        {{ this }}
-)
-{% endif %}
-
-qualify(ROW_NUMBER() over (PARTITION BY mint
-ORDER BY
-    _inserted_timestamp DESC)) = 1
+        {{ ref('bronze_api__parse_compressed_nft_mints') }},
+        TABLE(FLATTEN(responses)) AS r
+    WHERE
+        mint <> '' 
+    {% if is_incremental() %}
+    AND _inserted_timestamp >= (
+        SELECT
+            MAX(_inserted_timestamp)
+        FROM
+            {{ this }}
+    )
+    {% endif %}
+    qualify(ROW_NUMBER() over (PARTITION BY tx_id, mint
+    ORDER BY
+        _inserted_timestamp DESC)) = 1
 ),
 offchain_ordered AS (
     SELECT
         *,
         ROW_NUMBER() over (
-            PARTITION BY nft_collection_mint
+            PARTITION BY tx_id
             ORDER BY
-                page,
-                seq
-        ) AS ROW_NUMBER
+                mint_index,
+                mint_inner_index
+        ) AS instruction_order
     FROM
         offchain
 ),
+decoded AS (
+    SELECT
+        decoded_instruction :name :: STRING AS instruction_name,
+        *
+    FROM
+        {{ ref('silver__decoded_instructions') }}
+    WHERE
+        program_id = 'BGUMAp9Gq7iTEuizy4pqaxsTyUCBK68MDfK752saRPUY'
+    {% if is_incremental() %}
+    AND _inserted_timestamp >= (
+        SELECT
+            MAX(_inserted_timestamp)
+        FROM
+            {{ this }}
+    )
+    {% endif %}
+),
 onchain AS (
     SELECT
-        *,
+        m.*,
+        d.instruction_name,
         ROW_NUMBER() over (
-            PARTITION BY collection_mint
+            PARTITION BY m.tx_id
             ORDER BY
-                block_timestamp DESC
-        ) AS ROW_NUMBER
+                m.index,
+                m.inner_index
+        ) AS instruction_order
     FROM
         {{ ref('silver__nft_compressed_mints_onchain') }}
-
-{% if is_incremental() %}
-WHERE _inserted_timestamp >= (
-    SELECT
-        MAX(_inserted_timestamp)
-    FROM
-        {{ this }}
-)
-{% endif %}
+        m
+        LEFT OUTER JOIN decoded d
+        ON d.tx_id = m.tx_id
+        AND d.index = m.index
+        AND COALESCE(
+            d.inner_index,
+            -1
+        ) = COALESCE(
+            m.inner_index,
+            -1
+        )
+    {% if is_incremental() %}
+    WHERE _inserted_timestamp >= (
+        SELECT
+            MAX(_inserted_timestamp)
+        FROM
+            {{ this }}
+    )
+    {% endif %}
 )
 SELECT
     A.block_timestamp,
     A.block_id,
     A.succeeded,
-    A.tx_id,
+    b.tx_id,
     A.leaf_owner,
     A.collection_mint,
-    A._inserted_timestamp,
+    b._inserted_timestamp,
     A.creator_address AS purchaser,
     b.mint,
     b.mint_price,
     b.mint_currency,
     b.program_id,
+    A.instruction_name,
     {{ dbt_utils.generate_surrogate_key(
-        ['tx_id']
+        ['b.tx_id','b.mint']
     ) }} AS nft_compressed_mints_id,
     SYSDATE() AS inserted_timestamp,
     SYSDATE() AS modified_timestamp
 FROM
-    onchain A
-    LEFT JOIN offchain_ordered b
-    ON A.row_number = b.row_number
-    AND A.collection_mint = b.nft_collection_mint
+    offchain_ordered b
+    LEFT JOIN onchain A
+    ON A.tx_id = b.tx_id
+    AND A.instruction_order = b.instruction_order
 WHERE
-    b.row_number IS NOT NULL
+    (
+        A.instruction_name LIKE 'mint%'
+        OR A.instruction_name IS NULL
+    )

models/silver/nfts/silver__nft_compressed_mints.yml

@@ -8,6 +8,22 @@ models:
             - TX_ID
             - MINT
             - PURCHASER
+      - compare_model_subset:
+          name: silver__nft_compressed_mints_business_logic_test
+          compare_model: ref('testing__nft_compressed_mints')
+          compare_columns: 
+            - tx_id
+            - mint
+            - leaf_owner
+            - collection_mint
+            - purchaser
+            - instruction_name
+          model_condition: "where tx_id in ('43nKZtMkpcKAJF5Sj9sekrVqM3StnTzuQkBqghzdgvWNDxoG3hTtQ8H2qHLaiYy4ogjErYxrAGShpfbKashFtqG9',
+            '43cY1oaPwhJeQGbVCbhN8Pw9JsfN8CjHJ8xoPstEPebDVaWX9ZrvUDNfNXrdM1HxyWfkkMXogFm1VdmjPmhU4NqE',
+            '4pvJwVszQfX8JWpNMvv1hmvduTigCVnZabcs9KbqwusbPFjTWxLJzHCvzHZVCrg7svHgzuqaACp7uxMNMqemRc4',
+            '3hnRu4Sp6s3pPKZ2pRCkrM1zQkC6HxrruKhDfBmX3qNbKn4mu4UL2M5W8JonnEn7zQKT3fpsSjyGu4qb5JbhXNZf',
+            '2mHjoyQ1fCnjeJk51ZiTU3RWXejjSsbzEgrEbHkgeRjQX4b5Y36FzLQzsKxYXpP8RcLeYPbuyMWXj4qfVGqa8cFD',
+            'rQ4y9cvCKYKYUZPcxn6MJMpxPo3LYDSJ85Y4jpN59k7KPb1niZrWoFJ6XqShfDdWrxCFccRJuXMqwV3N3aNr7W2')" 
     columns:
       - name: BLOCK_TIMESTAMP
         description: "{{ doc('block_timestamp') }}"
@@ -47,6 +63,10 @@ models:
         description: "Address of the NFT collection"
         tests: 
           - not_null
+      - name: INSTRUCTION_NAME
+        description: "Name of instruction being performed by the program"
+        tests:
+          - not_null
       - name: _INSERTED_TIMESTAMP
         description: "{{ doc('_inserted_timestamp') }}"
         tests: 

models/silver/nfts/silver__nft_compressed_mints_onchain.sql

@@ -12,6 +12,8 @@ WITH bgum_mints AS (
     block_id,
     succeeded,
     tx_id,
+    e.index,
+    null as inner_index,
     instruction :accounts [1] :: STRING AS leaf_owner,
     instruction :accounts [8] :: STRING AS collection_mint,
     signers [0] :: STRING AS creator_address,
@@ -56,6 +58,8 @@ SELECT
   block_id,
   succeeded,
   tx_id,
+  e.index,
+  f.index as inner_index,
   f.value :accounts [1] :: STRING AS leaf_owner,
   f.value :accounts [8] :: STRING AS collection_mint,
   signers [0] :: STRING AS creator_address,