Topshot data (#440)

* livequery UDF model * streamline models * Add streamline model for TopShot metadata real-time processing * buyback table * add * topshot and all day * added streamline changes * realtime * update * updated * SL work - proposed changes (#441) * updates * . * rev * view to table * fixed a few bugs * sql limit --------- Co-authored-by: eric-laurello <102970824+eric-laurello@users.noreply.github.com> Co-authored-by: Eric Laurello <eric.laurello@flipsidecrypto.com>
2026-02-06 16:06:48 +00:00 · 2025-05-28 15:40:14 -05:00 · 2025-05-28 15:40:14 -05:00 · 3d13e03a45
commit 3d13e03a45
parent 0dd8d35c1f
13 changed files with 958 additions and 37 deletions
--- a/models/gold/nft/nft__dim_topshot_metadata.sql
+++ b/models/gold/nft/nft__dim_topshot_metadata.sql
@ -1,17 +1,46 @@
-{{ config (
-    materialized = 'view',
-    meta={
-    'database_tags':{
-        'table': {
-            'PURPOSE': 'NFT, TOPSHOT'
-            }
-        }
-    },
-    tag = ['scheduled']
+{{ config(
+    materialized = 'incremental',
+    incremental_strategy = 'merge',
+    merge_exclude_columns = ['inserted_timestamp'],
+    cluster_by = ['left(season,4)'],
+    unique_key = "nft_id",
+    post_hook = "ALTER TABLE {{ this }} ADD SEARCH OPTIMIZATION ON EQUALITY(nft_id,nbatopshot_id);",
+    tags = ['scheduled_non_core'],
+    meta ={ 'database_tags':{ 'table':{ 'PURPOSE': 'NFT, TOPSHOT' }} }
 ) }}
+-- depends_on: {{ ref('bronze__streamline_topshot_metadata') }}
+WITH

-WITH topshot AS (
+{% if is_incremental() %}
+{% else %}
+    topshot_old AS (

+        SELECT
+            nft_id,
+            nft_collection,
+            nbatopshot_id,
+            serial_number,
+            total_circulation,
+            moment_description,
+            player,
+            team,
+            season,
+            play_category,
+            play_type,
+            moment_date,
+            set_name,
+            set_series_number,
+            video_urls,
+            moment_stats_full,
+            player_stats_game,
+            player_stats_season_to_date,
+            nft_moment_metadata_topshot_id AS dim_topshot_metadata_id
+        FROM
+            {{ ref('silver__nft_topshot_metadata_view') }}
+    ),
+{% endif %}
+
+topshot AS (
    SELECT
        nft_id,
        nft_collection,
@ -31,13 +60,74 @@ WITH topshot AS (
        moment_stats_full,
        player_stats_game,
        player_stats_season_to_date,
-        nft_moment_metadata_topshot_id as dim_topshot_metadata_id,
-        inserted_timestamp,
-        modified_timestamp
+        nft_topshot_metadata_v2_id AS dim_topshot_metadata_id
    FROM
-        {{ ref('silver__nft_topshot_metadata') }}
+        {{ ref('silver__nft_topshot_metadata_v2') }}
+
+{% if is_incremental() %}
+WHERE
+    modified_timestamp >= (
+        SELECT
+            MAX(modified_timestamp) modified_timestamp
+        FROM
+            {{ this }}
+    )
+{% endif %}
+),
+ua AS (
+    SELECT
+        nft_id,
+        nft_collection,
+        nbatopshot_id,
+        serial_number,
+        total_circulation,
+        moment_description,
+        player,
+        team,
+        season,
+        play_category,
+        play_type,
+        moment_date,
+        set_name,
+        set_series_number,
+        video_urls,
+        moment_stats_full,
+        player_stats_game,
+        player_stats_season_to_date,
+        dim_topshot_metadata_id
+    FROM
+        topshot
+
+{% if is_incremental() %}
+{% else %}
+    UNION ALL
+    SELECT
+        nft_id,
+        nft_collection,
+        nbatopshot_id,
+        serial_number,
+        total_circulation,
+        moment_description,
+        player,
+        team,
+        season,
+        play_category,
+        play_type,
+        moment_date,
+        set_name,
+        set_series_number,
+        video_urls,
+        moment_stats_full,
+        player_stats_game,
+        player_stats_season_to_date,
+        dim_topshot_metadata_id
+    FROM
+        topshot_old
+    {% endif %}
 )
 SELECT
-    *
+    *,
+    SYSDATE() AS inserted_timestamp,
+    SYSDATE() AS modified_timestamp
 FROM
-    topshot
+    ua
--- a/models/gold/nft/nft__fact_topshot_buybacks.sql
+++ b/models/gold/nft/nft__fact_topshot_buybacks.sql
@ -0,0 +1,124 @@
+{{ config(
+    materialized = 'incremental',
+    incremental_strategy = 'merge',
+    merge_exclude_columns = ['inserted_timestamp'],
+    incremental_predicates = ["COALESCE(DBT_INTERNAL_DEST.block_timestamp::DATE,'2099-12-31') >= (select min(block_timestamp::DATE) from " ~ generate_tmp_view_name(this) ~ ")"],
+    cluster_by = ['block_timestamp::date', 'modified_timestamp::date'],
+    unique_key = "topshot_buyback_id",
+    tags = ['nft', 'topshot', 'scheduled'],
+    meta = { 'database_tags': { 'table': { 'PURPOSE': 'NFT, TOPSHOT' } } }
+) }}
+
+WITH flowty_sales AS (
+  SELECT
+    tx_id AS tx_id,
+    block_timestamp,
+    EVENT_DATA:buyer :: string AS buyer,
+    event_data:storefrontAddress :: string AS seller,
+    CAST(EVENT_DATA:"salePrice" AS DECIMAL(18, 2)) AS price,
+    event_data:salePaymentVaultType as currency, 
+    'A.3cdbb3d569211ff3.NFTStorefrontV2' as marketplace,
+    'FLOWTY' as sale_type,
+    EVENT_DATA:nftType :: string as nft_collection,
+    EVENT_DATA:nftID :: string as nft_id,
+    modified_timestamp
+  FROM
+    {{ ref('core__fact_events') }} AS events
+  WHERE
+    EVENT_CONTRACT IN ('A.3cdbb3d569211ff3.NFTStorefrontV2')
+    AND EVENT_TYPE = 'ListingCompleted'
+    AND TX_SUCCEEDED = TRUE
+    AND EVENT_DATA:purchased :: string = 'true'
+    AND CAST(EVENT_DATA:"salePrice" AS DECIMAL(18, 2)) > 0
+    AND EVENT_DATA:nftType :: string = 'A.0b2a3299cc857e29.TopShot'
+    
+    {% if is_incremental() %}
+    AND modified_timestamp >= (
+        SELECT
+            MAX(modified_timestamp)
+        FROM
+            {{ this }}
+    )
+    {% endif %}
+),
+
+all_sales AS (
+    SELECT
+        tx_id AS tx_id,
+        CONVERT_TIMEZONE('UTC', 'America/New_York', BLOCK_TIMESTAMP) as block_timestamp,
+        buyer AS buyer,
+        seller AS seller,
+        price AS price,
+        nft_collection AS nft_collection,
+        nft_id AS nft_id,
+        modified_timestamp
+    FROM {{ ref('nft__ez_nft_sales') }} AS sales
+    WHERE nft_collection = 'A.0b2a3299cc857e29.TopShot'
+        AND TX_SUCCEEDED = TRUE
+        
+        {% if is_incremental() %}
+        AND modified_timestamp >= (
+            SELECT
+                MAX(modified_timestamp)
+            FROM
+                {{ this }}
+        )
+        {% endif %}
+    
+    UNION ALL
+    
+    SELECT
+        tx_id AS tx_id,
+        block_timestamp AS block_timestamp,
+        buyer AS buyer,
+        seller AS seller,
+        price AS price,
+        nft_collection AS nft_collection,
+        nft_id AS nft_id,
+        modified_timestamp
+    FROM flowty_sales AS fs
+),
+
+    sales_with_running_total AS (
+        SELECT
+            block_timestamp AS block_timestamp,
+            DATE_TRUNC('DAY', block_timestamp) as block_day,
+            tx_id AS tx_id,
+            nft_id AS nft_id,
+            buyer AS buyer,
+            seller AS seller,
+            price AS price,
+            1 as sale_count,
+            ROW_NUMBER() OVER (PARTITION BY tx_id, nft_id ORDER BY block_timestamp) as rn,
+            SUM(price) OVER (ORDER BY block_timestamp ROWS UNBOUNDED PRECEDING) as running_total,
+            modified_timestamp
+        FROM all_sales AS asales
+        WHERE buyer = '0xe1f2a091f7bb5245'  -- Filtering for TopShot buyback wallet
+    )
+
+    SELECT
+        s.block_timestamp AS block_timestamp,
+        s.block_height AS block_height,
+        s.tx_id AS tx_id,
+        s.nft_id AS nft_id,
+        COALESCE(ts.player, mm.metadata:player::string) as player,
+        COALESCE(ts.team, mm.metadata:team::string) as team,
+        COALESCE(ts.season, mm.metadata:season::string) as season,
+        COALESCE(ts.set_name, mm.set_name) as set_name,
+        s.buyer AS buyer,
+        s.seller AS seller,
+        s.price AS price,
+        s.sale_count as sale,
+        s.running_total as total,
+        CONCAT($$https://nbatopshot.com/moment/$$, s.nft_id) AS URL,
+        {{ dbt_utils.generate_surrogate_key(['s.tx_id', 's.nft_id']) }} AS topshot_buyback_id,
+        SYSDATE() AS inserted_timestamp,
+        SYSDATE() AS modified_timestamp,
+        '{{ invocation_id }}' AS _invocation_id
+    FROM sales_with_running_total s
+    LEFT JOIN {{ ref('nft__dim_topshot_metadata') }} ts
+        ON s.nft_id = ts.nft_id
+    LEFT JOIN {{ ref('nft__dim_moment_metadata') }} mm
+        ON s.nft_id = mm.nft_id
+        AND ts.player IS NULL -- Only pull from moment_metadata if topshot_metadata is empty
+    WHERE s.rn = 1  -- Deduplicate if needed
--- a/models/gold/nft/nft__fact_topshot_buybacks.yml
+++ b/models/gold/nft/nft__fact_topshot_buybacks.yml
@ -0,0 +1,155 @@
+version: 2
+
+models:
+  - name: nft__fact_topshot_buybacks
+    description: |-
+      This table captures NBA TopShot buyback activity - transactions where the TopShot buyback wallet purchases moments from users.
+      The model combines sales from both standard marketplace and Flowty, and includes a running total of amount spent over time.
+    tests:
+      - dbt_utils.recency:
+          datepart: day
+          field: block_timestamp
+          interval: 3
+
+    columns:
+      - name: block_timestamp
+        description: "Timestamp of the block in Eastern time (America/New_York)"
+        tests:
+          - not_null
+          - dbt_expectations.expect_column_values_to_be_in_type_list:
+              column_type_list:
+                - TIMESTAMP_NTZ
+
+      - name: block_height
+        description: "Height of the block in the blockchain"
+        tests:
+          - not_null
+          - dbt_expectations.expect_column_values_to_be_in_type_list:
+              column_type_list:
+                - NUMBER
+                - INTEGER
+
+      - name: tx_id
+        description: "{{ doc('tx_id') }}"
+        tests:
+          - not_null
+
+      - name: nft_id
+        description: "{{ doc('nft_id') }}"
+        tests:
+          - not_null
+
+      - name: player
+        description: "Player name from TopShot metadata or moment metadata"
+        tests:
+          - not_null
+          - dbt_expectations.expect_column_values_to_be_in_type_list:
+              column_type_list:
+                - STRING
+                - VARCHAR
+
+      - name: team
+        description: "Team name from TopShot metadata or moment metadata"
+        tests:
+          - not_null
+          - dbt_expectations.expect_column_values_to_be_in_type_list:
+              column_type_list:
+                - STRING
+                - VARCHAR
+
+      - name: season
+        description: "Season from TopShot metadata or moment metadata"
+        tests:
+          - not_null
+          - dbt_expectations.expect_column_values_to_be_in_type_list:
+              column_type_list:
+                - STRING
+                - VARCHAR
+
+      - name: set_name
+        description: "Set name from TopShot metadata or moment metadata"
+        tests:
+          - not_null
+          - dbt_expectations.expect_column_values_to_be_in_type_list:
+              column_type_list:
+                - STRING
+                - VARCHAR
+
+      - name: buyer
+        description: "Buyer address (TopShot buyback wallet)"
+        tests:
+          - not_null
+          - dbt_expectations.expect_column_values_to_be_in_type_list:
+              column_type_list:
+                - STRING
+                - VARCHAR
+          - accepted_values:
+              values: ['0xe1f2a091f7bb5245']
+
+      - name: seller
+        description: "Seller address"
+        tests:
+          - not_null
+          - dbt_expectations.expect_column_values_to_be_in_type_list:
+              column_type_list:
+                - STRING
+                - VARCHAR
+
+      - name: price
+        description: "Price of the moment in USD"
+        tests:
+          - not_null
+          - dbt_expectations.expect_column_values_to_be_in_type_list:
+              column_type_list:
+                - NUMBER
+                - FLOAT
+                - DECIMAL
+
+      - name: sale
+        description: "Counter field, always 1 representing a single sale transaction"
+        tests:
+          - not_null
+          - dbt_expectations.expect_column_values_to_be_in_type_list:
+              column_type_list:
+                - NUMBER
+                - INTEGER
+
+      - name: total
+        description: "Running total of all buyback purchases up to this transaction"
+        tests:
+          - not_null
+          - dbt_expectations.expect_column_values_to_be_in_type_list:
+              column_type_list:
+                - NUMBER
+                - FLOAT
+                - DECIMAL
+
+      - name: URL
+        description: "Direct link to the TopShot moment on the website"
+        tests:
+          - not_null
+          - dbt_expectations.expect_column_values_to_be_in_type_list:
+              column_type_list:
+                - STRING
+                - VARCHAR
+
+      - name: topshot_buyback_id
+        description: "Surrogate key generated from tx_id and nft_id"
+        tests:
+          - not_null
+          - unique
+
+      - name: inserted_timestamp
+        description: "Timestamp when the record was inserted"
+        tests:
+          - not_null
+
+      - name: modified_timestamp
+        description: "Timestamp when the record was last modified"
+        tests:
+          - not_null
+
+      - name: _invocation_id
+        description: "Unique identifier for the dbt run that created this record"
+        tests:
+          - not_null
--- a/models/silver/nft/metadata/livequery/topshot/silver__nft_topshot_metadata.sql
+++ b/models/silver/nft/metadata/livequery/topshot/silver__nft_topshot_metadata.sql
@ -4,22 +4,23 @@
    cluster_by = ['_inserted_timestamp::DATE'],
    unique_key = 'nft_id',
    tags = ['livequery', 'topshot'],
-    post_hook = "ALTER TABLE {{ this }} ADD SEARCH OPTIMIZATION ON EQUALITY(nft_id,nbatopshot_id);",
-    full_refresh = False
+    full_refresh = False,
+    enabled = false
 ) }}
-{# NFT Metadata from legacy process lives in external table, deleted CTE and set FR=False 
-to limit / avoid unnecessary table scans #}
+{# NFT Metadata from legacy process lives in external table, deleted CTE and set FR=False
+TO

-WITH metadata_lq AS (
-
-    SELECT
-        _res_id,
-        'A.0b2a3299cc857e29.TopShot' AS contract,
-        moment_id,
-        DATA :data :data :: variant AS DATA,
-        _inserted_timestamp
-    FROM
-        {{ ref('livequery__request_topshot_metadata') }}
+LIMIT
+    / avoid unnecessary TABLE scans #}
+    WITH metadata_lq AS (
+        SELECT
+            _res_id,
+            'A.0b2a3299cc857e29.TopShot' AS contract,
+            moment_id,
+            DATA :data :data :: variant AS DATA,
+            _inserted_timestamp
+        FROM
+            {{ ref('livequery__request_topshot_metadata') }}

 {% if is_incremental() %}
 WHERE
@ -33,8 +34,8 @@ WHERE
 ),
 lq_final AS (
    SELECT
-        moment_id AS nft_id,
-        contract AS nft_collection,
+        moment_id :: STRING AS nft_id,
+        contract :: STRING AS nft_collection,
        DATA :getMintedMoment :data :id :: STRING AS nbatopshot_id,
        DATA :getMintedMoment :data :flowSerialNumber :: NUMBER AS serial_number,
        DATA :getMintedMoment :data :setPlay :circulationCount :: NUMBER AS total_circulation,
@ -60,11 +61,11 @@ lq_final AS (
 SELECT
    *,
    {{ dbt_utils.generate_surrogate_key(
-            ['nft_id']
-        ) }} AS nft_moment_metadata_topshot_id,
+        ['nft_id']
+    ) }} AS nft_moment_metadata_topshot_id,
    SYSDATE() AS inserted_timestamp,
    SYSDATE() AS modified_timestamp,
-    '{{ invocation_id }}' AS _invocation_id   
+    '{{ invocation_id }}' AS _invocation_id
 FROM
    lq_final qualify ROW_NUMBER() over (
        PARTITION BY nft_id
--- a/models/silver/nft/metadata/livequery/topshot/silver__nft_topshot_metadata_v2.sql
+++ b/models/silver/nft/metadata/livequery/topshot/silver__nft_topshot_metadata_v2.sql
@ -0,0 +1,64 @@
+{{ config(
+    materialized = 'incremental',
+    incremental_strategy = 'delete+insert',
+    cluster_by = ['_inserted_timestamp::DATE'],
+    unique_key = 'nft_id',
+    tags = ['streamline', 'topshot']
+) }}
+-- depends_on: {{ ref('bronze__streamline_topshot_metadata') }}
+WITH metadata_from_streamline AS (
+
+    SELECT
+        VALUE :CONTRACT AS contract,
+        VALUE :ID AS moment_id,
+        DATA,
+        _inserted_timestamp
+    FROM
+
+{% if is_incremental() %}
+{{ ref('bronze__streamline_topshot_metadata') }}
+WHERE
+    _inserted_timestamp >= (
+        SELECT
+            MAX(_inserted_timestamp) _inserted_timestamp
+        FROM
+            {{ this }}
+    )
+{% else %}
+    {{ ref('bronze__streamline_topshot_metadata_FR') }}
+{% endif %}
+)
+SELECT
+    moment_id :: STRING AS nft_id,
+    contract :: STRING AS nft_collection,
+    DATA :data :getMintedMoment :data :id :: STRING AS nbatopshot_id,
+    DATA :data :getMintedMoment :data :flowSerialNumber :: NUMBER AS serial_number,
+    DATA :data :getMintedMoment :data :setPlay :circulationCount :: NUMBER AS total_circulation,
+    DATA :data :getMintedMoment :data :play :description :: VARCHAR AS moment_description,
+    DATA :data :getMintedMoment :data :play :stats :playerName :: STRING AS player,
+    DATA :data :getMintedMoment :data :play :stats :teamAtMoment :: STRING AS team,
+    DATA :data :getMintedMoment :data :play :stats :nbaSeason :: STRING AS season,
+    DATA :data :getMintedMoment :data :play :stats :playCategory :: STRING AS play_category,
+    DATA :data :getMintedMoment :data :play :stats :playType :: STRING AS play_type,
+    DATA :data :getMintedMoment :data :play :stats :dateOfMoment :: TIMESTAMP AS moment_date,
+    DATA :data :getMintedMoment :data :set :flowName :: STRING AS set_name,
+    DATA :data :getMintedMoment :data :set :flowSeriesNumber :: NUMBER AS set_series_number,
+    DATA :data :getMintedMoment :data :play :assets :videos :: ARRAY AS video_urls,
+    DATA :data :getMintedMoment :data :play :stats :: OBJECT AS moment_stats_full,
+    DATA :data :getMintedMoment :data :play :statsPlayerGameScores :: OBJECT AS player_stats_game,
+    DATA :data :getMintedMoment :data :play :statsPlayerSeasonAverageScores :: OBJECT AS player_stats_season_to_date,
+    _inserted_timestamp,
+    {{ dbt_utils.generate_surrogate_key(
+        ['nft_id']
+    ) }} AS nft_topshot_metadata_v2_id,
+    SYSDATE() AS inserted_timestamp,
+    SYSDATE() AS modified_timestamp,
+    '{{ invocation_id }}' AS _invocation_id
+FROM
+    metadata_from_streamline
+WHERE
+    DATA :errors IS NULL qualify ROW_NUMBER() over (
+        PARTITION BY nft_id
+        ORDER BY
+            _inserted_timestamp DESC
+    ) = 1
--- a/models/silver/nft/metadata/livequery/topshot/silver__nft_topshot_metadata_view.sql
+++ b/models/silver/nft/metadata/livequery/topshot/silver__nft_topshot_metadata_view.sql
@ -0,0 +1,33 @@
+{{ config(
+    materialized = 'view'
+) }}
+
+SELECT
+    nft_id,
+    nft_collection,
+    nbatopshot_id,
+    serial_number,
+    total_circulation,
+    moment_description,
+    player,
+    team,
+    season,
+    play_category,
+    play_type,
+    moment_date,
+    set_name,
+    set_series_number,
+    video_urls,
+    moment_stats_full,
+    player_stats_game,
+    player_stats_season_to_date,
+    _INSERTED_TIMESTAMP,
+    nft_moment_metadata_topshot_id,
+    inserted_timestamp,
+    modified_timestamp,
+    _INVOCATION_ID
+FROM
+    {{ source(
+        'silver',
+        'nft_topshot_metadata'
+    ) }}
--- a/models/sources.yml
+++ b/models/sources.yml
@ -132,6 +132,7 @@ sources:
      - name: contract_abis
      - name: evm_traces_v2
      - name: evm_decoded_logs
+      - name: topshot_metadata

  - name: crosschain_silver
    database: crosschain
@ -184,4 +185,10 @@ sources:
      - name: evm_known_event_sigs
      - name: evm_known_event_names
      - name: evm_event_sigs
-      - name: dates
+      - name: dates
+
+
+  - name: silver
+    schema: silver
+    tables:
+      - name: nft_topshot_metadata
--- a/models/streamline/external/topshot/bronze__streamline_topshot_metadata.sql
+++ b/models/streamline/external/topshot/bronze__streamline_topshot_metadata.sql
@ -0,0 +1,8 @@
+{{ config (
+    materialized = 'view',
+    tags = ['streamline_non_core']
+) }}
+{{ streamline_external_table_query_v2(
+    model = "topshot_metadata",
+    partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 3), '_', 1) AS INTEGER )"
+) }}
--- a/models/streamline/external/topshot/bronze__streamline_topshot_metadata_FR.sql
+++ b/models/streamline/external/topshot/bronze__streamline_topshot_metadata_FR.sql
@ -0,0 +1,8 @@
+{{ config (
+    materialized = 'view',
+    tags = ['streamline_non_core']
+) }}
+{{ streamline_external_table_FR_query_v2(
+    model = "topshot_metadata",
+    partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 3), '_', 1) AS INTEGER )"
+) }}
--- a/models/streamline/external/topshot/streamline__topshot_metadata_complete.sql
+++ b/models/streamline/external/topshot/streamline__topshot_metadata_complete.sql
@ -0,0 +1,68 @@
+-- depends_on: {{ ref('bronze__streamline_topshot_metadata') }}
+-- depends_on: {{ ref('bronze__streamline_topshot_metadata_FR') }}
+{{ config (
+    materialized = "incremental",
+    unique_key = "topshot_metadata_complete_id",
+    merge_exclude_columns = ["inserted_timestamp"],
+    tags = ['streamline_complete']
+) }}
+
+SELECT
+    VALUE :CONTRACT :: STRING AS event_contract,
+    VALUE :id :: STRING AS moment_id,
+    DATA :errors :extensions :error_reason :: STRING AS error_reason,
+    partition_key :: STRING AS partition_key,
+    _inserted_timestamp,
+    {{ dbt_utils.generate_surrogate_key(
+        ['event_contract','moment_id']
+    ) }} AS topshot_metadata_complete_id,
+    SYSDATE() AS inserted_timestamp,
+    SYSDATE() AS modified_timestamp,
+    '{{ invocation_id }}' AS _invocation_id
+FROM
+
+{% if is_incremental() %}
+{{ ref('bronze__streamline_topshot_metadata') }}
+WHERE
+    _inserted_timestamp >= COALESCE(
+        (
+            SELECT
+                MAX(_inserted_timestamp) _inserted_timestamp
+            FROM
+                {{ this }}
+        ),
+        '1900-01-01' :: timestamp_ntz
+    )
+{% else %}
+    {{ ref('bronze__streamline_topshot_metadata_FR') }}
+{% endif %}
+
+{% if is_incremental() %}
+{% else %}
+    UNION ALL
+    SELECT
+        contract AS event_contract,
+        id AS moment_id,
+        CASE
+            WHEN len(
+                DATA :getMintedMoment
+            ) IS NULL THEN 'null data'
+        END error_reason,
+        to_char(TO_TIMESTAMP_NTZ(SYSDATE()), 'YYYYMMDD') AS partition_key,
+        _INSERTED_DATE AS _inserted_timestamp,
+        {{ dbt_utils.generate_surrogate_key(
+            ['event_contract','moment_id']
+        ) }} AS topshot_metadata_complete_id,
+        SYSDATE() AS inserted_timestamp,
+        SYSDATE() AS modified_timestamp,
+        '{{ invocation_id }}' AS _invocation_id
+    FROM
+        {{ source(
+            'bronze_streamline',
+            'moments_minted_metadata_api'
+        ) }}
+    {% endif %}
+
+    qualify(ROW_NUMBER() over (PARTITION BY topshot_metadata_complete_id
+ORDER BY
+    _inserted_timestamp DESC)) = 1
--- a/models/streamline/external/topshot/streamline__topshot_metadata_realtime.sql
+++ b/models/streamline/external/topshot/streamline__topshot_metadata_realtime.sql
@ -0,0 +1,71 @@
+{{ config(
+    materialized = 'view',
+    tags = ['streamline', 'topshot', 'moments_metadata', 'backfill'],
+    post_hook = fsc_utils.if_data_call_function_v2(
+        func = '{{this.schema}}.udf_bulk_rest_api_v2',
+        target = "{{this.schema}}.{{this.identifier}}",
+        params ={ "external_table": "topshot_metadata",
+        "sql_limit": "500",
+        "producer_batch_size": "100",
+        "worker_batch_size": "100",
+        "sql_source": "{{this.identifier}}",
+        "async_concurrent_requests": "10" }
+    )
+) }}
+
+WITH api_parameters AS (
+    -- Use the same parameters as for realtime
+
+    SELECT
+        base_url,
+        query
+    FROM
+        {{ ref('streamline__topshot_parameters') }}
+    WHERE
+        contract = 'A.0b2a3299cc857e29.TopShot'
+),
+work_todo AS (
+    SELECT
+        event_contract,
+        moment_id
+    FROM
+        {{ ref('streamline__topshot_moments') }}
+    EXCEPT
+    SELECT
+        event_contract,
+        moment_id
+    FROM
+        {{ ref('streamline__topshot_metadata_complete') }}
+)
+SELECT
+    to_char(TO_TIMESTAMP_NTZ(SYSDATE()), 'YYYYMMDD') AS partition_key,
+    m.event_contract AS contract,
+    m.moment_id AS id,
+    {{ target.database }}.live.udf_api(
+        'POST',
+        p.base_url,
+        OBJECT_CONSTRUCT(
+            'Accept',
+            'application/json',
+            'Accept-Encoding',
+            'gzip',
+            'Connection',
+            'keep-alive',
+            'Content-Type',
+            'application/json',
+            'User-Agent',
+            'Flipside_Flow_metadata/0.1'
+        ),
+        OBJECT_CONSTRUCT(
+            'query',
+            p.query,
+            'variables',
+            OBJECT_CONSTRUCT(
+                'momentId',
+                m.moment_id
+            )
+        )
+    ) AS request
+FROM
+    work_todo m
+    CROSS JOIN api_parameters p
--- a/models/streamline/external/topshot/streamline__topshot_moments.sql
+++ b/models/streamline/external/topshot/streamline__topshot_moments.sql
@ -0,0 +1,85 @@
+{{ config (
+    materialized = "incremental",
+    unique_key = ["moment_id","event_contract"],
+    merge_exclude_columns = ["inserted_timestamp"],
+    tags = ['streamline_complete_evm']
+) }}
+
+WITH mints AS (
+
+    SELECT
+        event_contract,
+        event_data :momentID :: STRING AS moment_id,
+        _inserted_timestamp
+    FROM
+        {{ ref('silver__nft_moments_s') }}
+    WHERE
+        event_contract = 'A.0b2a3299cc857e29.TopShot'
+        AND event_type = 'MomentMinted'
+
+{% if is_incremental() %}
+AND _inserted_timestamp >= COALESCE(
+    (
+        SELECT
+            MAX(_inserted_timestamp) _inserted_timestamp
+        FROM
+            {{ this }}
+    ),
+    '1900-01-01' :: timestamp_ntz
+)
+{% endif %}
+),
+sales AS (
+    SELECT
+        nft_collection AS event_contract,
+        nft_id AS moment_id,
+        _inserted_timestamp
+    FROM
+        {{ ref('silver__nft_sales_s') }}
+    WHERE
+        nft_collection ILIKE '%topshot%'
+
+{% if is_incremental() %}
+AND _inserted_timestamp >= COALESCE(
+    (
+        SELECT
+            MAX(_inserted_timestamp) _inserted_timestamp
+        FROM
+            {{ this }}
+    ),
+    '1900-01-01' :: timestamp_ntz
+)
+{% endif %}
+),
+all_topshots AS (
+    SELECT
+        event_contract,
+        moment_id,
+        _inserted_timestamp
+    FROM
+        mints
+    UNION ALL
+    SELECT
+        event_contract,
+        moment_id,
+        _inserted_timestamp
+    FROM
+        sales
+)
+SELECT
+    event_contract,
+    moment_id,
+    _inserted_timestamp,
+    {{ dbt_utils.generate_surrogate_key(
+        ['event_contract','moment_id']
+    ) }} AS topshot_moments_id,
+    SYSDATE() AS inserted_timestamp,
+    SYSDATE() AS modified_timestamp,
+    '{{ invocation_id }}' AS _invocation_id
+FROM
+    all_topshots qualify ROW_NUMBER() over (
+        PARTITION BY event_contract,
+        moment_id
+        ORDER BY
+            _inserted_timestamp DESC
+    ) = 1
--- a/models/streamline/external/topshot/streamline__topshot_parameters.sql
+++ b/models/streamline/external/topshot/streamline__topshot_parameters.sql
@ -0,0 +1,207 @@
+    {{ config(
+        materialized = 'table',
+        unique_key = 'contract',
+        tags = ['livequery', 'topshot', 'allday', 'moment_metadata']
+    ) }}
+
+    SELECT
+        'A.0b2a3299cc857e29.TopShot' AS contract,
+        'https://public-api.nbatopshot.com/graphql' as base_url,
+        'query getMintedMoment($momentId: ID!) {
+            getMintedMoment(momentId: $momentId) {
+                data {
+                id
+                version
+                sortID
+                set {
+                    id
+                    sortID
+                    version
+                    flowId
+                    flowName
+                    flowSeriesNumber
+                    flowLocked
+                    setVisualId
+                    assetPath
+                    assets {
+                    images {
+                        type
+                        url
+                    }
+                    }
+                }
+                play {
+                    id
+                    version
+                    description
+                    flowID
+                    sortID
+                    status
+                    assets {
+                    videos {
+                        type
+                        url
+                        videoLength
+                    }
+                    videoLengthInMilliseconds
+                    }
+                    stats {
+                    playerID
+                    playerName
+                    firstName
+                    lastName
+                    jerseyNumber
+                    teamAtMoment
+                    awayTeamName
+                    awayTeamScore
+                    homeTeamName
+                    homeTeamScore
+                    dateOfMoment
+                    totalYearsExperience
+                    teamAtMomentNbaId
+                    height
+                    weight
+                    currentTeam
+                    currentTeamId
+                    primaryPosition
+                    homeTeamNbaId
+                    awayTeamNbaId
+                    nbaSeason
+                    draftYear
+                    draftSelection
+                    draftRound
+                    birthplace
+                    birthdate
+                    draftTeam
+                    draftTeamNbaId
+                    playCategory
+                    playType
+                    quarter
+                    }
+                    statsPlayerGameScores {
+                    blocks
+                    points
+                    steals
+                    assists
+                    minutes
+                    rebounds
+                    turnovers
+                    plusMinus
+                    flagrantFouls
+                    personalFouls
+                    technicalFouls
+                    twoPointsMade
+                    blockedAttempts
+                    fieldGoalsMade
+                    freeThrowsMade
+                    threePointsMade
+                    defensiveRebounds
+                    offensiveRebounds
+                    pointsOffTurnovers
+                    twoPointsAttempted
+                    assistTurnoverRatio
+                    fieldGoalsAttempted
+                    freeThrowsAttempted
+                    twoPointsPercentage
+                    fieldGoalsPercentage
+                    freeThrowsPercentage
+                    threePointsAttempted
+                    threePointsPercentage
+                    playerPosition
+                    }
+                    statsPlayerSeasonAverageScores {
+                    minutes
+                    blocks
+                    points
+                    steals
+                    assists
+                    rebounds
+                    turnovers
+                    plusMinus
+                    flagrantFouls
+                    personalFouls
+                    technicalFouls
+                    twoPointsMade
+                    blockedAttempts
+                    fieldGoalsMade
+                    freeThrowsMade
+                    threePointsMade
+                    defensiveRebounds
+                    offensiveRebounds
+                    pointsOffTurnovers
+                    twoPointsAttempted
+                    assistTurnoverRatio
+                    fieldGoalsAttempted
+                    freeThrowsAttempted
+                    twoPointsPercentage
+                    fieldGoalsPercentage
+                    freeThrowsPercentage
+                    threePointsAttempted
+                    threePointsPercentage
+                    efficiency
+                    true_shooting_attempts
+                    points_in_paint_made
+                    points_in_paint_attempted
+                    points_in_paint
+                    fouls_drawn
+                    offensive_fouls
+                    fast_break_points
+                    fast_break_points_attempted
+                    fast_break_points_made
+                    second_chance_points
+                    second_chance_points_attempted
+                    second_chance_points_made
+                    }
+                    tags {
+                    id
+                    name
+                    title
+                    visible
+                    hardcourt
+                    level
+                    }
+                }
+                flowId
+                flowSerialNumber
+                price
+                forSale
+                listingOrderID
+                owner {
+                    dapperID
+                    email
+                    flowAddress
+                    username
+                    profileImageUrl
+                    twitterHandle
+                    segmentID
+                }
+                assetPathPrefix
+                setPlay {
+                    id: ID
+                    setID
+                    playID
+                    flowRetired
+                    circulationCount
+                    tags {
+                    id
+                    name
+                    title
+                    visible
+                    hardcourt
+                    level
+                    }
+                }
+                createdAt
+                acquiredAt
+                packListingID
+                tags {
+                    id
+                    name
+                    title
+                    visible
+                    hardcourt
+                    level
+                }
+                }
+            }
+            }' AS query