AN-5454/ABIs (#388)

* abi streamline pipeline - in prog

* add coalesce

* gha workflow

* CRs

.github/workflows/dbt_run_streamline_evm_daily_realtime.yml

@@ -0,0 +1,46 @@
+name: dbt_run_streamline_evm_daily_realtime
+run-name: dbt_run_streamline_evm_daily_realtime
+
+on:
+  workflow_dispatch:
+  schedule:
+    # Daily at 00:00 UTC (see https://crontab.guru)
+    - cron: "0 0 * * *"
+
+env:
+  USE_VARS: "${{ vars.USE_VARS }}"
+  DBT_PROFILES_DIR: "${{ vars.DBT_PROFILES_DIR }}"
+  ACCOUNT: "${{ vars.ACCOUNT }}"
+  ROLE: "${{ vars.ROLE }}"
+  USER: "${{ vars.USER }}"
+  PASSWORD: "${{ secrets.PASSWORD }}"
+  REGION: "${{ vars.REGION }}"
+  DATABASE: "${{ vars.DATABASE }}"
+  WAREHOUSE: "${{ vars.WAREHOUSE }}"
+  SCHEMA: "${{ vars.SCHEMA }}"
+
+concurrency:
+  group: ${{ github.workflow }}
+
+jobs:
+  run_dbt_jobs:
+    runs-on: ubuntu-latest
+    environment:
+      name: workflow_prod
+
+    steps:
+      - uses: actions/checkout@v3
+
+      - uses: actions/setup-python@v4
+        with:
+          python-version: "3.10"
+          cache: "pip"
+
+      - name: install dependencies
+        run: |
+          pip install -r requirements.txt
+          dbt deps
+
+      - name: Run DBT Jobs
+        run: |
+          dbt run --vars '{"STREAMLINE_INVOKE_STREAMS":True}' -s 1+streamline__contract_abis_realtime

.github/workflows/dbt_run_streamline_evm_daily_silver.yml

@@ -0,0 +1,46 @@
+name: dbt_run_streamline_evm_daily_silver
+run-name: dbt_run_streamline_evm_daily_silver
+
+on:
+  workflow_dispatch:
+  schedule:
+    # Daily at 01:00 UTC (see https://crontab.guru)
+    - cron: "0 1 * * *"
+
+env:
+  USE_VARS: "${{ vars.USE_VARS }}"
+  DBT_PROFILES_DIR: "${{ vars.DBT_PROFILES_DIR }}"
+  ACCOUNT: "${{ vars.ACCOUNT }}"
+  ROLE: "${{ vars.ROLE }}"
+  USER: "${{ vars.USER }}"
+  PASSWORD: "${{ secrets.PASSWORD }}"
+  REGION: "${{ vars.REGION }}"
+  DATABASE: "${{ vars.DATABASE }}"
+  WAREHOUSE: "${{ vars.WAREHOUSE }}"
+  SCHEMA: "${{ vars.SCHEMA }}"
+
+concurrency:
+  group: ${{ github.workflow }}
+
+jobs:
+  run_dbt_jobs:
+    runs-on: ubuntu-latest
+    environment:
+      name: workflow_prod
+
+    steps:
+      - uses: actions/checkout@v3
+
+      - uses: actions/setup-python@v4
+        with:
+          python-version: "3.10"
+          cache: "pip"
+
+      - name: install dependencies
+        run: |
+          pip install -r requirements.txt
+          dbt deps
+
+      - name: Run DBT Jobs
+        run: |
+          dbt run -s tag:streamline_evm_non_core

models/evm/streamline/external/flowscan/bronze_api__FR_contract_abis.sql

@@ -0,0 +1,9 @@
+{{ config (
+    materialized = 'view',
+    tags = ['streamline_evm_non_core', 'contract_abis']
+) }}
+
+{{ streamline_external_table_FR_query_v2(
+    model = "contract_abis",
+    partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 3), '_', 1) AS INTEGER )"
+) }}

models/evm/streamline/external/flowscan/bronze_api__contract_abis.sql

@@ -0,0 +1,9 @@
+{{ config (
+    materialized = 'view',
+    tags = ['streamline_evm_non_core', 'contract_abis']
+) }}
+
+{{ streamline_external_table_query_v2(
+    model = "contract_abis",
+    partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 3), '_', 1) AS INTEGER )"
+) }}

models/evm/streamline/external/flowscan/silver_api__contract_abis.sql

