nautilus_blockchain/cache/

database.rs

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

use std::{collections::BTreeSet, fmt::Display, pin::Pin};

use alloy::primitives::{Address, U256};
use anyhow::Context;
use futures_util::{Stream, StreamExt};
use nautilus_model::{
    defi::{
        Block, Chain, DexType, Pool, PoolIdentifier, PoolLiquidityUpdate, PoolSwap, SharedChain,
        SharedDex, Token,
        data::{
            DexPoolData, PoolFeeCollect, PoolFeeProtocolCollect, PoolFeeProtocolUpdate, PoolFlash,
            block::{BLOCK_SCOPED_SNAPSHOT_INDEX, BlockPosition},
        },
        pool_analysis::{
            position::PoolPosition,
            snapshot::{PoolAnalytics, PoolSnapshot, PoolState},
        },
        tick_map::tick::PoolTick,
        validation::validate_address,
    },
    identifiers::InstrumentId,
};
use rust_decimal::Decimal;
use sqlx::{
    AssertSqlSafe, PgPool, Postgres, Row, Transaction,
    postgres::{PgAdvisoryLock, PgAdvisoryLockKey, PgConnectOptions},
};

use crate::{
    cache::{
        PoolEventSyncState,
        consistency::CachedBlocksConsistencyStatus,
        copy::PostgresCopyHandler,
        rows::{
            BlockTimestampRow, ExecutionIntentInsert, ExecutionIntentRow,
            ExecutionTransactionHashRow, ExecutionTransactionRow, PoolRow, TokenRow,
            parse_cached_block_timestamp, transform_row_to_dex_pool_data,
        },
        types::{U128Pg, U256Pg},
    },
    events::initialize::InitializeEvent,
    execution::{
        sealing::{
            PayloadKeySet, PayloadPolicy, authenticate_payload, authenticate_retained_payload,
            envelope_key_id, payload_context, retained_payload_requires_policy,
        },
        transaction::{EXECUTION_SCHEMA_VERSION, TransactionStatus},
    },
    rpc::verification::VERIFICATION_SCHEMA_VERSION,
};

const EXECUTION_PAYLOAD_COMPONENT: &str = "evm_execution_payload";
const EXECUTION_PAYLOAD_PROTOCOL_VERSION: i16 = 1;
const EXECUTION_PAYLOAD_BATCH_SIZE: i64 = 100;
const EXECUTION_PAYLOAD_MAX_SEALS: i64 = 4_294_967_296;

pub(crate) struct ExecutionVerificationBootstrap<'a> {
    pub chain_id: u32,
    pub wallet_address: &'a str,
    pub manifest_version: &'a str,
    pub manifest_digest: &'a str,
    pub checkpoint_number: u64,
    pub checkpoint_hash: &'a str,
    pub checkpoint_parent_hash: &'a str,
    pub checkpoint_timestamp: u64,
    pub checkpoint_base_fee_per_gas: Option<u128>,
    pub finalized_headers: &'a [ExecutionVerifiedHeader],
    pub next_canonical_nonce: u64,
    pub observed_canonical_nonce: u64,
    pub provider_ids: &'a [String],
    pub operator_ids: &'a [String],
    pub failure_domain_ids: &'a [String],
    pub decisions: &'a [ExecutionVerificationDecision],
    pub migration: Option<&'a ExecutionVerificationMigration>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ExecutionVerificationMigrationSnapshot {
    pub intents: Vec<ExecutionIntentRow>,
    pub hashes: Vec<ExecutionTransactionHashRow>,
}

pub(crate) struct ExecutionVerificationMigration {
    pub snapshot: ExecutionVerificationMigrationSnapshot,
    pub records: Vec<ExecutionVerificationMigrationRecord>,
}

pub(crate) struct ExecutionVerificationMigrationRecord {
    pub intent_id: i64,
    pub nonce: Option<u64>,
    pub transaction_hash: Option<String>,
    pub terminal_status: Option<TransactionStatus>,
    pub block_number: Option<u64>,
    pub block_hash: Option<String>,
    pub receipt_success: Option<bool>,
    pub gas_used: Option<u64>,
    pub effective_gas_price: Option<String>,
    pub recover_prepared: bool,
    pub decisions: Vec<ExecutionVerificationDecision>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ExecutionVerifiedHeader {
    pub number: u64,
    pub hash: String,
    pub parent_hash: String,
    pub timestamp: u64,
    pub base_fee_per_gas: Option<u128>,
}

pub(crate) struct ExecutionVerificationPosition {
    pub next_canonical_nonce: u64,
    pub revision: u64,
    pub finalized_tip: ExecutionVerifiedHeader,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ExecutionVerificationDecision {
    pub read_class: &'static str,
    pub height_start: Option<u64>,
    pub height_end: Option<u64>,
    pub normalized_value_digest: String,
}

pub(crate) struct ExecutionNonceAssignment<'a> {
    pub intent_id: i64,
    pub chain_id: u32,
    pub wallet_address: &'a str,
    pub nonce: u64,
    pub manifest_version: &'a str,
    pub manifest_digest: &'a str,
    pub provider_ids: &'a [String],
    pub operator_ids: &'a [String],
    pub failure_domain_ids: &'a [String],
    pub decisions: &'a [ExecutionVerificationDecision],
}

pub(crate) struct ExecutionFinalityTransition<'a> {
    pub intent_id: i64,
    pub chain_id: u32,
    pub wallet_address: &'a str,
    pub nonce: u64,
    pub transaction_hash: &'a str,
    pub status: TransactionStatus,
    pub block_number: u64,
    pub block_hash: &'a str,
    pub receipt_success: bool,
    pub gas_used: u64,
    pub effective_gas_price: &'a str,
    pub manifest_version: &'a str,
    pub manifest_digest: &'a str,
    pub provider_ids: &'a [String],
    pub operator_ids: &'a [String],
    pub failure_domain_ids: &'a [String],
    pub decisions: &'a [ExecutionVerificationDecision],
    pub finalized_headers: &'a [ExecutionVerifiedHeader],
}

pub(crate) struct ExecutionVerificationBatch<'a> {
    pub intent_id: i64,
    pub chain_id: u32,
    pub wallet_address: &'a str,
    pub nonce: u64,
    pub decision_class: &'a str,
    pub manifest_version: &'a str,
    pub manifest_digest: &'a str,
    pub provider_ids: &'a [String],
    pub operator_ids: &'a [String],
    pub failure_domain_ids: &'a [String],
    pub decisions: &'a [ExecutionVerificationDecision],
}

pub(crate) struct ExecutionReplacementScan<'a> {
    pub intent_id: i64,
    pub chain_id: u32,
    pub wallet_address: &'a str,
    pub nonce: u64,
    pub finalized_cursor: Option<&'a ExecutionVerifiedHeader>,
    pub matched_transaction_hash: Option<&'a str>,
    pub manifest_version: &'a str,
    pub manifest_digest: &'a str,
    pub provider_ids: &'a [String],
    pub operator_ids: &'a [String],
    pub failure_domain_ids: &'a [String],
    pub decisions: &'a [ExecutionVerificationDecision],
}
const EXECUTION_PAYLOAD_INFRASTRUCTURE_QUERY: &str = "
    SELECT
        EXISTS (
            SELECT 1
            FROM pg_catalog.pg_trigger AS t
            JOIN pg_catalog.pg_class AS c ON c.oid = t.tgrelid
            JOIN pg_catalog.pg_namespace AS n ON n.oid = c.relnamespace
            WHERE n.nspname = current_schema()
              AND c.relname = 'execution_transaction_hash'
              AND t.tgname = 'execution_transaction_payload_fence'
              AND NOT t.tgisinternal
        ),
        EXISTS (
            SELECT 1
            FROM pg_catalog.pg_constraint AS con
            JOIN pg_catalog.pg_class AS c ON c.oid = con.conrelid
            JOIN pg_catalog.pg_namespace AS n ON n.oid = c.relnamespace
            WHERE n.nspname = current_schema()
              AND c.relname = 'execution_transaction_hash'
              AND con.conname = 'execution_transaction_payload_protected_check'
              AND con.contype = 'c'
        )
";

#[derive(Debug)]
struct ExecutionPayloadState {
    deployment_id: String,
    protocol_version: i16,
    operation: String,
    active_key_id: Vec<u8>,
}

/// Transaction-held lease which prevents payload maintenance from starting during an action.
pub(crate) struct ExecutionPayloadLease {
    _transaction: Transaction<'static, Postgres>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ExecutionPayloadCheck {
    pub protected: bool,
    pub deployment_id: Option<String>,
    pub plaintext_rows: u64,
    pub original_rows: u64,
    pub replacement_rows: u64,
    pub authenticated_rows: u64,
    pub key_ids: Vec<String>,
    pub read_roles: Vec<String>,
}

/// Database interface for persisting and retrieving blockchain entities and domain objects.
#[derive(Debug, Clone)]
pub struct BlockchainCacheDatabase {
    /// PostgreSQL connection pool used for database operations.
    pool: PgPool,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ExecutionIntentReservationStage {
    BeforeCommit,
    Commit,
}

#[derive(Debug)]
struct ExecutionIntentReservationError {
    stage: ExecutionIntentReservationStage,
    source: anyhow::Error,
}

impl Display for ExecutionIntentReservationError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self.stage {
            ExecutionIntentReservationStage::BeforeCommit => {
                f.write_str("Execution intent reservation failed before commit")
            }
            ExecutionIntentReservationStage::Commit => f.write_str(
                "Execution intent reservation commit outcome is unknown; reconciliation is required",
            ),
        }
    }
}

impl std::error::Error for ExecutionIntentReservationError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        Some(self.source.as_ref())
    }
}

pub(crate) fn reservation_failure_proven_not_committed(error: &anyhow::Error) -> bool {
    error
        .downcast_ref::<ExecutionIntentReservationError>()
        .is_some_and(|e| e.stage == ExecutionIntentReservationStage::BeforeCommit)
}

// Shared SELECT column list for `pool` row queries (load_pools, load_pool).
const POOL_ROW_COLUMNS: &str = "
    address,
    pool_identifier,
    dex_name,
    creation_block,
    COALESCE(
        (SELECT timestamp::TEXT FROM block WHERE block.chain_id = pool.chain_id AND block.number = pool.creation_block),
        (SELECT timestamp::TEXT FROM pool_event_block WHERE pool_event_block.chain_id = pool.chain_id AND pool_event_block.number = pool.creation_block)
    ) as creation_block_timestamp,
    token0_chain,
    token0_address,
    token1_chain,
    token1_address,
    fee,
    tick_spacing,
    initial_tick,
    initial_sqrt_price_x96,
    hook_address
";

impl BlockchainCacheDatabase {
    /// Initializes a new database instance by establishing a connection to PostgreSQL.
    ///
    /// # Panics
    ///
    /// Panics if unable to connect to PostgreSQL with the provided options.
    pub async fn init(pg_options: PgConnectOptions) -> Self {
        Self::connect(pg_options)
            .await
            .expect("Error connecting to Postgres")
    }

    /// Establishes a connection to PostgreSQL and returns a new database instance.
    ///
    /// # Errors
    ///
    /// Returns an error if a connection cannot be established with the provided options.
    pub async fn connect(pg_options: PgConnectOptions) -> anyhow::Result<Self> {
        let pool = sqlx::postgres::PgPoolOptions::new()
            .max_connections(32) // Increased from default 10
            .min_connections(5) // Keep some connections warm
            .acquire_timeout(std::time::Duration::from_secs(3))
            .connect_with(pg_options)
            .await?;
        Ok(Self { pool })
    }

