AN-6276/add gold view to snag/transaction_entries workflow (#454)

* add gold view to snag/transaction_entries workflow

* add sortDir param

* backfill

* upd gha to hourly

.github/workflows/dbt_run_scheduled_streamline_non_core.yml

@@ -4,8 +4,8 @@ run-name: dbt_run_scheduled_streamline_non_core
 on:
   workflow_dispatch:
   schedule:
-    # Daily at 04:00 UTC
-    - cron: "0 4 * * *"
+    # Daily at 06:00 UTC
+    - cron: "0 6 * * *"
 
 env:
   SLACK_WEBHOOK_URL: "${{ secrets.SLACK_WEBHOOK_URL }}"

.github/workflows/dbt_run_streamline_external_points_balances_realtime.yml

@@ -7,8 +7,8 @@ description: |
 on:
   workflow_dispatch:
   schedule:
-    # Daily at 0, 1, 2, 3 UTC
-    - cron: "0 0,1,2,3 * * *"
+    # Daily at 0, 1, 2, 3, 4 UTC
+    - cron: "0 0,1,2,3,4 * * *"
 
 env:
   USE_VARS: "${{ vars.USE_VARS }}"

.github/workflows/dbt_run_streamline_external_realtime.yml

@@ -6,8 +6,8 @@ description: |
 on:
   workflow_dispatch:
   schedule:
-    # Daily at 00:00 UTC
-    - cron: "0 0 * * *"
+    # Runs hourly
+    - cron: "0 * * * *"
 
 env:
   USE_VARS: "${{ vars.USE_VARS }}"

models/descriptions/DATA.md

@@ -0,0 +1,5 @@
+{% docs DATA %}
+
+Remaining raw data from the transaction entry, with key fields removed.
+
+{% enddocs %} 

models/descriptions/INDEX.md

@@ -0,0 +1,5 @@
+{% docs INDEX %}
+
+Index of the transaction entry within the batch or partition.
+
+{% enddocs %} 

models/descriptions/_inserted_timestamp.md

@@ -1,5 +1,5 @@
 {% docs _inserted_timestamp %}
 
-Time the record was inserted into snowflake
+Time the record was inserted into the warehouse.
 
 {% enddocs %}

models/descriptions/account_id.md

@@ -0,0 +1,5 @@
+{% docs account_id %}
+
+Loyalty account identifier involved in the transaction.
+
+{% enddocs %} 

models/descriptions/amount.md

@@ -1,5 +1,5 @@
 {% docs amount %}
 
-The amount of the asset involved in the transaction.
+Number of points transferred in the transaction.
 
 {% enddocs %}

models/descriptions/amount_end.md

@@ -0,0 +1,5 @@
+{% docs amount_end %}
+
+Ending balance of the loyalty account after the transaction.
+
+{% enddocs %} 

models/descriptions/amount_start.md

@@ -0,0 +1,5 @@
+{% docs amount_start %}
+
+Starting balance of the loyalty account before the transaction.
+
+{% enddocs %} 

models/descriptions/created_at.md

@@ -0,0 +1,5 @@
+{% docs created_at %}
+
+Timestamp when the transaction entry was created.
+
+{% enddocs %} 

models/descriptions/direction.md

@@ -1,5 +1,5 @@
 {% docs direction %}
 
-For bridging transactions, inbound refers to tokens bridged to the Flow blockchain, while outbound indicated tokens were sent from Flow to another (i.e. to Ethereum, BSC, etc.).
+Direction of the points transfer (e.g., debit or credit).
 
 {% enddocs %}

models/descriptions/entry_id.md

@@ -0,0 +1,5 @@
+{% docs entry_id %}
+
+Unique identifier for the transaction entry record.
+
+{% enddocs %} 

models/descriptions/idempotency_key.md

@@ -0,0 +1,5 @@
+{% docs idempotency_key %}
+
+Key used to ensure idempotency of the transaction entry.
+
+{% enddocs %} 

models/descriptions/organization_id.md

@@ -0,0 +1,5 @@
+{% docs organization_id %}
+
+Identifier for the organization associated with the transaction.
+
+{% enddocs %} 

models/descriptions/partition_key.md

@@ -0,0 +1,5 @@
+{% docs partition_key %}
+
+Partition key used for efficient querying and incremental loading.
+
+{% enddocs %} 

models/descriptions/reward_points_spend_id.md

@@ -0,0 +1,5 @@
+{% docs reward_points_spend_id %}
+
+Surrogate key for the reward points spend event, generated from entry_id and partition_key.
+
+{% enddocs %} 

models/descriptions/rule_description.md

@@ -0,0 +1,5 @@
+{% docs rule_description %}
+
+Description of the loyalty rule applied to the transaction.
+
+{% enddocs %} 

models/descriptions/rule_id.md

@@ -0,0 +1,5 @@
+{% docs rule_id %}
+
+Identifier for the loyalty rule applied to the transaction.
+
+{% enddocs %} 

models/descriptions/rule_metadata.md

@@ -0,0 +1,5 @@
+{% docs rule_metadata %}
+
+Metadata associated with the loyalty rule applied to the transaction.
+
+{% enddocs %} 

models/descriptions/rule_name.md

@@ -0,0 +1,5 @@
+{% docs rule_name %}
+
+Name of the loyalty rule applied to the transaction.
+
+{% enddocs %} 

models/descriptions/rule_type.md

@@ -0,0 +1,5 @@
+{% docs rule_type %}
+
+Type of the loyalty rule applied to the transaction.
+
+{% enddocs %} 

models/descriptions/transaction_description.md

@@ -0,0 +1,5 @@
+{% docs transaction_description %}
+
+Description of the loyalty transaction.
+
+{% enddocs %} 

models/descriptions/transaction_id.md

@@ -0,0 +1,5 @@
+{% docs transaction_id %}
+
+Identifier for the loyalty transaction.
+
+{% enddocs %} 

models/descriptions/transaction_type.md

@@ -0,0 +1,5 @@
+{% docs transaction_type %}
+
+Type of the loyalty transaction (e.g., earn, spend).
+
+{% enddocs %} 

models/descriptions/user_id.md

@@ -0,0 +1,5 @@
+{% docs user_id %}
+
+User identifier associated with the loyalty account.
+
+{% enddocs %} 

models/descriptions/user_wallet_address.md

@@ -0,0 +1,5 @@
+{% docs user_wallet_address %}
+
+Wallet address of the user associated with the loyalty account.
+
+{% enddocs %} 

models/descriptions/website_id.md

@@ -0,0 +1,5 @@
+{% docs website_id %}
+
+Identifier for the website associated with the transaction.
+
+{% enddocs %} 

models/gold/rewards/rewards__fact_transaction_entries.sql

@@ -0,0 +1,25 @@
+{{ config(
+    materialized = 'view',
+    tags = ['streamline_non_core']
+)}}
+
+SELECT
+    entry_id,
+    created_at,
+    direction,
+    amount,
+    amount_start,
+    amount_end,
+    account_id,
+    user_id,
+    user_wallet_address,
+    transaction_id,
+    DATA,
+    partition_key,
+    INDEX,
+    _inserted_timestamp,
+    reward_points_spend_id AS fact_transaction_entries_id,
+    inserted_timestamp,
+    modified_timestamp
+FROM 
+    {{ ref('silver_api__reward_points_spend') }}

models/gold/rewards/rewards__fact_transaction_entries.yml

@@ -0,0 +1,46 @@
+version: 2
+
+models:
+  - name: rewards__fact_transaction_entries
+    description: "individual flow points transfer events flattened from responses from the snag transaction entries endpoint"
+    tests:
+      - dbt_utils.recency:
+          datepart: day
+          field: INSERTED_TIMESTAMP
+          interval: 1
+
+    columns:
+      - name: ENTRY_ID
+        description: '{{ doc("entry_id") }}'
+      - name: CREATED_AT
+        description: '{{ doc("created_at") }}'
+      - name: DIRECTION
+        description: '{{ doc("direction") }}'
+      - name: AMOUNT
+        description: '{{ doc("amount") }}'
+      - name: AMOUNT_START
+        description: '{{ doc("amount_start") }}'
+      - name: AMOUNT_END
+        description: '{{ doc("amount_end") }}'
+      - name: ACCOUNT_ID
+        description: '{{ doc("account_id") }}'
+      - name: USER_ID
+        description: '{{ doc("user_id") }}'
+      - name: USER_WALLET_ADDRESS
+        description: '{{ doc("user_wallet_address") }}'
+      - name: TRANSACTION_ID
+        description: '{{ doc("transaction_id") }}'
+      - name: DATA
+        description: '{{ doc("DATA") }}'
+      - name: PARTITION_KEY
+        description: '{{ doc("partition_key") }}'
+      - name: INDEX
+        description: '{{ doc("INDEX") }}'
+      - name: _INSERTED_TIMESTAMP
+        description: '{{ doc("_inserted_timestamp") }}'
+      - name: FACT_TRANSACTION_ENTRIES_ID
+        description: '{{ doc("pk_id") }}'
+      - name: INSERTED_TIMESTAMP
+        description: '{{ doc("inserted_timestamp") }}'
+      - name: MODIFIED_TIMESTAMP
+        description: '{{ doc("modified_timestamp") }}'

models/sources.yml

@@ -191,4 +191,11 @@ sources:
   - name: silver
     schema: silver
     tables:
-      - name: nft_topshot_metadata
+      - name: nft_topshot_metadata
+
+  - name: flow_seeds
+    database: |
+      {{ "FLOW" if target.database == 'FLOW' else "FLOW_DEV" }}
+    schema: seeds
+    tables:
+      - name: transaction_entries

models/streamline/external/snag/transaction_entries/silver_api__reward_points_spend.sql

@@ -5,7 +5,7 @@
     merge_exclude_columns = ["inserted_timestamp"],
     cluster_by = ['_inserted_timestamp :: DATE'],
     post_hook = [ "ALTER TABLE {{ this }} ADD SEARCH OPTIMIZATION on equality(user_wallet_address)" ],
-    tags = ['rewards_points_spend']
+    tags = ['rewards_points_spend', 'streamline_non_core']
 ) }}
 
 WITH silver_responses AS (
@@ -43,38 +43,13 @@ SELECT
     DATA :loyaltyAccountStartAmount :: NUMBER AS amount_start,
     DATA :loyaltyAccountEndAmount :: NUMBER AS amount_end,
 
-    DATA :idempotencyKey :: STRING AS idempotency_key,
-    DATA :organizationId :: STRING AS organization_id,
-    DATA :websiteId :: STRING AS website_id,
-
     DATA :loyaltyAccountId :: STRING AS account_id,
     DATA :loyaltyAccount :user :id :: STRING AS user_id,
     DATA :loyaltyAccount :user :walletAddress :: STRING AS user_wallet_address,
 
     DATA :loyaltyTransactionId :: STRING AS transaction_id,
-    DATA :loyaltyTransaction :description :: STRING AS transaction_description,
-    DATA :loyaltyTransaction :type :: STRING AS transaction_type,
 
-    DATA :loyaltyTransaction :loyaltyRule :id :: STRING AS rule_id,
-    DATA :loyaltyTransaction :loyaltyRule :type :: STRING AS rule_type,
-    DATA :loyaltyTransaction :loyaltyRule :name :: STRING AS rule_name,
-    DATA :loyaltyTransaction :loyaltyRule :description :: STRING AS rule_description,
-    DATA :loyaltyTransaction :loyaltyRule :metadata :: variant AS rule_metadata,
-
-    OBJECT_DELETE(
-        DATA,
-        'amount',
-        'createdAt',
-        'direction',
-        'idempotencyKey',
-        'loyaltyAccount',
-        'loyaltyAccountId',
-        'loyaltyAccountEndAmount',
-        'loyaltyAccountStartAmount',
-        'loyaltyTransactionId',
-        'organizationId',
-        'websiteId'
-    ) AS DATA,
+    DATA,
     partition_key,
     INDEX,
     _inserted_timestamp,

models/streamline/external/snag/transaction_entries/silver_api__reward_points_spend.yml

@@ -35,16 +35,6 @@ models:
         tests:
           - not_null
 
-      - name: idempotency_key
-
-      - name: organization_id
-        tests:
-          - not_null
-
-      - name: website_id
-        tests:
-          - not_null
-
       - name: account_id
         tests:
           - not_null
@@ -62,26 +52,6 @@ models:
           - not_null
           - unique
 
-      - name: transaction_description
-
-      - name: transaction_type
-        tests:
-          - not_null
-
-      - name: rule_id
-        tests:
-          - not_null
-
-      - name: rule_type
-        tests:
-          - not_null
-
-      - name: rule_name
-
-      - name: rule_description
-
-      - name: rule_metadata
-
       - name: data
 
       - name: partition_key

models/streamline/external/snag/transaction_entries/silver_api__transaction_entries.sql

@@ -6,9 +6,33 @@
     incremental_strategy = 'merge',
     merge_exclude_columns = ["inserted_timestamp"],
     cluster_by = ['_inserted_timestamp :: DATE'],
-    tags = ['rewards_points_spend']
+    tags = ['rewards_points_spend', 'streamline_non_core']
 ) }}
 
