AN-5418/Flow x EVM address mapping (#377)

* address mapping

* gold

* upd comment, recency threshold, kill distinct

models/descriptions/address.md

@@ -1,5 +1,5 @@
 {% docs address %}
 
-The on-chain address. See more in the Flow docs on accounts and addresses: https://developers.flow.com/build/basics/accounts
+The on-chain address. See more in the Flow docs on accounts and addresses: https://developers.flow.com/build/basics/accounts and https://developers.flow.com/evm/accounts
 
 {% enddocs %}

models/gold/core/core__dim_address_mapping.sql

@@ -0,0 +1,30 @@
+{{ config (
+    materialized = 'incremental',
+    incremental_strategy = 'merge',
+    merge_exclude_columns = ['inserted_timestamp'],
+    unique_key = 'dim_address_mapping_id',
+    cluster_by = ['block_timestamp_associated::date'],
+    post_hook = 'ALTER TABLE {{ this }} ADD SEARCH OPTIMIZATION ON EQUALITY(evm_address, flow_address);',
+    tags = ['scheduled_non_core']
+) }}
+
+SELECT
+    block_timestamp AS block_timestamp_associated,
+    block_height AS block_height_associated,
+    flow_address,
+    evm_address,
+    flow_evm_address_map_id AS dim_address_mapping_id,
+    SYSDATE() AS modified_timestamp,
+    SYSDATE() AS inserted_timestamp
+FROM
+    {{ ref('silver__flow_evm_address_map') }}
+
+{% if is_incremental() %}
+WHERE
+    modified_timestamp >= (
+        SELECT
+            MAX(modified_timestamp)
+        FROM
+            {{ this }}
+    )
+{% endif %}

models/gold/core/core__dim_address_mapping.yml

@@ -0,0 +1,46 @@
+version: 2
+
+models:
+  - name: core__dim_address_mapping
+    description: -|
+      A table that maps EVM addresses to Flow addresses based on COA Creation events. A single Flow address may have multiple EVM addresses associated with it.
+    tests:
+      - dbt_utils.recency:
+          datepart: hours
+          field: BLOCK_TIMESTAMP_ASSOCIATED
+          interval: 24
+
+    columns:
+      - name: BLOCK_HEIGHT_ASSOCIATED
+        description: "{{ doc('block_number') }}"
+        tests:
+          - not_null
+
+      - name: BLOCK_TIMESTAMP_ASSOCIATED
+        description: "{{ doc('block_timestamp') }}"
+        tests:
+          - not_null
+
+      - name: EVM_ADDRESS
+        description: "{{ doc('address') }}"
+        tests:
+          - not_null
+          - unique
+
+      - name: FLOW_ADDRESS
+        description: "{{ doc('address') }}"
+        tests:
+          - not_null
+
+      - name: DIM_ADDRESS_MAPPING_ID
+        description: "{{ doc('pk_id') }}"
+        tests:
+          - not_null
+          - unique
+
+      - name: INSERTED_TIMESTAMP
+        description: "{{ doc('inserted_timestamp') }}"
+
+      - name: MODIFIED_TIMESTAMP
+        description: "{{ doc('modified_timestamp') }}"
+

models/silver/curated/silver__flow_evm_address_map.sql

@@ -0,0 +1,123 @@
+{{ config(
+    materialized = 'incremental',
+    unique_key = 'flow_evm_address_map_id',
+    incremental_strategy = 'merge',
+    merge_exclude_columns = ['inserted_timestamp'],
+    cluster_by = ['block_timestamp::date', 'modified_timestamp::date'],
+    tags = ['scheduled_non_core']
+) }}
+
+WITH events AS (
+
+    SELECT
+        *
+    FROM
+        {{ ref('silver__streamline_events') }}
+    WHERE
+        block_timestamp :: DATE >= '2024-09-02'
+
+{% if is_incremental() %}
+AND modified_timestamp >= (
+    SELECT
+        MAX(modified_timestamp)
+    FROM
+        {{ this }}
+)
+{% endif %}
+),
+coa_creation AS (
+    SELECT
+        tx_id,
+        block_timestamp,
+        block_height,
+        event_index,
+        CONCAT(
+            '0x',
+            event_data :address :: STRING
+        ) AS evm_address
+    FROM
+        events
+    WHERE
+        event_contract = 'A.e467b9dd11fa00df.EVM'
+        AND event_type = 'CadenceOwnedAccountCreated'
+),
+txs AS (
+    SELECT
+        tx_id,
+        block_height,
+        authorizers
+    FROM
+        {{ ref('silver__streamline_transactions_final') }}
+    WHERE
+        block_timestamp :: DATE >= '2024-09-02'
+        AND tx_id IN (
+            SELECT
+                tx_id
+            FROM
+                events
+        )
+
+{% if is_incremental() %}
+AND modified_timestamp >= (
+    SELECT
+        MAX(modified_timestamp)
+    FROM
+        {{ this }}
+)
+{% endif %}
+),
+get_flow_address AS (
+    SELECT
+        tx_id,
+        block_height,
+        event_index,
+        event_type,
+        event_data :address :: STRING AS flow_address
+    FROM
+        events
+    WHERE
+        event_contract = 'flow'
+        AND event_type = 'CapabilityPublished'
+        AND tx_id IN (
+            SELECT
+                tx_id
+            FROM
+                coa_creation
+        ) -- a transaction may emit multiple CapabilityPublished events
+        qualify(ROW_NUMBER() over (PARTITION BY tx_id
+    ORDER BY
+        event_index) = 1)
+),
+map_addresses AS (
+    SELECT
+        A.tx_id,
+        A.block_timestamp,
+        A.block_height,
+        A.evm_address,
+        COALESCE(
+            b.flow_address,
+            C.authorizers [0] :: STRING
+        ) AS flow_address,
+        b.flow_address IS NULL AS used_authorizer
+    FROM
+        coa_creation A
+        LEFT JOIN get_flow_address b
+        ON A.tx_id = b.tx_id
+        AND A.block_height = b.block_height
+        LEFT JOIN txs C
+        ON A.tx_id = C.tx_id
+        AND A.block_height = C.block_height
+)
+SELECT
+    tx_id,
+    block_timestamp,
+    block_height,
+    evm_address,
+    flow_address,
+    used_authorizer,
+    {{ dbt_utils.generate_surrogate_key(['evm_address', 'flow_address']) }} AS flow_evm_address_map_id,
+    SYSDATE() AS modified_timestamp,
+    SYSDATE() AS inserted_timestamp,
+    '{{ invocation_id }}' AS invocation_id
+FROM
+    map_addresses

models/silver/curated/silver__flow_evm_address_map.yml

@@ -0,0 +1,53 @@
+version: 2
+
+models:
+  - name: silver__flow_evm_address_map
+    description: -|
+      A table that maps EVM addresses to Flow addresses based on COA Creation events.
+    tests:
+      - dbt_utils.recency:
+          datepart: hours
+          field: block_timestamp
+          interval: 24
+
+    columns:
+      - name: BLOCK_HEIGHT
+        description: "{{ doc('block_number') }}"
+        tests:
+          - not_null
+
+      - name: BLOCK_TIMESTAMP
+        description: "{{ doc('block_timestamp') }}"
+        tests:
+          - not_null
+
+      - name: TX_ID
+        description: "{{ doc('tx_id') }}"
+        tests:
+          - not_null
+
+      - name: EVM_ADDRESS
+        description: "{{ doc('address') }}"
+        tests:
+          - not_null
+          - unique
+
+      - name: FLOW_ADDRESS
+        description: "{{ doc('address') }}"
+        tests:
+          - not_null
+
+      - name: FLOW_EVM_ADDRESS_MAP_ID
+        description: "{{ doc('pk_id') }}"
+        tests:
+          - not_null
+          - unique
+
+      - name: INSERTED_TIMESTAMP
+        description: "{{ doc('inserted_timestamp') }}"
+
+      - name: MODIFIED_TIMESTAMP
+        description: "{{ doc('modified_timestamp') }}"
+
+      - name: _INVOCATION_ID
+        description: "{{ doc('invocation_id') }}"