    /// Seeds the database with a blockchain chain record.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn seed_chain(&self, chain: &Chain) -> anyhow::Result<()> {
        sqlx::query(
            "
            INSERT INTO chain (
                chain_id, name
            ) VALUES ($1,$2)
            ON CONFLICT (chain_id)
            DO NOTHING
        ",
        )
        .bind(chain.chain_id as i32)
        .bind(chain.name.to_string())
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to seed chain table: {e}"))
    }

    /// Creates a table partition for the block table specific to the given chain
    /// by calling the existing PostgreSQL function `create_block_partition`.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn create_block_partition(&self, chain: &Chain) -> anyhow::Result<String> {
        let result: (String,) = sqlx::query_as("SELECT create_block_partition($1)")
            .bind(chain.chain_id as i32)
            .fetch_one(&self.pool)
            .await
            .map_err(|e| {
                anyhow::anyhow!(
                    "Failed to call create_block_partition for chain {}: {e}",
                    chain.chain_id
                )
            })?;

        Ok(result.0)
    }

    /// Creates a table partition for the token table specific to the given chain
    /// by calling the existing PostgreSQL function `create_token_partition`.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn create_token_partition(&self, chain: &Chain) -> anyhow::Result<String> {
        let result: (String,) = sqlx::query_as("SELECT create_token_partition($1)")
            .bind(chain.chain_id as i32)
            .fetch_one(&self.pool)
            .await
            .map_err(|e| {
                anyhow::anyhow!(
                    "Failed to call create_token_partition for chain {}: {e}",
                    chain.chain_id
                )
            })?;

        Ok(result.0)
    }

    /// Returns the highest block number that maintains data continuity in the database.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    pub async fn get_block_consistency_status(
        &self,
        chain: &Chain,
    ) -> anyhow::Result<CachedBlocksConsistencyStatus> {
        log::debug!("Fetching block consistency status");

        let result: (i64, i64) = sqlx::query_as(
            "
            SELECT
                COALESCE((SELECT number FROM block WHERE chain_id = $1 ORDER BY number DESC LIMIT 1), 0) as max_block,
                get_last_continuous_block($1) as last_continuous_block
            "
        )
        .bind(chain.chain_id as i32)
        .fetch_one(&self.pool)
        .await
        .map_err(|e| {
            anyhow::anyhow!(
                "Failed to get block info for chain {}: {}",
                chain.chain_id,
                e
            )
        })?;

        Ok(CachedBlocksConsistencyStatus::new(
            result.0 as u64,
            result.1 as u64,
        ))
    }

    /// Inserts or updates a block record in the database.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn add_block(&self, chain_id: u32, block: &Block) -> anyhow::Result<()> {
        sqlx::query(
            "
            INSERT INTO block (
                chain_id, number, hash, parent_hash, miner, gas_limit, gas_used, timestamp,
                base_fee_per_gas, blob_gas_used, excess_blob_gas,
                l1_gas_price, l1_gas_used, l1_fee_scalar
            ) VALUES (
                $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14
            )
            ON CONFLICT (chain_id, number)
            DO UPDATE
            SET
                hash = $3,
                parent_hash = $4,
                miner = $5,
                gas_limit = $6,
                gas_used = $7,
                timestamp = $8,
                base_fee_per_gas = $9,
                blob_gas_used = $10,
                excess_blob_gas = $11,
                l1_gas_price = $12,
                l1_gas_used = $13,
                l1_fee_scalar = $14
        ",
        )
        .bind(chain_id as i32)
        .bind(block.number as i64)
        .bind(block.hash.as_str())
        .bind(block.parent_hash.as_str())
        .bind(block.miner.as_str())
        .bind(block.gas_limit as i64)
        .bind(block.gas_used as i64)
        .bind(block.timestamp.to_string())
        .bind(block.base_fee_per_gas.as_ref().map(U256::to_string))
        .bind(block.blob_gas_used.as_ref().map(U256::to_string))
        .bind(block.excess_blob_gas.as_ref().map(U256::to_string))
        .bind(block.l1_gas_price.as_ref().map(U256::to_string))
        .bind(block.l1_gas_used.map(|v| v as i64))
        .bind(block.l1_fee_scalar.map(|v| v as i64))
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to insert into block table: {e}"))
    }

    /// Inserts multiple blocks in a single database operation using UNNEST for optimal performance.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn add_blocks_batch(&self, chain_id: u32, blocks: &[Block]) -> anyhow::Result<()> {
        if blocks.is_empty() {
            return Ok(());
        }

        // Prepare vectors for each column
        let mut numbers: Vec<i64> = Vec::with_capacity(blocks.len());
        let mut hashes: Vec<String> = Vec::with_capacity(blocks.len());
        let mut parent_hashes: Vec<String> = Vec::with_capacity(blocks.len());
        let mut miners: Vec<String> = Vec::with_capacity(blocks.len());
        let mut gas_limits: Vec<i64> = Vec::with_capacity(blocks.len());
        let mut gas_useds: Vec<i64> = Vec::with_capacity(blocks.len());
        let mut timestamps: Vec<String> = Vec::with_capacity(blocks.len());
        let mut base_fee_per_gases: Vec<Option<String>> = Vec::with_capacity(blocks.len());
        let mut blob_gas_useds: Vec<Option<String>> = Vec::with_capacity(blocks.len());
        let mut excess_blob_gases: Vec<Option<String>> = Vec::with_capacity(blocks.len());
        let mut l1_gas_prices: Vec<Option<String>> = Vec::with_capacity(blocks.len());
        let mut l1_gas_useds: Vec<Option<i64>> = Vec::with_capacity(blocks.len());
        let mut l1_fee_scalars: Vec<Option<i64>> = Vec::with_capacity(blocks.len());

        // Fill vectors from blocks
        for block in blocks {
            numbers.push(block.number as i64);
            hashes.push(block.hash.clone());
            parent_hashes.push(block.parent_hash.clone());
            miners.push(block.miner.to_string());
            gas_limits.push(block.gas_limit as i64);
            gas_useds.push(block.gas_used as i64);
            timestamps.push(block.timestamp.to_string());
            base_fee_per_gases.push(block.base_fee_per_gas.as_ref().map(U256::to_string));
            blob_gas_useds.push(block.blob_gas_used.as_ref().map(U256::to_string));
            excess_blob_gases.push(block.excess_blob_gas.as_ref().map(U256::to_string));
            l1_gas_prices.push(block.l1_gas_price.as_ref().map(U256::to_string));
            l1_gas_useds.push(block.l1_gas_used.map(|v| v as i64));
            l1_fee_scalars.push(block.l1_fee_scalar.map(|v| v as i64));
        }

        // Execute batch insert with UNNEST
        sqlx::query(
            "
            INSERT INTO block (
                chain_id, number, hash, parent_hash, miner, gas_limit, gas_used, timestamp,
                base_fee_per_gas, blob_gas_used, excess_blob_gas,
                l1_gas_price, l1_gas_used, l1_fee_scalar
            )
            SELECT
                $1, *
            FROM UNNEST(
                $2::int8[], $3::text[], $4::text[], $5::text[],
                $6::int8[], $7::int8[], $8::text[],
                $9::text[], $10::text[], $11::text[],
                $12::text[], $13::int8[], $14::int8[]
            )
            ON CONFLICT (chain_id, number) DO NOTHING
           ",
        )
        .bind(chain_id as i32)
        .bind(&numbers[..])
        .bind(&hashes[..])
        .bind(&parent_hashes[..])
        .bind(&miners[..])
        .bind(&gas_limits[..])
        .bind(&gas_useds[..])
        .bind(&timestamps[..])
        .bind(&base_fee_per_gases as &[Option<String>])
        .bind(&blob_gas_useds as &[Option<String>])
        .bind(&excess_blob_gases as &[Option<String>])
        .bind(&l1_gas_prices as &[Option<String>])
        .bind(&l1_gas_useds as &[Option<i64>])
        .bind(&l1_fee_scalars as &[Option<i64>])
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to batch insert into block table: {e}"))
    }

    /// Inserts block timestamps observed while streaming pool events.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn add_pool_event_blocks_batch(
        &self,
        chain_id: u32,
        blocks: &[Block],
    ) -> anyhow::Result<()> {
        if blocks.is_empty() {
            return Ok(());
        }

        let chain_id_db = i32::try_from(chain_id)
            .with_context(|| format!("Chain ID {chain_id} exceeds PostgreSQL INTEGER"))?;
        let mut numbers: Vec<i64> = Vec::with_capacity(blocks.len());
        let mut hashes: Vec<String> = Vec::with_capacity(blocks.len());
        let mut timestamps: Vec<String> = Vec::with_capacity(blocks.len());

        for block in blocks {
            numbers.push(i64::try_from(block.number).with_context(|| {
                format!(
                    "Pool event block {} exceeds PostgreSQL BIGINT",
                    block.number
                )
            })?);
            hashes.push(block.hash.clone());
            timestamps.push(block.timestamp.to_string());
        }

        sqlx::query(
            "
            INSERT INTO pool_event_block (
                chain_id, number, hash, timestamp
            )
            SELECT
                $1, *
            FROM UNNEST(
                $2::int8[], $3::text[], $4::text[]
            )
            ON CONFLICT (chain_id, number)
            DO UPDATE SET hash = EXCLUDED.hash, timestamp = EXCLUDED.timestamp
           ",
        )
        .bind(chain_id_db)
        .bind(&numbers[..])
        .bind(&hashes[..])
        .bind(&timestamps[..])
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to batch insert into pool_event_block table: {e}"))
    }

    /// Adds block-hash storage to databases created before hash-bound profiler checkpoints.
    ///
    /// # Errors
    ///
    /// Returns an error if the schema update fails.
    pub async fn ensure_pool_event_block_hash_schema(&self) -> anyhow::Result<()> {
        sqlx::query("ALTER TABLE pool_event_block ADD COLUMN IF NOT EXISTS hash TEXT")
            .execute(&self.pool)
            .await
            .map(|_| ())
            .map_err(|e| anyhow::anyhow!("Failed to add pool event block hash storage: {e}"))
    }

    /// Inserts blocks using PostgreSQL COPY BINARY for maximum performance.
    ///
    /// This method is significantly faster than INSERT for bulk operations as it bypasses
    /// SQL parsing and uses PostgreSQL's native binary protocol.
    ///
    /// # Errors
    ///
    /// Returns an error if the COPY operation fails.
    pub async fn add_blocks_copy(&self, chain_id: u32, blocks: &[Block]) -> anyhow::Result<()> {
        let copy_handler = PostgresCopyHandler::new(&self.pool);
        copy_handler.copy_blocks(chain_id, blocks).await
    }

    /// Inserts tokens using PostgreSQL COPY BINARY for maximum performance.
    ///
    /// # Errors
    ///
    /// Returns an error if the COPY operation fails.
    pub async fn add_tokens_copy(&self, chain_id: u32, tokens: &[Token]) -> anyhow::Result<()> {
        let copy_handler = PostgresCopyHandler::new(&self.pool);
        copy_handler.copy_tokens(chain_id, tokens).await
    }

    /// Inserts pools using PostgreSQL COPY BINARY for maximum performance.
    ///
    /// # Errors
    ///
    /// Returns an error if the COPY operation fails.
    pub async fn add_pools_copy(&self, chain_id: u32, pools: &[Pool]) -> anyhow::Result<()> {
        let copy_handler = PostgresCopyHandler::new(&self.pool);
        copy_handler.copy_pools(chain_id, pools).await
    }

    /// Inserts pool swaps using PostgreSQL COPY BINARY for maximum performance.
    ///
    /// This method is significantly faster than INSERT for bulk operations as it bypasses
    /// SQL parsing and uses PostgreSQL's native binary protocol.
    ///
    /// # Errors
    ///
    /// Returns an error if the COPY operation fails.
    pub async fn add_pool_swaps_copy(
        &self,
        chain_id: u32,
        swaps: &[PoolSwap],
    ) -> anyhow::Result<()> {
        let copy_handler = PostgresCopyHandler::new(&self.pool);
        copy_handler.copy_pool_swaps(chain_id, swaps).await
    }

    /// Inserts pool liquidity updates using PostgreSQL COPY BINARY for maximum performance.
    ///
    /// This method is significantly faster than INSERT for bulk operations as it bypasses
    /// SQL parsing and uses PostgreSQL's native binary protocol.
    ///
    /// # Errors
    ///
    /// Returns an error if the COPY operation fails.
    pub async fn add_pool_liquidity_updates_copy(
        &self,
        chain_id: u32,
        updates: &[PoolLiquidityUpdate],
    ) -> anyhow::Result<()> {
        let copy_handler = PostgresCopyHandler::new(&self.pool);
        copy_handler
            .copy_pool_liquidity_updates(chain_id, updates)
            .await
    }

    /// Inserts pool fee collect events using PostgreSQL COPY BINARY for maximum performance.
    ///
    /// This method is significantly faster than INSERT for bulk operations as it bypasses
    /// SQL parsing and most database validation checks.
    ///
    /// # Errors
    ///
    /// Returns an error if the COPY operation fails.
    pub async fn copy_pool_fee_collects_batch(
        &self,
        chain_id: u32,
        collects: &[PoolFeeCollect],
    ) -> anyhow::Result<()> {
        let copy_handler = PostgresCopyHandler::new(&self.pool);
        copy_handler.copy_pool_collects(chain_id, collects).await
    }

    /// Retrieves block timestamps for a given chain starting from a specific block number.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    pub async fn load_block_timestamps(
        &self,
        chain: SharedChain,
        from_block: u64,
    ) -> anyhow::Result<Vec<BlockTimestampRow>> {
        sqlx::query_as::<_, BlockTimestampRow>(
            "
            SELECT DISTINCT ON (number)
                number,
                timestamp
            FROM (
                SELECT number, timestamp, 0 AS source_order
                FROM block
                WHERE chain_id = $1 AND number >= $2 AND timestamp IS NOT NULL
                UNION ALL
                SELECT number, timestamp, 1 AS source_order
                FROM pool_event_block
                WHERE chain_id = $1 AND number >= $2 AND timestamp IS NOT NULL
            ) AS block_timestamps
            ORDER BY number ASC, source_order ASC
            ",
        )
        .bind(chain.chain_id as i32)
        .bind(from_block as i64)
        .fetch_all(&self.pool)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to load block timestamps: {e}"))
    }

    /// Adds or updates a DEX (Decentralized Exchange) record in the database.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn add_dex(&self, dex: SharedDex) -> anyhow::Result<()> {
        sqlx::query(
            "
            INSERT INTO dex (
                chain_id, name, factory_address, creation_block
            ) VALUES ($1, $2, $3, $4)
            ON CONFLICT (chain_id, name)
            DO UPDATE
            SET
                factory_address = $3,
                creation_block = $4
        ",
        )
        .bind(dex.chain.chain_id as i32)
        .bind(dex.name.to_string())
        .bind(dex.factory.to_string())
        .bind(dex.factory_creation_block as i64)
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to insert into dex table: {e}"))
    }

    /// Adds or updates a liquidity pool/pair record in the database.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn add_pool(&self, pool: &Pool) -> anyhow::Result<()> {
        sqlx::query(
            "
            INSERT INTO pool (
                chain_id, address, pool_identifier, dex_name, creation_block,
                token0_chain, token0_address,
                token1_chain, token1_address,
                fee, tick_spacing, initial_tick, initial_sqrt_price_x96, hook_address
            ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
            ON CONFLICT (chain_id, dex_name, pool_identifier)
            DO UPDATE
            SET
                address = $2,
                creation_block = $5,
                token0_chain = $6,
                token0_address = $7,
                token1_chain = $8,
                token1_address = $9,
                fee = $10,
                tick_spacing = $11,
                initial_tick = $12,
                initial_sqrt_price_x96 = $13,
                hook_address = $14
        ",
        )
        .bind(pool.chain.chain_id as i32)
        .bind(pool.address.to_string())
        .bind(pool.pool_identifier.as_ref())
        .bind(pool.dex.name.to_string())
        .bind(pool.creation_block as i64)
        .bind(pool.token0.chain.chain_id as i32)
        .bind(pool.token0.address.to_string())
        .bind(pool.token1.chain.chain_id as i32)
        .bind(pool.token1.address.to_string())
        .bind(pool.fee.map(|fee| fee as i32))
        .bind(pool.tick_spacing.map(|tick_spacing| tick_spacing as i32))
        .bind(pool.initial_tick)
        .bind(pool.initial_sqrt_price_x96.as_ref().map(|p| p.to_string()))
        .bind(pool.hooks.as_ref().map(|h| h.to_string()))
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to insert into pool table: {e}"))
    }

    /// Inserts multiple pools in a single database operation using UNNEST for optimal performance.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn add_pools_batch(&self, pools: &[Pool]) -> anyhow::Result<()> {
        if pools.is_empty() {
            return Ok(());
        }

        // Prepare vectors for each column
        let len = pools.len();
        let mut addresses: Vec<String> = Vec::with_capacity(len);
        let mut pool_identifiers: Vec<String> = Vec::with_capacity(len);
        let mut dex_names: Vec<String> = Vec::with_capacity(len);
        let mut creation_blocks: Vec<i64> = Vec::with_capacity(len);
        let mut token0_chains: Vec<i32> = Vec::with_capacity(len);
        let mut token0_addresses: Vec<String> = Vec::with_capacity(len);
        let mut token1_chains: Vec<i32> = Vec::with_capacity(len);
        let mut token1_addresses: Vec<String> = Vec::with_capacity(len);
        let mut fees: Vec<Option<i32>> = Vec::with_capacity(len);
        let mut tick_spacings: Vec<Option<i32>> = Vec::with_capacity(len);
        let mut initial_ticks: Vec<Option<i32>> = Vec::with_capacity(len);
        let mut initial_sqrt_price_x96s: Vec<Option<String>> = Vec::with_capacity(len);
        let mut hook_addresses: Vec<Option<String>> = Vec::with_capacity(len);
        let mut chain_ids: Vec<i32> = Vec::with_capacity(len);

        // Fill vectors from pools
        for pool in pools {
            chain_ids.push(pool.chain.chain_id as i32);
            addresses.push(pool.address.to_string());
            pool_identifiers.push(pool.pool_identifier.to_string());
            dex_names.push(pool.dex.name.to_string());
            creation_blocks.push(pool.creation_block as i64);
            token0_chains.push(pool.token0.chain.chain_id as i32);
            token0_addresses.push(pool.token0.address.to_string());
            token1_chains.push(pool.token1.chain.chain_id as i32);
            token1_addresses.push(pool.token1.address.to_string());
            fees.push(pool.fee.map(|fee| fee as i32));
            tick_spacings.push(pool.tick_spacing.map(|tick_spacing| tick_spacing as i32));
            initial_ticks.push(pool.initial_tick);
            initial_sqrt_price_x96s
                .push(pool.initial_sqrt_price_x96.as_ref().map(|p| p.to_string()));
            hook_addresses.push(pool.hooks.as_ref().map(|h| h.to_string()));
        }

        // Execute batch insert with UNNEST
        sqlx::query(
            "
            INSERT INTO pool (
                chain_id, address, pool_identifier, dex_name, creation_block,
                token0_chain, token0_address,
                token1_chain, token1_address,
                fee, tick_spacing, initial_tick, initial_sqrt_price_x96, hook_address
            )
            SELECT *
            FROM UNNEST(
                $1::int4[], $2::text[], $3::text[], $4::text[], $5::int8[],
                $6::int4[], $7::text[], $8::int4[], $9::text[],
                $10::int4[], $11::int4[], $12::int4[], $13::text[], $14::text[]
            )
            ON CONFLICT (chain_id, dex_name, pool_identifier) DO NOTHING
           ",
        )
        .bind(&chain_ids[..])
        .bind(&addresses[..])
        .bind(&pool_identifiers[..])
        .bind(&dex_names[..])
        .bind(&creation_blocks[..])
        .bind(&token0_chains[..])
        .bind(&token0_addresses[..])
        .bind(&token1_chains[..])
        .bind(&token1_addresses[..])
        .bind(&fees[..])
        .bind(&tick_spacings[..])
        .bind(&initial_ticks[..])
        .bind(&initial_sqrt_price_x96s[..])
        .bind(&hook_addresses as &[Option<String>])
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to batch insert into pool table: {e}"))
    }

    /// Inserts multiple pool swaps in a single database operation using UNNEST for optimal performance.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn add_pool_swaps_batch(
        &self,
        chain_id: u32,
        swaps: &[PoolSwap],
    ) -> anyhow::Result<()> {
        if swaps.is_empty() {
            return Ok(());
        }

        // Prepare vectors for each column
        let len = swaps.len();
        let mut chain_ids: Vec<i32> = Vec::with_capacity(len);
        let mut dex_names: Vec<String> = Vec::with_capacity(len);
        let mut pool_identifiers: Vec<String> = Vec::with_capacity(len);
        let mut blocks: Vec<i64> = Vec::with_capacity(len);
        let mut transaction_hashes: Vec<String> = Vec::with_capacity(len);
        let mut transaction_indices: Vec<i32> = Vec::with_capacity(len);
        let mut log_indices: Vec<i32> = Vec::with_capacity(len);
        let mut senders: Vec<String> = Vec::with_capacity(len);
        let mut recipients: Vec<String> = Vec::with_capacity(len);
        let mut sqrt_price_x96s: Vec<String> = Vec::with_capacity(len);
        let mut liquidities: Vec<String> = Vec::with_capacity(len);
        let mut ticks: Vec<i32> = Vec::with_capacity(len);
        let mut amount0s: Vec<String> = Vec::with_capacity(len);
        let mut amount1s: Vec<String> = Vec::with_capacity(len);
        let mut order_sides: Vec<Option<String>> = Vec::with_capacity(len);
        let mut base_quantities: Vec<Option<Decimal>> = Vec::with_capacity(len);
        let mut quote_quantities: Vec<Option<Decimal>> = Vec::with_capacity(len);
        let mut spot_prices: Vec<Option<Decimal>> = Vec::with_capacity(len);
        let mut execution_prices: Vec<Option<Decimal>> = Vec::with_capacity(len);

        // Fill vectors from swaps
        for swap in swaps {
            chain_ids.push(chain_id as i32);
            dex_names.push(swap.dex.name.to_string());
            pool_identifiers.push(swap.pool_identifier.to_string());
            blocks.push(swap.block as i64);
            transaction_hashes.push(swap.transaction_hash.clone());
            transaction_indices.push(swap.transaction_index as i32);
            log_indices.push(swap.log_index as i32);
            senders.push(swap.sender.to_string());
            recipients.push(swap.recipient.to_string());
            sqrt_price_x96s.push(swap.sqrt_price_x96.to_string());
            liquidities.push(swap.liquidity.to_string());
            ticks.push(swap.tick);
            amount0s.push(swap.amount0.to_string());
            amount1s.push(swap.amount1.to_string());

            // Extract trade_info fields if available
            if let Some(ref trade_info) = swap.trade_info {
                order_sides.push(Some(trade_info.order_side.to_string()));
                base_quantities.push(Some(trade_info.quantity_base.as_decimal()));
                quote_quantities.push(Some(trade_info.quantity_quote.as_decimal()));
                spot_prices.push(Some(trade_info.spot_price.as_decimal()));
                execution_prices.push(Some(trade_info.execution_price.as_decimal()));
            } else {
                order_sides.push(None);
                base_quantities.push(None);
                quote_quantities.push(None);
                spot_prices.push(None);
                execution_prices.push(None);
            }
        }

        // Execute batch insert with UNNEST
        sqlx::query(
            "
            INSERT INTO pool_swap_event (
                chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index,
                log_index, sender, recipient, sqrt_price_x96, liquidity, tick, amount0, amount1,
                order_side, base_quantity, quote_quantity, spot_price, execution_price
            )
            SELECT
                chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index, log_index, sender, recipient,
                sqrt_price_x96::U160, liquidity::U128, tick, amount0::I256, amount1::I256,
                order_side, base_quantity, quote_quantity, spot_price, execution_price
            FROM UNNEST(
                $1::INT[], $2::TEXT[], $3::TEXT[], $4::BIGINT[], $5::TEXT[], $6::INT[], $7::INT[],
                $8::TEXT[], $9::TEXT[], $10::TEXT[], $11::TEXT[], $12::INT[], $13::TEXT[], $14::TEXT[],
                $15::TEXT[], $16, $17, $18, $19
            ) AS t(chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index,
                   log_index, sender, recipient, sqrt_price_x96, liquidity, tick, amount0, amount1,
                   order_side, base_quantity, quote_quantity, spot_price, execution_price)
            ON CONFLICT (chain_id, transaction_hash, log_index) DO NOTHING
           ",
        )
        .bind(&chain_ids[..])
        .bind(&dex_names[..])
        .bind(&pool_identifiers[..])
        .bind(&blocks[..])
        .bind(&transaction_hashes[..])
        .bind(&transaction_indices[..])
        .bind(&log_indices[..])
        .bind(&senders[..])
        .bind(&recipients[..])
        .bind(&sqrt_price_x96s[..])
        .bind(&liquidities[..])
        .bind(&ticks[..])
        .bind(&amount0s[..])
        .bind(&amount1s[..])
        .bind(&order_sides[..])
        .bind(&base_quantities[..])
        .bind(&quote_quantities[..])
        .bind(&spot_prices[..])
        .bind(&execution_prices[..])
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to batch insert into pool_swap_event table: {e}"))
    }

    /// Inserts multiple pool liquidity updates in a single database operation using UNNEST for optimal performance.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn add_pool_liquidity_updates_batch(
        &self,
        chain_id: u32,
        updates: &[PoolLiquidityUpdate],
    ) -> anyhow::Result<()> {
        if updates.is_empty() {
            return Ok(());
        }

        // Prepare vectors for each column
        let len = updates.len();
        let mut chain_ids: Vec<i32> = Vec::with_capacity(len);
        let mut dex_names: Vec<String> = Vec::with_capacity(len);
        let mut pool_identifiers: Vec<String> = Vec::with_capacity(len);
        let mut blocks: Vec<i64> = Vec::with_capacity(len);
        let mut transaction_hashes: Vec<String> = Vec::with_capacity(len);
        let mut transaction_indices: Vec<i32> = Vec::with_capacity(len);
        let mut log_indices: Vec<i32> = Vec::with_capacity(len);
        let mut event_types: Vec<String> = Vec::with_capacity(len);
        let mut senders: Vec<Option<String>> = Vec::with_capacity(len);
        let mut owners: Vec<String> = Vec::with_capacity(len);
        let mut position_liquidities: Vec<String> = Vec::with_capacity(len);
        let mut amount0s: Vec<String> = Vec::with_capacity(len);
        let mut amount1s: Vec<String> = Vec::with_capacity(len);
        let mut tick_lowers: Vec<i32> = Vec::with_capacity(len);
        let mut tick_uppers: Vec<i32> = Vec::with_capacity(len);

        // Fill vectors from updates
        for update in updates {
            chain_ids.push(chain_id as i32);
            dex_names.push(update.dex.name.to_string());
            pool_identifiers.push(update.pool_identifier.to_string());
            blocks.push(update.block as i64);
            transaction_hashes.push(update.transaction_hash.clone());
            transaction_indices.push(update.transaction_index as i32);
            log_indices.push(update.log_index as i32);
            event_types.push(update.kind.to_string());
            senders.push(update.sender.map(|s| s.to_string()));
            owners.push(update.owner.to_string());
            position_liquidities.push(update.position_liquidity.to_string());
            amount0s.push(update.amount0.to_string());
            amount1s.push(update.amount1.to_string());
            tick_lowers.push(update.tick_lower);
            tick_uppers.push(update.tick_upper);
        }

        // Execute batch insert with UNNEST
        sqlx::query(
            "
            INSERT INTO pool_liquidity_event (
                chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index,
                log_index, event_type, sender, owner, position_liquidity,
                amount0, amount1, tick_lower, tick_upper
            )
            SELECT
                chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index,
                log_index, event_type, sender, owner, position_liquidity::u128,
                amount0::U256, amount1::U256, tick_lower, tick_upper
            FROM UNNEST(
                $1::INT[], $2::TEXT[], $3::TEXT[], $4::INT[], $5::TEXT[], $6::INT[],
                $7::INT[], $8::TEXT[], $9::TEXT[], $10::TEXT[], $11::TEXT[],
                $12::TEXT[], $13::TEXT[], $14::INT[], $15::INT[]
            ) AS t(chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index,
                   log_index, event_type, sender, owner, position_liquidity,
                   amount0, amount1, tick_lower, tick_upper)
            ON CONFLICT (chain_id, transaction_hash, log_index) DO NOTHING
           ",
        )
        .bind(&chain_ids[..])
        .bind(&dex_names[..])
        .bind(&pool_identifiers[..])
        .bind(&blocks[..])
        .bind(&transaction_hashes[..])
        .bind(&transaction_indices[..])
        .bind(&log_indices[..])
        .bind(&event_types[..])
        .bind(&senders[..])
        .bind(&owners[..])
        .bind(&position_liquidities[..])
        .bind(&amount0s[..])
        .bind(&amount1s[..])
        .bind(&tick_lowers[..])
        .bind(&tick_uppers[..])
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to batch insert into pool_liquidity_event table: {e}"))
    }

    /// Adds or updates a token record in the database.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn add_token(&self, token: &Token) -> anyhow::Result<()> {
        sqlx::query(
            "
            INSERT INTO token (
                chain_id, address, name, symbol, decimals
            ) VALUES ($1, $2, $3, $4, $5)
            ON CONFLICT (chain_id, address)
            DO UPDATE
            SET
                name = $3,
                symbol = $4,
                decimals = $5
        ",
        )
        .bind(token.chain.chain_id as i32)
        .bind(token.address.to_string())
        .bind(token.name.as_str())
        .bind(token.symbol.as_str())
        .bind(i32::from(token.decimals))
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to insert into token table: {e}"))
    }

    /// Records an invalid token address with associated error information.
    ///
    /// # Errors
    ///
    /// Returns an error if the database insertion fails.
    pub async fn add_invalid_token(
        &self,
        chain_id: u32,
        address: &Address,
        error_string: &str,
    ) -> anyhow::Result<()> {
        sqlx::query(
            "
            INSERT INTO token (
                chain_id, address, error
            ) VALUES ($1, $2, $3)
            ON CONFLICT (chain_id, address)
            DO NOTHING;
        ",
        )
        .bind(chain_id as i32)
        .bind(address.to_string())
        .bind(error_string)
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to insert into token table: {e}"))
    }

    /// Persists a token swap transaction event to the `pool_swap` table.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn add_swap(&self, chain_id: u32, swap: &PoolSwap) -> anyhow::Result<()> {
        // Extract trade_info fields if available
        let (order_side, base_quantity, quote_quantity, spot_price, execution_price) =
            if let Some(ref trade_info) = swap.trade_info {
                (
                    Some(trade_info.order_side.to_string()),
                    Some(trade_info.quantity_base.as_decimal()),
                    Some(trade_info.quantity_quote.as_decimal()),
                    Some(trade_info.spot_price.as_decimal()),
                    Some(trade_info.execution_price.as_decimal()),
                )
            } else {
                (None, None, None, None, None)
            };

        sqlx::query(
            "
            INSERT INTO pool_swap_event (
                chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index,
                log_index, sender, recipient, sqrt_price_x96, liquidity, tick, amount0, amount1,
                order_side, base_quantity, quote_quantity, spot_price, execution_price
            ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::U160, $11::U128, $12, $13::I256, $14::I256, $15, $16, $17, $18, $19)
            ON CONFLICT (chain_id, transaction_hash, log_index)
            DO NOTHING
        ",
        )
        .bind(chain_id as i32)
        .bind(swap.dex.name.to_string())
        .bind(swap.pool_identifier.as_str())
        .bind(swap.block as i64)
        .bind(swap.transaction_hash.as_str())
        .bind(swap.transaction_index as i32)
        .bind(swap.log_index as i32)
        .bind(swap.sender.to_string())
        .bind(swap.recipient.to_string())
        .bind(swap.sqrt_price_x96.to_string())
        .bind(swap.liquidity.to_string())
        .bind(swap.tick)
        .bind(swap.amount0.to_string())
        .bind(swap.amount1.to_string())
        .bind(order_side)
        .bind(base_quantity)
        .bind(quote_quantity)
        .bind(spot_price)
        .bind(execution_price)
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to insert into pool_swap table: {e}"))
    }

    /// Persists a liquidity position change (mint/burn) event to the `pool_liquidity` table.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn add_pool_liquidity_update(
        &self,
        chain_id: u32,
        liquidity_update: &PoolLiquidityUpdate,
    ) -> anyhow::Result<()> {
        sqlx::query(
            "
            INSERT INTO pool_liquidity_event (
                chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index, log_index,
                event_type, sender, owner, position_liquidity, amount0, amount1, tick_lower, tick_upper
            ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)
            ON CONFLICT (chain_id, transaction_hash, log_index)
            DO NOTHING
        ",
        )
        .bind(chain_id as i32)
        .bind(liquidity_update.dex.name.to_string())
        .bind(liquidity_update.pool_identifier.as_str())
        .bind(liquidity_update.block as i64)
        .bind(liquidity_update.transaction_hash.as_str())
        .bind(liquidity_update.transaction_index as i32)
        .bind(liquidity_update.log_index as i32)
        .bind(liquidity_update.kind.to_string())
        .bind(liquidity_update.sender.map(|sender| sender.to_string()))
        .bind(liquidity_update.owner.to_string())
        .bind(U128Pg(liquidity_update.position_liquidity))
        .bind(U256Pg(liquidity_update.amount0))
        .bind(U256Pg(liquidity_update.amount1))
        .bind(liquidity_update.tick_lower)
        .bind(liquidity_update.tick_upper)
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to insert into pool_liquidity table: {e}"))
    }

    /// Retrieves all valid token records for the given chain and converts them into `Token` domain objects.
    ///
    /// Only returns tokens that do not contain error information, filtering out invalid tokens
    /// that were previously recorded with error details.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    pub async fn load_tokens(&self, chain: SharedChain) -> anyhow::Result<Vec<Token>> {
        sqlx::query_as::<_, TokenRow>("SELECT * FROM token WHERE chain_id = $1 AND error IS NULL")
            .bind(chain.chain_id as i32)
            .fetch_all(&self.pool)
            .await
            .map(|rows| {
                rows.into_iter()
                    .map(|token_row| {
                        Token::new(
                            chain.clone(),
                            token_row.address,
                            token_row.name,
                            token_row.symbol,
                            token_row.decimals,
                        )
                    })
                    .collect::<Vec<_>>()
            })
            .map_err(|e| anyhow::anyhow!("Failed to load tokens: {e}"))
    }

    /// Retrieves all invalid token addresses for a given chain.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails or address validation fails.
    pub async fn load_invalid_token_addresses(
        &self,
        chain_id: u32,
    ) -> anyhow::Result<Vec<Address>> {
        sqlx::query_as::<_, (String,)>(
            "SELECT address FROM token WHERE chain_id = $1 AND error IS NOT NULL",
        )
        .bind(chain_id as i32)
        .fetch_all(&self.pool)
        .await?
        .into_iter()
        .map(|(address,)| validate_address(&address))
        .collect::<Result<Vec<_>, _>>()
        .map_err(|e| anyhow::anyhow!("Failed to load invalid token addresses: {e}"))
    }

    /// Loads pool data from the database for the specified chain and DEX.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails, the connection to the database is lost, or the query parameters are invalid.
    pub async fn load_pools(
        &self,
        chain: SharedChain,
        dex_id: &str,
    ) -> anyhow::Result<Vec<PoolRow>> {
        sqlx::query_as::<_, PoolRow>(AssertSqlSafe(format!(
            "SELECT {POOL_ROW_COLUMNS} FROM pool WHERE chain_id = $1 AND dex_name = $2 ORDER BY creation_block ASC"
        )))
        .bind(chain.chain_id as i32)
        .bind(dex_id)
        .fetch_all(&self.pool)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to load pools: {e}"))
    }

    /// Loads a single pool row by its identifier.
    ///
    /// Returns `None` when the pool is not present in the database. Lets per-pool tools load only
    /// the pool they analyze instead of the whole DEX pool set (see [`load_pools`]).
    ///
    /// [`load_pools`]: Self::load_pools
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    pub async fn load_pool(
        &self,
        chain: SharedChain,
        dex_id: &str,
        pool_identifier: &PoolIdentifier,
    ) -> anyhow::Result<Option<PoolRow>> {
        sqlx::query_as::<_, PoolRow>(AssertSqlSafe(format!(
            "SELECT {POOL_ROW_COLUMNS} FROM pool WHERE chain_id = $1 AND dex_name = $2 AND pool_identifier = $3"
        )))
        .bind(chain.chain_id as i32)
        .bind(dex_id)
        .bind(pool_identifier.as_ref())
        .fetch_optional(&self.pool)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to load pool {pool_identifier}: {e}"))
    }

    /// Toggles performance optimization settings for sync operations.
    ///
    /// When enabled (true), applies settings for maximum write performance:
    /// - `synchronous_commit` = OFF
    /// - `work_mem` increased for bulk operations
    ///
    /// When disabled (false), restores default safe settings:
    /// - `synchronous_commit` = ON (data safety)
    /// - `work_mem` back to default
    ///
    /// # Errors
    ///
    /// Returns an error if the database operations fail.
    pub async fn toggle_perf_sync_settings(&self, enable: bool) -> anyhow::Result<()> {
        if enable {
            log::debug!("Enabling performance sync settings for bulk operations");

            // Set synchronous_commit to OFF for maximum write performance
            sqlx::query("SET synchronous_commit = OFF")
                .execute(&self.pool)
                .await
                .map_err(|e| anyhow::anyhow!("Failed to set synchronous_commit OFF: {e}"))?;

            // Increase work_mem for bulk operations
            sqlx::query("SET work_mem = '256MB'")
                .execute(&self.pool)
                .await
                .map_err(|e| anyhow::anyhow!("Failed to set work_mem: {e}"))?;

            log::debug!("Performance settings enabled: synchronous_commit=OFF, work_mem=256MB");
        } else {
            log::debug!("Restoring default safe database performance settings");

            // Restore synchronous_commit to ON for data safety
            sqlx::query("SET synchronous_commit = ON")
                .execute(&self.pool)
                .await
                .map_err(|e| anyhow::anyhow!("Failed to set synchronous_commit ON: {e}"))?;

            // Reset work_mem to default
            sqlx::query("RESET work_mem")
                .execute(&self.pool)
                .await
                .map_err(|e| anyhow::anyhow!("Failed to reset work_mem: {e}"))?;
        }

        Ok(())
    }

    /// Saves the checkpoint block number indicating the last completed pool synchronization for a specific DEX.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn update_dex_last_synced_block(
        &self,
        chain_id: u32,
        dex: &DexType,
        block_number: u64,
    ) -> anyhow::Result<()> {
        sqlx::query(
            "
            UPDATE dex
            SET last_full_sync_pools_block_number = $3
            WHERE chain_id = $1 AND name = $2
            ",
        )
        .bind(chain_id as i32)
        .bind(dex.to_string())
        .bind(block_number as i64)
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to update dex last synced block: {e}"))
    }

    /// Updates the last synced block number for a pool.
    ///
    /// # Errors
    ///
    /// Returns an error if the database update fails.
    pub async fn update_pool_last_synced_block(
        &self,
        chain_id: u32,
        dex: &DexType,
        pool_identifier: &PoolIdentifier,
        block_number: u64,
    ) -> anyhow::Result<()> {
        sqlx::query(
            "
            UPDATE pool
            SET last_full_sync_block_number = $4
            WHERE chain_id = $1
            AND dex_name = $2
            AND pool_identifier = $3
            ",
        )
        .bind(chain_id as i32)
        .bind(dex.to_string())
        .bind(pool_identifier.as_ref())
        .bind(block_number as i64)
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to update pool last synced block: {e}"))
    }

    /// Retrieves the saved checkpoint block number from the last completed pool synchronization for a specific DEX.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    pub async fn get_dex_last_synced_block(
        &self,
        chain_id: u32,
        dex: &DexType,
    ) -> anyhow::Result<Option<u64>> {
        let result = sqlx::query_as::<_, (Option<i64>,)>(
            "
            SELECT
                last_full_sync_pools_block_number
            FROM dex
            WHERE chain_id = $1
            AND name = $2
            ",
        )
        .bind(chain_id as i32)
        .bind(dex.to_string())
        .fetch_optional(&self.pool)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to get dex last synced block: {e}"))?;

        Ok(result.and_then(|(block_number,)| block_number.map(|b| b as u64)))
    }

    /// Retrieves the last synced block number for a pool.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    pub async fn get_pool_last_synced_block(
        &self,
        chain_id: u32,
        dex: &DexType,
        pool_identifier: &PoolIdentifier,
    ) -> anyhow::Result<Option<u64>> {
        let result = sqlx::query_as::<_, (Option<i64>,)>(
            "
            SELECT
                last_full_sync_block_number
            FROM pool
            WHERE chain_id = $1
            AND dex_name = $2
            AND pool_identifier = $3
            ",
        )
        .bind(chain_id as i32)
        .bind(dex.to_string())
        .bind(pool_identifier.as_ref())
        .fetch_optional(&self.pool)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to get pool last synced block: {e}"))?;

        Ok(result.and_then(|(block_number,)| block_number.map(|b| b as u64)))
    }

    pub(crate) async fn get_pool_event_sync_state(
        &self,
        chain_id: u32,
        dex: &DexType,
        pool_identifier: &PoolIdentifier,
    ) -> anyhow::Result<PoolEventSyncState> {
        let pool_state = sqlx::query_as::<_, (i32, Option<i64>)>(
            "
            SELECT event_sync_version, last_full_sync_block_number
            FROM pool
            WHERE chain_id = $1
            AND dex_name = $2
            AND pool_identifier = $3
            ",
        )
        .bind(chain_id as i32)
        .bind(dex.to_string())
        .bind(pool_identifier.as_ref())
        .fetch_optional(&self.pool)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to get pool event sync state: {e}"))?;

        let Some((version, last_full_sync_block)) = pool_state else {
            return Ok(PoolEventSyncState::default());
        };

        let family_blocks = sqlx::query_as::<_, (String, i64)>(
            "
            SELECT event_family, last_full_sync_block_number
            FROM pool_event_sync
            WHERE chain_id = $1
            AND dex_name = $2
            AND pool_identifier = $3
            ORDER BY event_family
            ",
        )
        .bind(chain_id as i32)
        .bind(dex.to_string())
        .bind(pool_identifier.as_ref())
        .fetch_all(&self.pool)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to get pool event-family checkpoints: {e}"))?
        .into_iter()
        .map(|(family, block)| (family, block as u64))
        .collect();

        Ok(PoolEventSyncState {
            version: version as u32,
            last_full_sync_block: last_full_sync_block.map(|block| block as u64),
            family_blocks,
        })
    }

    pub(crate) async fn update_pool_event_sync(
        &self,
        chain_id: u32,
        dex: &DexType,
        pool_identifier: &PoolIdentifier,
        event_families: &[&str],
        block_number: u64,
        version: Option<u32>,
    ) -> anyhow::Result<()> {
        let mut transaction = self.pool.begin().await?;

        for event_family in event_families {
            sqlx::query(
                "
                INSERT INTO pool_event_sync (
                    chain_id, dex_name, pool_identifier, event_family,
                    last_full_sync_block_number
                )
                VALUES ($1, $2, $3, $4, $5)
                ON CONFLICT (chain_id, dex_name, pool_identifier, event_family)
                DO UPDATE SET last_full_sync_block_number =
                    GREATEST(
                        pool_event_sync.last_full_sync_block_number,
                        EXCLUDED.last_full_sync_block_number
                    )
                ",
            )
            .bind(chain_id as i32)
            .bind(dex.to_string())
            .bind(pool_identifier.as_ref())
            .bind(event_family)
            .bind(block_number as i64)
            .execute(&mut *transaction)
            .await
            .map_err(|e| {
                anyhow::anyhow!("Failed to update {event_family} pool event-family checkpoint: {e}")
            })?;
        }

        if let Some(version) = version {
            sqlx::query(
                "
                UPDATE pool
                SET
                    event_sync_version = GREATEST(event_sync_version, $4),
                    last_full_sync_block_number =
                        GREATEST(COALESCE(last_full_sync_block_number, $5), $5)
                WHERE chain_id = $1
                AND dex_name = $2
                AND pool_identifier = $3
                ",
            )
            .bind(chain_id as i32)
            .bind(dex.to_string())
            .bind(pool_identifier.as_ref())
            .bind(version as i32)
            .bind(block_number as i64)
            .execute(&mut *transaction)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to finalize pool event sync progress: {e}"))?;
        }

        transaction.commit().await?;
        Ok(())
    }

    /// Retrieves the maximum block number from a specific table for a given pool.
    /// This is useful to detect orphaned data where events were inserted but progress wasn't updated.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    pub async fn get_table_last_block(
        &self,
        chain_id: u32,
        table_name: &str,
        pool_identifier: &PoolIdentifier,
    ) -> anyhow::Result<Option<u64>> {
        let query = format!(
            "SELECT MAX(block) FROM {table_name} WHERE chain_id = $1 AND pool_identifier = $2"
        );
        let result = sqlx::query_as::<_, (Option<i64>,)>(AssertSqlSafe(query))
            .bind(chain_id as i32)
            .bind(pool_identifier.as_ref())
            .fetch_optional(&self.pool)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to get table last block for {table_name}: {e}"))?;

        Ok(result.and_then(|(block_number,)| block_number.map(|b| b as u64)))
    }

    /// Adds a batch of pool fee collect events to the database using batch operations.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn add_pool_collects_batch(
        &self,
        chain_id: u32,
        collects: &[PoolFeeCollect],
    ) -> anyhow::Result<()> {
        if collects.is_empty() {
            return Ok(());
        }

        // Prepare vectors for each column
        let len = collects.len();
        let mut chain_ids: Vec<i32> = Vec::with_capacity(len);
        let mut dex_names: Vec<String> = Vec::with_capacity(len);
        let mut pool_identifiers: Vec<String> = Vec::with_capacity(len);
        let mut blocks: Vec<i64> = Vec::with_capacity(len);
        let mut transaction_hashes: Vec<String> = Vec::with_capacity(len);
        let mut transaction_indices: Vec<i32> = Vec::with_capacity(len);
        let mut log_indices: Vec<i32> = Vec::with_capacity(len);
        let mut owners: Vec<String> = Vec::with_capacity(len);
        let mut amount0s: Vec<String> = Vec::with_capacity(len);
        let mut amount1s: Vec<String> = Vec::with_capacity(len);
        let mut tick_lowers: Vec<i32> = Vec::with_capacity(len);
        let mut tick_uppers: Vec<i32> = Vec::with_capacity(len);

        // Fill vectors from collects
        for collect in collects {
            chain_ids.push(chain_id as i32);
            dex_names.push(collect.dex.name.to_string());
            pool_identifiers.push(collect.pool_identifier.to_string());
            blocks.push(collect.block as i64);
            transaction_hashes.push(collect.transaction_hash.clone());
            transaction_indices.push(collect.transaction_index as i32);
            log_indices.push(collect.log_index as i32);
            owners.push(collect.owner.to_string());
            amount0s.push(collect.amount0.to_string());
            amount1s.push(collect.amount1.to_string());
            tick_lowers.push(collect.tick_lower);
            tick_uppers.push(collect.tick_upper);
        }

        // Execute batch insert with UNNEST
        sqlx::query(
            "
            INSERT INTO pool_collect_event (
                chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index,
                log_index, owner, amount0, amount1, tick_lower, tick_upper
            )
            SELECT
                chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index,
                log_index, owner, amount0::U256, amount1::U256, tick_lower, tick_upper
            FROM UNNEST(
                $1::INT[], $2::TEXT[], $3::TEXT[], $4::INT[], $5::TEXT[], $6::INT[],
                $7::INT[], $8::TEXT[], $9::TEXT[], $10::TEXT[], $11::INT[], $12::INT[]
            ) AS t(chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index,
                   log_index, owner, amount0, amount1, tick_lower, tick_upper)
            ON CONFLICT (chain_id, transaction_hash, log_index) DO NOTHING
           ",
        )
        .bind(&chain_ids[..])
        .bind(&dex_names[..])
        .bind(&pool_identifiers[..])
        .bind(&blocks[..])
        .bind(&transaction_hashes[..])
        .bind(&transaction_indices[..])
        .bind(&log_indices[..])
        .bind(&owners[..])
        .bind(&amount0s[..])
        .bind(&amount1s[..])
        .bind(&tick_lowers[..])
        .bind(&tick_uppers[..])
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to batch insert into pool_fee_collect table: {e}"))
    }

    /// Inserts multiple pool flash events in a single database operation using UNNEST for optimal performance.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn add_pool_flash_batch(
        &self,
        chain_id: u32,
        flash_events: &[PoolFlash],
    ) -> anyhow::Result<()> {
        if flash_events.is_empty() {
            return Ok(());
        }

        // Prepare vectors for each column
        let len = flash_events.len();
        let mut chain_ids: Vec<i32> = Vec::with_capacity(len);
        let mut dex_names: Vec<String> = Vec::with_capacity(len);
        let mut pool_identifiers: Vec<String> = Vec::with_capacity(len);
        let mut blocks: Vec<i64> = Vec::with_capacity(len);
        let mut transaction_hashes: Vec<String> = Vec::with_capacity(len);
        let mut transaction_indices: Vec<i32> = Vec::with_capacity(len);
        let mut log_indices: Vec<i32> = Vec::with_capacity(len);
        let mut senders: Vec<String> = Vec::with_capacity(len);
        let mut recipients: Vec<String> = Vec::with_capacity(len);
        let mut amount0s: Vec<String> = Vec::with_capacity(len);
        let mut amount1s: Vec<String> = Vec::with_capacity(len);
        let mut paid0s: Vec<String> = Vec::with_capacity(len);
        let mut paid1s: Vec<String> = Vec::with_capacity(len);

        // Fill vectors from flash events
        for flash in flash_events {
            chain_ids.push(chain_id as i32);
            dex_names.push(flash.dex.name.to_string());
            pool_identifiers.push(flash.pool_identifier.to_string());
            blocks.push(flash.block as i64);
            transaction_hashes.push(flash.transaction_hash.clone());
            transaction_indices.push(flash.transaction_index as i32);
            log_indices.push(flash.log_index as i32);
            senders.push(flash.sender.to_string());
            recipients.push(flash.recipient.to_string());
            amount0s.push(flash.amount0.to_string());
            amount1s.push(flash.amount1.to_string());
            paid0s.push(flash.paid0.to_string());
            paid1s.push(flash.paid1.to_string());
        }

        // Execute batch insert with UNNEST
        sqlx::query(
            "
            INSERT INTO pool_flash_event (
                chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index,
                log_index, sender, recipient, amount0, amount1, paid0, paid1
            )
            SELECT
                chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index,
                log_index, sender, recipient, amount0::U256, amount1::U256, paid0::U256, paid1::U256
            FROM UNNEST(
                $1::INT[], $2::TEXT[], $3::TEXT[], $4::INT[], $5::TEXT[], $6::INT[],
                $7::INT[], $8::TEXT[], $9::TEXT[], $10::TEXT[], $11::TEXT[], $12::TEXT[], $13::TEXT[]
            ) AS t(chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index,
                   log_index, sender, recipient, amount0, amount1, paid0, paid1)
            ON CONFLICT (chain_id, transaction_hash, log_index) DO NOTHING
           ",
        )
        .bind(&chain_ids[..])
        .bind(&dex_names[..])
        .bind(&pool_identifiers[..])
        .bind(&blocks[..])
        .bind(&transaction_hashes[..])
        .bind(&transaction_indices[..])
        .bind(&log_indices[..])
        .bind(&senders[..])
        .bind(&recipients[..])
        .bind(&amount0s[..])
        .bind(&amount1s[..])
        .bind(&paid0s[..])
        .bind(&paid1s[..])
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to batch insert into pool_flash_event table: {e}"))
    }

    /// Inserts multiple pool fee-protocol update events in a single database operation using UNNEST.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn add_pool_fee_protocol_updates_batch(
        &self,
        chain_id: u32,
        updates: &[PoolFeeProtocolUpdate],
    ) -> anyhow::Result<()> {
        if updates.is_empty() {
            return Ok(());
        }

        // Prepare vectors for each column
        let len = updates.len();
        let mut chain_ids: Vec<i32> = Vec::with_capacity(len);
        let mut dex_names: Vec<String> = Vec::with_capacity(len);
        let mut pool_identifiers: Vec<String> = Vec::with_capacity(len);
        let mut blocks: Vec<i64> = Vec::with_capacity(len);
        let mut transaction_hashes: Vec<String> = Vec::with_capacity(len);
        let mut transaction_indices: Vec<i32> = Vec::with_capacity(len);
        let mut log_indices: Vec<i32> = Vec::with_capacity(len);
        let mut fee_protocol0s: Vec<i32> = Vec::with_capacity(len);
        let mut fee_protocol1s: Vec<i32> = Vec::with_capacity(len);

        // Fill vectors from updates
        for update in updates {
            chain_ids.push(chain_id as i32);
            dex_names.push(update.dex.name.to_string());
            pool_identifiers.push(update.pool_identifier.to_string());
            blocks.push(update.block as i64);
            transaction_hashes.push(update.transaction_hash.clone());
            transaction_indices.push(update.transaction_index as i32);
            log_indices.push(update.log_index as i32);
            fee_protocol0s.push(i32::try_from(update.fee_protocol0_new).map_err(|e| {
                anyhow::anyhow!(
                    "Invalid fee_protocol0_new '{}': {e}",
                    update.fee_protocol0_new
                )
            })?);
            fee_protocol1s.push(i32::try_from(update.fee_protocol1_new).map_err(|e| {
                anyhow::anyhow!(
                    "Invalid fee_protocol1_new '{}': {e}",
                    update.fee_protocol1_new
                )
            })?);
        }

        // Execute batch insert with UNNEST
        sqlx::query(
            "
            INSERT INTO pool_fee_protocol_update_event (
                chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index,
                log_index, fee_protocol0_new, fee_protocol1_new
            )
            SELECT
                chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index,
                log_index, fee_protocol0_new, fee_protocol1_new
            FROM UNNEST(
                $1::INT[], $2::TEXT[], $3::TEXT[], $4::INT[], $5::TEXT[], $6::INT[],
                $7::INT[], $8::INT[], $9::INT[]
            ) AS t(chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index,
                   log_index, fee_protocol0_new, fee_protocol1_new)
            ON CONFLICT (chain_id, transaction_hash, log_index) DO NOTHING
           ",
        )
        .bind(&chain_ids[..])
        .bind(&dex_names[..])
        .bind(&pool_identifiers[..])
        .bind(&blocks[..])
        .bind(&transaction_hashes[..])
        .bind(&transaction_indices[..])
        .bind(&log_indices[..])
        .bind(&fee_protocol0s[..])
        .bind(&fee_protocol1s[..])
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| {
            anyhow::anyhow!("Failed to batch insert into pool_fee_protocol_update_event table: {e}")
        })
    }

    /// Inserts multiple pool protocol-fee withdrawal events in a single database operation using UNNEST.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn add_pool_fee_protocol_collect_batch(
        &self,
        chain_id: u32,
        collects: &[PoolFeeProtocolCollect],
    ) -> anyhow::Result<()> {
        if collects.is_empty() {
            return Ok(());
        }

        // Prepare vectors for each column
        let len = collects.len();
        let mut chain_ids: Vec<i32> = Vec::with_capacity(len);
        let mut dex_names: Vec<String> = Vec::with_capacity(len);
        let mut pool_identifiers: Vec<String> = Vec::with_capacity(len);
        let mut blocks: Vec<i64> = Vec::with_capacity(len);
        let mut transaction_hashes: Vec<String> = Vec::with_capacity(len);
        let mut transaction_indices: Vec<i32> = Vec::with_capacity(len);
        let mut log_indices: Vec<i32> = Vec::with_capacity(len);
        let mut senders: Vec<String> = Vec::with_capacity(len);
        let mut recipients: Vec<String> = Vec::with_capacity(len);
        let mut amount0s: Vec<String> = Vec::with_capacity(len);
        let mut amount1s: Vec<String> = Vec::with_capacity(len);

        // Fill vectors from collects
        for collect in collects {
            chain_ids.push(chain_id as i32);
            dex_names.push(collect.dex.name.to_string());
            pool_identifiers.push(collect.pool_identifier.to_string());
            blocks.push(collect.block as i64);
            transaction_hashes.push(collect.transaction_hash.clone());
            transaction_indices.push(collect.transaction_index as i32);
            log_indices.push(collect.log_index as i32);
            senders.push(collect.sender.to_string());
            recipients.push(collect.recipient.to_string());
            amount0s.push(collect.amount0.to_string());
            amount1s.push(collect.amount1.to_string());
        }

        // Execute batch insert with UNNEST
        sqlx::query(
            "
            INSERT INTO pool_fee_protocol_collect_event (
                chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index,
                log_index, sender, recipient, amount0, amount1
            )
            SELECT
                chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index,
                log_index, sender, recipient, amount0::U256, amount1::U256
            FROM UNNEST(
                $1::INT[], $2::TEXT[], $3::TEXT[], $4::INT[], $5::TEXT[], $6::INT[],
                $7::INT[], $8::TEXT[], $9::TEXT[], $10::TEXT[], $11::TEXT[]
            ) AS t(chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index,
                   log_index, sender, recipient, amount0, amount1)
            ON CONFLICT (chain_id, transaction_hash, log_index) DO NOTHING
           ",
        )
        .bind(&chain_ids[..])
        .bind(&dex_names[..])
        .bind(&pool_identifiers[..])
        .bind(&blocks[..])
        .bind(&transaction_hashes[..])
        .bind(&transaction_indices[..])
        .bind(&log_indices[..])
        .bind(&senders[..])
        .bind(&recipients[..])
        .bind(&amount0s[..])
        .bind(&amount1s[..])
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| {
            anyhow::anyhow!(
                "Failed to batch insert into pool_fee_protocol_collect_event table: {e}"
            )
        })
    }

    /// Adds a pool snapshot to the database.
    ///
    /// # Errors
    ///
    /// Returns an error if the database insert fails.
    pub async fn add_pool_snapshot(
        &self,
        chain_id: u32,
        dex_name: &DexType,
        pool_identifier: &PoolIdentifier,
        snapshot: &PoolSnapshot,
    ) -> anyhow::Result<()> {
        sqlx::query(
            "
            INSERT INTO pool_snapshot (
                chain_id, dex_name, pool_identifier, block, transaction_index, log_index, transaction_hash,
                current_tick, price_sqrt_ratio_x96, liquidity,
                protocol_fees_token0, protocol_fees_token1, fee_protocol,
                fee_protocol0_basis_points, fee_protocol1_basis_points,
                fee_growth_global_0, fee_growth_global_1,
                total_amount0_deposited, total_amount1_deposited,
                total_amount0_collected, total_amount1_collected,
                total_swaps, total_mints, total_burns, total_fee_collects, total_flashes,
                liquidity_utilization_rate
            ) VALUES (
                $1, $2, $3, $4, $5, $6, $7,
                $8, $9::U160, $10::U128, $11::U256, $12::U256, $13, $14, $15,
                $16::U256, $17::U256, $18::U256, $19::U256, $20::U256, $21::U256,
                $22, $23, $24, $25, $26, $27
            )
            ON CONFLICT (chain_id, pool_identifier, block, transaction_index, log_index)
            DO UPDATE SET
                dex_name = EXCLUDED.dex_name,
                transaction_hash = EXCLUDED.transaction_hash,
                current_tick = EXCLUDED.current_tick,
                price_sqrt_ratio_x96 = EXCLUDED.price_sqrt_ratio_x96,
                liquidity = EXCLUDED.liquidity,
                protocol_fees_token0 = EXCLUDED.protocol_fees_token0,
                protocol_fees_token1 = EXCLUDED.protocol_fees_token1,
                fee_protocol = EXCLUDED.fee_protocol,
                fee_protocol0_basis_points = EXCLUDED.fee_protocol0_basis_points,
                fee_protocol1_basis_points = EXCLUDED.fee_protocol1_basis_points,
                fee_growth_global_0 = EXCLUDED.fee_growth_global_0,
                fee_growth_global_1 = EXCLUDED.fee_growth_global_1,
                total_amount0_deposited = EXCLUDED.total_amount0_deposited,
                total_amount1_deposited = EXCLUDED.total_amount1_deposited,
                total_amount0_collected = EXCLUDED.total_amount0_collected,
                total_amount1_collected = EXCLUDED.total_amount1_collected,
                total_swaps = EXCLUDED.total_swaps,
                total_mints = EXCLUDED.total_mints,
                total_burns = EXCLUDED.total_burns,
                total_fee_collects = EXCLUDED.total_fee_collects,
                total_flashes = EXCLUDED.total_flashes,
                liquidity_utilization_rate = EXCLUDED.liquidity_utilization_rate
            ",
        )
        .bind(chain_id as i32)
        .bind(dex_name.to_string())
        .bind(pool_identifier.as_ref())
        .bind(snapshot.block_position.number as i64)
        .bind(snapshot.block_position.transaction_index as i32)
        .bind(snapshot.block_position.log_index as i32)
        .bind(snapshot.block_position.transaction_hash.clone())
        .bind(snapshot.state.current_tick)
        .bind(snapshot.state.price_sqrt_ratio_x96.to_string())
        .bind(snapshot.state.liquidity.to_string())
        .bind(snapshot.state.protocol_fees_token0.to_string())
        .bind(snapshot.state.protocol_fees_token1.to_string())
        .bind(snapshot.state.fee_protocol as i16)
        .bind(snapshot.state.fee_protocol0_basis_points.map(|v| v as i32))
        .bind(snapshot.state.fee_protocol1_basis_points.map(|v| v as i32))
        .bind(snapshot.state.fee_growth_global_0.to_string())
        .bind(snapshot.state.fee_growth_global_1.to_string())
        .bind(snapshot.analytics.total_amount0_deposited.to_string())
        .bind(snapshot.analytics.total_amount1_deposited.to_string())
        .bind(snapshot.analytics.total_amount0_collected.to_string())
        .bind(snapshot.analytics.total_amount1_collected.to_string())
        .bind(snapshot.analytics.total_swaps as i32)
        .bind(snapshot.analytics.total_mints as i32)
        .bind(snapshot.analytics.total_burns as i32)
        .bind(snapshot.analytics.total_fee_collects as i32)
        .bind(snapshot.analytics.total_flashes as i32)
        .bind(snapshot.analytics.liquidity_utilization_rate)
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to insert into pool_snapshot table: {e}"))
    }

    /// Inserts multiple pool positions in a single database operation using UNNEST for optimal performance.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn add_pool_positions_batch(
        &self,
        chain_id: u32,
        snapshot_block: u64,
        snapshot_transaction_index: u32,
        snapshot_log_index: u32,
        positions: &[(PoolIdentifier, PoolPosition)],
    ) -> anyhow::Result<()> {
        if positions.is_empty() {
            return Ok(());
        }

        // Prepare vectors for each column
        let len = positions.len();
        let mut pool_identifiers: Vec<String> = Vec::with_capacity(len);
        let mut owners: Vec<String> = Vec::with_capacity(len);
        let mut tick_lowers: Vec<i32> = Vec::with_capacity(len);
        let mut tick_uppers: Vec<i32> = Vec::with_capacity(len);
        let mut liquidities: Vec<String> = Vec::with_capacity(len);
        let mut fee_growth_inside_0_lasts: Vec<String> = Vec::with_capacity(len);
        let mut fee_growth_inside_1_lasts: Vec<String> = Vec::with_capacity(len);
        let mut tokens_owed_0s: Vec<String> = Vec::with_capacity(len);
        let mut tokens_owed_1s: Vec<String> = Vec::with_capacity(len);
        let mut total_amount0_depositeds: Vec<Option<String>> = Vec::with_capacity(len);
        let mut total_amount1_depositeds: Vec<Option<String>> = Vec::with_capacity(len);
        let mut total_amount0_collecteds: Vec<Option<String>> = Vec::with_capacity(len);
        let mut total_amount1_collecteds: Vec<Option<String>> = Vec::with_capacity(len);

        // Fill vectors from positions
        for (pool_address, position) in positions {
            pool_identifiers.push(pool_address.to_string());
            owners.push(position.owner.to_string());
            tick_lowers.push(position.tick_lower);
            tick_uppers.push(position.tick_upper);
            liquidities.push(position.liquidity.to_string());
            fee_growth_inside_0_lasts.push(position.fee_growth_inside_0_last.to_string());
            fee_growth_inside_1_lasts.push(position.fee_growth_inside_1_last.to_string());
            tokens_owed_0s.push(position.tokens_owed_0.to_string());
            tokens_owed_1s.push(position.tokens_owed_1.to_string());
            total_amount0_depositeds.push(Some(position.total_amount0_deposited.to_string()));
            total_amount1_depositeds.push(Some(position.total_amount1_deposited.to_string()));
            total_amount0_collecteds.push(Some(position.total_amount0_collected.to_string()));
            total_amount1_collecteds.push(Some(position.total_amount1_collected.to_string()));
        }

        // Execute batch insert with UNNEST
        sqlx::query(
            "
            INSERT INTO pool_position (
                chain_id, pool_identifier, snapshot_block, snapshot_transaction_index, snapshot_log_index,
                owner, tick_lower, tick_upper,
                liquidity, fee_growth_inside_0_last, fee_growth_inside_1_last,
                tokens_owed_0, tokens_owed_1,
                total_amount0_deposited, total_amount1_deposited,
                total_amount0_collected, total_amount1_collected
            )
            SELECT
                $1, pool_identifier, $2, $3, $4,
                owner, tick_lower, tick_upper,
                liquidity::U128, fee_growth_inside_0_last::U256, fee_growth_inside_1_last::U256,
                tokens_owed_0::U128, tokens_owed_1::U128,
                total_amount0_deposited::U256, total_amount1_deposited::U256,
                total_amount0_collected::U128, total_amount1_collected::U128
            FROM UNNEST(
                $5::TEXT[], $6::TEXT[], $7::INT[], $8::INT[], $9::TEXT[], $10::TEXT[],
                $11::TEXT[], $12::TEXT[], $13::TEXT[], $14::TEXT[], $15::TEXT[],
                $16::TEXT[], $17::TEXT[]
            ) AS t(pool_identifier, owner, tick_lower, tick_upper,
                   liquidity, fee_growth_inside_0_last, fee_growth_inside_1_last,
                   tokens_owed_0, tokens_owed_1,
                   total_amount0_deposited, total_amount1_deposited,
                   total_amount0_collected, total_amount1_collected)
            ON CONFLICT (chain_id, pool_identifier, snapshot_block, snapshot_transaction_index, snapshot_log_index, owner, tick_lower, tick_upper)
            DO UPDATE SET
                liquidity = EXCLUDED.liquidity,
                fee_growth_inside_0_last = EXCLUDED.fee_growth_inside_0_last,
                fee_growth_inside_1_last = EXCLUDED.fee_growth_inside_1_last,
                tokens_owed_0 = EXCLUDED.tokens_owed_0,
                tokens_owed_1 = EXCLUDED.tokens_owed_1,
                total_amount0_deposited = EXCLUDED.total_amount0_deposited,
                total_amount1_deposited = EXCLUDED.total_amount1_deposited,
                total_amount0_collected = EXCLUDED.total_amount0_collected,
                total_amount1_collected = EXCLUDED.total_amount1_collected
           ",
        )
        .bind(chain_id as i32)
        .bind(snapshot_block as i64)
        .bind(snapshot_transaction_index as i32)
        .bind(snapshot_log_index as i32)
        .bind(&pool_identifiers[..])
        .bind(&owners[..])
        .bind(&tick_lowers[..])
        .bind(&tick_uppers[..])
        .bind(&liquidities[..])
        .bind(&fee_growth_inside_0_lasts[..])
        .bind(&fee_growth_inside_1_lasts[..])
        .bind(&tokens_owed_0s[..])
        .bind(&tokens_owed_1s[..])
        .bind(&total_amount0_depositeds as &[Option<String>])
        .bind(&total_amount1_depositeds as &[Option<String>])
        .bind(&total_amount0_collecteds as &[Option<String>])
        .bind(&total_amount1_collecteds as &[Option<String>])
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to batch insert into pool_position table: {e}"))
    }

    /// Inserts multiple pool ticks in a single database operation using UNNEST for optimal performance.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn add_pool_ticks_batch(
        &self,
        chain_id: u32,
        snapshot_block: u64,
        snapshot_transaction_index: u32,
        snapshot_log_index: u32,
        ticks: &[(PoolIdentifier, &PoolTick)],
    ) -> anyhow::Result<()> {
        if ticks.is_empty() {
            return Ok(());
        }

        // Prepare vectors for each column
        let len = ticks.len();
        let mut pool_identifiers: Vec<String> = Vec::with_capacity(len);
        let mut tick_values: Vec<i32> = Vec::with_capacity(len);
        let mut liquidity_grosses: Vec<String> = Vec::with_capacity(len);
        let mut liquidity_nets: Vec<String> = Vec::with_capacity(len);
        let mut fee_growth_outside_0s: Vec<String> = Vec::with_capacity(len);
        let mut fee_growth_outside_1s: Vec<String> = Vec::with_capacity(len);
        let mut initializeds: Vec<bool> = Vec::with_capacity(len);
        let mut last_updated_blocks: Vec<i64> = Vec::with_capacity(len);

        // Fill vectors from ticks
        for (pool_address, tick) in ticks {
            pool_identifiers.push(pool_address.to_string());
            tick_values.push(tick.value);
            liquidity_grosses.push(tick.liquidity_gross.to_string());
            liquidity_nets.push(tick.liquidity_net.to_string());
            fee_growth_outside_0s.push(tick.fee_growth_outside_0.to_string());
            fee_growth_outside_1s.push(tick.fee_growth_outside_1.to_string());
            initializeds.push(tick.initialized);
            last_updated_blocks.push(tick.last_updated_block as i64);
        }

        // Execute batch insert with UNNEST
        sqlx::query(
            "
            INSERT INTO pool_tick (
                chain_id, pool_identifier, snapshot_block, snapshot_transaction_index, snapshot_log_index,
                tick_value, liquidity_gross, liquidity_net,
                fee_growth_outside_0, fee_growth_outside_1, initialized, last_updated_block
            )
            SELECT
                $1, pool_identifier, $2, $3, $4,
                tick_value, liquidity_gross::U128, liquidity_net::I128,
                fee_growth_outside_0::U256, fee_growth_outside_1::U256, initialized, last_updated_block
            FROM UNNEST(
                $5::TEXT[], $6::INT[], $7::TEXT[], $8::TEXT[], $9::TEXT[],
                $10::TEXT[], $11::BOOLEAN[], $12::BIGINT[]
            ) AS t(pool_identifier, tick_value, liquidity_gross, liquidity_net,
                   fee_growth_outside_0, fee_growth_outside_1, initialized, last_updated_block)
            ON CONFLICT (chain_id, pool_identifier, snapshot_block, snapshot_transaction_index, snapshot_log_index, tick_value)
            DO UPDATE SET
                liquidity_gross = EXCLUDED.liquidity_gross,
                liquidity_net = EXCLUDED.liquidity_net,
                fee_growth_outside_0 = EXCLUDED.fee_growth_outside_0,
                fee_growth_outside_1 = EXCLUDED.fee_growth_outside_1,
                initialized = EXCLUDED.initialized,
                last_updated_block = EXCLUDED.last_updated_block
           ",
        )
        .bind(chain_id as i32)
        .bind(snapshot_block as i64)
        .bind(snapshot_transaction_index as i32)
        .bind(snapshot_log_index as i32)
        .bind(&pool_identifiers[..])
        .bind(&tick_values[..])
        .bind(&liquidity_grosses[..])
        .bind(&liquidity_nets[..])
        .bind(&fee_growth_outside_0s[..])
        .bind(&fee_growth_outside_1s[..])
        .bind(&initializeds[..])
        .bind(&last_updated_blocks[..])
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to batch insert into pool_tick table: {e}"))
    }

    /// Updates the initial price and tick for a pool.
    ///
    /// # Errors
    ///
    /// Returns an error if the database update fails.
    pub async fn update_pool_initial_price_tick(
        &self,
        chain_id: u32,
        initialize_event: &InitializeEvent,
    ) -> anyhow::Result<()> {
        sqlx::query(
            "
            UPDATE pool
            SET
                initial_tick = $4,
                initial_sqrt_price_x96 = $5
            WHERE chain_id = $1
            AND dex_name = $2
            AND pool_identifier = $3
            ",
        )
        .bind(chain_id as i32)
        .bind(initialize_event.dex.name.to_string())
        .bind(initialize_event.pool_identifier.as_ref())
        .bind(initialize_event.tick)
        .bind(initialize_event.sqrt_price_x96.to_string())
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to update pool initial price and tick: {e}"))
    }

    /// Loads the latest usable pool snapshot from the database.
    ///
    /// Returns the most recent snapshot usable as a replay start point: on-chain validated or
    /// replay-derived. Snapshots that failed on-chain validation are excluded.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    pub async fn load_latest_valid_pool_snapshot(
        &self,
        chain_id: u32,
        pool_identifier: &PoolIdentifier,
    ) -> anyhow::Result<Option<PoolSnapshot>> {
        self.load_latest_pool_snapshot(chain_id, pool_identifier, None, true)
            .await
    }

    /// Loads the latest pool snapshot from the database, optionally bounded by block.
    ///
    /// When `max_block` is `Some`, only snapshots at or before that block are considered, so a
    /// backtest can restore pool state as of a replay start. When `require_valid` is `true`,
    /// snapshots that failed on-chain validation are excluded (both on-chain validated and
    /// replay-derived snapshots are returned).
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    pub async fn load_latest_pool_snapshot(
        &self,
        chain_id: u32,
        pool_identifier: &PoolIdentifier,
        max_block: Option<u64>,
        require_valid: bool,
    ) -> anyhow::Result<Option<PoolSnapshot>> {
        let allow_invalid = !require_valid;
        let result = sqlx::query(
            "
            SELECT
                block, transaction_index, log_index, transaction_hash,
                COALESCE(
                    (SELECT hash FROM pool_event_block WHERE pool_event_block.chain_id = pool_snapshot.chain_id AND pool_event_block.number = pool_snapshot.block),
                    (SELECT hash FROM block WHERE block.chain_id = pool_snapshot.chain_id AND block.number = pool_snapshot.block)
                ) as block_hash,
                current_tick, price_sqrt_ratio_x96::TEXT, liquidity::TEXT,
                protocol_fees_token0::TEXT, protocol_fees_token1::TEXT, fee_protocol,
                fee_protocol0_basis_points, fee_protocol1_basis_points,
                fee_growth_global_0::TEXT, fee_growth_global_1::TEXT,
                total_amount0_deposited::TEXT, total_amount1_deposited::TEXT,
                total_amount0_collected::TEXT, total_amount1_collected::TEXT,
                total_swaps, total_mints, total_burns, total_fee_collects, total_flashes,
                liquidity_utilization_rate,
                (SELECT dex_name FROM pool WHERE chain_id = $1 AND address = $2) as dex_name,
                COALESCE(
                    (SELECT timestamp::TEXT FROM block WHERE block.chain_id = pool_snapshot.chain_id AND block.number = pool_snapshot.block),
                    (SELECT timestamp::TEXT FROM pool_event_block WHERE pool_event_block.chain_id = pool_snapshot.chain_id AND pool_event_block.number = pool_snapshot.block)
                ) as block_timestamp
            FROM pool_snapshot
            WHERE chain_id = $1 AND pool_identifier = $2
                AND ($3::BIGINT IS NULL OR block <= $3)
                AND ($4 OR validation_state <> 'invalid')
            ORDER BY block DESC, transaction_index DESC, log_index DESC
            LIMIT 1
            ",
        )
        .bind(chain_id as i32)
        .bind(pool_identifier.as_ref())
        .bind(max_block.map(|b| b as i64))
        .bind(allow_invalid)
        .fetch_optional(&self.pool)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to load latest valid pool snapshot: {e}"))?;

        if let Some(row) = result {
            // Parse snapshot state
            let block: i64 = row.get("block");
            let transaction_index: i32 = row.get("transaction_index");
            let log_index: i32 = row.get("log_index");
            let transaction_hash: String = row.get("transaction_hash");
            let observed_block_hash = row.try_get::<Option<String>, _>("block_hash")?;
            let block =
                u64::try_from(block).with_context(|| "Pool snapshot block number is negative")?;
            let transaction_index = u32::try_from(transaction_index)
                .with_context(|| "Pool snapshot transaction index is negative")?;
            let log_index =
                u32::try_from(log_index).with_context(|| "Pool snapshot log index is negative")?;
            let block_hash = if transaction_index == BLOCK_SCOPED_SNAPSHOT_INDEX
                && log_index == BLOCK_SCOPED_SNAPSHOT_INDEX
            {
                Some(transaction_hash.clone())
            } else {
                observed_block_hash
            };
            let block_timestamp = row
                .try_get::<Option<String>, _>("block_timestamp")?
                .ok_or_else(|| {
                    anyhow::anyhow!(
                        "Missing block timestamp for pool snapshot {} at block {}",
                        pool_identifier,
                        block
                    )
                })?;
            let timestamp = parse_cached_block_timestamp(&block_timestamp)
                .map_err(|e| anyhow::anyhow!("Invalid block timestamp '{block_timestamp}': {e}"))?;

            let block_position =
                BlockPosition::new(block, transaction_hash, transaction_index, log_index)
                    .with_block_hash(block_hash);

            let fee_protocol_value = row.get::<i16, _>("fee_protocol");
            let fee_protocol = u8::try_from(fee_protocol_value).with_context(|| {
                format!("Invalid pool snapshot fee protocol {fee_protocol_value}")
            })?;
            let fee_protocol0_basis_points = row
                .get::<Option<i32>, _>("fee_protocol0_basis_points")
                .map(|value| {
                    u32::try_from(value).with_context(|| {
                        format!("Invalid token0 fee protocol basis points {value}")
                    })
                })
                .transpose()?;
            let fee_protocol1_basis_points = row
                .get::<Option<i32>, _>("fee_protocol1_basis_points")
                .map(|value| {
                    u32::try_from(value).with_context(|| {
                        format!("Invalid token1 fee protocol basis points {value}")
                    })
                })
                .transpose()?;

            let state = PoolState {
                current_tick: row.get("current_tick"),
                price_sqrt_ratio_x96: row.get::<String, _>("price_sqrt_ratio_x96").parse()?,
                liquidity: row.get::<String, _>("liquidity").parse()?,
                protocol_fees_token0: row.get::<String, _>("protocol_fees_token0").parse()?,
                protocol_fees_token1: row.get::<String, _>("protocol_fees_token1").parse()?,
                fee_protocol,
                fee_protocol0_basis_points,
                fee_protocol1_basis_points,
                fee_growth_global_0: row.get::<String, _>("fee_growth_global_0").parse()?,
                fee_growth_global_1: row.get::<String, _>("fee_growth_global_1").parse()?,
            };

            let analytics = PoolAnalytics {
                total_amount0_deposited: row.get::<String, _>("total_amount0_deposited").parse()?,
                total_amount1_deposited: row.get::<String, _>("total_amount1_deposited").parse()?,
                total_amount0_collected: row.get::<String, _>("total_amount0_collected").parse()?,
                total_amount1_collected: row.get::<String, _>("total_amount1_collected").parse()?,
                total_swaps: u64::try_from(row.get::<i32, _>("total_swaps"))
                    .with_context(|| "Pool snapshot total swaps is negative")?,
                total_mints: u64::try_from(row.get::<i32, _>("total_mints"))
                    .with_context(|| "Pool snapshot total mints is negative")?,
                total_burns: u64::try_from(row.get::<i32, _>("total_burns"))
                    .with_context(|| "Pool snapshot total burns is negative")?,
                total_fee_collects: u64::try_from(row.get::<i32, _>("total_fee_collects"))
                    .with_context(|| "Pool snapshot total fee collects is negative")?,
                total_flashes: u64::try_from(row.get::<i32, _>("total_flashes"))
                    .with_context(|| "Pool snapshot total flashes is negative")?,
                liquidity_utilization_rate: row.get::<f64, _>("liquidity_utilization_rate"),
            };

            // Load positions and ticks
            let positions = self
                .load_pool_positions_for_snapshot(
                    chain_id,
                    pool_identifier,
                    block,
                    transaction_index,
                    log_index,
                )
                .await?;

            let ticks = self
                .load_pool_ticks_for_snapshot(
                    chain_id,
                    pool_identifier,
                    block,
                    transaction_index,
                    log_index,
                )
                .await?;

            let dex_name = row
                .try_get::<Option<String>, _>("dex_name")?
                .ok_or_else(|| {
                    anyhow::anyhow!("Missing dex_name for pool snapshot {}", pool_identifier)
                })?;
            let chain = Chain::from_chain_id(chain_id)
                .ok_or_else(|| anyhow::anyhow!("Unknown chain_id: {chain_id}"))?;

            let dex_type = DexType::from_dex_name(&dex_name)
                .ok_or_else(|| anyhow::anyhow!("Unknown dex_name: {dex_name}"))?;

            let dex_extended = crate::exchanges::get_dex_extended(chain.name, &dex_type)
                .ok_or_else(|| {
                    anyhow::anyhow!("No DEX extended found for {} on {}", dex_name, chain.name)
                })?;

            let instrument_id =
                Pool::create_instrument_id(chain.name, &dex_extended.dex, pool_identifier.as_ref());

            Ok(Some(PoolSnapshot::new(
                instrument_id,
                state,
                positions,
                ticks,
                analytics,
                block_position,
                timestamp, // ts_event
                timestamp, // ts_init (same block timestamp)
            )))
        } else {
            Ok(None)
        }
    }

    /// Sets the validation state of a pool snapshot after a validation attempt.
    ///
    /// `state` is one of `on_chain` (hydrated and matched), `replay` (replay-derived, not checked),
    /// or `invalid` (hydrated and mismatched).
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn set_pool_snapshot_validation_state(
        &self,
        chain_id: u32,
        pool_identifier: &PoolIdentifier,
        block: u64,
        transaction_index: u32,
        log_index: u32,
        state: &str,
    ) -> anyhow::Result<()> {
        sqlx::query(
            "
            UPDATE pool_snapshot
            SET validation_state = $6
            WHERE chain_id = $1
            AND pool_identifier = $2
            AND block = $3
            AND transaction_index = $4
            AND log_index = $5
            ",
        )
        .bind(chain_id as i32)
        .bind(pool_identifier.as_ref())
        .bind(block as i64)
        .bind(transaction_index as i32)
        .bind(log_index as i32)
        .bind(state)
        .execute(&self.pool)
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("Failed to set pool snapshot validation state: {e}"))
    }

    /// Reads the stored `validation_state` for the snapshot at the given watermark.
    ///
    /// Returns `None` when no snapshot row exists at that position. Used to report the persisted
    /// verdict (rather than re-deriving `replay`) when on-chain validation cannot reach the block.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    pub async fn get_pool_snapshot_validation_state(
        &self,
        chain_id: u32,
        pool_identifier: &PoolIdentifier,
        block: u64,
        transaction_index: u32,
        log_index: u32,
    ) -> anyhow::Result<Option<String>> {
        let row = sqlx::query(
            "
            SELECT validation_state
            FROM pool_snapshot
            WHERE chain_id = $1
            AND pool_identifier = $2
            AND block = $3
            AND transaction_index = $4
            AND log_index = $5
            ",
        )
        .bind(chain_id as i32)
        .bind(pool_identifier.as_ref())
        .bind(block as i64)
        .bind(transaction_index as i32)
        .bind(log_index as i32)
        .fetch_optional(&self.pool)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to get pool snapshot validation state: {e}"))?;

        Ok(row.map(|row| row.get::<String, _>("validation_state")))
    }

    /// Loads all positions for a specific snapshot.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    pub async fn load_pool_positions_for_snapshot(
        &self,
        chain_id: u32,
        pool_identifier: &PoolIdentifier,
        snapshot_block: u64,
        snapshot_transaction_index: u32,
        snapshot_log_index: u32,
    ) -> anyhow::Result<Vec<PoolPosition>> {
        let rows = sqlx::query(
            "
            SELECT
                owner, tick_lower, tick_upper,
                liquidity::TEXT, fee_growth_inside_0_last::TEXT, fee_growth_inside_1_last::TEXT,
                tokens_owed_0::TEXT, tokens_owed_1::TEXT,
                total_amount0_deposited::TEXT, total_amount1_deposited::TEXT,
                total_amount0_collected::TEXT, total_amount1_collected::TEXT
            FROM pool_position
            WHERE chain_id = $1
            AND pool_identifier = $2
            AND snapshot_block = $3
            AND snapshot_transaction_index = $4
            AND snapshot_log_index = $5
            ",
        )
        .bind(chain_id as i32)
        .bind(pool_identifier.as_ref())
        .bind(snapshot_block as i64)
        .bind(snapshot_transaction_index as i32)
        .bind(snapshot_log_index as i32)
        .fetch_all(&self.pool)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to load pool positions: {e}"))?;

        rows.iter()
            .map(|row| {
                let owner: String = row.get("owner");
                let position = PoolPosition {
                    owner: validate_address(&owner)?,
                    tick_lower: row.get("tick_lower"),
                    tick_upper: row.get("tick_upper"),
                    liquidity: row.get::<String, _>("liquidity").parse()?,
                    fee_growth_inside_0_last: row
                        .get::<String, _>("fee_growth_inside_0_last")
                        .parse()?,
                    fee_growth_inside_1_last: row
                        .get::<String, _>("fee_growth_inside_1_last")
                        .parse()?,
                    tokens_owed_0: row.get::<String, _>("tokens_owed_0").parse()?,
                    tokens_owed_1: row.get::<String, _>("tokens_owed_1").parse()?,
                    total_amount0_deposited: row
                        .get::<String, _>("total_amount0_deposited")
                        .parse()?,
                    total_amount1_deposited: row
                        .get::<String, _>("total_amount1_deposited")
                        .parse()?,
                    total_amount0_collected: row
                        .get::<String, _>("total_amount0_collected")
                        .parse()?,
                    total_amount1_collected: row
                        .get::<String, _>("total_amount1_collected")
                        .parse()?,
                };
                Ok(position)
            })
            .collect()
    }

    /// Loads all ticks for a specific snapshot.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    pub async fn load_pool_ticks_for_snapshot(
        &self,
        chain_id: u32,
        pool_identifier: &PoolIdentifier,
        snapshot_block: u64,
        snapshot_transaction_index: u32,
        snapshot_log_index: u32,
    ) -> anyhow::Result<Vec<PoolTick>> {
        let rows = sqlx::query(
            "
            SELECT
                tick_value, liquidity_gross::TEXT, liquidity_net::TEXT,
                fee_growth_outside_0::TEXT, fee_growth_outside_1::TEXT, initialized,
                last_updated_block
            FROM pool_tick
            WHERE chain_id = $1
            AND pool_identifier = $2
            AND snapshot_block = $3
            AND snapshot_transaction_index = $4
            AND snapshot_log_index = $5
            ",
        )
        .bind(chain_id as i32)
        .bind(pool_identifier.as_ref())
        .bind(snapshot_block as i64)
        .bind(snapshot_transaction_index as i32)
        .bind(snapshot_log_index as i32)
        .fetch_all(&self.pool)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to load pool ticks: {e}"))?;

        rows.iter()
            .map(|row| {
                let tick = PoolTick::new(
                    row.get("tick_value"),
                    row.get::<String, _>("liquidity_gross").parse()?,
                    row.get::<String, _>("liquidity_net").parse()?,
                    row.get::<String, _>("fee_growth_outside_0").parse()?,
                    row.get::<String, _>("fee_growth_outside_1").parse()?,
                    row.get("initialized"),
                    u64::try_from(row.get::<i64, _>("last_updated_block"))
                        .with_context(|| "Pool tick last updated block is negative")?,
                );
                Ok(tick)
            })
            .collect()
    }

    /// Streams pool events from all event tables (swap, liquidity, collect) for a specific pool.
    ///
    /// Creates a unified stream of pool events from multiple tables, ordering them chronologically
    /// by block number, transaction index, and log index. Optionally resumes from a specific
    /// block position and stops at a maximum block.
    ///
    /// # Returns
    ///
    /// A stream of `DexPoolData` events in chronological order.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails or if event transformation fails.
    pub fn stream_pool_events<'a>(
        &'a self,
        chain: SharedChain,
        dex: SharedDex,
        instrument_id: InstrumentId,
        pool_identifier: PoolIdentifier,
        from_position: Option<BlockPosition>,
        to_block: Option<u64>,
    ) -> Pin<Box<dyn Stream<Item = Result<DexPoolData, anyhow::Error>> + Send + 'a>> {
        const QUERY_ALL: &str = "
            SELECT events.*,
                COALESCE(
                    (SELECT hash FROM pool_event_block WHERE pool_event_block.chain_id = events.chain_id AND pool_event_block.number = events.block),
                    (SELECT hash FROM block WHERE block.chain_id = events.chain_id AND block.number = events.block)
                ) as block_hash
            FROM (
            (SELECT
                'swap' as event_type,
                chain_id,
                pool_identifier,
                block,
                transaction_hash,
                transaction_index,
                log_index,
                COALESCE(
                    (SELECT timestamp::TEXT FROM block WHERE block.chain_id = pool_swap_event.chain_id AND block.number = pool_swap_event.block),
                    (SELECT timestamp::TEXT FROM pool_event_block WHERE pool_event_block.chain_id = pool_swap_event.chain_id AND pool_event_block.number = pool_swap_event.block)
                ) as block_timestamp,
                sender,
                recipient,
                NULL::TEXT as owner,
                sqrt_price_x96::TEXT,
                liquidity::TEXT as swap_liquidity,
                tick as swap_tick,
                amount0::TEXT as swap_amount0,
                amount1::TEXT as swap_amount1,
                NULL::TEXT as position_liquidity,
                NULL::TEXT as amount0,
                NULL::TEXT as amount1,
                NULL::INT as tick_lower,
                NULL::INT as tick_upper,
                NULL::TEXT as liquidity_event_type,
                NULL::TEXT as flash_amount0,
                NULL::TEXT as flash_amount1,
                NULL::TEXT as flash_paid0,
                NULL::TEXT as flash_paid1,
                NULL::INTEGER as fee_protocol0_new,
                NULL::INTEGER as fee_protocol1_new
            FROM pool_swap_event
            WHERE chain_id = $1 AND pool_identifier = $2
            AND ($3::BIGINT IS NULL OR block <= $3))
            UNION ALL
            (SELECT
                'liquidity' as event_type,
                chain_id,
                pool_identifier,
                block,
                transaction_hash,
                transaction_index,
                log_index,
                COALESCE(
                    (SELECT timestamp::TEXT FROM block WHERE block.chain_id = pool_liquidity_event.chain_id AND block.number = pool_liquidity_event.block),
                    (SELECT timestamp::TEXT FROM pool_event_block WHERE pool_event_block.chain_id = pool_liquidity_event.chain_id AND pool_event_block.number = pool_liquidity_event.block)
                ) as block_timestamp,
                sender,
                NULL::TEXT as recipient,
                owner,
                NULL::text as sqrt_price_x96,
                NULL::TEXT as swap_liquidity,
                NULL::INT as swap_tick,
                amount0::TEXT as swap_amount0,
                amount1::TEXT as swap_amount1,
                position_liquidity::TEXT,
                amount0::TEXT,
                amount1::TEXT,
                tick_lower::INT,
                tick_upper::INT,
                event_type as liquidity_event_type,
                NULL::TEXT as flash_amount0,
                NULL::TEXT as flash_amount1,
                NULL::TEXT as flash_paid0,
                NULL::TEXT as flash_paid1,
                NULL::INTEGER as fee_protocol0_new,
                NULL::INTEGER as fee_protocol1_new
            FROM pool_liquidity_event
            WHERE chain_id = $1 AND pool_identifier = $2
            AND ($3::BIGINT IS NULL OR block <= $3))
            UNION ALL
            (SELECT
                'collect' as event_type,
                chain_id,
                pool_identifier,
                block,
                transaction_hash,
                transaction_index,
                log_index,
                COALESCE(
                    (SELECT timestamp::TEXT FROM block WHERE block.chain_id = pool_collect_event.chain_id AND block.number = pool_collect_event.block),
                    (SELECT timestamp::TEXT FROM pool_event_block WHERE pool_event_block.chain_id = pool_collect_event.chain_id AND pool_event_block.number = pool_collect_event.block)
                ) as block_timestamp,
                NULL::TEXT as sender,
                NULL::TEXT as recipient,
                owner,
                NULL::TEXT as sqrt_price_x96,
                NULL::TEXT as swap_liquidity,
                NULL::INT AS swap_tick,
                amount0::TEXT as swap_amount0,
                amount1::TEXT as swap_amount1,
                NULL::TEXT as position_liquidity,
                amount0::TEXT,
                amount1::TEXT,
                tick_lower::INT,
                tick_upper::INT,
                NULL::TEXT as liquidity_event_type,
                NULL::TEXT as flash_amount0,
                NULL::TEXT as flash_amount1,
                NULL::TEXT as flash_paid0,
                NULL::TEXT as flash_paid1,
                NULL::INTEGER as fee_protocol0_new,
                NULL::INTEGER as fee_protocol1_new
            FROM pool_collect_event
            WHERE chain_id = $1 AND pool_identifier = $2
            AND ($3::BIGINT IS NULL OR block <= $3))
            UNION ALL
            (SELECT
                'fee_protocol_update' as event_type,
                chain_id,
                pool_identifier,
                block,
                transaction_hash,
                transaction_index,
                log_index,
                COALESCE(
                    (SELECT timestamp::TEXT FROM block WHERE block.chain_id = pool_fee_protocol_update_event.chain_id AND block.number = pool_fee_protocol_update_event.block),
                    (SELECT timestamp::TEXT FROM pool_event_block WHERE pool_event_block.chain_id = pool_fee_protocol_update_event.chain_id AND pool_event_block.number = pool_fee_protocol_update_event.block)
                ) as block_timestamp,
                NULL::TEXT as sender,
                NULL::TEXT as recipient,
                NULL::TEXT as owner,
                NULL::TEXT as sqrt_price_x96,
                NULL::TEXT as swap_liquidity,
                NULL::INT AS swap_tick,
                NULL::TEXT as swap_amount0,
                NULL::TEXT as swap_amount1,
                NULL::TEXT as position_liquidity,
                NULL::TEXT as amount0,
                NULL::TEXT as amount1,
                NULL::INT as tick_lower,
                NULL::INT as tick_upper,
                NULL::TEXT as liquidity_event_type,
                NULL::TEXT as flash_amount0,
                NULL::TEXT as flash_amount1,
                NULL::TEXT as flash_paid0,
                NULL::TEXT as flash_paid1,
                fee_protocol0_new::INTEGER,
                fee_protocol1_new::INTEGER
            FROM pool_fee_protocol_update_event
            WHERE chain_id = $1 AND pool_identifier = $2
            AND ($3::BIGINT IS NULL OR block <= $3))
            UNION ALL
            (SELECT
                'fee_protocol_collect' as event_type,
                chain_id,
                pool_identifier,
                block,
                transaction_hash,
                transaction_index,
                log_index,
                COALESCE(
                    (SELECT timestamp::TEXT FROM block WHERE block.chain_id = pool_fee_protocol_collect_event.chain_id AND block.number = pool_fee_protocol_collect_event.block),
                    (SELECT timestamp::TEXT FROM pool_event_block WHERE pool_event_block.chain_id = pool_fee_protocol_collect_event.chain_id AND pool_event_block.number = pool_fee_protocol_collect_event.block)
                ) as block_timestamp,
                sender,
                recipient,
                NULL::TEXT as owner,
                NULL::TEXT as sqrt_price_x96,
                NULL::TEXT as swap_liquidity,
                NULL::INT AS swap_tick,
                NULL::TEXT as swap_amount0,
                NULL::TEXT as swap_amount1,
                NULL::TEXT as position_liquidity,
                amount0::TEXT,
                amount1::TEXT,
                NULL::INT as tick_lower,
                NULL::INT as tick_upper,
                NULL::TEXT as liquidity_event_type,
                NULL::TEXT as flash_amount0,
                NULL::TEXT as flash_amount1,
                NULL::TEXT as flash_paid0,
                NULL::TEXT as flash_paid1,
                NULL::INTEGER as fee_protocol0_new,
                NULL::INTEGER as fee_protocol1_new
            FROM pool_fee_protocol_collect_event
            WHERE chain_id = $1 AND pool_identifier = $2
            AND ($3::BIGINT IS NULL OR block <= $3))
            UNION ALL
            (SELECT
                'flash' as event_type,
                chain_id,
                pool_identifier,
                block,
                transaction_hash,
                transaction_index,
                log_index,
                COALESCE(
                    (SELECT timestamp::TEXT FROM block WHERE block.chain_id = pool_flash_event.chain_id AND block.number = pool_flash_event.block),
                    (SELECT timestamp::TEXT FROM pool_event_block WHERE pool_event_block.chain_id = pool_flash_event.chain_id AND pool_event_block.number = pool_flash_event.block)
                ) as block_timestamp,
                sender,
                recipient,
                NULL::TEXT as owner,
                NULL::TEXT as sqrt_price_x96,
                NULL::TEXT as swap_liquidity,
                NULL::INT AS swap_tick,
                NULL::TEXT as swap_amount0,
                NULL::TEXT as swap_amount1,
                NULL::TEXT as position_liquidity,
                NULL::TEXT as amount0,
                NULL::TEXT as amount1,
                NULL::INT as tick_lower,
                NULL::INT as tick_upper,
                NULL::TEXT as liquidity_event_type,
                amount0::TEXT as flash_amount0,
                amount1::TEXT as flash_amount1,
                paid0::TEXT as flash_paid0,
                paid1::TEXT as flash_paid1,
                NULL::INTEGER as fee_protocol0_new,
                NULL::INTEGER as fee_protocol1_new
            FROM pool_flash_event
            WHERE chain_id = $1 AND pool_identifier = $2
            AND ($3::BIGINT IS NULL OR block <= $3))
            ) AS events
            ORDER BY events.block, events.transaction_index, events.log_index";

        const QUERY_FROM_POSITION: &str = "
            SELECT events.*,
                COALESCE(
                    (SELECT hash FROM pool_event_block WHERE pool_event_block.chain_id = events.chain_id AND pool_event_block.number = events.block),
                    (SELECT hash FROM block WHERE block.chain_id = events.chain_id AND block.number = events.block)
                ) as block_hash
            FROM (
            (SELECT
                'swap' as event_type,
                chain_id,
                pool_identifier,
                block,
                transaction_hash,
                transaction_index,
                log_index,
                COALESCE(
                    (SELECT timestamp::TEXT FROM block WHERE block.chain_id = pool_swap_event.chain_id AND block.number = pool_swap_event.block),
                    (SELECT timestamp::TEXT FROM pool_event_block WHERE pool_event_block.chain_id = pool_swap_event.chain_id AND pool_event_block.number = pool_swap_event.block)
                ) as block_timestamp,
                sender,
                recipient,
                NULL::TEXT as owner,
                sqrt_price_x96::TEXT,
                liquidity::TEXT as swap_liquidity,
                tick as swap_tick,
                amount0::TEXT as swap_amount0,
                amount1::TEXT as swap_amount1,
                NULL::TEXT as position_liquidity,
                NULL::TEXT as amount0,
                NULL::TEXT as amount1,
                NULL::INT as tick_lower,
                NULL::INT as tick_upper,
                NULL::TEXT as liquidity_event_type,
                NULL::TEXT as flash_amount0,
                NULL::TEXT as flash_amount1,
                NULL::TEXT as flash_paid0,
                NULL::TEXT as flash_paid1,
                NULL::INTEGER as fee_protocol0_new,
                NULL::INTEGER as fee_protocol1_new
            FROM pool_swap_event
            WHERE chain_id = $1 AND pool_identifier = $2
            AND (block > $3 OR (block = $3 AND transaction_index > $4) OR (block = $3 AND transaction_index = $4 AND log_index > $5))
            AND ($6::BIGINT IS NULL OR block <= $6))
            UNION ALL
            (SELECT
                'liquidity' as event_type,
                chain_id,
                pool_identifier,
                block,
                transaction_hash,
                transaction_index,
                log_index,
                COALESCE(
                    (SELECT timestamp::TEXT FROM block WHERE block.chain_id = pool_liquidity_event.chain_id AND block.number = pool_liquidity_event.block),
                    (SELECT timestamp::TEXT FROM pool_event_block WHERE pool_event_block.chain_id = pool_liquidity_event.chain_id AND pool_event_block.number = pool_liquidity_event.block)
                ) as block_timestamp,
                sender,
                NULL::TEXT as recipient,
                owner,
                NULL::text as sqrt_price_x96,
                NULL::TEXT as swap_liquidity,
                NULL::INT as swap_tick,
                amount0::TEXT as swap_amount0,
                amount1::TEXT as swap_amount1,
                position_liquidity::TEXT,
                amount0::TEXT,
                amount1::TEXT,
                tick_lower::INT,
                tick_upper::INT,
                event_type as liquidity_event_type,
                NULL::TEXT as flash_amount0,
                NULL::TEXT as flash_amount1,
                NULL::TEXT as flash_paid0,
                NULL::TEXT as flash_paid1,
                NULL::INTEGER as fee_protocol0_new,
                NULL::INTEGER as fee_protocol1_new
            FROM pool_liquidity_event
            WHERE chain_id = $1 AND pool_identifier = $2
            AND (block > $3 OR (block = $3 AND transaction_index > $4) OR (block = $3 AND transaction_index = $4 AND log_index > $5))
            AND ($6::BIGINT IS NULL OR block <= $6))
            UNION ALL
            (SELECT
                'collect' as event_type,
                chain_id,
                pool_identifier,
                block,
                transaction_hash,
                transaction_index,
                log_index,
                COALESCE(
                    (SELECT timestamp::TEXT FROM block WHERE block.chain_id = pool_collect_event.chain_id AND block.number = pool_collect_event.block),
                    (SELECT timestamp::TEXT FROM pool_event_block WHERE pool_event_block.chain_id = pool_collect_event.chain_id AND pool_event_block.number = pool_collect_event.block)
                ) as block_timestamp,
                NULL::TEXT as sender,
                NULL::TEXT as recipient,
                owner,
                NULL::TEXT as sqrt_price_x96,
                NULL::TEXT as swap_liquidity,
                NULL::INT AS swap_tick,
                amount0::TEXT as swap_amount0,
                amount1::TEXT as swap_amount1,
                NULL::TEXT as position_liquidity,
                amount0::TEXT,
                amount1::TEXT,
                tick_lower::INT,
                tick_upper::INT,
                NULL::TEXT as liquidity_event_type,
                NULL::TEXT as flash_amount0,
                NULL::TEXT as flash_amount1,
                NULL::TEXT as flash_paid0,
                NULL::TEXT as flash_paid1,
                NULL::INTEGER as fee_protocol0_new,
                NULL::INTEGER as fee_protocol1_new
            FROM pool_collect_event
            WHERE chain_id = $1 AND pool_identifier = $2
            AND (block > $3 OR (block = $3 AND transaction_index > $4) OR (block = $3 AND transaction_index = $4 AND log_index > $5))
            AND ($6::BIGINT IS NULL OR block <= $6))
            UNION ALL
            (SELECT
                'fee_protocol_update' as event_type,
                chain_id,
                pool_identifier,
                block,
                transaction_hash,
                transaction_index,
                log_index,
                COALESCE(
                    (SELECT timestamp::TEXT FROM block WHERE block.chain_id = pool_fee_protocol_update_event.chain_id AND block.number = pool_fee_protocol_update_event.block),
                    (SELECT timestamp::TEXT FROM pool_event_block WHERE pool_event_block.chain_id = pool_fee_protocol_update_event.chain_id AND pool_event_block.number = pool_fee_protocol_update_event.block)
                ) as block_timestamp,
                NULL::TEXT as sender,
                NULL::TEXT as recipient,
                NULL::TEXT as owner,
                NULL::TEXT as sqrt_price_x96,
                NULL::TEXT as swap_liquidity,
                NULL::INT AS swap_tick,
                NULL::TEXT as swap_amount0,
                NULL::TEXT as swap_amount1,
                NULL::TEXT as position_liquidity,
                NULL::TEXT as amount0,
                NULL::TEXT as amount1,
                NULL::INT as tick_lower,
                NULL::INT as tick_upper,
                NULL::TEXT as liquidity_event_type,
                NULL::TEXT as flash_amount0,
                NULL::TEXT as flash_amount1,
                NULL::TEXT as flash_paid0,
                NULL::TEXT as flash_paid1,
                fee_protocol0_new::INTEGER,
                fee_protocol1_new::INTEGER
            FROM pool_fee_protocol_update_event
            WHERE chain_id = $1 AND pool_identifier = $2
            AND (block > $3 OR (block = $3 AND transaction_index > $4) OR (block = $3 AND transaction_index = $4 AND log_index > $5))
            AND ($6::BIGINT IS NULL OR block <= $6))
            UNION ALL
            (SELECT
                'fee_protocol_collect' as event_type,
                chain_id,
                pool_identifier,
                block,
                transaction_hash,
                transaction_index,
                log_index,
                COALESCE(
                    (SELECT timestamp::TEXT FROM block WHERE block.chain_id = pool_fee_protocol_collect_event.chain_id AND block.number = pool_fee_protocol_collect_event.block),
                    (SELECT timestamp::TEXT FROM pool_event_block WHERE pool_event_block.chain_id = pool_fee_protocol_collect_event.chain_id AND pool_event_block.number = pool_fee_protocol_collect_event.block)
                ) as block_timestamp,
                sender,
                recipient,
                NULL::TEXT as owner,
                NULL::TEXT as sqrt_price_x96,
                NULL::TEXT as swap_liquidity,
                NULL::INT AS swap_tick,
                NULL::TEXT as swap_amount0,
                NULL::TEXT as swap_amount1,
                NULL::TEXT as position_liquidity,
                amount0::TEXT,
                amount1::TEXT,
                NULL::INT as tick_lower,
                NULL::INT as tick_upper,
                NULL::TEXT as liquidity_event_type,
                NULL::TEXT as flash_amount0,
                NULL::TEXT as flash_amount1,
                NULL::TEXT as flash_paid0,
                NULL::TEXT as flash_paid1,
                NULL::INTEGER as fee_protocol0_new,
                NULL::INTEGER as fee_protocol1_new
            FROM pool_fee_protocol_collect_event
            WHERE chain_id = $1 AND pool_identifier = $2
            AND (block > $3 OR (block = $3 AND transaction_index > $4) OR (block = $3 AND transaction_index = $4 AND log_index > $5))
            AND ($6::BIGINT IS NULL OR block <= $6))
            UNION ALL
            (SELECT
                'flash' as event_type,
                chain_id,
                pool_identifier,
                block,
                transaction_hash,
                transaction_index,
                log_index,
                COALESCE(
                    (SELECT timestamp::TEXT FROM block WHERE block.chain_id = pool_flash_event.chain_id AND block.number = pool_flash_event.block),
                    (SELECT timestamp::TEXT FROM pool_event_block WHERE pool_event_block.chain_id = pool_flash_event.chain_id AND pool_event_block.number = pool_flash_event.block)
                ) as block_timestamp,
                sender,
                recipient,
                NULL::TEXT as owner,
                NULL::TEXT as sqrt_price_x96,
                NULL::TEXT as swap_liquidity,
                NULL::INT AS swap_tick,
                NULL::TEXT as swap_amount0,
                NULL::TEXT as swap_amount1,
                NULL::TEXT as position_liquidity,
                NULL::TEXT as amount0,
                NULL::TEXT as amount1,
                NULL::INT as tick_lower,
                NULL::INT as tick_upper,
                NULL::TEXT as liquidity_event_type,
                amount0::TEXT as flash_amount0,
                amount1::TEXT as flash_amount1,
                paid0::TEXT as flash_paid0,
                paid1::TEXT as flash_paid1,
                NULL::INTEGER as fee_protocol0_new,
                NULL::INTEGER as fee_protocol1_new
            FROM pool_flash_event
            WHERE chain_id = $1 AND pool_identifier = $2
            AND (block > $3 OR (block = $3 AND transaction_index > $4) OR (block = $3 AND transaction_index = $4 AND log_index > $5))
            AND ($6::BIGINT IS NULL OR block <= $6))
            ) AS events
            ORDER BY events.block, events.transaction_index, events.log_index";

        // Build query with appropriate bindings
        let query = if let Some(pos) = from_position {
            sqlx::query(QUERY_FROM_POSITION)
                .bind(chain.chain_id as i32)
                .bind(pool_identifier.to_string())
                .bind(pos.number as i64)
                .bind(pos.transaction_index as i32)
                .bind(pos.log_index as i32)
                .bind(to_block.map(|block| block as i64))
                .fetch(&self.pool)
        } else {
            sqlx::query(QUERY_ALL)
                .bind(chain.chain_id as i32)
                .bind(pool_identifier.to_string())
                .bind(to_block.map(|block| block as i64))
                .fetch(&self.pool)
        };

        // Transform rows to events
        let stream = query.map(move |row_result| match row_result {
            Ok(row) => {
                transform_row_to_dex_pool_data(&row, chain.clone(), dex.clone(), instrument_id)
                    .map_err(|e| anyhow::anyhow!("Steam pool event transform error: {e}"))
            }
            Err(e) => Err(anyhow::anyhow!("Stream pool events database error: {e}")),
        });

        Box::pin(stream)
    }

    /// Persists an execution transaction record to the `execution_transaction` table.
    ///
    /// Records are written before broadcast so a signed transaction is never forgotten;
    /// the unique `(chain_id, transaction_hash)` constraint makes an exact re-insertion
    /// idempotent. Signer nonce ownership and order IDs are unique before broadcast. Order
    /// submission records carry the client order ID; operator transactions (wrap, approve)
    /// store `NULL`.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    #[expect(
        clippy::too_many_arguments,
        reason = "the parameters mirror the persisted execution transaction fields"
    )]
    pub async fn add_execution_transaction(
        &self,
        chain_id: u32,
        wallet_address: &str,
        nonce: u64,
        transaction_hash: &str,
        purpose: &str,
        status: &str,
        client_order_id: Option<&str>,
    ) -> anyhow::Result<()> {
        let result = sqlx::query(
            "
            INSERT INTO execution_transaction (
                chain_id,
                wallet_address,
                nonce,
                transaction_hash,
                purpose,
                status,
                client_order_id
            )
            VALUES ($1, $2, $3, $4, $5, $6, $7)
            ON CONFLICT (chain_id, transaction_hash)
            DO UPDATE SET transaction_hash = EXCLUDED.transaction_hash
            WHERE execution_transaction.wallet_address = EXCLUDED.wallet_address
              AND execution_transaction.nonce = EXCLUDED.nonce
              AND execution_transaction.purpose = EXCLUDED.purpose
              AND execution_transaction.status = EXCLUDED.status
              AND execution_transaction.client_order_id IS NOT DISTINCT FROM EXCLUDED.client_order_id
        ",
        )
        .bind(chain_id as i32)
        .bind(wallet_address)
        .bind(nonce as i64)
        .bind(transaction_hash)
        .bind(purpose)
        .bind(status)
        .bind(client_order_id)
        .execute(&self.pool)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to insert into execution_transaction table: {e}"))?;

        anyhow::ensure!(
            result.rows_affected() == 1,
            "Execution transaction {transaction_hash} conflicts with its persisted record"
        );
        Ok(())
    }

    /// Installs execution schema version 2 without changing existing transaction rows.
    ///
    /// The migration locks the legacy transaction table, refuses unresolved version 1 rows,
    /// installs the versioned intent and hash-history tables, and fences older writers before
    /// releasing the lock. This prevents a mixed-version process from bypassing the new signer
    /// ownership constraints.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn ensure_execution_transaction_schema(&self) -> anyhow::Result<()> {
        let mut transaction = self
            .pool
            .begin()
            .await
            .map_err(|e| anyhow::anyhow!("Failed to start execution schema migration: {e}"))?;

        sqlx::query("LOCK TABLE execution_transaction IN ACCESS EXCLUSIVE MODE")
            .execute(&mut *transaction)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to lock legacy execution transactions: {e}"))?;

        for statement in [
            "
            ALTER TABLE execution_transaction
            ADD COLUMN IF NOT EXISTS client_order_id TEXT
            ",
            "
            ALTER TABLE execution_transaction
            ADD COLUMN IF NOT EXISTS wallet_address TEXT
            ",
            "
            ALTER TABLE execution_transaction
            ALTER COLUMN wallet_address DROP NOT NULL
            ",
            "
            CREATE TABLE IF NOT EXISTS execution_schema_version (
                component TEXT PRIMARY KEY,
                version SMALLINT NOT NULL CHECK (version > 0)
            )
            ",
            "
            CREATE TABLE IF NOT EXISTS execution_intent (
                id BIGSERIAL PRIMARY KEY,
                schema_version SMALLINT NOT NULL,
                chain_id INTEGER NOT NULL REFERENCES chain(chain_id) ON DELETE RESTRICT,
                wallet_address TEXT NOT NULL,
                nonce BIGINT,
                purpose TEXT NOT NULL,
                status TEXT NOT NULL,
                client_order_id TEXT,
                trader_id TEXT,
                strategy_id TEXT,
                account_id TEXT,
                instrument_id TEXT,
                pool_address TEXT,
                transaction_to TEXT NOT NULL,
                transaction_input TEXT NOT NULL,
                transaction_value TEXT NOT NULL,
                amount_in TEXT,
                created_block BIGINT NOT NULL,
                acknowledgement_emitted BOOLEAN NOT NULL DEFAULT FALSE,
                fill_emitted BOOLEAN NOT NULL DEFAULT FALSE,
                terminal_emitted BOOLEAN NOT NULL DEFAULT FALSE,
                active BOOLEAN NOT NULL DEFAULT TRUE,
                created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                CHECK (schema_version = 2),
                CHECK (nonce IS NULL OR nonce >= 0),
                CHECK (created_block >= 0),
                CHECK (status IN (
                    'prepared', 'signed', 'broadcast', 'included', 'finalized',
                    'reverted', 'replaced', 'dropped', 'reorged', 'recoverable'
                )),
                CONSTRAINT execution_intent_active_check CHECK (
                    NOT active
                    OR status NOT IN ('finalized', 'reverted', 'recoverable')
                    OR (
                        status IN ('finalized', 'reverted')
                        AND NOT (fill_emitted OR terminal_emitted)
                    )
                ),
                CHECK (NOT (fill_emitted AND terminal_emitted)),
                CHECK (
                    purpose <> 'swap'
                    OR (
                        client_order_id IS NOT NULL
                        AND trader_id IS NOT NULL
                        AND strategy_id IS NOT NULL
                        AND account_id IS NOT NULL
                        AND instrument_id IS NOT NULL
                        AND pool_address IS NOT NULL
                        AND amount_in IS NOT NULL
                    )
                ),
                UNIQUE (id, chain_id)
            )
            ",
            "
            CREATE UNIQUE INDEX IF NOT EXISTS execution_intent_active_signer_key
            ON execution_intent (chain_id, wallet_address)
            WHERE active
            ",
            "
            CREATE UNIQUE INDEX IF NOT EXISTS execution_intent_active_nonce_key
            ON execution_intent (chain_id, wallet_address, nonce)
            WHERE active AND nonce IS NOT NULL
            ",
            "
            CREATE UNIQUE INDEX IF NOT EXISTS execution_intent_client_order_key
            ON execution_intent (chain_id, wallet_address, client_order_id)
            WHERE client_order_id IS NOT NULL
            ",
            "
            ALTER TABLE execution_intent
            DROP CONSTRAINT IF EXISTS execution_intent_active_check
            ",
            "
            ALTER TABLE execution_intent
            ADD CONSTRAINT execution_intent_active_check CHECK (
                NOT active
                OR status NOT IN ('finalized', 'reverted', 'recoverable')
                OR (
                    status IN ('finalized', 'reverted')
                    AND NOT (fill_emitted OR terminal_emitted)
                )
            )
            ",
            "
            CREATE TABLE IF NOT EXISTS execution_transaction_hash (
                id BIGSERIAL PRIMARY KEY,
                intent_id BIGINT NOT NULL,
                chain_id INTEGER NOT NULL,
                transaction_hash TEXT NOT NULL,
                payload_expected BOOLEAN NOT NULL DEFAULT TRUE,
                raw_transaction BYTEA,
                sealed_transaction BYTEA,
                status TEXT NOT NULL,
                block_number BIGINT,
                block_hash TEXT,
                receipt_success BOOLEAN,
                gas_used BIGINT,
                effective_gas_price TEXT,
                current BOOLEAN NOT NULL DEFAULT TRUE,
                created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                FOREIGN KEY (intent_id, chain_id)
                    REFERENCES execution_intent(id, chain_id) ON DELETE RESTRICT,
                CHECK (block_number IS NULL OR block_number >= 0),
                CHECK (gas_used IS NULL OR gas_used >= 0),
                CONSTRAINT execution_transaction_raw_size_check
                    CHECK (raw_transaction IS NULL OR octet_length(raw_transaction) <= 131072),
                CONSTRAINT execution_transaction_sealed_size_check
                    CHECK (sealed_transaction IS NULL OR octet_length(sealed_transaction) <= 131133),
                CHECK (status IN (
                    'signed', 'broadcast', 'included', 'finalized', 'reverted',
                    'replaced', 'dropped', 'reorged'
                )),
                UNIQUE (chain_id, transaction_hash),
                UNIQUE (intent_id, transaction_hash)
            )
            ",
            "
            ALTER TABLE execution_transaction_hash
            ADD COLUMN IF NOT EXISTS payload_expected BOOLEAN
            ",
            "
            UPDATE execution_transaction_hash
            SET payload_expected = raw_transaction IS NOT NULL
            WHERE payload_expected IS NULL
            ",
            "
            ALTER TABLE execution_transaction_hash
            ALTER COLUMN payload_expected SET DEFAULT TRUE
            ",
            "
            ALTER TABLE execution_transaction_hash
            ALTER COLUMN payload_expected SET NOT NULL
            ",
            "
            ALTER TABLE execution_transaction_hash
            ADD COLUMN IF NOT EXISTS sealed_transaction BYTEA
            ",
            "
            ALTER TABLE execution_transaction_hash
            DROP CONSTRAINT IF EXISTS execution_transaction_raw_size_check
            ",
            "
            ALTER TABLE execution_transaction_hash
            ADD CONSTRAINT execution_transaction_raw_size_check
            CHECK (raw_transaction IS NULL OR octet_length(raw_transaction) <= 131072)
            ",
            "
            ALTER TABLE execution_transaction_hash
            DROP CONSTRAINT IF EXISTS execution_transaction_sealed_size_check
            ",
            "
            ALTER TABLE execution_transaction_hash
            ADD CONSTRAINT execution_transaction_sealed_size_check
            CHECK (sealed_transaction IS NULL OR octet_length(sealed_transaction) <= 131133)
            ",
            "
            CREATE TABLE IF NOT EXISTS execution_payload_state (
                component TEXT PRIMARY KEY CHECK (component = 'signed_transactions'),
                deployment_id TEXT NOT NULL CHECK (deployment_id <> ''),
                protocol_version SMALLINT NOT NULL CHECK (protocol_version = 1),
                operation TEXT NOT NULL CHECK (operation IN ('migrate', 'ready', 'rewrap', 'rollback')),
                active_key_id BYTEA NOT NULL CHECK (octet_length(active_key_id) = 32),
                progress_id BIGINT NOT NULL DEFAULT 0 CHECK (progress_id >= 0),
                updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
            )
            ",
            "
            CREATE TABLE IF NOT EXISTS execution_payload_key_state (
                key_id BYTEA PRIMARY KEY CHECK (octet_length(key_id) = 32),
                seals BIGINT NOT NULL DEFAULT 0 CHECK (seals >= 0 AND seals < 4294967296)
            )
            ",
            "
            CREATE UNIQUE INDEX IF NOT EXISTS execution_transaction_hash_current_key
            ON execution_transaction_hash (intent_id)
            WHERE current
            ",
            "
            CREATE TABLE IF NOT EXISTS execution_transaction_transition (
                id BIGSERIAL PRIMARY KEY,
                intent_id BIGINT NOT NULL REFERENCES execution_intent(id) ON DELETE RESTRICT,
                transaction_hash_id BIGINT REFERENCES execution_transaction_hash(id) ON DELETE RESTRICT,
                transition_key TEXT NOT NULL,
                from_status TEXT,
                to_status TEXT NOT NULL,
                block_number BIGINT,
                block_hash TEXT,
                observed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                CHECK (block_number IS NULL OR block_number >= 0),
                CHECK (to_status IN (
                    'prepared', 'signed', 'broadcast', 'included', 'finalized',
                    'reverted', 'replaced', 'dropped', 'reorged', 'recoverable'
                )),
                UNIQUE (intent_id, transition_key)
            )
            ",
        ] {
            sqlx::query(statement)
                .execute(&mut *transaction)
                .await
                .map_err(|e| {
                    anyhow::anyhow!("Failed to migrate execution_transaction table: {e}")
                })?;
        }

        let installed_version = sqlx::query_scalar::<_, i16>(
            "SELECT version FROM execution_schema_version WHERE component = 'evm_execution'",
        )
        .fetch_optional(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to read execution schema version: {e}"))?;
        if let Some(installed_version) = installed_version
            && installed_version > EXECUTION_SCHEMA_VERSION
        {
            anyhow::bail!(
                "Execution schema version {} is newer than supported version {EXECUTION_SCHEMA_VERSION}",
                installed_version
            );
        }

        let unresolved_legacy = sqlx::query_scalar::<_, i64>(
            "
            SELECT COUNT(*)
            FROM execution_transaction
            WHERE status IN ('pending', 'included', 'reverted')
            ",
        )
        .fetch_one(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to inspect legacy execution transactions: {e}"))?;
        anyhow::ensure!(
            unresolved_legacy == 0,
            "Cannot safely migrate {unresolved_legacy} unresolved execution schema version 1 transaction(s); resolve them with the prior version before enabling version {EXECUTION_SCHEMA_VERSION}"
        );

        for statement in [
            "
            CREATE OR REPLACE FUNCTION execution_transaction_v2_fence()
            RETURNS TRIGGER AS $$
            BEGIN
                RAISE EXCEPTION 'Legacy execution writer refused after schema version 2 activation';
            END;
            $$ LANGUAGE plpgsql
            ",
            "DROP TRIGGER IF EXISTS execution_transaction_v2_fence ON execution_transaction",
            "
            CREATE TRIGGER execution_transaction_v2_fence
            BEFORE INSERT OR UPDATE OR DELETE ON execution_transaction
            FOR EACH STATEMENT EXECUTE FUNCTION execution_transaction_v2_fence()
            ",
            "
            CREATE OR REPLACE FUNCTION execution_transition_append_only()
            RETURNS TRIGGER AS $$
            BEGIN
                RAISE EXCEPTION 'Execution transitions are append-only';
            END;
            $$ LANGUAGE plpgsql
            ",
            "DROP TRIGGER IF EXISTS execution_transition_append_only ON execution_transaction_transition",
            "
            CREATE TRIGGER execution_transition_append_only
            BEFORE UPDATE OR DELETE ON execution_transaction_transition
            FOR EACH STATEMENT EXECUTE FUNCTION execution_transition_append_only()
            ",
            "
            INSERT INTO execution_schema_version (component, version)
            VALUES ('evm_execution', 2)
            ON CONFLICT (component) DO UPDATE SET version = EXCLUDED.version
            WHERE execution_schema_version.version <= EXCLUDED.version
            ",
        ] {
            sqlx::query(statement)
                .execute(&mut *transaction)
                .await
                .map_err(|e| {
                    anyhow::anyhow!("Failed to activate execution schema version 2: {e}")
                })?;
        }

        transaction
            .commit()
            .await
            .map_err(|e| anyhow::anyhow!("Failed to commit execution schema migration: {e}"))?;

        Ok(())
    }

    /// Loads the durable finalized-header and nonce position when verification is active.
    pub(crate) async fn load_execution_verification_position(
        &self,
        chain_id: u32,
        wallet_address: &str,
        manifest_version: &str,
        manifest_digest: &str,
    ) -> anyhow::Result<Option<ExecutionVerificationPosition>> {
        let installed = sqlx::query_scalar::<_, i16>(
            "SELECT version FROM execution_schema_version \
             WHERE component = 'evm_execution_verification'",
        )
        .fetch_optional(&self.pool)
        .await
        .context("failed to inspect execution verification schema")?;
        let Some(installed) = installed else {
            return Ok(None);
        };
        anyhow::ensure!(
            installed <= VERIFICATION_SCHEMA_VERSION,
            "Unsupported execution verification schema version {installed}"
        );
        let chain_id =
            i32::try_from(chain_id).context("Verification chain ID exceeds PostgreSQL INTEGER")?;
        let current = sqlx::query_as::<_, (String, String, i64, i64)>(
            "
            SELECT manifest_version, manifest_digest, next_canonical_nonce, revision
            FROM execution_verification_nonce
            WHERE chain_id = $1 AND wallet_address = $2
            ",
        )
        .bind(chain_id)
        .bind(wallet_address)
        .fetch_optional(&self.pool)
        .await
        .context("failed to load execution verification nonce position")?;
        let Some((stored_version, stored_digest, nonce, revision)) = current else {
            return Ok(None);
        };
        anyhow::ensure!(
            stored_version == manifest_version && stored_digest == manifest_digest,
            "Execution verification manifest identity changed"
        );
        let row = sqlx::query_as::<_, (i64, String, String, i64, Option<String>, String)>(
            "
            SELECT number, hash, parent_hash, timestamp, base_fee_per_gas, manifest_digest
            FROM execution_verified_finalized_header
            WHERE chain_id = $1 AND wallet_address = $2
            ORDER BY number DESC
            LIMIT 1
            ",
        )
        .bind(chain_id)
        .bind(wallet_address)
        .fetch_optional(&self.pool)
        .await
        .context("failed to load verified finalized header tip")?
        .ok_or_else(|| anyhow::anyhow!("Verified finalized header ledger is empty"))?;
        let (number, hash, parent_hash, timestamp, base_fee, digest) = row;
        anyhow::ensure!(
            digest == manifest_digest,
            "Finalized header manifest identity changed"
        );
        Ok(Some(ExecutionVerificationPosition {
            next_canonical_nonce: u64::try_from(nonce).context("Canonical nonce is negative")?,
            revision: u64::try_from(revision).context("Canonical nonce revision is negative")?,
            finalized_tip: ExecutionVerifiedHeader {
                number: u64::try_from(number).context("Finalized header number is negative")?,
                hash,
                parent_hash,
                timestamp: u64::try_from(timestamp)
                    .context("Finalized header timestamp is negative")?,
                base_fee_per_gas: base_fee
                    .map(|value| {
                        value
                            .parse::<u128>()
                            .map_err(|_| anyhow::anyhow!("Finalized header base fee is invalid"))
                    })
                    .transpose()?,
            },
        }))
    }

    pub(crate) async fn load_execution_verified_header(
        &self,
        chain_id: u32,
        wallet_address: &str,
        number: u64,
        manifest_digest: &str,
    ) -> anyhow::Result<Option<ExecutionVerifiedHeader>> {
        let chain_id =
            i32::try_from(chain_id).context("Verification chain ID exceeds PostgreSQL INTEGER")?;
        let number =
            i64::try_from(number).context("Finalized header number exceeds PostgreSQL BIGINT")?;
        let row = sqlx::query_as::<_, (String, String, i64, Option<String>, String)>(
            "
            SELECT hash, parent_hash, timestamp, base_fee_per_gas, manifest_digest
            FROM execution_verified_finalized_header
            WHERE chain_id = $1 AND wallet_address = $2 AND number = $3
            ",
        )
        .bind(chain_id)
        .bind(wallet_address)
        .bind(number)
        .fetch_optional(&self.pool)
        .await
        .context("failed to load verified finalized header")?;
        let Some((hash, parent_hash, timestamp, base_fee, digest)) = row else {
            return Ok(None);
        };
        anyhow::ensure!(
            digest == manifest_digest,
            "Finalized header manifest identity changed"
        );
        Ok(Some(ExecutionVerifiedHeader {
            number: u64::try_from(number).context("Finalized header number is negative")?,
            hash,
            parent_hash,
            timestamp: u64::try_from(timestamp)
                .context("Finalized header timestamp is negative")?,
            base_fee_per_gas: base_fee
                .map(|value| {
                    value
                        .parse::<u128>()
                        .map_err(|_| anyhow::anyhow!("Finalized header base fee is invalid"))
                })
                .transpose()?,
        }))
    }

    /// Loads the retained execution state that a first verification bootstrap must classify.
    pub(crate) async fn load_execution_verification_migration_snapshot(
        &self,
        chain_id: u32,
        wallet_address: &str,
    ) -> anyhow::Result<ExecutionVerificationMigrationSnapshot> {
        let chain_id =
            i32::try_from(chain_id).context("Verification chain ID exceeds PostgreSQL INTEGER")?;
        let intents = sqlx::query_as::<_, ExecutionIntentRow>(
            "
            SELECT
                id, schema_version, chain_id, wallet_address, nonce, purpose, status,
                client_order_id, trader_id, strategy_id, account_id, instrument_id,
                pool_address, transaction_to, transaction_input, transaction_value,
                amount_in, created_block, acknowledgement_emitted, fill_emitted,
                terminal_emitted, active
            FROM execution_intent
            WHERE chain_id = $1 AND wallet_address = $2
            ORDER BY id
            ",
        )
        .bind(chain_id)
        .bind(wallet_address)
        .fetch_all(&self.pool)
        .await
        .context("failed to load retained execution intents for verification migration")?;
        let hashes = sqlx::query_as::<_, ExecutionTransactionHashRow>(
            "
            SELECT
                hash.id, hash.intent_id, hash.chain_id, hash.transaction_hash,
                hash.payload_expected, hash.raw_transaction, hash.sealed_transaction,
                hash.status, hash.block_number, hash.block_hash, hash.receipt_success,
                hash.gas_used, hash.effective_gas_price, hash.current
            FROM execution_transaction_hash AS hash
            JOIN execution_intent AS intent ON intent.id = hash.intent_id
            WHERE intent.chain_id = $1 AND intent.wallet_address = $2
            ORDER BY hash.id
            ",
        )
        .bind(chain_id)
        .bind(wallet_address)
        .fetch_all(&self.pool)
        .await
        .context("failed to load retained execution hashes for verification migration")?;
        Ok(ExecutionVerificationMigrationSnapshot { intents, hashes })
    }

    /// Installs or validates the independent verification ledger for an execution signer.
    ///
    /// A database with retained execution intents but no verification ledger requires an archive-
    /// verified migration which classifies every retained intent. An empty database initializes
    /// its canonical nonce from a verified finalized-height transaction count and its header
    /// ledger from the trusted checkpoint.
    ///
    /// # Errors
    ///
    /// Returns an error if the schema, migration snapshot, historical reconstruction, or retained
    /// evidence is inconsistent, or persistence fails.
    pub(crate) async fn ensure_execution_verification_schema(
        &self,
        bootstrap: &ExecutionVerificationBootstrap<'_>,
    ) -> anyhow::Result<()> {
        let first_header = bootstrap
            .finalized_headers
            .first()
            .ok_or_else(|| anyhow::anyhow!("Verified finalized header ledger is empty"))?;
        anyhow::ensure!(
            bootstrap.finalized_headers.windows(2).all(|headers| {
                headers[1].number == headers[0].number.saturating_add(1)
                    && headers[1].parent_hash == headers[0].hash
            }),
            "Verified finalized headers are not one continuous parent-linked chain"
        );
        anyhow::ensure!(
            bootstrap.provider_ids.len() == 3
                && bootstrap.operator_ids.len() == 3
                && bootstrap.failure_domain_ids.len() >= 3
                && !bootstrap.decisions.is_empty(),
            "Connect verification evidence is incomplete"
        );
        let chain_id = i32::try_from(bootstrap.chain_id)
            .context("Verification chain ID exceeds PostgreSQL INTEGER")?;
        let checkpoint_number = i64::try_from(bootstrap.checkpoint_number)
            .context("Verification checkpoint exceeds PostgreSQL BIGINT")?;
        let checkpoint_timestamp = i64::try_from(bootstrap.checkpoint_timestamp)
            .context("Verification checkpoint timestamp exceeds PostgreSQL BIGINT")?;
        let next_canonical_nonce = i64::try_from(bootstrap.next_canonical_nonce)
            .context("Canonical nonce exceeds PostgreSQL BIGINT")?;
        let observed_canonical_nonce = i64::try_from(bootstrap.observed_canonical_nonce)
            .context("Observed canonical nonce exceeds PostgreSQL BIGINT")?;
        let base_fee = bootstrap
            .checkpoint_base_fee_per_gas
            .map(|value| value.to_string());
        let mut transaction = self.pool.begin().await.map_err(|e| {
            anyhow::anyhow!("Failed to start execution verification migration: {e}")
        })?;

        for statement in [
            "
            CREATE TABLE IF NOT EXISTS execution_verification_nonce (
                chain_id INTEGER NOT NULL REFERENCES chain(chain_id) ON DELETE RESTRICT,
                wallet_address TEXT NOT NULL,
                manifest_version TEXT NOT NULL CHECK (manifest_version <> ''),
                manifest_digest TEXT NOT NULL CHECK (manifest_digest <> ''),
                next_canonical_nonce BIGINT NOT NULL CHECK (next_canonical_nonce >= 0),
                revision BIGINT NOT NULL DEFAULT 0 CHECK (revision >= 0),
                updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                PRIMARY KEY (chain_id, wallet_address)
            )
            ",
            "
            CREATE TABLE IF NOT EXISTS execution_verified_finalized_header (
                chain_id INTEGER NOT NULL,
                wallet_address TEXT NOT NULL,
                number BIGINT NOT NULL CHECK (number >= 0),
                hash TEXT NOT NULL CHECK (hash <> ''),
                parent_hash TEXT NOT NULL CHECK (parent_hash <> ''),
                timestamp BIGINT NOT NULL CHECK (timestamp >= 0),
                base_fee_per_gas TEXT,
                manifest_digest TEXT NOT NULL CHECK (manifest_digest <> ''),
                observed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                PRIMARY KEY (chain_id, wallet_address, number),
                UNIQUE (chain_id, wallet_address, hash),
                FOREIGN KEY (chain_id, wallet_address)
                    REFERENCES execution_verification_nonce(chain_id, wallet_address)
                    ON DELETE RESTRICT
            )
            ",
            "
            CREATE TABLE IF NOT EXISTS execution_verification_decision (
                id BIGSERIAL PRIMARY KEY,
                intent_id BIGINT REFERENCES execution_intent(id) ON DELETE RESTRICT,
                nonce BIGINT CHECK (nonce IS NULL OR nonce >= 0),
                decision_class TEXT NOT NULL CHECK (decision_class <> ''),
                read_class TEXT NOT NULL CHECK (read_class <> ''),
                height_start BIGINT CHECK (height_start IS NULL OR height_start >= 0),
                height_end BIGINT CHECK (height_end IS NULL OR height_end >= 0),
                manifest_version TEXT NOT NULL CHECK (manifest_version <> ''),
                manifest_digest TEXT NOT NULL CHECK (manifest_digest <> ''),
                provider_ids TEXT[] NOT NULL,
                operator_ids TEXT[] NOT NULL,
                failure_domain_ids TEXT[] NOT NULL,
                response_class TEXT NOT NULL CHECK (response_class <> ''),
                normalized_value_digest TEXT,
                safe_value JSONB,
                nonce_revision BIGINT CHECK (nonce_revision IS NULL OR nonce_revision >= 0),
                outcome TEXT NOT NULL CHECK (outcome IN (
                    'verified', 'disagreement', 'unavailable', 'retryable', 'locally_invalid'
                )),
                transition_key TEXT,
                observed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                CHECK (cardinality(provider_ids) = 3),
                CHECK (cardinality(operator_ids) = 3),
                CHECK (cardinality(failure_domain_ids) >= 3),
                CHECK (safe_value IS NULL OR pg_column_size(safe_value) <= 16384),
                UNIQUE (intent_id, transition_key)
            )
            ",
            "
            CREATE TABLE IF NOT EXISTS execution_replacement_scan (
                intent_id BIGINT PRIMARY KEY REFERENCES execution_intent(id) ON DELETE RESTRICT,
                chain_id INTEGER NOT NULL,
                wallet_address TEXT NOT NULL,
                nonce BIGINT NOT NULL CHECK (nonce >= 0),
                finalized_cursor_number BIGINT NOT NULL CHECK (finalized_cursor_number >= 0),
                finalized_cursor_hash TEXT NOT NULL CHECK (finalized_cursor_hash <> ''),
                manifest_digest TEXT NOT NULL CHECK (manifest_digest <> ''),
                updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                UNIQUE (chain_id, wallet_address, nonce)
            )
            ",
        ] {
            sqlx::query(statement)
                .execute(&mut *transaction)
                .await
                .map_err(|e| anyhow::anyhow!("Failed to install verification schema: {e}"))?;
        }

        sqlx::query(
            "LOCK TABLE execution_intent, execution_transaction_hash, \
             execution_verification_nonce, execution_verified_finalized_header, \
             execution_verification_decision, execution_replacement_scan \
             IN ACCESS EXCLUSIVE MODE",
        )
        .execute(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to lock execution verification state: {e}"))?;

        let installed_version = sqlx::query_scalar::<_, i16>(
            "SELECT version FROM execution_schema_version \
             WHERE component = 'evm_execution_verification'",
        )
        .fetch_optional(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to read verification schema version: {e}"))?;
        if let Some(installed_version) = installed_version {
            anyhow::ensure!(
                installed_version <= VERIFICATION_SCHEMA_VERSION,
                "Execution verification schema version {installed_version} is newer than supported version {VERIFICATION_SCHEMA_VERSION}"
            );
        }

        let current = sqlx::query_as::<_, (String, String, i64, i64)>(
            "
            SELECT manifest_version, manifest_digest, next_canonical_nonce, revision
            FROM execution_verification_nonce
            WHERE chain_id = $1 AND wallet_address = $2
            FOR UPDATE
            ",
        )
        .bind(chain_id)
        .bind(bootstrap.wallet_address)
        .fetch_optional(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to read canonical nonce ledger: {e}"))?;
        let initialized = current.is_some();

        let revision = if let Some((manifest_version, manifest_digest, stored_nonce, revision)) =
            current
        {
            anyhow::ensure!(
                bootstrap.migration.is_none(),
                "Verification migration was supplied for an initialized signer"
            );
            anyhow::ensure!(
                manifest_version == bootstrap.manifest_version
                    && manifest_digest == bootstrap.manifest_digest,
                "Execution verification manifest identity changed"
            );
            anyhow::ensure!(
                stored_nonce == next_canonical_nonce,
                "Canonical nonce ledger changed during verification bootstrap"
            );

            if observed_canonical_nonce != stored_nonce {
                let expected_observed_nonce = stored_nonce
                    .checked_add(1)
                    .ok_or_else(|| anyhow::anyhow!("Canonical nonce overflow"))?;
                anyhow::ensure!(
                    observed_canonical_nonce == expected_observed_nonce,
                    "Verified finalized transaction count is outside the owned recovery range"
                );
                let recovery = sqlx::query_as::<_, (Option<i64>, String, i64)>(
                    "
                    SELECT
                        intent.nonce,
                        intent.status,
                        COUNT(hash.id) FILTER (
                            WHERE hash.current
                              AND hash.payload_expected
                              AND ((hash.raw_transaction IS NOT NULL)::INTEGER
                                   + (hash.sealed_transaction IS NOT NULL)::INTEGER) = 1
                              AND hash.status IN (
                                  'broadcast', 'included', 'replaced', 'dropped', 'reorged'
                              )
                        )
                    FROM execution_intent AS intent
                    LEFT JOIN execution_transaction_hash AS hash
                        ON hash.intent_id = intent.id
                    WHERE intent.chain_id = $1
                      AND intent.wallet_address = $2
                      AND intent.active
                    GROUP BY intent.id, intent.nonce, intent.status
                    ",
                )
                .bind(chain_id)
                .bind(bootstrap.wallet_address)
                .fetch_optional(&mut *transaction)
                .await
                .map_err(|e| anyhow::anyhow!("Failed to validate nonce recovery ownership: {e}"))?;
                let Some((intent_nonce, intent_status, payload_count)) = recovery else {
                    anyhow::bail!(
                        "Verified finalized transaction count advanced without an active owned intent"
                    );
                };
                anyhow::ensure!(
                    intent_nonce == Some(stored_nonce)
                        && matches!(
                            intent_status.as_str(),
                            "broadcast" | "included" | "replaced" | "dropped" | "reorged"
                        )
                        && payload_count == 1,
                    "Verified finalized transaction count advanced without one recoverable retained payload at the durable nonce"
                );
            }
            revision
        } else {
            anyhow::ensure!(
                next_canonical_nonce == observed_canonical_nonce,
                "Initial canonical nonce conflicts with the verified finalized transaction count"
            );
            let locked_intents = sqlx::query_as::<_, ExecutionIntentRow>(
                "
                SELECT
                    id, schema_version, chain_id, wallet_address, nonce, purpose, status,
                    client_order_id, trader_id, strategy_id, account_id, instrument_id,
                    pool_address, transaction_to, transaction_input, transaction_value,
                    amount_in, created_block, acknowledgement_emitted, fill_emitted,
                    terminal_emitted, active
                FROM execution_intent
                WHERE chain_id = $1 AND wallet_address = $2
                ORDER BY id
                ",
            )
            .bind(chain_id)
            .bind(bootstrap.wallet_address)
            .fetch_all(&mut *transaction)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to lock retained execution intents: {e}"))?;
            let locked_hashes = sqlx::query_as::<_, ExecutionTransactionHashRow>(
                "
                SELECT
                    hash.id, hash.intent_id, hash.chain_id, hash.transaction_hash,
                    hash.payload_expected, hash.raw_transaction, hash.sealed_transaction,
                    hash.status, hash.block_number, hash.block_hash, hash.receipt_success,
                    hash.gas_used, hash.effective_gas_price, hash.current
                FROM execution_transaction_hash AS hash
                JOIN execution_intent AS intent ON intent.id = hash.intent_id
                WHERE intent.chain_id = $1 AND intent.wallet_address = $2
                ORDER BY hash.id
                ",
            )
            .bind(chain_id)
            .bind(bootstrap.wallet_address)
            .fetch_all(&mut *transaction)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to lock retained execution hashes: {e}"))?;
            if locked_intents.is_empty() {
                anyhow::ensure!(
                    bootstrap.migration.is_none(),
                    "Verification migration was supplied for empty execution history"
                );
            } else {
                let migration = bootstrap.migration.ok_or_else(|| {
                    anyhow::anyhow!(
                        "Retained execution history requires archive verification before migration"
                    )
                })?;
                anyhow::ensure!(
                    migration.snapshot.intents == locked_intents
                        && migration.snapshot.hashes == locked_hashes,
                    "Retained execution history changed during verification migration"
                );
                let intent_ids = locked_intents
                    .iter()
                    .map(|intent| intent.id)
                    .collect::<BTreeSet<_>>();
                let record_ids = migration
                    .records
                    .iter()
                    .map(|record| record.intent_id)
                    .collect::<BTreeSet<_>>();
                anyhow::ensure!(
                    intent_ids == record_ids && record_ids.len() == migration.records.len(),
                    "Verification migration does not classify every retained intent exactly once"
                );
            }
            sqlx::query(
                "
                INSERT INTO execution_verification_nonce (
                    chain_id, wallet_address, manifest_version, manifest_digest,
                    next_canonical_nonce
                )
                VALUES ($1, $2, $3, $4, $5)
                ",
            )
            .bind(chain_id)
            .bind(bootstrap.wallet_address)
            .bind(bootstrap.manifest_version)
            .bind(bootstrap.manifest_digest)
            .bind(next_canonical_nonce)
            .execute(&mut *transaction)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to initialize canonical nonce ledger: {e}"))?;

            if let Some(migration) = bootstrap.migration {
                for record in &migration.records {
                    anyhow::ensure!(
                        !record.decisions.is_empty(),
                        "Verification migration evidence is empty for intent {}",
                        record.intent_id
                    );

                    if record.recover_prepared {
                        anyhow::ensure!(
                            record.terminal_status.is_none()
                                && record.transaction_hash.is_none()
                                && record.nonce.is_none(),
                            "Prepared recovery migration record is inconsistent"
                        );
                        let result = sqlx::query(
                            "
                            UPDATE execution_intent
                            SET status = 'recoverable', active = FALSE, updated_at = NOW()
                            WHERE id = $1 AND active AND status = 'prepared' AND nonce IS NULL
                              AND NOT EXISTS (
                                  SELECT 1 FROM execution_transaction_hash WHERE intent_id = $1
                              )
                            ",
                        )
                        .bind(record.intent_id)
                        .execute(&mut *transaction)
                        .await
                        .map_err(|e| {
                            anyhow::anyhow!("Failed to recover migrated prepared intent: {e}")
                        })?;
                        anyhow::ensure!(
                            result.rows_affected() == 1,
                            "Prepared intent {} changed during verification migration",
                            record.intent_id
                        );
                    }

                    if let Some(status) = record.terminal_status {
                        anyhow::ensure!(
                            matches!(
                                status,
                                TransactionStatus::Finalized | TransactionStatus::Reverted
                            ),
                            "Migration terminal status is not finalized or reverted"
                        );
                        let transaction_hash =
                            record.transaction_hash.as_deref().ok_or_else(|| {
                                anyhow::anyhow!("Terminal migration record has no transaction hash")
                            })?;
                        let block_number = i64::try_from(record.block_number.ok_or_else(|| {
                            anyhow::anyhow!("Terminal migration record has no block number")
                        })?)
                        .context("Migration block exceeds PostgreSQL BIGINT")?;
                        let block_hash = record.block_hash.as_deref().ok_or_else(|| {
                            anyhow::anyhow!("Terminal migration record has no block hash")
                        })?;
                        let gas_used = i64::try_from(record.gas_used.ok_or_else(|| {
                            anyhow::anyhow!("Terminal migration record has no gas usage")
                        })?)
                        .context("Migration gas usage exceeds PostgreSQL BIGINT")?;
                        let receipt_success = record.receipt_success.ok_or_else(|| {
                            anyhow::anyhow!("Terminal migration record has no receipt status")
                        })?;
                        anyhow::ensure!(
                            receipt_success == (status == TransactionStatus::Finalized),
                            "Migration receipt status conflicts with terminal status"
                        );
                        let current_status = sqlx::query_scalar::<_, String>(
                            "SELECT status FROM execution_intent WHERE id = $1 FOR UPDATE",
                        )
                        .bind(record.intent_id)
                        .fetch_one(&mut *transaction)
                        .await
                        .map_err(|e| anyhow::anyhow!("Failed to lock migrated intent: {e}"))?;
                        let hash_result = sqlx::query(
                            "
                            UPDATE execution_transaction_hash
                            SET status = $3, block_number = $4, block_hash = $5,
                                receipt_success = $6, gas_used = $7,
                                effective_gas_price = $8, updated_at = NOW()
                            WHERE intent_id = $1 AND transaction_hash = $2 AND current
                              AND payload_expected
                            ",
                        )
                        .bind(record.intent_id)
                        .bind(transaction_hash)
                        .bind(status.as_str())
                        .bind(block_number)
                        .bind(block_hash)
                        .bind(receipt_success)
                        .bind(gas_used)
                        .bind(record.effective_gas_price.as_deref())
                        .execute(&mut *transaction)
                        .await
                        .map_err(|e| anyhow::anyhow!("Failed to reconstruct migrated hash: {e}"))?;
                        anyhow::ensure!(
                            hash_result.rows_affected() == 1,
                            "Migrated terminal transaction hash was not found"
                        );
                        sqlx::query(
                            "UPDATE execution_intent SET status = $2, updated_at = NOW() WHERE id = $1",
                        )
                        .bind(record.intent_id)
                        .bind(status.as_str())
                        .execute(&mut *transaction)
                        .await
                        .map_err(|e| anyhow::anyhow!("Failed to reconstruct migrated intent: {e}"))?;
                        if current_status != status.as_str() {
                            sqlx::query(
                                "
                                INSERT INTO execution_transaction_transition (
                                    intent_id, transaction_hash_id, transition_key,
                                    from_status, to_status, block_number, block_hash
                                )
                                SELECT $1, id, $3, $4, $5, $6, $7
                                FROM execution_transaction_hash
                                WHERE intent_id = $1 AND transaction_hash = $2
                                ",
                            )
                            .bind(record.intent_id)
                            .bind(transaction_hash)
                            .bind(format!("migration:{}:{transaction_hash}", status.as_str()))
                            .bind(current_status)
                            .bind(status.as_str())
                            .bind(block_number)
                            .bind(block_hash)
                            .execute(&mut *transaction)
                            .await
                            .map_err(|e| {
                                anyhow::anyhow!(
                                    "Failed to record migrated terminal transition: {e}"
                                )
                            })?;
                        }
                    }

                    let nonce = record
                        .nonce
                        .map(i64::try_from)
                        .transpose()
                        .context("Migration nonce exceeds PostgreSQL BIGINT")?;

                    for (index, decision) in record.decisions.iter().enumerate() {
                        let height_start = decision
                            .height_start
                            .map(i64::try_from)
                            .transpose()
                            .context("Migration evidence height exceeds PostgreSQL BIGINT")?;
                        let height_end = decision
                            .height_end
                            .map(i64::try_from)
                            .transpose()
                            .context("Migration evidence height exceeds PostgreSQL BIGINT")?;
                        sqlx::query(
                            "
                            INSERT INTO execution_verification_decision (
                                intent_id, nonce, decision_class, read_class,
                                height_start, height_end, manifest_version, manifest_digest,
                                provider_ids, operator_ids, failure_domain_ids,
                                response_class, normalized_value_digest, nonce_revision,
                                outcome, transition_key
                            )
                            VALUES (
                                $1, $2, 'migration', $3, $4, $5, $6, $7, $8, $9, $10,
                                'all_valid', $11, 0, 'verified', $12
                            )
                            ",
                        )
                        .bind(record.intent_id)
                        .bind(nonce)
                        .bind(decision.read_class)
                        .bind(height_start)
                        .bind(height_end)
                        .bind(bootstrap.manifest_version)
                        .bind(bootstrap.manifest_digest)
                        .bind(bootstrap.provider_ids)
                        .bind(bootstrap.operator_ids)
                        .bind(bootstrap.failure_domain_ids)
                        .bind(&decision.normalized_value_digest)
                        .bind(format!("migration:{}:{index}", record.intent_id))
                        .execute(&mut *transaction)
                        .await
                        .map_err(|e| {
                            anyhow::anyhow!("Failed to persist migration evidence: {e}")
                        })?;
                    }
                }
            }
            0
        };

        sqlx::query(
            "
            INSERT INTO execution_verified_finalized_header (
                chain_id, wallet_address, number, hash, parent_hash, timestamp,
                base_fee_per_gas, manifest_digest
            )
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
            ON CONFLICT (chain_id, wallet_address, number) DO NOTHING
            ",
        )
        .bind(chain_id)
        .bind(bootstrap.wallet_address)
        .bind(checkpoint_number)
        .bind(bootstrap.checkpoint_hash)
        .bind(bootstrap.checkpoint_parent_hash)
        .bind(checkpoint_timestamp)
        .bind(base_fee)
        .bind(bootstrap.manifest_digest)
        .execute(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to initialize finalized header ledger: {e}"))?;
        let stored_checkpoint = sqlx::query_as::<_, (String, String, i64, Option<String>)>(
            "
            SELECT hash, parent_hash, timestamp, base_fee_per_gas
            FROM execution_verified_finalized_header
            WHERE chain_id = $1 AND wallet_address = $2 AND number = $3
            ",
        )
        .bind(chain_id)
        .bind(bootstrap.wallet_address)
        .bind(checkpoint_number)
        .fetch_one(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to validate finalized checkpoint ledger: {e}"))?;
        anyhow::ensure!(
            stored_checkpoint
                == (
                    bootstrap.checkpoint_hash.to_string(),
                    bootstrap.checkpoint_parent_hash.to_string(),
                    checkpoint_timestamp,
                    bootstrap
                        .checkpoint_base_fee_per_gas
                        .map(|value| value.to_string()),
                ),
            "Finalized checkpoint ledger conflicts with the trusted chain anchor"
        );

        if initialized {
            let stored_tip =
                sqlx::query_as::<_, (i64, String, String, i64, Option<String>, String)>(
                    "
                SELECT number, hash, parent_hash, timestamp, base_fee_per_gas, manifest_digest
                FROM execution_verified_finalized_header
                WHERE chain_id = $1 AND wallet_address = $2
                ORDER BY number DESC
                LIMIT 1
                ",
                )
                .bind(chain_id)
                .bind(bootstrap.wallet_address)
                .fetch_one(&mut *transaction)
                .await
                .map_err(|e| anyhow::anyhow!("Failed to lock finalized header tip: {e}"))?;
            anyhow::ensure!(
                stored_tip
                    == (
                        i64::try_from(first_header.number)
                            .context("Verified finalized height exceeds PostgreSQL BIGINT")?,
                        first_header.hash.clone(),
                        first_header.parent_hash.clone(),
                        i64::try_from(first_header.timestamp)
                            .context("Verified finalized timestamp exceeds PostgreSQL BIGINT")?,
                        first_header.base_fee_per_gas.map(|value| value.to_string()),
                        bootstrap.manifest_digest.to_string(),
                    ),
                "Verified finalized header extension does not start at the durable tip"
            );
        } else {
            anyhow::ensure!(
                first_header.number == bootstrap.checkpoint_number
                    && first_header.hash == bootstrap.checkpoint_hash
                    && first_header.parent_hash == bootstrap.checkpoint_parent_hash
                    && first_header.timestamp == bootstrap.checkpoint_timestamp
                    && first_header.base_fee_per_gas == bootstrap.checkpoint_base_fee_per_gas,
                "Verified finalized headers do not start at the trusted checkpoint"
            );
        }

        for header in bootstrap.finalized_headers.iter().skip(1) {
            let number = i64::try_from(header.number)
                .context("Verified finalized height exceeds PostgreSQL BIGINT")?;
            let timestamp = i64::try_from(header.timestamp)
                .context("Verified finalized timestamp exceeds PostgreSQL BIGINT")?;
            let base_fee = header.base_fee_per_gas.map(|value| value.to_string());
            sqlx::query(
                "
                INSERT INTO execution_verified_finalized_header (
                    chain_id, wallet_address, number, hash, parent_hash, timestamp,
                    base_fee_per_gas, manifest_digest
                )
                VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
                ON CONFLICT (chain_id, wallet_address, number) DO NOTHING
                ",
            )
            .bind(chain_id)
            .bind(bootstrap.wallet_address)
            .bind(number)
            .bind(&header.hash)
            .bind(&header.parent_hash)
            .bind(timestamp)
            .bind(&base_fee)
            .bind(bootstrap.manifest_digest)
            .execute(&mut *transaction)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to extend finalized header ledger: {e}"))?;
            let stored = sqlx::query_as::<_, (String, String, i64, Option<String>, String)>(
                "
                SELECT hash, parent_hash, timestamp, base_fee_per_gas, manifest_digest
                FROM execution_verified_finalized_header
                WHERE chain_id = $1 AND wallet_address = $2 AND number = $3
                ",
            )
            .bind(chain_id)
            .bind(bootstrap.wallet_address)
            .bind(number)
            .fetch_one(&mut *transaction)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to validate finalized header ledger: {e}"))?;
            anyhow::ensure!(
                stored
                    == (
                        header.hash.clone(),
                        header.parent_hash.clone(),
                        timestamp,
                        base_fee,
                        bootstrap.manifest_digest.to_string(),
                    ),
                "Finalized header ledger conflicts at height {}",
                header.number
            );
        }

        let finalized_height = bootstrap
            .finalized_headers
            .last()
            .expect("verified finalized headers are nonempty")
            .number;

        for (index, decision) in bootstrap.decisions.iter().enumerate() {
            let height_start = decision
                .height_start
                .map(i64::try_from)
                .transpose()
                .context("Connect verification height exceeds PostgreSQL BIGINT")?;
            let height_end = decision
                .height_end
                .map(i64::try_from)
                .transpose()
                .context("Connect verification height exceeds PostgreSQL BIGINT")?;
            let transition_key = format!("connect:{finalized_height}:{revision}:{index}");
            sqlx::query(
                "
                INSERT INTO execution_verification_decision (
                    intent_id, nonce, decision_class, read_class, height_start, height_end,
                    manifest_version, manifest_digest, provider_ids, operator_ids,
                    failure_domain_ids, response_class, normalized_value_digest,
                    nonce_revision, outcome, transition_key
                )
                VALUES (
                    NULL, NULL, 'connect', $1, $2, $3, $4, $5, $6, $7, $8,
                    'all_valid', $9, $10, 'verified', $11
                )
                ",
            )
            .bind(decision.read_class)
            .bind(height_start)
            .bind(height_end)
            .bind(bootstrap.manifest_version)
            .bind(bootstrap.manifest_digest)
            .bind(bootstrap.provider_ids)
            .bind(bootstrap.operator_ids)
            .bind(bootstrap.failure_domain_ids)
            .bind(&decision.normalized_value_digest)
            .bind(revision)
            .bind(transition_key)
            .execute(&mut *transaction)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to persist connect verification: {e}"))?;
        }

        for statement in [
            "
            CREATE OR REPLACE FUNCTION execution_verification_append_only()
            RETURNS TRIGGER AS $$
            BEGIN
                RAISE EXCEPTION 'Execution verification evidence is append-only';
            END;
            $$ LANGUAGE plpgsql
            ",
            "DROP TRIGGER IF EXISTS execution_verification_decision_append_only \
             ON execution_verification_decision",
            "
            CREATE TRIGGER execution_verification_decision_append_only
            BEFORE UPDATE OR DELETE ON execution_verification_decision
            FOR EACH STATEMENT EXECUTE FUNCTION execution_verification_append_only()
            ",
            "DROP TRIGGER IF EXISTS execution_finalized_header_append_only \
             ON execution_verified_finalized_header",
            "
            CREATE TRIGGER execution_finalized_header_append_only
            BEFORE UPDATE OR DELETE ON execution_verified_finalized_header
            FOR EACH STATEMENT EXECUTE FUNCTION execution_verification_append_only()
            ",
        ] {
            sqlx::query(statement)
                .execute(&mut *transaction)
                .await
                .map_err(|e| anyhow::anyhow!("Failed to activate verification schema: {e}"))?;
        }
        sqlx::query(
            "
            INSERT INTO execution_schema_version (component, version)
            VALUES ('evm_execution_verification', $1)
            ON CONFLICT (component) DO UPDATE SET version = EXCLUDED.version
            WHERE execution_schema_version.version <= EXCLUDED.version
            ",
        )
        .bind(VERIFICATION_SCHEMA_VERSION)
        .execute(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to activate verification schema: {e}"))?;

        transaction
            .commit()
            .await
            .map_err(|e| anyhow::anyhow!("Failed to commit verification migration: {e}"))?;
        Ok(())
    }

    /// Activates or resumes protected signed-transaction storage for this database.
    pub(crate) async fn ensure_execution_payload_storage(
        &self,
        keys: &PayloadKeySet,
    ) -> anyhow::Result<()> {
        let marker = self.execution_payload_marker().await?;
        match marker {
            None => self.activate_execution_payload_storage(keys).await?,
            Some(version) => anyhow::ensure!(
                version == EXECUTION_PAYLOAD_PROTOCOL_VERSION,
                "Execution payload protection version {version} is newer than supported version {EXECUTION_PAYLOAD_PROTOCOL_VERSION}"
            ),
        }

        loop {
            let state = self.execution_payload_state().await?.ok_or_else(|| {
                anyhow::anyhow!("Execution payload marker exists without durable state")
            })?;
            validate_execution_payload_state(&state, keys)?;
            match state.operation.as_str() {
                "migrate" => {
                    if self
                        .migrate_execution_payload_batch(keys, EXECUTION_PAYLOAD_BATCH_SIZE)
                        .await?
                    {
                        break;
                    }
                }
                "ready" => break,
                operation => anyhow::bail!(
                    "Execution payload storage is in {operation} maintenance; complete or roll back that operation before connecting"
                ),
            }
        }

        self.validate_execution_payload_ready(keys).await
    }

    /// Requires ready protected storage and authenticates every persisted signed payload.
    pub(crate) async fn require_execution_payload_storage(
        &self,
        keys: &PayloadKeySet,
        policy: PayloadPolicy,
        batch_size: i64,
    ) -> anyhow::Result<ExecutionPayloadLease> {
        self.require_execution_payload_storage_ready(keys).await?;
        let (check, transaction) = self
            .inspect_execution_payload_storage(Some(keys), Some(policy), batch_size)
            .await?;
        anyhow::ensure!(
            check.protected,
            "Postgres execution requires protected payload storage"
        );
        Ok(ExecutionPayloadLease {
            _transaction: transaction,
        })
    }

    /// Requires protected storage to be ready before execution schema initialization.
    pub(crate) async fn require_execution_payload_storage_ready(
        &self,
        keys: &PayloadKeySet,
    ) -> anyhow::Result<()> {
        anyhow::ensure!(
            self.execution_payload_marker().await?.is_some(),
            "Postgres execution requires protected payload storage"
        );
        self.validate_execution_payload_ready(keys).await
    }

    /// Acquires a shared state lease for signing, payload access, acknowledgment, or broadcast.
    pub(crate) async fn acquire_execution_payload_lease(
        &self,
        keys: &PayloadKeySet,
    ) -> anyhow::Result<ExecutionPayloadLease> {
        let mut transaction = self
            .pool
            .begin()
            .await
            .context("failed to start execution payload action lease")?;
        let marker = sqlx::query_scalar::<_, i16>(
            "SELECT version FROM execution_schema_version WHERE component = $1",
        )
        .bind(EXECUTION_PAYLOAD_COMPONENT)
        .fetch_optional(&mut *transaction)
        .await
        .context("failed to read execution payload marker")?;

        let version = marker.ok_or_else(|| {
            anyhow::anyhow!("Postgres execution requires protected payload storage")
        })?;
        anyhow::ensure!(
            version == EXECUTION_PAYLOAD_PROTOCOL_VERSION,
            "Unsupported execution payload protection version {version}"
        );
        let row = sqlx::query(
            "SELECT deployment_id, protocol_version, operation, active_key_id \
             FROM execution_payload_state WHERE component = 'signed_transactions' \
             FOR SHARE",
        )
        .fetch_optional(&mut *transaction)
        .await
        .context("failed to lock execution payload state")?
        .ok_or_else(|| anyhow::anyhow!("Execution payload marker exists without state"))?;
        let state = execution_payload_state_from_row(&row)?;
        validate_execution_payload_state(&state, keys)?;
        anyhow::ensure!(
            state.operation == "ready",
            "Execution payload storage is in {} maintenance",
            state.operation
        );

        Ok(ExecutionPayloadLease {
            _transaction: transaction,
        })
    }

    /// Reserves one nonce use under the active payload key.
    pub(crate) async fn reserve_execution_payload_seal(
        &self,
        keys: &PayloadKeySet,
    ) -> anyhow::Result<()> {
        let mut transaction = self
            .pool
            .begin()
            .await
            .context("failed to start payload seal reservation")?;
        let row = sqlx::query(
            "SELECT deployment_id, protocol_version, operation, active_key_id \
             FROM execution_payload_state WHERE component = 'signed_transactions' FOR SHARE",
        )
        .fetch_optional(&mut *transaction)
        .await
        .context("failed to lock execution payload state")?
        .ok_or_else(|| anyhow::anyhow!("Execution payload protection is not active"))?;
        let state = execution_payload_state_from_row(&row)?;
        validate_execution_payload_state(&state, keys)?;
        anyhow::ensure!(
            state.operation == "ready",
            "Execution payload storage is in {} maintenance",
            state.operation
        );
        reserve_execution_payload_seal(&mut transaction, keys.active_key_id()).await?;
        transaction
            .commit()
            .await
            .context("failed to commit payload seal reservation")?;
        Ok(())
    }

    async fn execution_payload_marker(&self) -> anyhow::Result<Option<i16>> {
        sqlx::query_scalar::<_, i16>(
            "SELECT version FROM execution_schema_version WHERE component = $1",
        )
        .bind(EXECUTION_PAYLOAD_COMPONENT)
        .fetch_optional(&self.pool)
        .await
        .context("failed to read execution payload marker")
    }

    async fn execution_payload_state(&self) -> anyhow::Result<Option<ExecutionPayloadState>> {
        let row = sqlx::query(
            "SELECT deployment_id, protocol_version, operation, active_key_id \
             FROM execution_payload_state WHERE component = 'signed_transactions'",
        )
        .fetch_optional(&self.pool)
        .await
        .context("failed to read execution payload state")?;
        row.as_ref()
            .map(execution_payload_state_from_row)
            .transpose()
    }

    async fn activate_execution_payload_storage(&self, keys: &PayloadKeySet) -> anyhow::Result<()> {
        let mut transaction = self
            .pool
            .begin()
            .await
            .context("failed to start execution payload activation")?;
        lock_execution_payload_operation(&mut transaction).await?;
        sqlx::query("LOCK TABLE execution_transaction_hash IN ACCESS EXCLUSIVE MODE")
            .execute(&mut *transaction)
            .await
            .context("failed to fence execution payload writers")?;
        let marker = sqlx::query_scalar::<_, i16>(
            "SELECT version FROM execution_schema_version WHERE component = $1",
        )
        .bind(EXECUTION_PAYLOAD_COMPONENT)
        .fetch_optional(&mut *transaction)
        .await
        .context("failed to recheck execution payload marker")?;
        if marker.is_some() {
            transaction
                .rollback()
                .await
                .context("failed to release concurrent payload activation")?;
            return Ok(());
        }
        let sealed_rows = sqlx::query_scalar::<_, i64>(
            "SELECT COUNT(*) FROM execution_transaction_hash WHERE sealed_transaction IS NOT NULL",
        )
        .fetch_one(&mut *transaction)
        .await
        .context("failed to inspect pre-activation envelopes")?;
        anyhow::ensure!(
            sealed_rows == 0,
            "Unprotected execution payload storage contains {sealed_rows} sealed row(s)"
        );

        for statement in [
            "
            CREATE OR REPLACE FUNCTION execution_transaction_payload_fence()
            RETURNS TRIGGER AS $$
            DECLARE payload_operation TEXT;
            BEGIN
                IF NEW.raw_transaction IS NOT NULL
                   AND (TG_OP = 'INSERT'
                        OR OLD.raw_transaction IS NULL
                        OR NEW.raw_transaction IS DISTINCT FROM OLD.raw_transaction) THEN
                    SELECT operation INTO payload_operation
                    FROM execution_payload_state
                    WHERE component = 'signed_transactions';
                    IF NOT (
                        TG_OP = 'UPDATE'
                        AND payload_operation = 'rollback'
                        AND OLD.payload_expected
                        AND OLD.raw_transaction IS NULL
                        AND NEW.raw_transaction IS NOT NULL
                        AND OLD.sealed_transaction IS NOT NULL
                        AND NEW.sealed_transaction = OLD.sealed_transaction
                    ) THEN
                        RAISE EXCEPTION 'Plaintext signed transaction writes are disabled';
                    END IF;
                END IF;
                RETURN NEW;
            END;
            $$ LANGUAGE plpgsql
            ",
            "DROP TRIGGER IF EXISTS execution_transaction_payload_fence ON execution_transaction_hash",
            "
            CREATE TRIGGER execution_transaction_payload_fence
            BEFORE INSERT OR UPDATE ON execution_transaction_hash
            FOR EACH ROW EXECUTE FUNCTION execution_transaction_payload_fence()
            ",
        ] {
            sqlx::query(statement)
                .execute(&mut *transaction)
                .await
                .context("failed to install execution payload write fence")?;
        }
        sqlx::query("INSERT INTO execution_schema_version (component, version) VALUES ($1, $2)")
            .bind(EXECUTION_PAYLOAD_COMPONENT)
            .bind(EXECUTION_PAYLOAD_PROTOCOL_VERSION)
            .execute(&mut *transaction)
            .await
            .context("failed to record execution payload marker")?;
        sqlx::query(
            "INSERT INTO execution_payload_state (component, deployment_id, protocol_version, operation, active_key_id) \
             VALUES ('signed_transactions', $1, $2, 'migrate', $3)",
        )
        .bind(keys.deployment_id())
        .bind(EXECUTION_PAYLOAD_PROTOCOL_VERSION)
        .bind(keys.active_key_id().as_slice())
        .execute(&mut *transaction)
        .await
        .context("failed to record execution payload migration state")?;
        sqlx::query(
            "INSERT INTO execution_payload_key_state (key_id, seals) VALUES ($1, 0) \
             ON CONFLICT (key_id) DO NOTHING",
        )
        .bind(keys.active_key_id().as_slice())
        .execute(&mut *transaction)
        .await
        .context("failed to initialize execution payload key state")?;
        transaction
            .commit()
            .await
            .context("failed to commit execution payload activation")?;
        Ok(())
    }

    async fn migrate_execution_payload_batch(
        &self,
        keys: &PayloadKeySet,
        batch_size: i64,
    ) -> anyhow::Result<bool> {
        anyhow::ensure!(
            batch_size > 0,
            "Execution payload batch size must be positive"
        );
        let mut transaction = self
            .pool
            .begin()
            .await
            .context("failed to start execution payload migration batch")?;
        lock_execution_payload_operation(&mut transaction).await?;
        let state_row = sqlx::query(
            "SELECT deployment_id, protocol_version, operation, active_key_id \
             FROM execution_payload_state WHERE component = 'signed_transactions' FOR UPDATE",
        )
        .fetch_optional(&mut *transaction)
        .await
        .context("failed to lock execution payload migration state")?
        .ok_or_else(|| anyhow::anyhow!("Execution payload migration state is missing"))?;
        let state = execution_payload_state_from_row(&state_row)?;
        validate_execution_payload_state(&state, keys)?;
        anyhow::ensure!(
            state.operation == "migrate",
            "Execution payload storage is {}, not migrating",
            state.operation
        );

        let rows = sqlx::query_as::<_, ExecutionTransactionHashRow>(
            "
            SELECT
                id, intent_id, chain_id, transaction_hash, payload_expected,
                raw_transaction, sealed_transaction, status, block_number, block_hash,
                receipt_success, gas_used, effective_gas_price, current
            FROM execution_transaction_hash
            WHERE payload_expected AND raw_transaction IS NOT NULL
            ORDER BY id
            LIMIT $1
            FOR UPDATE
            ",
        )
        .bind(batch_size)
        .fetch_all(&mut *transaction)
        .await
        .context("failed to load legacy signed transaction batch")?;

        if rows.is_empty() {
            self.complete_execution_payload_migration(&mut transaction, keys)
                .await?;
            transaction
                .commit()
                .await
                .context("failed to commit execution payload migration completion")?;
            return Ok(true);
        }

        for hash in rows {
            anyhow::ensure!(
                hash.sealed_transaction.is_none(),
                "Execution transaction {} contains both plaintext and sealed payloads",
                hash.id
            );
            let raw_transaction = hash
                .raw_transaction
                .as_deref()
                .expect("migration query requires plaintext");
            let intent = load_execution_intent(&mut transaction, hash.intent_id).await?;
            let context = authenticate_retained_payload(
                raw_transaction,
                &intent,
                &hash,
                keys.deployment_id(),
            )
            .with_context(|| format!("failed to authenticate execution payload {}", hash.id))?;
            reserve_execution_payload_seal(&mut transaction, keys.active_key_id()).await?;
            let envelope = keys.seal(raw_transaction, &context)?;
            let unsealed = keys.unseal(&envelope, &context)?;
            authenticate_retained_payload(&unsealed, &intent, &hash, keys.deployment_id())?;
            anyhow::ensure!(
                unsealed == raw_transaction,
                "Execution payload {} changed during seal round trip",
                hash.id
            );
            let result = sqlx::query(
                "UPDATE execution_transaction_hash \
                 SET sealed_transaction = $2, raw_transaction = NULL, updated_at = NOW() \
                 WHERE id = $1 AND raw_transaction = $3 AND sealed_transaction IS NULL",
            )
            .bind(hash.id)
            .bind(&envelope)
            .bind(raw_transaction)
            .execute(&mut *transaction)
            .await
            .context("failed to promote execution payload")?;
            anyhow::ensure!(
                result.rows_affected() == 1,
                "Execution payload {} changed during migration",
                hash.id
            );
            sqlx::query(
                "UPDATE execution_payload_state SET progress_id = $1, updated_at = NOW() \
                 WHERE component = 'signed_transactions'",
            )
            .bind(hash.id)
            .execute(&mut *transaction)
            .await
            .context("failed to record execution payload migration progress")?;
        }

        transaction
            .commit()
            .await
            .context("failed to commit execution payload migration batch")?;
        Ok(false)
    }

    async fn complete_execution_payload_migration(
        &self,
        transaction: &mut Transaction<'_, Postgres>,
        keys: &PayloadKeySet,
    ) -> anyhow::Result<()> {
        sqlx::query("LOCK TABLE execution_transaction_hash IN SHARE ROW EXCLUSIVE MODE")
            .execute(&mut **transaction)
            .await
            .context("failed to lock execution payload migration completion")?;
        let invalid = sqlx::query_scalar::<_, i64>(
            "SELECT COUNT(*) FROM execution_transaction_hash \
             WHERE (payload_expected AND (raw_transaction IS NOT NULL OR sealed_transaction IS NULL)) \
                OR (NOT payload_expected AND (raw_transaction IS NOT NULL OR sealed_transaction IS NOT NULL))",
        )
        .fetch_one(&mut **transaction)
        .await
        .context("failed to verify migrated execution payload representations")?;
        anyhow::ensure!(
            invalid == 0,
            "Execution payload migration found {invalid} invalid representation(s)"
        );
        validate_execution_payload_key_inventory(transaction, keys).await?;

        for statement in [
            "ALTER TABLE execution_transaction_hash \
             DROP CONSTRAINT IF EXISTS execution_transaction_payload_protected_check",
            "ALTER TABLE execution_transaction_hash \
             ADD CONSTRAINT execution_transaction_payload_protected_check CHECK ( \
                 (payload_expected AND raw_transaction IS NULL AND sealed_transaction IS NOT NULL) \
                 OR (NOT payload_expected AND raw_transaction IS NULL AND sealed_transaction IS NULL) \
             )",
        ] {
            sqlx::query(statement)
                .execute(&mut **transaction)
                .await
                .context("failed to install protected execution payload constraint")?;
        }
        sqlx::query(
            "UPDATE execution_payload_state SET operation = 'ready', updated_at = NOW() \
             WHERE component = 'signed_transactions' AND operation = 'migrate'",
        )
        .execute(&mut **transaction)
        .await
        .context("failed to mark execution payload storage ready")?;
        Ok(())
    }

    async fn validate_execution_payload_ready(&self, keys: &PayloadKeySet) -> anyhow::Result<()> {
        let mut transaction = self
            .pool
            .begin()
            .await
            .context("failed to start execution payload validation")?;
        let state_row = sqlx::query(
            "SELECT deployment_id, protocol_version, operation, active_key_id \
             FROM execution_payload_state WHERE component = 'signed_transactions' FOR SHARE",
        )
        .fetch_optional(&mut *transaction)
        .await
        .context("failed to lock execution payload state")?
        .ok_or_else(|| anyhow::anyhow!("Execution payload state is missing"))?;
        let state = execution_payload_state_from_row(&state_row)?;
        validate_execution_payload_state(&state, keys)?;
        anyhow::ensure!(
            state.operation == "ready",
            "Execution payload storage is not ready"
        );
        let (payload_fence, payload_constraint) =
            sqlx::query_as::<_, (bool, bool)>(EXECUTION_PAYLOAD_INFRASTRUCTURE_QUERY)
                .fetch_one(&mut *transaction)
                .await
                .context("failed to inspect execution payload protection infrastructure")?;
        anyhow::ensure!(
            payload_fence && payload_constraint,
            "Execution payload storage is marked ready without its write fence or constraint"
        );
        validate_execution_payload_key_inventory(&mut transaction, keys).await?;
        transaction
            .commit()
            .await
            .context("failed to complete execution payload validation")?;
        Ok(())
    }

    pub(crate) async fn check_execution_payload_storage(
        &self,
        keys: Option<&PayloadKeySet>,
        policy: Option<PayloadPolicy>,
        batch_size: i64,
    ) -> anyhow::Result<ExecutionPayloadCheck> {
        let (check, transaction) = self
            .inspect_execution_payload_storage(keys, policy, batch_size)
            .await?;
        transaction
            .commit()
            .await
            .context("failed to complete execution payload check")?;
        Ok(check)
    }

    async fn inspect_execution_payload_storage(
        &self,
        keys: Option<&PayloadKeySet>,
        policy: Option<PayloadPolicy>,
        batch_size: i64,
    ) -> anyhow::Result<(ExecutionPayloadCheck, Transaction<'static, Postgres>)> {
        anyhow::ensure!(
            batch_size > 0,
            "Execution payload check batch size must be positive"
        );
        let mut transaction = self
            .pool
            .begin()
            .await
            .context("failed to start execution payload check")?;
        sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ")
            .execute(&mut *transaction)
            .await
            .context("failed to stabilize execution payload snapshot")?;
        sqlx::query("LOCK TABLE execution_transaction_hash IN SHARE MODE")
            .execute(&mut *transaction)
            .await
            .context("failed to stabilize execution payload check")?;
        let marker = sqlx::query_scalar::<_, i16>(
            "SELECT version FROM execution_schema_version WHERE component = $1",
        )
        .bind(EXECUTION_PAYLOAD_COMPONENT)
        .fetch_optional(&mut *transaction)
        .await
        .context("failed to read execution payload marker")?;
        let deployment_id = match (marker, keys) {
            (None, _) => {
                let state = sqlx::query_scalar::<_, bool>(
                    "SELECT EXISTS (SELECT 1 FROM execution_payload_state) \
                     OR EXISTS (SELECT 1 FROM execution_transaction_hash \
                                WHERE sealed_transaction IS NOT NULL)",
                )
                .fetch_one(&mut *transaction)
                .await
                .context("failed to check legacy payload state")?;
                anyhow::ensure!(!state, "Legacy payload storage contains protected state");
                None
            }
            (Some(version), Some(keys)) => {
                anyhow::ensure!(
                    version == EXECUTION_PAYLOAD_PROTOCOL_VERSION,
                    "Unsupported execution payload protection version {version}"
                );
                let state_row = sqlx::query(
                    "SELECT deployment_id, protocol_version, operation, active_key_id \
                     FROM execution_payload_state WHERE component = 'signed_transactions' FOR SHARE",
                )
                .fetch_optional(&mut *transaction)
                .await
                .context("failed to lock execution payload state")?
                .ok_or_else(|| anyhow::anyhow!("Execution payload state is missing"))?;
                let state = execution_payload_state_from_row(&state_row)?;
                validate_execution_payload_state(&state, keys)?;
                anyhow::ensure!(
                    state.operation == "ready",
                    "Execution payload storage is not ready"
                );
                Some(state.deployment_id)
            }
            (Some(_), None) => anyhow::bail!(
                "Execution payload protection is active, but no payload key is configured"
            ),
        };

        let mut cursor = 0_i64;
        let mut plaintext_rows = 0_u64;
        let mut original_rows = 0_u64;
        let mut replacement_rows = 0_u64;
        let mut authenticated_rows = 0_u64;
        let mut key_ids = BTreeSet::new();

        loop {
            let rows = sqlx::query_as::<_, ExecutionTransactionHashRow>(
                "
                SELECT
                    id, intent_id, chain_id, transaction_hash, payload_expected,
                    raw_transaction, sealed_transaction, status, block_number, block_hash,
                    receipt_success, gas_used, effective_gas_price, current
                FROM execution_transaction_hash
                WHERE id > $1
                ORDER BY id
                LIMIT $2
                ",
            )
            .bind(cursor)
            .bind(batch_size)
            .fetch_all(&mut *transaction)
            .await
            .context("failed to load execution payload check batch")?;

            if rows.is_empty() {
                break;
            }

            for hash in &rows {
                cursor = hash.id;
                plaintext_rows += u64::from(hash.raw_transaction.is_some());
                if !hash.payload_expected {
                    replacement_rows += 1;
                    anyhow::ensure!(
                        hash.raw_transaction.is_none() && hash.sealed_transaction.is_none(),
                        "Replacement execution transaction {} retains signed bytes",
                        hash.id
                    );
                    continue;
                }
                original_rows += 1;
                let intent = load_execution_intent(&mut transaction, hash.intent_id).await?;
                let raw_transaction = if let (Some(keys), Some(deployment_id)) =
                    (keys, deployment_id.as_deref())
                {
                    anyhow::ensure!(
                        hash.raw_transaction.is_none(),
                        "Protected execution transaction {} contains plaintext",
                        hash.id
                    );
                    let envelope = hash.sealed_transaction.as_deref().ok_or_else(|| {
                        anyhow::anyhow!(
                            "Protected execution transaction {} has no envelope",
                            hash.id
                        )
                    })?;
                    let key_id = envelope_key_id(envelope)?;
                    anyhow::ensure!(
                        keys.contains_key(&key_id),
                        "Execution transaction {} requires an unavailable payload key",
                        hash.id
                    );
                    key_ids.insert(alloy::hex::encode(key_id));
                    let context = payload_context(&intent, hash, deployment_id)?;
                    keys.unseal(envelope, &context)?
                } else {
                    anyhow::ensure!(
                        hash.sealed_transaction.is_none(),
                        "Legacy execution transaction {} contains an envelope",
                        hash.id
                    );
                    hash.raw_transaction.clone().ok_or_else(|| {
                        anyhow::anyhow!("Legacy execution transaction {} has no plaintext", hash.id)
                    })?
                };
                authenticate_retained_payload(
                    &raw_transaction,
                    &intent,
                    hash,
                    deployment_id.as_deref().unwrap_or(""),
                )?;

                if let Some(policy) = policy
                    && retained_payload_requires_policy(&intent, hash, policy)?
                {
                    authenticate_payload(
                        &raw_transaction,
                        &intent,
                        hash,
                        policy,
                        deployment_id.as_deref().unwrap_or(""),
                    )
                    .with_context(|| {
                        format!(
                            "execution intent {} transaction {} violates current execution policy",
                            intent.id, hash.transaction_hash
                        )
                    })?;
                }
                authenticated_rows += 1;
            }
        }
        let read_roles = sqlx::query_scalar::<_, String>(
            "SELECT DISTINCT role_name FROM ( \
                 SELECT tableowner AS role_name FROM pg_tables \
                 WHERE schemaname = current_schema() AND tablename = 'execution_transaction_hash' \
                 UNION \
                 SELECT grantee AS role_name FROM information_schema.role_table_grants \
                 WHERE table_schema = current_schema() \
                   AND table_name = 'execution_transaction_hash' \
                   AND privilege_type = 'SELECT' \
             ) AS roles ORDER BY role_name",
        )
        .fetch_all(&mut *transaction)
        .await
        .context("failed to inspect execution payload read roles")?;
        Ok((
            ExecutionPayloadCheck {
                protected: deployment_id.is_some(),
                deployment_id,
                plaintext_rows,
                original_rows,
                replacement_rows,
                authenticated_rows,
                key_ids: key_ids.into_iter().collect(),
                read_roles,
            },
            transaction,
        ))
    }

    pub(crate) async fn rewrap_execution_payload_storage(
        &self,
        keys: &PayloadKeySet,
        batch_size: i64,
    ) -> anyhow::Result<()> {
        anyhow::ensure!(
            batch_size > 0,
            "Execution payload rewrap batch size must be positive"
        );
        self.begin_execution_payload_rewrap(keys).await?;

        loop {
            if self
                .rewrap_execution_payload_batch(keys, batch_size)
                .await?
            {
                return Ok(());
            }
        }
    }

    async fn begin_execution_payload_rewrap(&self, keys: &PayloadKeySet) -> anyhow::Result<()> {
        let mut transaction = self
            .pool
            .begin()
            .await
            .context("failed to start execution payload rewrap")?;
        lock_execution_payload_operation(&mut transaction).await?;
        let state_row = sqlx::query(
            "SELECT deployment_id, protocol_version, operation, active_key_id \
             FROM execution_payload_state WHERE component = 'signed_transactions' FOR UPDATE",
        )
        .fetch_optional(&mut *transaction)
        .await
        .context("failed to lock execution payload state for rewrap")?
        .ok_or_else(|| anyhow::anyhow!("Execution payload protection is not active"))?;
        let state = execution_payload_state_from_row(&state_row)?;
        anyhow::ensure!(
            state.protocol_version == EXECUTION_PAYLOAD_PROTOCOL_VERSION
                && state.deployment_id == keys.deployment_id(),
            "Execution payload rewrap context does not match this database"
        );

        match state.operation.as_str() {
            "ready" => {
                let current_id: [u8; 32] = state
                    .active_key_id
                    .as_slice()
                    .try_into()
                    .context("database active payload key ID is invalid")?;
                anyhow::ensure!(
                    keys.contains_key(&current_id),
                    "Current database payload key is not configured for rewrap"
                );
                validate_execution_payload_key_inventory(&mut transaction, keys).await?;
                if current_id == *keys.active_key_id() {
                    transaction
                        .commit()
                        .await
                        .context("failed to complete no-op execution payload rewrap")?;
                    return Ok(());
                }
                sqlx::query(
                    "UPDATE execution_payload_state \
                     SET operation = 'rewrap', active_key_id = $1, progress_id = 0, updated_at = NOW() \
                     WHERE component = 'signed_transactions'",
                )
                .bind(keys.active_key_id().as_slice())
                .execute(&mut *transaction)
                .await
                .context("failed to record execution payload rewrap target")?;
            }
            "rewrap" => validate_execution_payload_state(&state, keys)?,
            operation => anyhow::bail!(
                "Execution payload storage is in {operation} maintenance, not ready for rewrap"
            ),
        }
        sqlx::query(
            "INSERT INTO execution_payload_key_state (key_id, seals) VALUES ($1, 0) \
             ON CONFLICT (key_id) DO NOTHING",
        )
        .bind(keys.active_key_id().as_slice())
        .execute(&mut *transaction)
        .await
        .context("failed to initialize rewrap target key state")?;
        transaction
            .commit()
            .await
            .context("failed to commit execution payload rewrap state")?;
        Ok(())
    }

    async fn rewrap_execution_payload_batch(
        &self,
        keys: &PayloadKeySet,
        batch_size: i64,
    ) -> anyhow::Result<bool> {
        let mut transaction = self
            .pool
            .begin()
            .await
            .context("failed to start execution payload rewrap batch")?;
        lock_execution_payload_operation(&mut transaction).await?;
        let state_row = sqlx::query(
            "SELECT deployment_id, protocol_version, operation, active_key_id \
             FROM execution_payload_state WHERE component = 'signed_transactions' FOR UPDATE",
        )
        .fetch_optional(&mut *transaction)
        .await
        .context("failed to lock execution payload rewrap state")?
        .ok_or_else(|| anyhow::anyhow!("Execution payload rewrap state is missing"))?;
        let state = execution_payload_state_from_row(&state_row)?;
        validate_execution_payload_state(&state, keys)?;
        if state.operation == "ready" {
            transaction
                .commit()
                .await
                .context("failed to complete execution payload rewrap")?;
            return Ok(true);
        }
        anyhow::ensure!(
            state.operation == "rewrap",
            "Execution payload storage is not rewrapping"
        );
        let rows = sqlx::query_as::<_, ExecutionTransactionHashRow>(
            "
            SELECT
                id, intent_id, chain_id, transaction_hash, payload_expected,
                raw_transaction, sealed_transaction, status, block_number, block_hash,
                receipt_success, gas_used, effective_gas_price, current
            FROM execution_transaction_hash
            WHERE payload_expected
              AND substring(sealed_transaction FROM 2 FOR 32) <> $1
            ORDER BY id
            LIMIT $2
            FOR UPDATE
            ",
        )
        .bind(keys.active_key_id().as_slice())
        .bind(batch_size)
        .fetch_all(&mut *transaction)
        .await
        .context("failed to load execution payload rewrap batch")?;

        if rows.is_empty() {
            sqlx::query("LOCK TABLE execution_transaction_hash IN SHARE ROW EXCLUSIVE MODE")
                .execute(&mut *transaction)
                .await
                .context("failed to lock execution payload rewrap completion")?;
            let remaining = sqlx::query_scalar::<_, i64>(
                "SELECT COUNT(*) FROM execution_transaction_hash \
                 WHERE payload_expected \
                   AND (raw_transaction IS NOT NULL OR sealed_transaction IS NULL \
                        OR substring(sealed_transaction FROM 2 FOR 32) <> $1)",
            )
            .bind(keys.active_key_id().as_slice())
            .fetch_one(&mut *transaction)
            .await
            .context("failed to verify execution payload rewrap completion")?;
            anyhow::ensure!(
                remaining == 0,
                "Execution payload rewrap left {remaining} row(s)"
            );
            sqlx::query(
                "UPDATE execution_payload_state SET operation = 'ready', updated_at = NOW() \
                 WHERE component = 'signed_transactions' AND operation = 'rewrap'",
            )
            .execute(&mut *transaction)
            .await
            .context("failed to mark execution payload rewrap complete")?;
            transaction
                .commit()
                .await
                .context("failed to commit execution payload rewrap completion")?;
            return Ok(true);
        }

        for hash in rows {
            let envelope = hash.sealed_transaction.as_deref().ok_or_else(|| {
                anyhow::anyhow!(
                    "Execution payload {} has no envelope during rewrap",
                    hash.id
                )
            })?;
            anyhow::ensure!(
                hash.raw_transaction.is_none(),
                "Execution payload {} contains plaintext during rewrap",
                hash.id
            );
            let intent = load_execution_intent(&mut transaction, hash.intent_id).await?;
            let context = payload_context(&intent, &hash, keys.deployment_id())?;
            let raw_transaction = keys.unseal(envelope, &context)?;
            authenticate_retained_payload(&raw_transaction, &intent, &hash, keys.deployment_id())?;
            reserve_execution_payload_seal(&mut transaction, keys.active_key_id()).await?;
            let rewrapped = keys.seal(&raw_transaction, &context)?;
            let verified = keys.unseal(&rewrapped, &context)?;
            authenticate_retained_payload(&verified, &intent, &hash, keys.deployment_id())?;
            anyhow::ensure!(
                verified == raw_transaction,
                "Execution payload {} changed during rewrap",
                hash.id
            );
            let result = sqlx::query(
                "UPDATE execution_transaction_hash SET sealed_transaction = $2, updated_at = NOW() \
                 WHERE id = $1 AND sealed_transaction = $3 AND raw_transaction IS NULL",
            )
            .bind(hash.id)
            .bind(&rewrapped)
            .bind(envelope)
            .execute(&mut *transaction)
            .await
            .context("failed to persist rewrapped execution payload")?;
            anyhow::ensure!(
                result.rows_affected() == 1,
                "Execution payload {} changed during rewrap",
                hash.id
            );
            sqlx::query(
                "UPDATE execution_payload_state SET progress_id = $1, updated_at = NOW() \
                 WHERE component = 'signed_transactions'",
            )
            .bind(hash.id)
            .execute(&mut *transaction)
            .await
            .context("failed to record execution payload rewrap progress")?;
        }
        transaction
            .commit()
            .await
            .context("failed to commit execution payload rewrap batch")?;
        Ok(false)
    }

    pub(crate) async fn rollback_execution_payload_storage(
        &self,
        keys: &PayloadKeySet,
        batch_size: i64,
    ) -> anyhow::Result<()> {
        anyhow::ensure!(
            batch_size > 0,
            "Execution payload rollback batch size must be positive"
        );
        self.begin_execution_payload_rollback(keys).await?;

        loop {
            if self
                .rollback_execution_payload_batch(keys, batch_size)
                .await?
            {
                return Ok(());
            }
        }
    }

    async fn begin_execution_payload_rollback(&self, keys: &PayloadKeySet) -> anyhow::Result<()> {
        let mut transaction = self
            .pool
            .begin()
            .await
            .context("failed to start execution payload rollback")?;
        lock_execution_payload_operation(&mut transaction).await?;
        let state_row = sqlx::query(
            "SELECT deployment_id, protocol_version, operation, active_key_id \
             FROM execution_payload_state WHERE component = 'signed_transactions' FOR UPDATE",
        )
        .fetch_optional(&mut *transaction)
        .await
        .context("failed to lock execution payload state for rollback")?
        .ok_or_else(|| anyhow::anyhow!("Execution payload protection is not active"))?;
        let state = execution_payload_state_from_row(&state_row)?;
        validate_execution_payload_state(&state, keys)?;
        match state.operation.as_str() {
            "ready" => {
                sqlx::query(
                    "ALTER TABLE execution_transaction_hash \
                     DROP CONSTRAINT IF EXISTS execution_transaction_payload_protected_check",
                )
                .execute(&mut *transaction)
                .await
                .context("failed to open execution payload rollback representation")?;
                sqlx::query(
                    "UPDATE execution_payload_state \
                     SET operation = 'rollback', progress_id = 0, updated_at = NOW() \
                     WHERE component = 'signed_transactions'",
                )
                .execute(&mut *transaction)
                .await
                .context("failed to record execution payload rollback state")?;
            }
            "rollback" => {}
            operation => anyhow::bail!(
                "Execution payload storage is in {operation} maintenance, not ready for rollback"
            ),
        }
        transaction
            .commit()
            .await
            .context("failed to commit execution payload rollback state")?;
        Ok(())
    }

    async fn rollback_execution_payload_batch(
        &self,
        keys: &PayloadKeySet,
        batch_size: i64,
    ) -> anyhow::Result<bool> {
        let mut transaction = self
            .pool
            .begin()
            .await
            .context("failed to start execution payload rollback batch")?;
        lock_execution_payload_operation(&mut transaction).await?;
        let state_row = sqlx::query(
            "SELECT deployment_id, protocol_version, operation, active_key_id \
             FROM execution_payload_state WHERE component = 'signed_transactions' FOR UPDATE",
        )
        .fetch_optional(&mut *transaction)
        .await
        .context("failed to lock execution payload rollback state")?
        .ok_or_else(|| anyhow::anyhow!("Execution payload rollback state is missing"))?;
        let state = execution_payload_state_from_row(&state_row)?;
        validate_execution_payload_state(&state, keys)?;
        anyhow::ensure!(
            state.operation == "rollback",
            "Execution payload storage is not rolling back"
        );
        let rows = sqlx::query_as::<_, ExecutionTransactionHashRow>(
            "
            SELECT
                id, intent_id, chain_id, transaction_hash, payload_expected,
                raw_transaction, sealed_transaction, status, block_number, block_hash,
                receipt_success, gas_used, effective_gas_price, current
            FROM execution_transaction_hash
            WHERE payload_expected AND sealed_transaction IS NOT NULL
            ORDER BY id
            LIMIT $1
            FOR UPDATE
            ",
        )
        .bind(batch_size)
        .fetch_all(&mut *transaction)
        .await
        .context("failed to load execution payload rollback batch")?;

        if rows.is_empty() {
            sqlx::query("LOCK TABLE execution_transaction_hash IN SHARE ROW EXCLUSIVE MODE")
                .execute(&mut *transaction)
                .await
                .context("failed to lock execution payload rollback completion")?;
            let invalid = sqlx::query_scalar::<_, i64>(
                "SELECT COUNT(*) FROM execution_transaction_hash \
                 WHERE (payload_expected AND (raw_transaction IS NULL OR sealed_transaction IS NOT NULL)) \
                    OR (NOT payload_expected AND (raw_transaction IS NOT NULL OR sealed_transaction IS NOT NULL))",
            )
            .fetch_one(&mut *transaction)
            .await
            .context("failed to verify execution payload rollback completion")?;
            anyhow::ensure!(
                invalid == 0,
                "Execution payload rollback left {invalid} invalid row(s)"
            );

            for statement in [
                "DROP TRIGGER IF EXISTS execution_transaction_payload_fence ON execution_transaction_hash",
                "DELETE FROM execution_payload_state WHERE component = 'signed_transactions'",
                "DELETE FROM execution_schema_version WHERE component = 'evm_execution_payload'",
            ] {
                sqlx::query(statement)
                    .execute(&mut *transaction)
                    .await
                    .context("failed to complete execution payload rollback")?;
            }
            transaction
                .commit()
                .await
                .context("failed to commit execution payload rollback completion")?;
            return Ok(true);
        }

        for hash in rows {
            anyhow::ensure!(
                hash.raw_transaction.is_none(),
                "Execution payload {} contains both representations during rollback",
                hash.id
            );
            let envelope = hash
                .sealed_transaction
                .as_deref()
                .expect("rollback query requires envelope");
            let intent = load_execution_intent(&mut transaction, hash.intent_id).await?;
            let context = payload_context(&intent, &hash, keys.deployment_id())?;
            let raw_transaction = keys.unseal(envelope, &context)?;
            authenticate_retained_payload(&raw_transaction, &intent, &hash, keys.deployment_id())?;
            let result = sqlx::query(
                "UPDATE execution_transaction_hash SET raw_transaction = $2, updated_at = NOW() \
                 WHERE id = $1 AND raw_transaction IS NULL AND sealed_transaction = $3",
            )
            .bind(hash.id)
            .bind(&raw_transaction)
            .bind(envelope)
            .execute(&mut *transaction)
            .await
            .context("failed to recreate plaintext execution payload")?;
            anyhow::ensure!(
                result.rows_affected() == 1,
                "Execution payload {} changed during rollback",
                hash.id
            );
            authenticate_retained_payload(&raw_transaction, &intent, &hash, keys.deployment_id())?;
            let result = sqlx::query(
                "UPDATE execution_transaction_hash SET sealed_transaction = NULL, updated_at = NOW() \
                 WHERE id = $1 AND raw_transaction = $2 AND sealed_transaction = $3",
            )
            .bind(hash.id)
            .bind(&raw_transaction)
            .bind(envelope)
            .execute(&mut *transaction)
            .await
            .context("failed to clear rolled-back execution payload envelope")?;
            anyhow::ensure!(
                result.rows_affected() == 1,
                "Execution payload {} changed while clearing rollback envelope",
                hash.id
            );
            sqlx::query(
                "UPDATE execution_payload_state SET progress_id = $1, updated_at = NOW() \
                 WHERE component = 'signed_transactions'",
            )
            .bind(hash.id)
            .execute(&mut *transaction)
            .await
            .context("failed to record execution payload rollback progress")?;
        }
        transaction
            .commit()
            .await
            .context("failed to commit execution payload rollback batch")?;
        Ok(false)
    }

    /// Reserves durable ownership of a signer slot and optional client order before signing.
    ///
    /// # Errors
    ///
    /// Returns a stage-classified error if the signer or client order is already owned, or
    /// persistence fails. A commit-stage error does not prove the transaction rolled back.
    pub async fn reserve_execution_intent(
        &self,
        intent: &ExecutionIntentInsert,
    ) -> anyhow::Result<ExecutionIntentRow> {
        let (transaction, row) = async {
            let chain_id = i32::try_from(intent.chain_id).with_context(|| {
                format!("Chain ID {} exceeds PostgreSQL INTEGER", intent.chain_id)
            })?;
            let created_block = i64::try_from(intent.created_block).with_context(|| {
                format!(
                    "Execution creation block {} exceeds PostgreSQL BIGINT",
                    intent.created_block
                )
            })?;
            let mut transaction = self
                .pool
                .begin()
                .await
                .context("failed to start execution intent reservation")?;
            Self::lock_execution_signer(&mut transaction, intent.chain_id, &intent.wallet_address)
                .await?;
            let row = sqlx::query_as::<_, ExecutionIntentRow>(
                "
                INSERT INTO execution_intent (
                    schema_version, chain_id, wallet_address, purpose, status,
                    client_order_id, trader_id, strategy_id, account_id, instrument_id,
                    pool_address, transaction_to, transaction_input, transaction_value,
                    amount_in, created_block
                )
                VALUES ($1, $2, $3, $4, 'prepared', $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)
                RETURNING
                    id, schema_version, chain_id, wallet_address, nonce, purpose, status,
                    client_order_id, trader_id, strategy_id, account_id, instrument_id,
                    pool_address, transaction_to, transaction_input, transaction_value,
                    amount_in, created_block, acknowledgement_emitted, fill_emitted,
                    terminal_emitted, active
                ",
            )
            .bind(EXECUTION_SCHEMA_VERSION)
            .bind(chain_id)
            .bind(&intent.wallet_address)
            .bind(&intent.purpose)
            .bind(&intent.client_order_id)
            .bind(&intent.trader_id)
            .bind(&intent.strategy_id)
            .bind(&intent.account_id)
            .bind(&intent.instrument_id)
            .bind(&intent.pool_address)
            .bind(&intent.transaction_to)
            .bind(&intent.transaction_input)
            .bind(&intent.transaction_value)
            .bind(&intent.amount_in)
            .bind(created_block)
            .fetch_one(&mut *transaction)
            .await
            .context("failed to insert execution intent reservation")?;

            sqlx::query(
                "
                INSERT INTO execution_transaction_transition (
                    intent_id, transition_key, to_status, block_number
                ) VALUES ($1, 'prepared', 'prepared', $2)
                ",
            )
            .bind(row.id)
            .bind(created_block)
            .execute(&mut *transaction)
            .await
            .context("failed to record prepared execution intent")?;

            Ok::<_, anyhow::Error>((transaction, row))
        }
        .await
        .map_err(|source| {
            anyhow::Error::new(ExecutionIntentReservationError {
                stage: ExecutionIntentReservationStage::BeforeCommit,
                source,
            })
        })?;
        transaction.commit().await.map_err(|e| {
            anyhow::Error::new(ExecutionIntentReservationError {
                stage: ExecutionIntentReservationStage::Commit,
                source: anyhow::Error::new(e)
                    .context("failed to commit execution intent reservation"),
            })
        })?;
        Ok(row)
    }

    /// Assigns the signer nonce to a prepared execution intent.
    ///
    /// Repeating the same assignment is idempotent. A different nonce or non-prepared state
    /// fails closed.
    ///
    /// # Errors
    ///
    /// Returns an error if the intent cannot own the nonce or persistence fails.
    pub async fn assign_execution_intent_nonce(
        &self,
        intent_id: i64,
        nonce: u64,
    ) -> anyhow::Result<()> {
        let nonce_db = i64::try_from(nonce)
            .with_context(|| format!("Execution nonce {nonce} exceeds PostgreSQL BIGINT"))?;
        let result = sqlx::query(
            "
            UPDATE execution_intent
            SET nonce = $2, updated_at = NOW()
            WHERE id = $1
              AND status = 'prepared'
              AND (nonce IS NULL OR nonce = $2)
            ",
        )
        .bind(intent_id)
        .bind(nonce_db)
        .execute(&self.pool)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to assign nonce {nonce} to execution intent: {e}"))?;
        anyhow::ensure!(
            result.rows_affected() == 1,
            "Execution intent {intent_id} is not prepared for nonce {nonce}"
        );
        Ok(())
    }

    /// Assigns the canonical nonce and its authorizing verification evidence atomically.
    ///
    /// # Errors
    ///
    /// Returns an error if the durable nonce ledger, intent ownership, manifest identity, or
    /// evidence is inconsistent, or if persistence fails.
    pub(crate) async fn assign_execution_intent_nonce_verified(
        &self,
        assignment: &ExecutionNonceAssignment<'_>,
    ) -> anyhow::Result<()> {
        anyhow::ensure!(
            !assignment.decisions.is_empty(),
            "Verified nonce assignment requires decision evidence"
        );
        anyhow::ensure!(
            assignment.provider_ids.len() == 3 && assignment.operator_ids.len() == 3,
            "Verified nonce assignment requires exactly three provider and operator IDs"
        );
        anyhow::ensure!(
            assignment.failure_domain_ids.len() >= 3,
            "Verified nonce assignment requires the configured failure domains"
        );
        let chain_id = i32::try_from(assignment.chain_id)
            .context("Verification chain ID exceeds PostgreSQL INTEGER")?;
        let nonce =
            i64::try_from(assignment.nonce).context("Execution nonce exceeds PostgreSQL BIGINT")?;
        let mut transaction = self
            .pool
            .begin()
            .await
            .map_err(|e| anyhow::anyhow!("Failed to start verified nonce assignment: {e}"))?;
        let (manifest_version, manifest_digest, next_nonce, revision) =
            sqlx::query_as::<_, (String, String, i64, i64)>(
                "
                SELECT manifest_version, manifest_digest, next_canonical_nonce, revision
                FROM execution_verification_nonce
                WHERE chain_id = $1 AND wallet_address = $2
                FOR UPDATE
                ",
            )
            .bind(chain_id)
            .bind(assignment.wallet_address)
            .fetch_optional(&mut *transaction)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to lock canonical nonce ledger: {e}"))?
            .ok_or_else(|| anyhow::anyhow!("Canonical nonce ledger is not initialized"))?;
        anyhow::ensure!(
            manifest_version == assignment.manifest_version
                && manifest_digest == assignment.manifest_digest,
            "Verified nonce assignment manifest identity changed"
        );
        anyhow::ensure!(
            next_nonce == nonce,
            "Execution nonce {} does not match canonical nonce {next_nonce}",
            assignment.nonce
        );

        let (intent_chain_id, intent_wallet, intent_nonce, intent_status, intent_active) =
            sqlx::query_as::<_, (i32, String, Option<i64>, String, bool)>(
                "
            SELECT chain_id, wallet_address, nonce, status, active
            FROM execution_intent
            WHERE id = $1
            FOR UPDATE
            ",
            )
            .bind(assignment.intent_id)
            .fetch_optional(&mut *transaction)
            .await
            .map_err(|e| {
                anyhow::anyhow!("Failed to lock execution intent for nonce assignment: {e}")
            })?
            .ok_or_else(|| {
                anyhow::anyhow!("Execution intent {} was not found", assignment.intent_id)
            })?;
        anyhow::ensure!(
            intent_chain_id == chain_id
                && intent_wallet == assignment.wallet_address
                && intent_status == "prepared"
                && intent_active
                && intent_nonce.is_none_or(|assigned| assigned == nonce),
            "Execution intent {} cannot own canonical nonce {}",
            assignment.intent_id,
            assignment.nonce
        );

        for (index, decision) in assignment.decisions.iter().enumerate() {
            let height_start = decision
                .height_start
                .map(i64::try_from)
                .transpose()
                .context("Verification height exceeds PostgreSQL BIGINT")?;
            let height_end = decision
                .height_end
                .map(i64::try_from)
                .transpose()
                .context("Verification height exceeds PostgreSQL BIGINT")?;
            let transition_key = format!(
                "pre_sign:{}:{}:{index}",
                assignment.intent_id, decision.read_class
            );
            sqlx::query(
                "
                INSERT INTO execution_verification_decision (
                    intent_id, nonce, decision_class, read_class, height_start, height_end,
                    manifest_version, manifest_digest, provider_ids, operator_ids,
                    failure_domain_ids, response_class, normalized_value_digest,
                    nonce_revision, outcome, transition_key
                )
                VALUES (
                    $1, $2, 'pre_sign', $3, $4, $5, $6, $7, $8, $9, $10,
                    'all_valid', $11, $12, 'verified', $13
                )
                ",
            )
            .bind(assignment.intent_id)
            .bind(nonce)
            .bind(decision.read_class)
            .bind(height_start)
            .bind(height_end)
            .bind(assignment.manifest_version)
            .bind(assignment.manifest_digest)
            .bind(assignment.provider_ids)
            .bind(assignment.operator_ids)
            .bind(assignment.failure_domain_ids)
            .bind(&decision.normalized_value_digest)
            .bind(revision)
            .bind(transition_key)
            .execute(&mut *transaction)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to persist pre-sign verification: {e}"))?;
        }

        let result = sqlx::query(
            "
            UPDATE execution_intent
            SET nonce = $2, updated_at = NOW()
            WHERE id = $1
              AND status = 'prepared'
              AND active
              AND (nonce IS NULL OR nonce = $2)
            ",
        )
        .bind(assignment.intent_id)
        .bind(nonce)
        .execute(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to assign verified execution nonce: {e}"))?;
        anyhow::ensure!(
            result.rows_affected() == 1,
            "Execution intent {} is not prepared for canonical nonce {}",
            assignment.intent_id,
            assignment.nonce
        );
        transaction
            .commit()
            .await
            .map_err(|e| anyhow::anyhow!("Failed to commit verified nonce assignment: {e}"))?;
        Ok(())
    }

    /// Appends one verified decision batch before an action on an existing active intent.
    pub(crate) async fn record_execution_verification_batch(
        &self,
        batch: &ExecutionVerificationBatch<'_>,
    ) -> anyhow::Result<()> {
        anyhow::ensure!(
            !batch.decision_class.trim().is_empty() && !batch.decisions.is_empty(),
            "Verified action requires a decision class and evidence"
        );
        anyhow::ensure!(
            batch.provider_ids.len() == 3
                && batch.operator_ids.len() == 3
                && batch.failure_domain_ids.len() >= 3,
            "Verified action requires the configured provider identities"
        );
        let chain_id = i32::try_from(batch.chain_id)
            .context("Verification chain ID exceeds PostgreSQL INTEGER")?;
        let nonce =
            i64::try_from(batch.nonce).context("Execution nonce exceeds PostgreSQL BIGINT")?;
        let mut transaction = self
            .pool
            .begin()
            .await
            .map_err(|e| anyhow::anyhow!("Failed to start verified action evidence: {e}"))?;
        let (manifest_version, manifest_digest, revision) =
            sqlx::query_as::<_, (String, String, i64)>(
                "
                SELECT manifest_version, manifest_digest, revision
                FROM execution_verification_nonce
                WHERE chain_id = $1 AND wallet_address = $2
                FOR SHARE
                ",
            )
            .bind(chain_id)
            .bind(batch.wallet_address)
            .fetch_optional(&mut *transaction)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to read verified action nonce ledger: {e}"))?
            .ok_or_else(|| anyhow::anyhow!("Canonical nonce ledger is not initialized"))?;
        anyhow::ensure!(
            manifest_version == batch.manifest_version && manifest_digest == batch.manifest_digest,
            "Verified action manifest identity changed"
        );
        let intent_nonce = sqlx::query_scalar::<_, Option<i64>>(
            "
            SELECT nonce
            FROM execution_intent
            WHERE id = $1 AND chain_id = $2 AND wallet_address = $3 AND active
            FOR UPDATE
            ",
        )
        .bind(batch.intent_id)
        .bind(chain_id)
        .bind(batch.wallet_address)
        .fetch_optional(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to lock intent for verified action: {e}"))?
        .ok_or_else(|| anyhow::anyhow!("Active verified-action intent was not found"))?;
        anyhow::ensure!(
            intent_nonce == Some(nonce),
            "Verified action nonce does not match the active intent"
        );
        let attempt = sqlx::query_scalar::<_, i64>(
            "
            SELECT COUNT(*)
            FROM execution_verification_decision
            WHERE intent_id = $1 AND decision_class = $2
            ",
        )
        .bind(batch.intent_id)
        .bind(batch.decision_class)
        .fetch_one(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to number verified action evidence: {e}"))?;

        for (index, decision) in batch.decisions.iter().enumerate() {
            let height_start = decision
                .height_start
                .map(i64::try_from)
                .transpose()
                .context("Verification height exceeds PostgreSQL BIGINT")?;
            let height_end = decision
                .height_end
                .map(i64::try_from)
                .transpose()
                .context("Verification height exceeds PostgreSQL BIGINT")?;
            let transition_key = format!(
                "{}:{}:{attempt}:{}:{index}",
                batch.decision_class, batch.intent_id, decision.read_class
            );
            sqlx::query(
                "
                INSERT INTO execution_verification_decision (
                    intent_id, nonce, decision_class, read_class, height_start, height_end,
                    manifest_version, manifest_digest, provider_ids, operator_ids,
                    failure_domain_ids, response_class, normalized_value_digest,
                    nonce_revision, outcome, transition_key
                )
                VALUES (
                    $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11,
                    'all_valid', $12, $13, 'verified', $14
                )
                ",
            )
            .bind(batch.intent_id)
            .bind(nonce)
            .bind(batch.decision_class)
            .bind(decision.read_class)
            .bind(height_start)
            .bind(height_end)
            .bind(batch.manifest_version)
            .bind(batch.manifest_digest)
            .bind(batch.provider_ids)
            .bind(batch.operator_ids)
            .bind(batch.failure_domain_ids)
            .bind(&decision.normalized_value_digest)
            .bind(revision)
            .bind(transition_key)
            .execute(&mut *transaction)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to persist verified action evidence: {e}"))?;
        }
        transaction
            .commit()
            .await
            .map_err(|e| anyhow::anyhow!("Failed to commit verified action evidence: {e}"))?;
        Ok(())
    }

    pub(crate) async fn load_execution_replacement_cursor(
        &self,
        intent_id: i64,
        chain_id: u32,
        wallet_address: &str,
        nonce: u64,
        manifest_digest: &str,
    ) -> anyhow::Result<Option<ExecutionVerifiedHeader>> {
        let chain_id = i32::try_from(chain_id)
            .context("Replacement scan chain ID exceeds PostgreSQL INTEGER")?;
        let nonce =
            i64::try_from(nonce).context("Replacement scan nonce exceeds PostgreSQL BIGINT")?;
        let row = sqlx::query_as::<_, (i64, String, String, i64, Option<String>, String)>(
            "
            SELECT
                h.number, h.hash, h.parent_hash, h.timestamp, h.base_fee_per_gas,
                s.manifest_digest
            FROM execution_replacement_scan AS s
            JOIN execution_verified_finalized_header AS h
              ON h.chain_id = s.chain_id
             AND h.wallet_address = s.wallet_address
             AND h.number = s.finalized_cursor_number
             AND h.hash = s.finalized_cursor_hash
            WHERE s.intent_id = $1
              AND s.chain_id = $2
              AND s.wallet_address = $3
              AND s.nonce = $4
            ",
        )
        .bind(intent_id)
        .bind(chain_id)
        .bind(wallet_address)
        .bind(nonce)
        .fetch_optional(&self.pool)
        .await
        .context("failed to load replacement scan cursor")?;
        row.map(
            |(number, hash, parent_hash, timestamp, base_fee, stored_digest)| {
                anyhow::ensure!(
                    stored_digest == manifest_digest,
                    "Replacement scan manifest identity changed"
                );
                Ok(ExecutionVerifiedHeader {
                    number: u64::try_from(number)
                        .context("Replacement scan cursor number is negative")?,
                    hash,
                    parent_hash,
                    timestamp: u64::try_from(timestamp)
                        .context("Replacement scan cursor timestamp is negative")?,
                    base_fee_per_gas: base_fee
                        .map(|value| {
                            value.parse::<u128>().map_err(|_| {
                                anyhow::anyhow!("Replacement scan cursor base fee is invalid")
                            })
                        })
                        .transpose()?,
                })
            },
        )
        .transpose()
    }

    pub(crate) async fn record_execution_replacement_scan(
        &self,
        scan: &ExecutionReplacementScan<'_>,
    ) -> anyhow::Result<()> {
        anyhow::ensure!(
            !scan.decisions.is_empty()
                && scan.provider_ids.len() == 3
                && scan.operator_ids.len() == 3
                && scan.failure_domain_ids.len() >= 3,
            "Replacement scan verification evidence is incomplete"
        );
        let chain_id = i32::try_from(scan.chain_id)
            .context("Replacement scan chain ID exceeds PostgreSQL INTEGER")?;
        let nonce = i64::try_from(scan.nonce)
            .context("Replacement scan nonce exceeds PostgreSQL BIGINT")?;
        let mut transaction = self
            .pool
            .begin()
            .await
            .context("failed to start verified replacement scan transition")?;
        let (stored_version, stored_digest, stored_nonce, revision) =
            sqlx::query_as::<_, (String, String, i64, i64)>(
                "
                SELECT manifest_version, manifest_digest, next_canonical_nonce, revision
                FROM execution_verification_nonce
                WHERE chain_id = $1 AND wallet_address = $2
                FOR SHARE
                ",
            )
            .bind(chain_id)
            .bind(scan.wallet_address)
            .fetch_optional(&mut *transaction)
            .await
            .context("failed to read replacement scan nonce ledger")?
            .ok_or_else(|| anyhow::anyhow!("Canonical nonce ledger is not initialized"))?;
        anyhow::ensure!(
            stored_version == scan.manifest_version
                && stored_digest == scan.manifest_digest
                && stored_nonce == nonce,
            "Replacement scan conflicts with the canonical nonce or manifest ledger"
        );
        let current_status = sqlx::query_scalar::<_, String>(
            "
            SELECT status
            FROM execution_intent
            WHERE id = $1 AND chain_id = $2 AND wallet_address = $3
              AND nonce = $4 AND active
            FOR UPDATE
            ",
        )
        .bind(scan.intent_id)
        .bind(chain_id)
        .bind(scan.wallet_address)
        .bind(nonce)
        .fetch_optional(&mut *transaction)
        .await
        .context("failed to lock the replacement scan intent")?
        .ok_or_else(|| anyhow::anyhow!("Active replacement scan intent was not found"))?;

        if let Some(cursor) = scan.finalized_cursor {
            let number = i64::try_from(cursor.number)
                .context("Replacement scan cursor exceeds PostgreSQL BIGINT")?;
            let durable_hash = sqlx::query_scalar::<_, String>(
                "
                SELECT hash
                FROM execution_verified_finalized_header
                WHERE chain_id = $1 AND wallet_address = $2 AND number = $3
                ",
            )
            .bind(chain_id)
            .bind(scan.wallet_address)
            .bind(number)
            .fetch_optional(&mut *transaction)
            .await
            .context("failed to validate replacement cursor against finalized headers")?
            .ok_or_else(|| {
                anyhow::anyhow!(
                    "Replacement scan cursor {} is not durably finalized",
                    cursor.number
                )
            })?;
            anyhow::ensure!(
                durable_hash == cursor.hash,
                "Replacement scan cursor conflicts with the finalized header ledger"
            );
            let existing = sqlx::query_as::<_, (i64, String)>(
                "
                SELECT finalized_cursor_number, finalized_cursor_hash
                FROM execution_replacement_scan
                WHERE intent_id = $1
                FOR UPDATE
                ",
            )
            .bind(scan.intent_id)
            .fetch_optional(&mut *transaction)
            .await
            .context("failed to lock replacement scan progress")?;

            if let Some((stored_number, stored_hash)) = existing {
                anyhow::ensure!(
                    number > stored_number
                        || (number == stored_number && cursor.hash == stored_hash),
                    "Replacement scan cursor regressed or changed"
                );
            }
            sqlx::query(
                "
                INSERT INTO execution_replacement_scan (
                    intent_id, chain_id, wallet_address, nonce,
                    finalized_cursor_number, finalized_cursor_hash, manifest_digest
                ) VALUES ($1, $2, $3, $4, $5, $6, $7)
                ON CONFLICT (intent_id) DO UPDATE SET
                    finalized_cursor_number = EXCLUDED.finalized_cursor_number,
                    finalized_cursor_hash = EXCLUDED.finalized_cursor_hash,
                    updated_at = NOW()
                ",
            )
            .bind(scan.intent_id)
            .bind(chain_id)
            .bind(scan.wallet_address)
            .bind(nonce)
            .bind(number)
            .bind(&cursor.hash)
            .bind(scan.manifest_digest)
            .execute(&mut *transaction)
            .await
            .context("failed to persist replacement scan cursor")?;
        }

        let attempt = sqlx::query_scalar::<_, i64>(
            "
            SELECT COUNT(*)
            FROM execution_verification_decision
            WHERE intent_id = $1 AND decision_class = 'replacement_scan'
            ",
        )
        .bind(scan.intent_id)
        .fetch_one(&mut *transaction)
        .await
        .context("failed to number replacement scan evidence")?;
        for (index, decision) in scan.decisions.iter().enumerate() {
            let height_start = decision
                .height_start
                .map(i64::try_from)
                .transpose()
                .context("Replacement scan height exceeds PostgreSQL BIGINT")?;
            let height_end = decision
                .height_end
                .map(i64::try_from)
                .transpose()
                .context("Replacement scan height exceeds PostgreSQL BIGINT")?;
            let transition_key = format!(
                "replacement_scan:{}:{attempt}:{}:{index}",
                scan.intent_id, decision.read_class
            );
            sqlx::query(
                "
                INSERT INTO execution_verification_decision (
                    intent_id, nonce, decision_class, read_class, height_start, height_end,
                    manifest_version, manifest_digest, provider_ids, operator_ids,
                    failure_domain_ids, response_class, normalized_value_digest,
                    nonce_revision, outcome, transition_key
                ) VALUES (
                    $1, $2, 'replacement_scan', $3, $4, $5, $6, $7, $8, $9, $10,
                    'all_valid', $11, $12, 'verified', $13
                )
                ",
            )
            .bind(scan.intent_id)
            .bind(nonce)
            .bind(decision.read_class)
            .bind(height_start)
            .bind(height_end)
            .bind(scan.manifest_version)
            .bind(scan.manifest_digest)
            .bind(scan.provider_ids)
            .bind(scan.operator_ids)
            .bind(scan.failure_domain_ids)
            .bind(&decision.normalized_value_digest)
            .bind(revision)
            .bind(transition_key)
            .execute(&mut *transaction)
            .await
            .context("failed to persist replacement scan evidence")?;
        }

        if let Some(transaction_hash) = scan.matched_transaction_hash {
            anyhow::ensure!(
                execution_transition_allowed(&current_status, TransactionStatus::Replaced),
                "Intent cannot attach a verified replacement from status {current_status}"
            );
            let (hash_id, payload_expected, already_current) =
                sqlx::query_as::<_, (i64, bool, bool)>(
                    "
                    SELECT id, payload_expected, current
                    FROM execution_transaction_hash
                    WHERE intent_id = $1 AND chain_id = $2 AND transaction_hash = $3
                    FOR UPDATE
                    ",
                )
                .bind(scan.intent_id)
                .bind(chain_id)
                .bind(transaction_hash)
                .fetch_optional(&mut *transaction)
                .await
                .context("failed to lock authenticated replacement payload")?
                .ok_or_else(|| {
                    anyhow::anyhow!(
                        "Replacement transaction {transaction_hash} has no retained payload"
                    )
                })?;
            anyhow::ensure!(
                payload_expected,
                "Replacement transaction {transaction_hash} is not authenticated"
            );

            if !already_current {
                sqlx::query(
                    "
                    UPDATE execution_transaction_hash
                    SET current = FALSE, status = 'replaced', updated_at = NOW()
                    WHERE intent_id = $1 AND current
                    ",
                )
                .bind(scan.intent_id)
                .execute(&mut *transaction)
                .await
                .context("failed to retire the prior transaction hash")?;
                sqlx::query(
                    "
                    UPDATE execution_transaction_hash
                    SET current = TRUE, status = 'replaced', updated_at = NOW()
                    WHERE id = $1
                    ",
                )
                .bind(hash_id)
                .execute(&mut *transaction)
                .await
                .context("failed to attach the authenticated replacement hash")?;
                sqlx::query(
                    "UPDATE execution_intent SET status = 'replaced', updated_at = NOW() WHERE id = $1",
                )
                .bind(scan.intent_id)
                .execute(&mut *transaction)
                .await
                .context("failed to mark the execution intent replaced")?;
                sqlx::query(
                    "
                    INSERT INTO execution_transaction_transition (
                        intent_id, transaction_hash_id, transition_key, from_status, to_status
                    ) VALUES ($1, $2, $3, $4, 'replaced')
                    ON CONFLICT (intent_id, transition_key) DO NOTHING
                    ",
                )
                .bind(scan.intent_id)
                .bind(hash_id)
                .bind(format!("replacement_verified:{transaction_hash}"))
                .bind(current_status)
                .execute(&mut *transaction)
                .await
                .context("failed to record the verified replacement transition")?;
            }
        }

        transaction
            .commit()
            .await
            .context("failed to commit verified replacement scan")?;
        Ok(())
    }

    /// Releases an intent when no broadcast attempt can have occurred.
    ///
    /// # Errors
    ///
    /// Returns an error if the intent advanced to broadcast or persistence fails.
    pub async fn mark_execution_intent_recoverable(&self, intent_id: i64) -> anyhow::Result<()> {
        let mut transaction = self.pool.begin().await.map_err(|e| {
            anyhow::anyhow!("Failed to start recoverable execution transition: {e}")
        })?;
        let current_status = sqlx::query_scalar::<_, String>(
            "SELECT status FROM execution_intent WHERE id = $1 FOR UPDATE",
        )
        .bind(intent_id)
        .fetch_optional(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to lock recoverable execution intent: {e}"))?
        .ok_or_else(|| anyhow::anyhow!("Execution intent {intent_id} was not found"))?;
        anyhow::ensure!(
            current_status == "prepared",
            "Execution intent {intent_id} is {current_status}, not recoverable before signing"
        );
        let result = sqlx::query(
            "
            UPDATE execution_intent
            SET status = 'recoverable', active = FALSE, updated_at = NOW()
            WHERE id = $1 AND status = 'prepared'
            ",
        )
        .bind(intent_id)
        .execute(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to mark execution intent recoverable: {e}"))?;
        anyhow::ensure!(
            result.rows_affected() == 1,
            "Execution intent {intent_id} is not recoverable from preparation"
        );
        sqlx::query(
            "
            INSERT INTO execution_transaction_transition (
                intent_id, transition_key, from_status, to_status
            ) VALUES ($1, 'recoverable', $2, 'recoverable')
            ON CONFLICT (intent_id, transition_key) DO NOTHING
            ",
        )
        .bind(intent_id)
        .bind(current_status)
        .execute(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to record recoverable transition: {e}"))?;
        transaction
            .commit()
            .await
            .map_err(|e| anyhow::anyhow!("Failed to commit recoverable transition: {e}"))?;
        Ok(())
    }

    /// Persists a signed transaction and advances its intent before broadcast.
    ///
    /// # Errors
    ///
    /// Returns an error if the intent is not prepared, lacks a nonce, conflicts with a stored
    /// hash, or persistence fails.
    pub async fn add_execution_transaction_hash(
        &self,
        intent_id: i64,
        chain_id: u32,
        transaction_hash: &str,
        raw_transaction: &[u8],
    ) -> anyhow::Result<ExecutionTransactionHashRow> {
        self.add_execution_transaction_payload(
            intent_id,
            chain_id,
            transaction_hash,
            Some(raw_transaction),
            None,
        )
        .await
    }

    pub(crate) async fn add_execution_transaction_envelope(
        &self,
        intent_id: i64,
        chain_id: u32,
        transaction_hash: &str,
        sealed_transaction: &[u8],
    ) -> anyhow::Result<ExecutionTransactionHashRow> {
        self.add_execution_transaction_payload(
            intent_id,
            chain_id,
            transaction_hash,
            None,
            Some(sealed_transaction),
        )
        .await
    }

    async fn add_execution_transaction_payload(
        &self,
        intent_id: i64,
        chain_id: u32,
        transaction_hash: &str,
        raw_transaction: Option<&[u8]>,
        sealed_transaction: Option<&[u8]>,
    ) -> anyhow::Result<ExecutionTransactionHashRow> {
        anyhow::ensure!(
            raw_transaction.is_some() != sealed_transaction.is_some(),
            "A signed transaction requires exactly one payload representation"
        );
        let chain_id_db = i32::try_from(chain_id)
            .with_context(|| format!("Chain ID {chain_id} exceeds PostgreSQL INTEGER"))?;
        let mut transaction =
            self.pool.begin().await.map_err(|e| {
                anyhow::anyhow!("Failed to start signed transaction persistence: {e}")
            })?;

        if let Some(envelope) = sealed_transaction {
            let state_row = sqlx::query(
                "SELECT deployment_id, protocol_version, operation, active_key_id \
                 FROM execution_payload_state WHERE component = 'signed_transactions' FOR SHARE",
            )
            .fetch_optional(&mut *transaction)
            .await
            .context("failed to lock execution payload state for protected persistence")?
            .ok_or_else(|| anyhow::anyhow!("Execution payload protection is not active"))?;
            let state = execution_payload_state_from_row(&state_row)?;
            anyhow::ensure!(
                state.protocol_version == EXECUTION_PAYLOAD_PROTOCOL_VERSION
                    && state.operation == "ready",
                "Execution payload storage is not ready for protected persistence"
            );
            anyhow::ensure!(
                envelope_key_id(envelope)?.as_slice() == state.active_key_id.as_slice(),
                "Signed transaction envelope does not use the database active key"
            );
        } else {
            let marker = sqlx::query_scalar::<_, bool>(
                "SELECT EXISTS (SELECT 1 FROM execution_schema_version WHERE component = $1)",
            )
            .bind(EXECUTION_PAYLOAD_COMPONENT)
            .fetch_one(&mut *transaction)
            .await
            .context("failed to inspect execution payload marker")?;
            anyhow::ensure!(
                !marker,
                "Plaintext signed transaction persistence is disabled after payload activation"
            );
        }
        let current_status = sqlx::query_scalar::<_, String>(
            "SELECT status FROM execution_intent WHERE id = $1 FOR UPDATE",
        )
        .bind(intent_id)
        .fetch_optional(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to lock execution intent {intent_id}: {e}"))?
        .ok_or_else(|| anyhow::anyhow!("Execution intent {intent_id} was not found"))?;
        anyhow::ensure!(
            current_status == TransactionStatus::Prepared.as_str()
                || current_status == TransactionStatus::Signed.as_str(),
            "Execution intent {intent_id} is {current_status}, not prepared for signing"
        );

        let row = sqlx::query_as::<_, ExecutionTransactionHashRow>(
            "
            INSERT INTO execution_transaction_hash (
                intent_id, chain_id, transaction_hash, payload_expected,
                raw_transaction, sealed_transaction, status
            )
            SELECT id, chain_id, $3, TRUE, $4, $5, 'signed'
            FROM execution_intent
            WHERE id = $1 AND chain_id = $2 AND nonce IS NOT NULL
            ON CONFLICT (chain_id, transaction_hash) DO UPDATE
            SET transaction_hash = EXCLUDED.transaction_hash
            WHERE execution_transaction_hash.intent_id = EXCLUDED.intent_id
              AND execution_transaction_hash.payload_expected
              AND (
                  (
                      EXCLUDED.raw_transaction IS NOT NULL
                      AND execution_transaction_hash.raw_transaction = EXCLUDED.raw_transaction
                      AND execution_transaction_hash.sealed_transaction IS NULL
                  )
                  OR (
                      EXCLUDED.sealed_transaction IS NOT NULL
                      AND execution_transaction_hash.raw_transaction IS NULL
                      AND execution_transaction_hash.sealed_transaction IS NOT NULL
                  )
              )
            RETURNING
                id, intent_id, chain_id, transaction_hash, payload_expected,
                raw_transaction, sealed_transaction, status,
                block_number, block_hash, receipt_success, gas_used,
                effective_gas_price, current
            ",
        )
        .bind(intent_id)
        .bind(chain_id_db)
        .bind(transaction_hash)
        .bind(raw_transaction)
        .bind(sealed_transaction)
        .fetch_optional(&mut *transaction)
        .await
        .map_err(|e| {
            anyhow::anyhow!("Failed to persist signed transaction {transaction_hash}: {e}")
        })?
        .ok_or_else(|| {
            anyhow::anyhow!(
                "Signed transaction {transaction_hash} conflicts with its persisted identity"
            )
        })?;

        sqlx::query(
            "UPDATE execution_intent SET status = 'signed', updated_at = NOW() WHERE id = $1",
        )
        .bind(intent_id)
        .execute(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to mark execution intent signed: {e}"))?;
        sqlx::query(
            "
            INSERT INTO execution_transaction_transition (
                intent_id, transaction_hash_id, transition_key, from_status, to_status
            ) VALUES ($1, $2, $3, $4, 'signed')
            ON CONFLICT (intent_id, transition_key) DO NOTHING
            ",
        )
        .bind(intent_id)
        .bind(row.id)
        .bind(format!("signed:{transaction_hash}"))
        .bind(current_status)
        .execute(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to record signed transaction transition: {e}"))?;

        transaction
            .commit()
            .await
            .map_err(|e| anyhow::anyhow!("Failed to commit signed transaction: {e}"))?;
        Ok(row)
    }

    /// Records an idempotent transaction observation and advances its intent state.
    ///
    /// # Errors
    ///
    /// Returns an error if the transition is invalid, the hash is unknown, or persistence fails.
    #[expect(
        clippy::too_many_arguments,
        reason = "the parameters are the canonical receipt observation persisted atomically"
    )]
    pub async fn record_execution_status(
        &self,
        intent_id: i64,
        transaction_hash: &str,
        status: TransactionStatus,
        block_number: Option<u64>,
        block_hash: Option<&str>,
        receipt_success: Option<bool>,
        gas_used: Option<u64>,
        effective_gas_price: Option<&str>,
    ) -> anyhow::Result<()> {
        let block_number_db = block_number
            .map(i64::try_from)
            .transpose()
            .with_context(|| {
                format!(
                    "Execution block number {} exceeds PostgreSQL BIGINT",
                    block_number.unwrap_or_default()
                )
            })?;
        let gas_used_db = gas_used.map(i64::try_from).transpose().with_context(|| {
            format!(
                "Execution gas used {} exceeds PostgreSQL BIGINT",
                gas_used.unwrap_or_default()
            )
        })?;
        let mut transaction = self
            .pool
            .begin()
            .await
            .map_err(|e| anyhow::anyhow!("Failed to start execution status transition: {e}"))?;
        let (current_status, fill_emitted, terminal_emitted) =
            sqlx::query_as::<_, (String, bool, bool)>(
            "SELECT status, fill_emitted, terminal_emitted FROM execution_intent WHERE id = $1 FOR UPDATE",
        )
        .bind(intent_id)
        .fetch_optional(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to lock execution intent {intent_id}: {e}"))?
        .ok_or_else(|| anyhow::anyhow!("Execution intent {intent_id} was not found"))?;
        anyhow::ensure!(
            execution_transition_allowed(&current_status, status),
            "Invalid execution transition for intent {intent_id}: {current_status} -> {}",
            status.as_str()
        );

        let active = match status {
            TransactionStatus::Finalized | TransactionStatus::Reverted => {
                !fill_emitted && !terminal_emitted
            }
            TransactionStatus::Recoverable => false,
            _ => true,
        };
        let hash_result = sqlx::query(
            "
            UPDATE execution_transaction_hash
            SET status = $3,
                block_number = COALESCE($4, block_number),
                block_hash = COALESCE($5, block_hash),
                receipt_success = COALESCE($6, receipt_success),
                gas_used = COALESCE($7, gas_used),
                effective_gas_price = COALESCE($8, effective_gas_price),
                updated_at = NOW()
            WHERE intent_id = $1 AND transaction_hash = $2
            ",
        )
        .bind(intent_id)
        .bind(transaction_hash)
        .bind(status.as_str())
        .bind(block_number_db)
        .bind(block_hash)
        .bind(receipt_success)
        .bind(gas_used_db)
        .bind(effective_gas_price)
        .execute(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to update execution hash {transaction_hash}: {e}"))?;
        anyhow::ensure!(
            hash_result.rows_affected() == 1,
            "Execution transaction hash {transaction_hash} was not found for intent {intent_id}"
        );

        sqlx::query(
            "
            UPDATE execution_intent
            SET status = $2, active = $3, updated_at = NOW()
            WHERE id = $1
            ",
        )
        .bind(intent_id)
        .bind(status.as_str())
        .bind(active)
        .execute(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to update execution intent {intent_id}: {e}"))?;

        let transition_key = format!(
            "{}:{transaction_hash}:{}:{}",
            status.as_str(),
            block_number.map_or_else(|| "none".to_string(), |value| value.to_string()),
            block_hash.unwrap_or("none")
        );
        sqlx::query(
            "
            INSERT INTO execution_transaction_transition (
                intent_id, transaction_hash_id, transition_key, from_status, to_status,
                block_number, block_hash
            )
            SELECT $1, id, $3, $4, $5, $6, $7
            FROM execution_transaction_hash
            WHERE intent_id = $1 AND transaction_hash = $2
            ON CONFLICT (intent_id, transition_key) DO NOTHING
            ",
        )
        .bind(intent_id)
        .bind(transaction_hash)
        .bind(transition_key)
        .bind(current_status)
        .bind(status.as_str())
        .bind(block_number_db)
        .bind(block_hash)
        .execute(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to record execution transition: {e}"))?;

        transaction
            .commit()
            .await
            .map_err(|e| anyhow::anyhow!("Failed to commit execution transition: {e}"))?;
        Ok(())
    }

    /// Records verified final consumption and advances the canonical nonce in one transaction.
    ///
    /// # Errors
    ///
    /// Returns an error if the receipt transition, nonce ledger, manifest identity, or evidence
    /// is inconsistent, or if persistence fails.
    pub(crate) async fn record_execution_finality_verified(
        &self,
        finality: &ExecutionFinalityTransition<'_>,
    ) -> anyhow::Result<()> {
        anyhow::ensure!(
            matches!(
                finality.status,
                TransactionStatus::Finalized | TransactionStatus::Reverted
            ),
            "Verified finality requires a finalized or reverted status"
        );
        anyhow::ensure!(
            !finality.decisions.is_empty(),
            "Verified finality requires decision evidence"
        );
        anyhow::ensure!(
            !finality.finalized_headers.is_empty()
                && finality.finalized_headers.windows(2).all(|headers| {
                    headers[1].number == headers[0].number.saturating_add(1)
                        && headers[1].parent_hash == headers[0].hash
                })
                && finality
                    .finalized_headers
                    .last()
                    .is_some_and(|header| header.number >= finality.block_number),
            "Verified finality headers must form a continuous chain through the inclusion height"
        );
        let chain_id = i32::try_from(finality.chain_id)
            .context("Verification chain ID exceeds PostgreSQL INTEGER")?;
        let nonce =
            i64::try_from(finality.nonce).context("Execution nonce exceeds PostgreSQL BIGINT")?;
        let next_nonce = nonce
            .checked_add(1)
            .ok_or_else(|| anyhow::anyhow!("Canonical nonce overflow"))?;
        let block_number = i64::try_from(finality.block_number)
            .context("Finality block exceeds PostgreSQL BIGINT")?;
        let gas_used = i64::try_from(finality.gas_used)
            .context("Finality gas used exceeds PostgreSQL BIGINT")?;
        let mut transaction =
            self.pool.begin().await.map_err(|e| {
                anyhow::anyhow!("Failed to start verified finality transition: {e}")
            })?;
        let (manifest_version, manifest_digest, stored_nonce, revision) =
            sqlx::query_as::<_, (String, String, i64, i64)>(
                "
                SELECT manifest_version, manifest_digest, next_canonical_nonce, revision
                FROM execution_verification_nonce
                WHERE chain_id = $1 AND wallet_address = $2
                FOR UPDATE
                ",
            )
            .bind(chain_id)
            .bind(finality.wallet_address)
            .fetch_optional(&mut *transaction)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to lock finality nonce ledger: {e}"))?
            .ok_or_else(|| anyhow::anyhow!("Canonical nonce ledger is not initialized"))?;
        anyhow::ensure!(
            manifest_version == finality.manifest_version
                && manifest_digest == finality.manifest_digest,
            "Verified finality manifest identity changed"
        );
        anyhow::ensure!(
            stored_nonce == nonce,
            "Finalized nonce {} does not match canonical nonce {stored_nonce}",
            finality.nonce
        );
        let stored_tip = sqlx::query_as::<_, (i64, String, String, i64, Option<String>, String)>(
            "
            SELECT number, hash, parent_hash, timestamp, base_fee_per_gas, manifest_digest
            FROM execution_verified_finalized_header
            WHERE chain_id = $1 AND wallet_address = $2
            ORDER BY number DESC
            LIMIT 1
            ",
        )
        .bind(chain_id)
        .bind(finality.wallet_address)
        .fetch_one(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to lock finalized header tip: {e}"))?;
        let first_header = &finality.finalized_headers[0];
        anyhow::ensure!(
            stored_tip
                == (
                    i64::try_from(first_header.number)
                        .context("Verified finalized height exceeds PostgreSQL BIGINT")?,
                    first_header.hash.clone(),
                    first_header.parent_hash.clone(),
                    i64::try_from(first_header.timestamp)
                        .context("Verified finalized timestamp exceeds PostgreSQL BIGINT")?,
                    first_header.base_fee_per_gas.map(|value| value.to_string()),
                    finality.manifest_digest.to_string(),
                ),
            "Verified finality extension does not start at the durable tip"
        );

        for header in finality.finalized_headers.iter().skip(1) {
            let number = i64::try_from(header.number)
                .context("Verified finalized height exceeds PostgreSQL BIGINT")?;
            let timestamp = i64::try_from(header.timestamp)
                .context("Verified finalized timestamp exceeds PostgreSQL BIGINT")?;
            let base_fee = header.base_fee_per_gas.map(|value| value.to_string());
            sqlx::query(
                "
                INSERT INTO execution_verified_finalized_header (
                    chain_id, wallet_address, number, hash, parent_hash, timestamp,
                    base_fee_per_gas, manifest_digest
                )
                VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
                ON CONFLICT (chain_id, wallet_address, number) DO NOTHING
                ",
            )
            .bind(chain_id)
            .bind(finality.wallet_address)
            .bind(number)
            .bind(&header.hash)
            .bind(&header.parent_hash)
            .bind(timestamp)
            .bind(&base_fee)
            .bind(finality.manifest_digest)
            .execute(&mut *transaction)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to extend finalized header ledger: {e}"))?;
            let stored = sqlx::query_as::<_, (String, String, i64, Option<String>, String)>(
                "
                SELECT hash, parent_hash, timestamp, base_fee_per_gas, manifest_digest
                FROM execution_verified_finalized_header
                WHERE chain_id = $1 AND wallet_address = $2 AND number = $3
                ",
            )
            .bind(chain_id)
            .bind(finality.wallet_address)
            .bind(number)
            .fetch_one(&mut *transaction)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to validate finalized header ledger: {e}"))?;
            anyhow::ensure!(
                stored
                    == (
                        header.hash.clone(),
                        header.parent_hash.clone(),
                        timestamp,
                        base_fee,
                        finality.manifest_digest.to_string(),
                    ),
                "Finalized header ledger conflicts at height {}",
                header.number
            );
        }

        let (current_status, intent_nonce, fill_emitted, terminal_emitted) =
            sqlx::query_as::<_, (String, Option<i64>, bool, bool)>(
                "
                SELECT status, nonce, fill_emitted, terminal_emitted
                FROM execution_intent
                WHERE id = $1 AND chain_id = $2 AND wallet_address = $3 AND active
                FOR UPDATE
                ",
            )
            .bind(finality.intent_id)
            .bind(chain_id)
            .bind(finality.wallet_address)
            .fetch_optional(&mut *transaction)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to lock intent for verified finality: {e}"))?
            .ok_or_else(|| anyhow::anyhow!("Active finality intent was not found"))?;
        anyhow::ensure!(
            intent_nonce == Some(nonce)
                && execution_transition_allowed(&current_status, finality.status),
            "Intent cannot make the verified finality transition"
        );

        for (index, decision) in finality.decisions.iter().enumerate() {
            let height_start = decision
                .height_start
                .map(i64::try_from)
                .transpose()
                .context("Verification height exceeds PostgreSQL BIGINT")?;
            let height_end = decision
                .height_end
                .map(i64::try_from)
                .transpose()
                .context("Verification height exceeds PostgreSQL BIGINT")?;
            let transition_key = format!(
                "finality:{}:{}:{index}",
                finality.intent_id, decision.read_class
            );
            sqlx::query(
                "
                INSERT INTO execution_verification_decision (
                    intent_id, nonce, decision_class, read_class, height_start, height_end,
                    manifest_version, manifest_digest, provider_ids, operator_ids,
                    failure_domain_ids, response_class, normalized_value_digest,
                    nonce_revision, outcome, transition_key
                )
                VALUES (
                    $1, $2, 'finality', $3, $4, $5, $6, $7, $8, $9, $10,
                    'all_valid', $11, $12, 'verified', $13
                )
                ",
            )
            .bind(finality.intent_id)
            .bind(nonce)
            .bind(decision.read_class)
            .bind(height_start)
            .bind(height_end)
            .bind(finality.manifest_version)
            .bind(finality.manifest_digest)
            .bind(finality.provider_ids)
            .bind(finality.operator_ids)
            .bind(finality.failure_domain_ids)
            .bind(&decision.normalized_value_digest)
            .bind(revision)
            .bind(transition_key)
            .execute(&mut *transaction)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to persist finality verification: {e}"))?;
        }

        let hash_result = sqlx::query(
            "
            UPDATE execution_transaction_hash
            SET status = $3, block_number = $4, block_hash = $5,
                receipt_success = $6, gas_used = $7, effective_gas_price = $8,
                updated_at = NOW()
            WHERE intent_id = $1 AND transaction_hash = $2
            ",
        )
        .bind(finality.intent_id)
        .bind(finality.transaction_hash)
        .bind(finality.status.as_str())
        .bind(block_number)
        .bind(finality.block_hash)
        .bind(finality.receipt_success)
        .bind(gas_used)
        .bind(finality.effective_gas_price)
        .execute(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to record verified finality receipt: {e}"))?;
        anyhow::ensure!(
            hash_result.rows_affected() == 1,
            "Finality transaction hash was not found"
        );

        let active = !fill_emitted && !terminal_emitted;
        sqlx::query(
            "UPDATE execution_intent SET status = $2, active = $3, updated_at = NOW() WHERE id = $1",
        )
        .bind(finality.intent_id)
        .bind(finality.status.as_str())
        .bind(active)
        .execute(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to record verified finality intent: {e}"))?;

        let transition_key = format!(
            "{}:{}:{}:{}",
            finality.status.as_str(),
            finality.transaction_hash,
            finality.block_number,
            finality.block_hash
        );
        sqlx::query(
            "
            INSERT INTO execution_transaction_transition (
                intent_id, transaction_hash_id, transition_key, from_status, to_status,
                block_number, block_hash
            )
            SELECT $1, id, $3, $4, $5, $6, $7
            FROM execution_transaction_hash
            WHERE intent_id = $1 AND transaction_hash = $2
            ",
        )
        .bind(finality.intent_id)
        .bind(finality.transaction_hash)
        .bind(transition_key)
        .bind(current_status)
        .bind(finality.status.as_str())
        .bind(block_number)
        .bind(finality.block_hash)
        .execute(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to record verified finality transition: {e}"))?;

        let nonce_result = sqlx::query(
            "
            UPDATE execution_verification_nonce
            SET next_canonical_nonce = $3, revision = revision + 1, updated_at = NOW()
            WHERE chain_id = $1 AND wallet_address = $2
              AND next_canonical_nonce = $4 AND revision = $5
            ",
        )
        .bind(chain_id)
        .bind(finality.wallet_address)
        .bind(next_nonce)
        .bind(nonce)
        .bind(revision)
        .execute(&mut *transaction)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to advance canonical nonce ledger: {e}"))?;
        anyhow::ensure!(
            nonce_result.rows_affected() == 1,
            "Canonical nonce ledger changed during finality transition"
        );
        transaction
            .commit()
            .await
            .map_err(|e| anyhow::anyhow!("Failed to commit verified finality transition: {e}"))?;
        Ok(())
    }

    /// Loads one durable execution intent by ID.
    pub(crate) async fn get_execution_intent(
        &self,
        intent_id: i64,
    ) -> anyhow::Result<ExecutionIntentRow> {
        sqlx::query_as::<_, ExecutionIntentRow>(
            "
            SELECT
                id, schema_version, chain_id, wallet_address, nonce, purpose, status,
                client_order_id, trader_id, strategy_id, account_id, instrument_id,
                pool_address, transaction_to, transaction_input, transaction_value,
                amount_in, created_block, acknowledgement_emitted, fill_emitted,
                terminal_emitted, active
            FROM execution_intent
            WHERE id = $1
            ",
        )
        .bind(intent_id)
        .fetch_optional(&self.pool)
        .await
        .context("failed to load execution intent")?
        .ok_or_else(|| anyhow::anyhow!("Execution intent {intent_id} was not found"))
    }

    /// Loads the active intent owned by a signer after any concurrent reservation completes.
    ///
    /// # Errors
    ///
    /// Returns an error if the query fails.
    pub async fn get_active_execution_intent(
        &self,
        chain_id: u32,
        wallet_address: &str,
    ) -> anyhow::Result<Option<ExecutionIntentRow>> {
        let chain_id_db = i32::try_from(chain_id)
            .with_context(|| format!("Chain ID {chain_id} exceeds PostgreSQL INTEGER"))?;
        let mut transaction = self
            .pool
            .begin_with("BEGIN ISOLATION LEVEL READ COMMITTED")
            .await
            .context("failed to start active execution intent reconciliation")?;
        Self::lock_execution_signer(&mut transaction, chain_id, wallet_address).await?;
        let intent = sqlx::query_as::<_, ExecutionIntentRow>(
            "
            SELECT
                id, schema_version, chain_id, wallet_address, nonce, purpose, status,
                client_order_id, trader_id, strategy_id, account_id, instrument_id,
                pool_address, transaction_to, transaction_input, transaction_value,
                amount_in, created_block, acknowledgement_emitted, fill_emitted,
                terminal_emitted, active
            FROM execution_intent
            WHERE chain_id = $1 AND wallet_address = $2 AND active
            ",
        )
        .bind(chain_id_db)
        .bind(wallet_address)
        .fetch_optional(&mut *transaction)
        .await
        .context("failed to load active execution intent")?;
        transaction
            .commit()
            .await
            .context("failed to complete active execution intent reconciliation")?;
        Ok(intent)
    }

    async fn lock_execution_signer(
        transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
        chain_id: u32,
        wallet_address: &str,
    ) -> anyhow::Result<()> {
        let lock = PgAdvisoryLock::new(format!(
            "nautilus:blockchain:execution:{chain_id}:{}",
            wallet_address.to_ascii_lowercase()
        ));
        let PgAdvisoryLockKey::BigInt(lock_key) = lock.key() else {
            unreachable!("string advisory locks use the 64-bit key space");
        };

        // Transaction-scoped release makes row presence or absence authoritative to the next
        // holder after an ambiguous reservation commit.
        sqlx::query("SELECT pg_advisory_xact_lock($1)")
            .bind(*lock_key)
            .execute(&mut **transaction)
            .await
            .context("failed to acquire execution signer reservation fence")?;
        Ok(())
    }

    /// Reports whether a released legacy intent still retains a broadcastable signature.
    ///
    /// # Errors
    ///
    /// Returns an error if the query fails.
    pub async fn has_recoverable_signed_execution(
        &self,
        chain_id: u32,
        wallet_address: &str,
    ) -> anyhow::Result<bool> {
        let chain_id_db = i32::try_from(chain_id)
            .with_context(|| format!("Chain ID {chain_id} exceeds PostgreSQL INTEGER"))?;
        sqlx::query_scalar::<_, bool>(
            "
            SELECT EXISTS (
                SELECT 1
                FROM execution_intent AS intent
                JOIN execution_transaction_hash AS hash ON hash.intent_id = intent.id
                WHERE intent.chain_id = $1
                  AND intent.wallet_address = $2
                  AND intent.status = 'recoverable'
                  AND (hash.raw_transaction IS NOT NULL OR hash.sealed_transaction IS NOT NULL)
            )
            ",
        )
        .bind(chain_id_db)
        .bind(wallet_address)
        .fetch_one(&self.pool)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to inspect recoverable signed executions: {e}"))
    }

    /// Loads all transaction hashes for an intent in insertion order.
    ///
    /// # Errors
    ///
    /// Returns an error if the query fails.
    pub async fn get_execution_transaction_hashes(
        &self,
        intent_id: i64,
    ) -> anyhow::Result<Vec<ExecutionTransactionHashRow>> {
        sqlx::query_as::<_, ExecutionTransactionHashRow>(
            "
            SELECT
                id, intent_id, chain_id, transaction_hash, payload_expected,
                raw_transaction, sealed_transaction, status,
                block_number, block_hash, receipt_success, gas_used,
                effective_gas_price, current
            FROM execution_transaction_hash
            WHERE intent_id = $1
            ORDER BY id
            ",
        )
        .bind(intent_id)
        .fetch_all(&self.pool)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to load execution transaction hashes: {e}"))
    }

    /// Marks one order event as emitted after dispatch.
    ///
    /// # Errors
    ///
    /// Returns an error if the event kind is unknown, the intent is absent, the opposing
    /// terminal marker is already set, or persistence fails.
    pub async fn mark_execution_event_emitted(
        &self,
        intent_id: i64,
        event: &str,
    ) -> anyhow::Result<()> {
        let statement = match event {
            "acknowledgement" => {
                "UPDATE execution_intent SET acknowledgement_emitted = TRUE, updated_at = NOW() WHERE id = $1"
            }
            "fill" => {
                "UPDATE execution_intent SET fill_emitted = TRUE, active = CASE WHEN status = 'finalized' THEN FALSE ELSE active END, updated_at = NOW() WHERE id = $1 AND NOT terminal_emitted"
            }
            "terminal" => {
                "UPDATE execution_intent SET terminal_emitted = TRUE, active = CASE WHEN status IN ('finalized', 'reverted') THEN FALSE ELSE active END, updated_at = NOW() WHERE id = $1 AND NOT fill_emitted"
            }
            _ => anyhow::bail!("Unknown execution event marker {event}"),
        };
        let result = sqlx::query(statement)
            .bind(intent_id)
            .execute(&self.pool)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to mark execution {event} emitted: {e}"))?;
        anyhow::ensure!(
            result.rows_affected() == 1,
            "Execution intent {intent_id} cannot mark {event} emitted"
        );
        Ok(())
    }

    /// Updates the status of a persisted execution transaction record.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn update_execution_transaction_status(
        &self,
        chain_id: u32,
        transaction_hash: &str,
        status: &str,
    ) -> anyhow::Result<()> {
        let result = sqlx::query(
            "
            UPDATE execution_transaction
            SET status = $3
            WHERE chain_id = $1 AND transaction_hash = $2
        ",
        )
        .bind(chain_id as i32)
        .bind(transaction_hash)
        .bind(status)
        .execute(&self.pool)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to update execution_transaction table: {e}"))?;

        anyhow::ensure!(
            result.rows_affected() == 1,
            "Execution transaction {transaction_hash} was not found for status update"
        );
        Ok(())
    }

    /// Loads an execution transaction record by chain ID and transaction hash.
    ///
    /// # Errors
    ///
    /// Returns an error if the database operation fails.
    pub async fn get_execution_transaction(
        &self,
        chain_id: u32,
        transaction_hash: &str,
    ) -> anyhow::Result<Option<ExecutionTransactionRow>> {
        let chain_id_db = i32::try_from(chain_id)
            .with_context(|| format!("Chain ID {chain_id} exceeds PostgreSQL INTEGER"))?;
        sqlx::query_as::<_, ExecutionTransactionRow>(
            "
            SELECT wallet_address, nonce, transaction_hash, purpose, status, client_order_id
            FROM (
                SELECT
                    intent.wallet_address,
                    intent.nonce,
                    hash.transaction_hash,
                    intent.purpose,
                    intent.status,
                    intent.client_order_id,
                    0 AS source_priority
                FROM execution_transaction_hash AS hash
                JOIN execution_intent AS intent ON intent.id = hash.intent_id
                WHERE hash.chain_id = $1 AND hash.transaction_hash = $2
                UNION ALL
                SELECT
                    wallet_address,
                    nonce,
                    transaction_hash,
                    purpose,
                    status,
                    client_order_id,
                    1 AS source_priority
                FROM execution_transaction
                WHERE chain_id = $1 AND transaction_hash = $2
            ) AS record
            ORDER BY source_priority
            LIMIT 1
        ",
        )
        .bind(chain_id_db)
        .bind(transaction_hash)
        .fetch_optional(&self.pool)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to load from execution_transaction table: {e}"))
    }
}

fn execution_payload_state_from_row(
    row: &sqlx::postgres::PgRow,
) -> anyhow::Result<ExecutionPayloadState> {
    Ok(ExecutionPayloadState {
        deployment_id: row.try_get("deployment_id")?,
        protocol_version: row.try_get("protocol_version")?,
        operation: row.try_get("operation")?,
        active_key_id: row.try_get("active_key_id")?,
    })
}

fn validate_execution_payload_state(
    state: &ExecutionPayloadState,
    keys: &PayloadKeySet,
) -> anyhow::Result<()> {
    anyhow::ensure!(
        state.protocol_version == EXECUTION_PAYLOAD_PROTOCOL_VERSION,
        "Execution payload protocol version {} is not supported",
        state.protocol_version
    );
    anyhow::ensure!(
        state.deployment_id == keys.deployment_id(),
        "Execution payload deployment ID does not match this database"
    );
    anyhow::ensure!(
        state.active_key_id.as_slice() == keys.active_key_id(),
        "Configured active payload key does not match the database active key"
    );
    Ok(())
}

async fn lock_execution_payload_operation(
    transaction: &mut Transaction<'_, Postgres>,
) -> anyhow::Result<()> {
    let lock = PgAdvisoryLock::new("nautilus:blockchain:execution-payload");
    let PgAdvisoryLockKey::BigInt(lock_key) = lock.key() else {
        unreachable!("string advisory locks use the 64-bit key space");
    };
    sqlx::query("SELECT pg_advisory_xact_lock($1)")
        .bind(*lock_key)
        .execute(&mut **transaction)
        .await
        .context("failed to acquire execution payload operation fence")?;
    Ok(())
}

async fn reserve_execution_payload_seal(
    transaction: &mut Transaction<'_, Postgres>,
    key_id: &[u8; 32],
) -> anyhow::Result<()> {
    let seals = sqlx::query_scalar::<_, i64>(
        "INSERT INTO execution_payload_key_state (key_id, seals) VALUES ($1, 1) \
         ON CONFLICT (key_id) DO UPDATE SET seals = execution_payload_key_state.seals + 1 \
         WHERE execution_payload_key_state.seals < $2 \
         RETURNING seals",
    )
    .bind(key_id.as_slice())
    .bind(EXECUTION_PAYLOAD_MAX_SEALS - 1)
    .fetch_optional(&mut **transaction)
    .await
    .context("failed to reserve execution payload nonce use")?;
    anyhow::ensure!(
        seals.is_some(),
        "Execution payload key reached its seal limit; rotate the active key before continuing"
    );
    Ok(())
}

async fn validate_execution_payload_key_inventory(
    transaction: &mut Transaction<'_, Postgres>,
    keys: &PayloadKeySet,
) -> anyhow::Result<()> {
    let envelopes = sqlx::query_scalar::<_, Vec<u8>>(
        "SELECT DISTINCT substring(sealed_transaction FROM 1 FOR 33) \
         FROM execution_transaction_hash \
         WHERE sealed_transaction IS NOT NULL",
    )
    .fetch_all(&mut **transaction)
    .await
    .context("failed to inspect execution payload key inventory")?;

    for header in envelopes {
        anyhow::ensure!(
            header.len() == 33,
            "Stored execution payload has a truncated envelope header"
        );
        let mut envelope = header;
        envelope.extend_from_slice(&[0; 12 + 16]);
        let key_id = envelope_key_id(&envelope)?;
        anyhow::ensure!(
            keys.contains_key(&key_id),
            "Stored execution payload requires unavailable key {}",
            alloy::hex::encode(key_id)
        );
    }
    Ok(())
}

async fn load_execution_intent(
    transaction: &mut Transaction<'_, Postgres>,
    intent_id: i64,
) -> anyhow::Result<ExecutionIntentRow> {
    sqlx::query_as::<_, ExecutionIntentRow>(
        "
        SELECT
            id, schema_version, chain_id, wallet_address, nonce, purpose, status,
            client_order_id, trader_id, strategy_id, account_id, instrument_id,
            pool_address, transaction_to, transaction_input, transaction_value,
            amount_in, created_block, acknowledgement_emitted, fill_emitted,
            terminal_emitted, active
        FROM execution_intent
        WHERE id = $1
        FOR SHARE
        ",
    )
    .bind(intent_id)
    .fetch_optional(&mut **transaction)
    .await
    .context("failed to load execution intent for payload authentication")?
    .ok_or_else(|| anyhow::anyhow!("Execution intent {intent_id} was not found"))
}

fn execution_transition_allowed(current: &str, next: TransactionStatus) -> bool {
    if current == next.as_str() {
        return true;
    }

    match current {
        "prepared" => matches!(
            next,
            TransactionStatus::Signed | TransactionStatus::Recoverable
        ),
        "signed" | "broadcast" => matches!(
            next,
            TransactionStatus::Broadcast
                | TransactionStatus::Included
                | TransactionStatus::Finalized
                | TransactionStatus::Reverted
                | TransactionStatus::Replaced
                | TransactionStatus::Dropped
                | TransactionStatus::Reorged
        ),
        "included" => matches!(
            next,
            TransactionStatus::Finalized
                | TransactionStatus::Reverted
                | TransactionStatus::Reorged
                | TransactionStatus::Replaced
                | TransactionStatus::Dropped
        ),
        "replaced" | "dropped" | "reorged" => matches!(
            next,
            TransactionStatus::Included
                | TransactionStatus::Finalized
                | TransactionStatus::Reverted
                | TransactionStatus::Replaced
                | TransactionStatus::Dropped
                | TransactionStatus::Reorged
        ),
        "finalized" | "reverted" | "recoverable" => false,
        _ => false,
    }
}

#[cfg(test)]
pub(crate) mod tests {
    use std::time::Duration;

    use anyhow::Context;
    use sqlx::postgres::{PgConnectOptions, PgPoolOptions};

    use super::{BlockchainCacheDatabase, ExecutionVerifiedHeader};
    use crate::rpc::verification::VERIFICATION_SCHEMA_VERSION;

    pub(crate) struct ExecutionVerificationResume {
        pub next_canonical_nonce: u64,
        pub revision: u64,
        pub finalized_headers: Vec<ExecutionVerifiedHeader>,
    }

    pub(crate) async fn connect_test_database(
        pg_options: PgConnectOptions,
    ) -> anyhow::Result<BlockchainCacheDatabase> {
        let pool = PgPoolOptions::new()
            .max_connections(32)
            .min_connections(1)
            .acquire_timeout(Duration::from_secs(3))
            .connect_with(pg_options)
            .await?;
        Ok(BlockchainCacheDatabase { pool })
    }

    impl BlockchainCacheDatabase {
        /// Loads the complete durable finalized-header ledger for tests
        pub(crate) async fn load_execution_verification_resume(
            &self,
            chain_id: u32,
            wallet_address: &str,
            manifest_version: &str,
            manifest_digest: &str,
        ) -> anyhow::Result<Option<ExecutionVerificationResume>> {
            let installed = sqlx::query_scalar::<_, i16>(
                "SELECT version FROM execution_schema_version \
                 WHERE component = 'evm_execution_verification'",
            )
            .fetch_optional(&self.pool)
            .await
            .context("failed to inspect execution verification schema")?;
            let Some(installed) = installed else {
                return Ok(None);
            };
            anyhow::ensure!(
                installed <= VERIFICATION_SCHEMA_VERSION,
                "Unsupported execution verification schema version {installed}"
            );
            let chain_id = i32::try_from(chain_id)
                .context("Verification chain ID exceeds PostgreSQL INTEGER")?;
            let current = sqlx::query_as::<_, (String, String, i64, i64)>(
                "
                SELECT manifest_version, manifest_digest, next_canonical_nonce, revision
                FROM execution_verification_nonce
                WHERE chain_id = $1 AND wallet_address = $2
                ",
            )
            .bind(chain_id)
            .bind(wallet_address)
            .fetch_optional(&self.pool)
            .await
            .context("failed to load execution verification nonce position")?;
            let Some((stored_version, stored_digest, nonce, revision)) = current else {
                return Ok(None);
            };
            anyhow::ensure!(
                stored_version == manifest_version && stored_digest == manifest_digest,
                "Execution verification manifest identity changed"
            );
            let rows = sqlx::query_as::<_, (i64, String, String, i64, Option<String>, String)>(
                "
                SELECT number, hash, parent_hash, timestamp, base_fee_per_gas, manifest_digest
                FROM execution_verified_finalized_header
                WHERE chain_id = $1 AND wallet_address = $2
                ORDER BY number
                ",
            )
            .bind(chain_id)
            .bind(wallet_address)
            .fetch_all(&self.pool)
            .await
            .context("failed to load verified finalized header ledger")?;
            let finalized_headers = rows
                .into_iter()
                .map(|(number, hash, parent_hash, timestamp, base_fee, digest)| {
                    anyhow::ensure!(
                        digest == manifest_digest,
                        "Finalized header manifest identity changed"
                    );
                    Ok(ExecutionVerifiedHeader {
                        number: u64::try_from(number)
                            .context("Finalized header number is negative")?,
                        hash,
                        parent_hash,
                        timestamp: u64::try_from(timestamp)
                            .context("Finalized header timestamp is negative")?,
                        base_fee_per_gas: base_fee
                            .map(|value| {
                                value.parse::<u128>().map_err(|_| {
                                    anyhow::anyhow!("Finalized header base fee is invalid")
                                })
                            })
                            .transpose()?,
                    })
                })
                .collect::<anyhow::Result<Vec<_>>>()?;
            anyhow::ensure!(
                !finalized_headers.is_empty()
                    && finalized_headers.windows(2).all(|headers| {
                        headers[1].number == headers[0].number.saturating_add(1)
                            && headers[1].parent_hash == headers[0].hash
                    }),
                "Verified finalized header ledger is not continuous"
            );
            Ok(Some(ExecutionVerificationResume {
                next_canonical_nonce: u64::try_from(nonce)
                    .context("Canonical nonce is negative")?,
                revision: u64::try_from(revision)
                    .context("Canonical nonce revision is negative")?,
                finalized_headers,
            }))
        }
    }

    #[tokio::test]
    async fn connect_returns_err_for_unreachable_database() {
        // `connect` backs the Python `load_pool_snapshot` binding, so a connection
        // failure must surface as `Err` rather than panicking across the API boundary.
        let options = PgConnectOptions::new()
            .host("127.0.0.1")
            .port(1)
            .username("nautilus")
            .password("pass")
            .database("nautilus");

        let result = BlockchainCacheDatabase::connect(options).await;

        assert!(result.is_err());
    }
}