+{% if var('STOREFRONT_INITIAL_RUN', false) %}
+
+SELECT
+    partition_key,
+    NULL AS entry_count,
+    NULL AS starting_after,
+    NULL AS api_limit,
+    NULL AS first_entry_id,
+    NULL AS last_entry_id,
+    TRY_PARSE_JSON(DATA) :createdAt :: timestamp_ntz AS created_at,
+    TRY_PARSE_JSON(DATA) :id :: STRING AS entry_id,
+    index_in_batch :: INTEGER AS INDEX,
+    TRY_PARSE_JSON(DATA) AS DATA,
+    _inserted_timestamp :: timestamp_ntz AS _inserted_timestamp,
+    {{ dbt_utils.generate_surrogate_key(
+        ['entry_id', 'partition_key']
+    ) }} AS transaction_entries_id,
+    SYSDATE() AS inserted_timestamp,
+    SYSDATE() AS modified_timestamp,
+    '{{ invocation_id }}' AS _invocation_id
+FROM
+    {{ source('flow_seeds', 'transaction_entries') }}
+
+{% else %}
 WITH bronze AS (
 
     SELECT
@@ -61,3 +85,4 @@ FROM
         input => DATA :data :: ARRAY
     ) 
 
+{% endif %}

models/streamline/external/snag/transaction_entries/streamline__transaction_entries_realtime.sql

@@ -55,7 +55,7 @@ SELECT
     DATE_PART('EPOCH', SYSDATE()) :: INTEGER AS partition_key,
     {{ target.database }}.live.udf_api(
         'GET',
-        '{Service}/api/loyalty/transaction_entries' || '?organizationId=' || '{organization_id}' || '&websiteId=' || '{website_id}' || '&limit=' || api_limit{% if not var('STOREFRONT_INITIAL_RUN', false) %} || '&startingAfter=' || '{{ starting_after }}'{% endif %},
+        '{Service}/api/loyalty/transaction_entries' || '?organizationId=' || '{organization_id}' || '&websiteId=' || '{website_id}' || '&limit=' || api_limit{% if not var('STOREFRONT_INITIAL_RUN', false) %} || '&startingAfter=' || '{{ starting_after }}'{% endif %}{% if not var('IS_BACKFILL', false) %} || '&sortDir=asc'{% endif %},
         { 'x-api-key': '{Authentication}' },
         {},
         'Vault/prod/flow/snag-api'

tests/points/tests__unique_address_count_threshold.sql

@@ -18,4 +18,4 @@ SELECT
 FROM
     distinct_count
 WHERE
-    ct > 110000
+    ct > 150000