flow rewards api fixes (#401)

* flow rewards api fixes

* push silver back 2h

* add evm addrs before bal

* chg batch size

.github/workflows/dbt_run_scheduled_reward_points_realtime.yml

@@ -45,7 +45,7 @@ jobs:
 
       - name: Request User Points Balances and Storefront Items
         run: >
-          dbt run -s streamline__reward_points_realtime streamline__minting_assets_realtime --vars '{"STREAMLINE_INVOKE_STREAMS": True}'
+          dbt run -s 1+streamline__reward_points_realtime streamline__minting_assets_realtime --vars '{"STREAMLINE_INVOKE_STREAMS": True}'
 
       - name: Authenticate with Flow Points API
         run: |

.github/workflows/dbt_run_streamline_evm_daily_silver.yml

@@ -4,8 +4,8 @@ run-name: dbt_run_streamline_evm_daily_silver
 on:
   workflow_dispatch:
   schedule:
-    # Daily at 01:00 UTC (see https://crontab.guru)
-    - cron: "0 1 * * *"
+    # Daily at 02:00 UTC (see https://crontab.guru)
+    - cron: "0 2 * * *"
 
 env:
   USE_VARS: "${{ vars.USE_VARS }}"

models/streamline/external/balances/streamline__reward_points_realtime.sql

@@ -5,9 +5,9 @@
         target = "{{this.schema}}.{{this.identifier}}",
         params = {
             "external_table": "reward_points",
-            "sql_limit": "24000",
-            "producer_batch_size": "3000",
-            "worker_batch_size": "3000",
+            "sql_limit": "32000",
+            "producer_batch_size": "8000",
+            "worker_batch_size": "1600",
             "sql_source": "{{this.identifier}}"
         }
     )

models/streamline/external/streamline__evm_addresses.sql

@@ -29,7 +29,7 @@ onchain AS (
     SELECT
         DISTINCT from_address AS address
     FROM
-        {{ ref('silver_evm__transactions') }}
+        {{ ref('core_evm__fact_transactions') }}
 
 {% if is_incremental() %}
 WHERE

models/streamline/external/transfers/silver_api__points_transfers.sql

@@ -6,36 +6,21 @@
     tags = ['streamline_non_core']
 ) }}
 
-
-
 WITH points_transfers_raw AS (
 
     SELECT
         partition_key,
         request_date,
-        DATA,
+        from_address,
+        batch_index,
+        batch_id,
+        created_at,
+        batch_status,
+        batch_transfers,
         _inserted_timestamp
     FROM
         {{ ref('silver_api__points_transfers_response') }}
 
-),
-flatten_batches AS (
-    SELECT
-        partition_key,
-        request_date,
-        _inserted_timestamp,
-        DATA :address :: STRING AS from_address,
-        A.index AS batch_index,
-        A.value :createdAt :: TIMESTAMP_NTZ AS created_at,
-        A.value :batchId :: STRING AS batch_id,
-        A.value :status :: STRING AS batch_status,
-        A.value :transfers :: ARRAY AS batch_transfers
-    FROM
-        points_transfers_raw,
-        LATERAL FLATTEN(
-            DATA :transfers :: ARRAY
-        ) A
-
 ),
 flatten_transfers AS (
     SELECT
@@ -52,7 +37,7 @@ flatten_transfers AS (
         A.value :points :: NUMBER AS points,
         A.value :toAddressId :: STRING AS to_address
     FROM
-        flatten_batches,
+        points_transfers_raw,
         LATERAL FLATTEN(batch_transfers) A
 )
 SELECT
@@ -68,7 +53,7 @@ SELECT
     points,
     partition_key,
     {{ dbt_utils.generate_surrogate_key(
-        ['batch_id', 'transfer_index']
+        ['from_address', 'batch_id', 'transfer_index']
     ) }} AS points_transfers_id,
     SYSDATE() AS inserted_timestamp,
     SYSDATE() AS modified_timestamp,

models/streamline/external/transfers/silver_api__points_transfers_response.sql

@@ -29,11 +29,16 @@
 SELECT
     partition_key,
     TO_TIMESTAMP(partition_key) :: DATE AS request_date,
-    DATA,
+    VALUE :address :: STRING AS from_address,
+    ZEROIFNULL(VALUE :array_index :: INTEGER) AS batch_index,
+    DATA :batchId :: STRING AS batch_id,
+    DATA :createdAt :: TIMESTAMP_NTZ AS created_at,
+    DATA :secondsToFinalize :: INTEGER AS seconds_to_finalize,
+    DATA :status :: STRING AS batch_status,
+    DATA :transfers :: ARRAY AS batch_transfers,
     _inserted_timestamp,
-    ROUND(OCTET_LENGTH(DATA) / 1048576, 2) AS data_mb,
     {{ dbt_utils.generate_surrogate_key(
-        ['file_name', 'data :address :: STRING']
+        ['VALUE :address :: STRING', 'DATA :batchId :: STRING']
     ) }} AS points_transfers_id,
     SYSDATE() AS inserted_timestamp,
     SYSDATE() AS modified_timestamp,

models/streamline/external/transfers/streamline__points_transfers_realtime.sql

@@ -9,7 +9,8 @@
             "producer_batch_size": "1",
             "worker_batch_size": "1",
             "sql_source": "{{this.identifier}}",
-            "exploded_key": tojson(["result"])
+            "exploded_key": tojson(["transfers"]),
+            "include_top_level_json": tojson(["address"])
         }
     )
 ) }}
@@ -19,7 +20,7 @@ SELECT
     DATE_PART('EPOCH', SYSDATE()) :: INTEGER AS partition_key,
     {{ target.database }}.live.udf_api(
         'GET',
-        '{Service}/points/dapp/transfer/all',
+        '{Service}points/dapp/transfer/all',
         {
             'Authorization': 'Bearer ' || '{{ env_var("JWT", "") }}',
             'Accept': 'application/json',
@@ -30,4 +31,3 @@ SELECT
         {},
         'Vault/prod/flow/points-api/prod'
     ) AS request
-

tests/points/tests__unique_address_count_threshold.sql

@@ -18,4 +18,4 @@ SELECT
 FROM
     distinct_count
 WHERE
-    ct > 20000
+    ct > 30000