@@ -0,0 +1,48 @@
+-- depends_on: {{ ref('bronze_api__contract_abis') }}
+-- depends_on: {{ ref('bronze_api__FR_contract_abis') }}
+{{ config(
+    materialized = 'incremental',
+    unique_key = "contract_address",
+    incremental_strategy = 'merge',
+    merge_exclude_columns = ["inserted_timestamp"],
+    tags = ['streamline_evm_non_core']
+) }}
+
+WITH bronze AS (
+
+    SELECT
+        VALUE :CONTRACT_ADDRESS :: STRING AS contract_address,
+        partition_key,
+        DATA :abi ::VARIANT as ABI,
+        _inserted_timestamp
+    FROM
+
+{% if is_incremental() %}
+{{ ref('bronze_api__contract_abis') }}
+WHERE
+    _inserted_timestamp >= COALESCE(
+        (
+            SELECT
+                MAX(_inserted_timestamp) _inserted_timestamp
+            FROM
+                {{ this }}
+        ),
+        '1900-01-01' :: timestamp_ntz
+    )
+{% else %}
+    {{ ref('bronze_api__FR_contract_abis') }}
+{% endif %}
+)
+SELECT
+    contract_address,
+    ABI,
+    partition_key,
+    _inserted_timestamp,
+    {{ dbt_utils.generate_surrogate_key(
+        ['contract_address']
+    ) }} AS contract_abis_id,
+    SYSDATE() AS inserted_timestamp,
+    SYSDATE() AS modified_timestamp,
+    '{{ invocation_id }}' AS _invocation_id
+FROM
+    bronze

models/evm/streamline/external/flowscan/streamline__complete_contract_abis.sql

@@ -0,0 +1,41 @@
+-- depends_on: {{ ref('bronze_api__contract_abis') }}
+-- depends_on: {{ ref('bronze_api__FR_contract_abis') }}
+
+{{ config (
+    materialized = "incremental",
+    unique_key = "contract_address",
+    merge_exclude_columns = ["inserted_timestamp"],
+    tags = ['streamline_evm_non_core']
+) }}
+
+SELECT
+    VALUE :CONTRACT_ADDRESS :: STRING AS contract_address,
+    partition_key,
+    _inserted_timestamp,
+    {{ dbt_utils.generate_surrogate_key(
+        ['contract_address']
+    ) }} AS complete_contract_abis_id,
+    SYSDATE() AS inserted_timestamp,
+    SYSDATE() AS modified_timestamp,
+    '{{ invocation_id }}' AS _invocation_id
+FROM
+
+{% if is_incremental() %}
+{{ ref('bronze_api__contract_abis') }}
+WHERE
+    _inserted_timestamp >= COALESCE(
+        (
+            SELECT
+                MAX(_inserted_timestamp) _inserted_timestamp
+            FROM
+                {{ this }}
+        ),
+        '1900-01-01' :: timestamp_ntz
+    )
+{% else %}
+    {{ ref('bronze_api__FR_contract_abis') }}
+{% endif %}
+
+qualify(ROW_NUMBER() over (PARTITION BY contract_address
+ORDER BY
+    _inserted_timestamp DESC)) = 1

models/evm/streamline/external/flowscan/streamline__contract_abis_realtime.sql

@@ -0,0 +1,28 @@
+{{ config (
+    materialized = "view",
+    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": "contract_abis",
+        "sql_limit": "1000",
+        "producer_batch_size": "250",
+        "worker_batch_size": "250",
+        "sql_source": "{{this.identifier}}" }
+    )
+) }}
+with verified_contracts as (
+    select address_hash AS contract_address from {{ ref('seeds__evm_verified_contracts') }}
+    EXCEPT
+    select contract_address from {{ ref('streamline__complete_contract_abis') }}
+)
+SELECT
+    contract_address,
+    DATE_PART('EPOCH', SYSDATE()) :: INTEGER AS partition_key,
+    {{ target.database }}.live.udf_api(
+        'GET',
+        'https://evm.flowscan.io/api/v2/smart-contracts/' || contract_address,
+        {},
+        {}
+    ) AS request
+FROM
+    verified_contracts

models/sources.yml

@@ -129,6 +129,7 @@ sources:
       - name: transaction_entries
       - name: points_transfers
       - name: minting_assets
+      - name: contract_abis
 
   - name: crosschain_silver
     database: crosschain