//! Preimage-specific pipeline tests for storage v2 selfdestruct behavior around Cancun.

use alloy_consensus::{
    constants::{EMPTY_WITHDRAWALS, ETH_TO_WEI},
    Header, TxEip1559, TxReceipt,
};
use alloy_eips::eip1559::INITIAL_BASE_FEE;
use alloy_genesis::{Genesis, GenesisAccount};
use alloy_primitives::{bytes, keccak256, Address, Bytes, TxKind, B256, U256};
use reth_chainspec::{
    ChainSpecBuilder, ChainSpecProvider, EthereumHardfork, ForkCondition, MAINNET,
};
use reth_config::config::StageConfig;
use reth_consensus::noop::NoopConsensus;
use reth_db::tables;
use reth_db_api::{
    cursor::{DbCursorRO, DbDupCursorRO},
    transaction::DbTx,
};
use reth_db_common::init::{init_genesis, init_genesis_with_settings};
use reth_downloaders::{
    bodies::bodies::BodiesDownloaderBuilder, file_client::FileClient,
    headers::reverse_headers::ReverseHeadersDownloaderBuilder,
};
use reth_ethereum_primitives::{Block, BlockBody, Transaction, TransactionSigned};
use reth_evm::{execute::Executor, ConfigureEvm};
use reth_evm_ethereum::EthEvmConfig;
use reth_libmdbx::{Environment, EnvironmentFlags, Mode};
use reth_network_p2p::{
    bodies::downloader::BodyDownloader,
    headers::downloader::{HeaderDownloader, SyncTarget},
};
use reth_primitives_traits::{
    crypto::secp256k1::public_key_to_address,
    proofs::{calculate_receipt_root, calculate_transaction_root},
    RecoveredBlock, SealedBlock,
};
use reth_provider::{
    test_utils::create_test_provider_factory_with_chain_spec, BlockNumReader, DBProvider,
    DatabaseProviderFactory, HashedPostStateProvider, HeaderProvider, OriginalValuesKnown,
    StateWriter, StoragePath,
};
use reth_prune_types::PruneModes;
use reth_revm::database::StateProviderDatabase;
use reth_stages::{
    sets::{ExecutionStages, HashingStages, OnlineStages},
    stages::FinishStage,
};
use reth_stages_api::{Pipeline, StageSet};
use reth_static_file::StaticFileProducer;
use reth_storage_api::{StorageChangeSetReader, StorageSettings, StorageSettingsCache};
use reth_testing_utils::generators::{self, generate_key, sign_tx_with_key_pair};
use reth_trie::{HashedPostState, KeccakKeyHasher, StateRoot};
use reth_trie_db::DatabaseStateRoot;
use std::{collections::BTreeMap, path::Path, sync::Arc};
use tokio::sync::watch;

type TestProviderFactory =
    reth_provider::ProviderFactory<reth_provider::test_utils::MockNodeTypesWithDB>;

const TEST_SELFDESTRUCT_BENEFICIARY: Address = Address::new([0x77; 20]);
const TEST_CREATE2_SALT: B256 = B256::with_last_byte(0x42);

/// Scenario coverage:
/// 1. Cross-batch pre-Cancun wipe writes plain slot keys into storage changesets.
/// 2. The destroyed account's hashed storage is removed.
/// 3. Preimage DB exists and is usable across multiple stage executions.
/// 4. Post-Cancun execution removes the preimage DB directory.
///
/// Verifies v2 selfdestruct handling across a pre-/post-Cancun boundary.
///
/// Test flow:
/// 1. Run block 1 (pre-Cancun) and assert the `preimage/` MDBX directory exists and contains
///    `keccak(slot) -> slot` rows for the two written storage slots.
/// 2. Run block 2 (pre-Cancun selfdestruct) and assert the account's hashed storage is empty and
///    its storage changesets contain exactly those two slots as **plain** keys with the expected
///    prior values (`0x2a`, `0x99`).
/// 3. Run block 3 (post-Cancun) and assert `preimage/` is removed, since this auxiliary DB is no
///    longer needed after Cancun semantics are active.
#[tokio::test(flavor = "multi_thread")]
async fn test_pipeline_v2_selfdestruct_changesets_use_plain_slots() -> eyre::Result<()> {
    reth_tracing::init_test_tracing();

    // Build a 3-block scenario:
    // - block 1/2 are pre-Cancun (selfdestruct still clears storage)
    // - block 3 is post-Cancun (no storage-destroy semantics; preimage DB should be cleaned up)
    let scenario = setup_selfdestruct_scenario()?;

    let (pipeline_provider_factory, pipeline_genesis) =
        init_v2_pipeline_provider_factory(scenario.chain_spec.clone())?;

    run_pipeline_range(
        pipeline_provider_factory.clone(),
        create_file_client_from_blocks(vec![scenario.blocks[0].clone()]),
        pipeline_genesis,
        1..=1,
        1,
    )
    .await?;

    // Phase 1 (pre-Cancun): preimage DB should be created and contain slot preimages.
    let provider = pipeline_provider_factory.provider()?;
    assert_eq!(provider.last_block_number()?, 1, "pipeline should sync block 1");
    assert!(provider.cached_storage_settings().storage_v2, "test requires storage.v2 mode");

    let preimage_path = provider.storage_path().join("preimage");
    let expected_slots = scenario.expected_slots;
    assert!(preimage_path.exists(), "preimage dir should exist after first pre-Cancun run");
    assert_preimage_rows(&preimage_path, &expected_slots)?;
    assert_hashed_storage_entries(
        &provider,
        scenario.selfdestruct_contract,
        &[
            (B256::with_last_byte(0x01), U256::from(0x2a)),
            (B256::with_last_byte(0x02), U256::from(0x99)),
        ],
    )?;

    let local_head =
        pipeline_provider_factory.sealed_header(1)?.expect("block 1 header should exist");
    run_pipeline_range(
        pipeline_provider_factory.clone(),
        create_file_client_from_blocks(vec![
            scenario.blocks[0].clone(),
            scenario.blocks[1].clone(),
        ]),
        local_head,
        2..=2,
        2,
    )
    .await?;

    // Phase 2 (pre-Cancun selfdestruct): changeset keys for destroyed account must be plain slots.
    let provider = pipeline_provider_factory.provider()?;
    assert_eq!(provider.last_block_number()?, 2, "pipeline should sync block 2");
    assert!(preimage_path.exists(), "preimage dir should still exist after second pre-Cancun run");
    assert_preimage_rows(&preimage_path, &expected_slots)?;
    assert_destroyed_changeset_entries(&provider, scenario.selfdestruct_contract)?;
    assert_hashed_storage_entries(&provider, scenario.selfdestruct_contract, &[])?;

    let third_local_head =
        pipeline_provider_factory.sealed_header(2)?.expect("block 2 header should exist");
    run_pipeline_range(
        pipeline_provider_factory.clone(),
        create_file_client_from_blocks(scenario.blocks),
        third_local_head,
        3..=3,
        3,
    )
    .await?;

    // Phase 3 (post-Cancun): execution path removes the now-unneeded preimage DB directory.
    let provider = pipeline_provider_factory.provider()?;
    assert_eq!(provider.last_block_number()?, 3, "pipeline should sync block 3");
    assert!(!preimage_path.exists(), "preimage dir should be removed after post-Cancun execution");

    Ok(())
}

/// Prefunding ensures the CREATE2 target has original account state, while its init code performs
/// no `SSTORE`. This verifies that destruction alone does not create an otherwise-empty
/// [`reth_trie::HashedStorage`] solely to carry `wiped = true`.
#[tokio::test(flavor = "multi_thread")]
async fn test_pipeline_v2_prefunded_create2_selfdestruct_does_not_wipe_storage() -> eyre::Result<()>
{
    reth_tracing::init_test_tracing();

    let scenario = setup_create2_selfdestruct_scenario()?;
    let hashed_address = keccak256(scenario.child_contract);

    assert!(scenario.child_was_destroyed);
    let storage = scenario.hashed_state.storages.get(&hashed_address);
    assert!(storage.is_none());

    let (pipeline_provider_factory, pipeline_genesis) =
        init_v2_pipeline_provider_factory(scenario.chain_spec)?;
    run_pipeline_range(
        pipeline_provider_factory.clone(),
        create_file_client_from_blocks(vec![scenario.block]),
        pipeline_genesis,
        1..=1,
        1,
    )
    .await?;

    let provider = pipeline_provider_factory.provider()?;
    assert_eq!(provider.last_block_number()?, 1, "pipeline should sync block 1");
    assert!(provider.tx_ref().get::<tables::HashedAccounts>(hashed_address)?.is_none());
    assert_hashed_storage_entries(&provider, scenario.child_contract, &[])?;
    assert_destroyed_changeset_entries_in_block(&provider, 1, scenario.child_contract, &[])?;

    Ok(())
}

/// Scenario coverage:
/// 1. Single execution batch (`1..=2`) where block 1 writes and block 2 wipes.
/// 2. Block-by-block storage changesets still contain plain keys for wipe entries.
///
/// Regression coverage for single execution-batch behavior.
#[tokio::test(flavor = "multi_thread")]
async fn test_pipeline_v2_single_batch_write_then_selfdestruct_changesets_plain_slots(
) -> eyre::Result<()> {
    reth_tracing::init_test_tracing();

    let scenario = setup_selfdestruct_scenario()?;
    let (pipeline_provider_factory, pipeline_genesis) =
        init_v2_pipeline_provider_factory(scenario.chain_spec.clone())?;

    run_pipeline_range(
        pipeline_provider_factory.clone(),
        create_file_client_from_blocks(vec![
            scenario.blocks[0].clone(),
            scenario.blocks[1].clone(),
        ]),
        pipeline_genesis,
        1..=2,
        2,
    )
    .await?;

    let provider = pipeline_provider_factory.provider()?;
    assert_eq!(provider.last_block_number()?, 2, "pipeline should sync blocks 1..=2");
    assert_preimage_rows_for_provider(&provider, &scenario.expected_slots)?;
    assert_destroyed_changeset_entries(&provider, scenario.selfdestruct_contract)?;

    Ok(())
}

/// Scenario coverage:
/// 1. A slot appears only in intermediate reverts (not final bundle state).
/// 2. A later block in the same execution batch wipes the account.
/// 3. Wipe changesets still contain the required plain slot key/value.
///
/// Covers the edge case where a slot appears in intermediate block reverts but not in the final
/// bundle state, then gets wiped by a later selfdestruct in the same execution batch.
#[tokio::test(flavor = "multi_thread")]
async fn test_pipeline_v2_single_batch_reverted_slot_then_selfdestruct_changesets_plain_slots(
) -> eyre::Result<()> {
    reth_tracing::init_test_tracing();

    let scenario = setup_reverted_slot_selfdestruct_scenario()?;
    let (pipeline_provider_factory, pipeline_genesis) =
        init_v2_pipeline_provider_factory(scenario.chain_spec.clone())?;

    run_pipeline_range(
        pipeline_provider_factory.clone(),
        create_file_client_from_blocks(scenario.blocks),
        pipeline_genesis,
        1..=3,
        3,
    )
    .await?;

    let provider = pipeline_provider_factory.provider()?;
    assert_eq!(provider.last_block_number()?, 3, "pipeline should sync blocks 1..=3");
    assert_preimage_rows_for_provider(&provider, &[scenario.expected_slot.0])?;
    assert_destroyed_changeset_entries_in_block(
        &provider,
        3,
        scenario.selfdestruct_contract,
        &[scenario.expected_slot],
    )?;

    Ok(())
}

/// Scenario coverage:
/// 1. Same address is destroyed, recreated via CREATE2, then destroyed again in one execution
///    batch.
/// 2. Both wipe blocks emit plain storage keys for the destroyed account.
#[tokio::test(flavor = "multi_thread")]
async fn test_pipeline_v2_single_batch_same_address_double_wipe_changesets_plain_slots(
) -> eyre::Result<()> {
    reth_tracing::init_test_tracing();

    let scenario = setup_same_address_double_wipe_scenario()?;
    let (pipeline_provider_factory, pipeline_genesis) =
        init_v2_pipeline_provider_factory(scenario.chain_spec.clone())?;

    run_pipeline_range(
        pipeline_provider_factory.clone(),
        create_file_client_from_blocks(scenario.blocks),
        pipeline_genesis,
        1..=6,
        6,
    )
    .await?;

    let provider = pipeline_provider_factory.provider()?;
    assert_eq!(provider.last_block_number()?, 6, "pipeline should sync blocks 1..=6");
    let expected_preimages = scenario.expected_slots.map(|(slot, _)| slot);
    assert_preimage_rows_for_provider(&provider, &expected_preimages)?;
    assert_destroyed_changeset_entries_in_block(
        &provider,
        3,
        scenario.child_contract,
        &scenario.expected_slots,
    )?;
    assert_destroyed_changeset_entries_in_block(
        &provider,
        6,
        scenario.child_contract,
        &scenario.expected_slots,
    )?;

    Ok(())
}

/// Scenario coverage:
/// 1. Same address is destroyed at block N.
/// 2. In block N+1, it is recreated and writes new storage in the same block (multi-tx block).
/// 3. At block N+2, it is destroyed again.
/// 4. Both wipe blocks emit plain storage keys for the destroyed account.
#[tokio::test(flavor = "multi_thread")]
async fn test_pipeline_v2_single_batch_same_address_recreate_and_write_same_block_then_wipe_plain_slots(
) -> eyre::Result<()> {
    reth_tracing::init_test_tracing();

    let scenario = setup_same_address_recreate_and_write_same_block_then_wipe_scenario()?;
    let (pipeline_provider_factory, pipeline_genesis) =
        init_v2_pipeline_provider_factory(scenario.chain_spec.clone())?;

    run_pipeline_range(
        pipeline_provider_factory.clone(),
        create_file_client_from_blocks(scenario.blocks),
        pipeline_genesis,
        1..=5,
        5,
    )
    .await?;

    let provider = pipeline_provider_factory.provider()?;
    assert_eq!(provider.last_block_number()?, 5, "pipeline should sync blocks 1..=5");
    assert_preimage_rows_for_provider(&provider, &scenario.expected_preimage_slots)?;
    assert_destroyed_changeset_entries_in_block(
        &provider,
        3,
        scenario.child_contract,
        &scenario.expected_slots_first_wipe,
    )?;
    assert_destroyed_changeset_entries_in_block(
        &provider,
        5,
        scenario.child_contract,
        &scenario.expected_slots_second_wipe,
    )?;

    Ok(())
}

/// Scenario coverage:
/// 1. Intra-block multi-tx net-zero path: tx1 writes slots, tx2 wipes in the same block.
/// 2. Intra-tx net-zero path: one tx writes, restores, and wipes.
/// 3. Net-zero wipe paths emit no storage changeset rows for those accounts.
#[tokio::test(flavor = "multi_thread")]
async fn test_pipeline_v2_single_block_intra_block_and_intra_tx_wipes_use_plain_slots(
) -> eyre::Result<()> {
    reth_tracing::init_test_tracing();

    let scenario = setup_intra_block_and_intra_tx_selfdestruct_scenario()?;
    let (pipeline_provider_factory, pipeline_genesis) =
        init_v2_pipeline_provider_factory(scenario.chain_spec.clone())?;

    run_pipeline_range(
        pipeline_provider_factory.clone(),
        create_file_client_from_blocks(scenario.blocks),
        pipeline_genesis,
        1..=1,
        1,
    )
    .await?;

    let provider = pipeline_provider_factory.provider()?;
    assert_eq!(provider.last_block_number()?, 1, "pipeline should sync block 1");
    // Net-zero wipe scenarios do not require any specific preimage rows for changeset emission.
    assert_preimage_rows_for_provider(&provider, &[])?;
    assert_destroyed_changeset_entries_in_block(
        &provider,
        1,
        scenario.multi_tx_contract,
        &scenario.expected_multi_tx_slots,
    )?;
    assert_destroyed_changeset_entries_in_block(&provider, 1, scenario.intra_tx_contract, &[])?;
    Ok(())
}

struct SelfdestructScenario {
    chain_spec: Arc<reth_chainspec::ChainSpec>,
    blocks: Vec<SealedBlock<Block>>,
    selfdestruct_contract: Address,
    expected_slots: [B256; 2],
}

struct Create2SelfdestructScenario {
    chain_spec: Arc<reth_chainspec::ChainSpec>,
    block: SealedBlock<Block>,
    child_contract: Address,
    child_was_destroyed: bool,
    hashed_state: HashedPostState,
}

fn setup_selfdestruct_scenario() -> eyre::Result<SelfdestructScenario> {
    let mut rng = generators::rng();
    let key_pair = generate_key(&mut rng);
    let signer_address = public_key_to_address(key_pair.public_key());
    let selfdestruct_contract = Address::new([0x66; 20]);
    let chain_spec = build_selfdestruct_chain_spec(signer_address, selfdestruct_contract);
    let blocks = {
        // Build blocks via direct execution first, so each header has a valid state root.
        // The pipeline test then replays these exact blocks in phase-separated ranges.
        let provider_factory = create_test_provider_factory_with_chain_spec(chain_spec.clone());
        init_genesis(&provider_factory).expect("init genesis");

        let genesis = provider_factory.sealed_header(0)?.expect("genesis should exist");
        let evm_config = EthEvmConfig::new(chain_spec.clone());
        let mut blocks = Vec::new();
        let mut parent_hash = genesis.hash();
        let gas_price = INITIAL_BASE_FEE as u128;

        for (block_num, timestamp, nonce, input, to, gas_limit, value) in [
            (
                1_u64,
                12_u64,
                0_u64,
                Bytes::new(),
                TxKind::Call(selfdestruct_contract),
                100_000_u64,
                U256::ZERO,
            ),
            (
                2_u64,
                24_u64,
                1_u64,
                Bytes::from(vec![0x01]),
                TxKind::Call(selfdestruct_contract),
                100_000_u64,
                U256::ZERO,
            ),
            (
                3_u64,
                36_u64,
                2_u64,
                Bytes::new(),
                TxKind::Call(TEST_SELFDESTRUCT_BENEFICIARY),
                21_000_u64,
                U256::from(1),
            ),
        ] {
            // Block behavior by timestamp:
            // - block 1 (ts=12): writes two storage slots
            // - block 2 (ts=24): triggers SELFDESTRUCT (pre-Cancun semantics)
            // - block 3 (ts=36): post-Cancun no-op transfer path
            let tx = sign_tx_with_key_pair(
                key_pair,
                Transaction::Eip1559(TxEip1559 {
                    chain_id: chain_spec.chain.id(),
                    nonce,
                    gas_limit,
                    max_fee_per_gas: gas_price,
                    max_priority_fee_per_gas: 0,
                    to,
                    value,
                    input,
                    ..Default::default()
                }),
            );
            let block = execute_and_commit_block(
                &provider_factory,
                &evm_config,
                signer_address,
                parent_hash,
                block_num,
                timestamp,
                vec![tx],
            )?;
            parent_hash = block.hash();
            blocks.push(block);
        }

        blocks
    };

    Ok(SelfdestructScenario {
        chain_spec,
        blocks,
        selfdestruct_contract,
        expected_slots: expected_destroyed_slots(),
    })
}

fn setup_create2_selfdestruct_scenario() -> eyre::Result<Create2SelfdestructScenario> {
    let mut rng = generators::rng();
    let key_pair = generate_key(&mut rng);
    let signer_address = public_key_to_address(key_pair.public_key());
    let factory_contract = Address::new([0xaa; 20]);
    let child_init = CREATE2_SELFDESTRUCT_INIT_CODE;
    let child_contract = create2_address(factory_contract, TEST_CREATE2_SALT, &child_init);
    let chain_spec = Arc::new(
        ChainSpecBuilder::default()
            .chain(MAINNET.chain)
            .genesis(Genesis {
                alloc: [
                    (
                        signer_address,
                        GenesisAccount {
                            balance: U256::from(ETH_TO_WEI) * U256::from(1000),
                            ..Default::default()
                        },
                    ),
                    (
                        factory_contract,
                        GenesisAccount {
                            code: Some(CREATE2_FACTORY_RUNTIME_CODE),
                            ..Default::default()
                        },
                    ),
                    (child_contract, GenesisAccount { balance: U256::ONE, ..Default::default() }),
                ]
                .into(),
                ..MAINNET.genesis.clone()
            })
            .shanghai_activated()
            .with_fork(EthereumHardfork::Cancun, ForkCondition::Timestamp(30))
            .build(),
    );

    let provider_factory = create_test_provider_factory_with_chain_spec(chain_spec.clone());
    init_genesis(&provider_factory)?;
    let genesis = provider_factory.sealed_header(0)?.expect("genesis should exist");
    let evm_config = EthEvmConfig::new(chain_spec.clone());
    let tx = sign_tx_with_key_pair(
        key_pair,
        Transaction::Eip1559(TxEip1559 {
            chain_id: chain_spec.chain.id(),
            nonce: 0,
            gas_limit: 400_000,
            max_fee_per_gas: INITIAL_BASE_FEE as u128,
            max_priority_fee_per_gas: 0,
            to: TxKind::Call(factory_contract),
            input: child_init,
            ..Default::default()
        }),
    );

    let preview_block = RecoveredBlock::new_unhashed(
        Block::new(
            build_execution_header(genesis.hash(), 1, 12),
            BlockBody { transactions: vec![tx.clone()], ommers: Vec::new(), withdrawals: None },
        ),
        vec![signer_address],
    );
    let output = {
        let provider = provider_factory.database_provider_rw()?;
        let state_provider = provider.latest();
        let db = StateProviderDatabase::new(&*state_provider);
        evm_config.batch_executor(db).execute(&preview_block)?
    };
    let child_was_destroyed =
        output.state.account(&child_contract).is_some_and(|account| account.was_destroyed());
    let hashed_state = HashedPostState::from_bundle_state::<KeccakKeyHasher>(output.state.state());
    let block = execute_and_commit_block(
        &provider_factory,
        &evm_config,
        signer_address,
        genesis.hash(),
        1,
        12,
        vec![tx],
    )?;

    Ok(Create2SelfdestructScenario {
        chain_spec,
        block,
        child_contract,
        child_was_destroyed,
        hashed_state,
    })
}

struct RevertedSlotSelfdestructScenario {
    chain_spec: Arc<reth_chainspec::ChainSpec>,
    blocks: Vec<SealedBlock<Block>>,
    selfdestruct_contract: Address,
    expected_slot: (B256, U256),
}

struct SameAddressDoubleWipeScenario {
    chain_spec: Arc<reth_chainspec::ChainSpec>,
    blocks: Vec<SealedBlock<Block>>,
    child_contract: Address,
    expected_slots: [(B256, U256); 2],
}

struct SameAddressDifferentSlotsDoubleWipeScenario {
    chain_spec: Arc<reth_chainspec::ChainSpec>,
    blocks: Vec<SealedBlock<Block>>,
    child_contract: Address,
    expected_slots_first_wipe: [(B256, U256); 2],
    expected_slots_second_wipe: [(B256, U256); 2],
    expected_preimage_slots: [B256; 4],
}

struct IntraBlockAndIntraTxSelfdestructScenario {
    chain_spec: Arc<reth_chainspec::ChainSpec>,
    blocks: Vec<SealedBlock<Block>>,
    multi_tx_contract: Address,
    intra_tx_contract: Address,
    expected_multi_tx_slots: Vec<(B256, U256)>,
}

fn setup_reverted_slot_selfdestruct_scenario() -> eyre::Result<RevertedSlotSelfdestructScenario> {
    let mut rng = generators::rng();
    let key_pair = generate_key(&mut rng);
    let signer_address = public_key_to_address(key_pair.public_key());
    let selfdestruct_contract = Address::new([0x33; 20]);
    let slot = B256::with_last_byte(0x03);
    let original_value = B256::with_last_byte(0x07);

    let chain_spec = Arc::new(
        ChainSpecBuilder::default()
            .chain(MAINNET.chain)
            .genesis(Genesis {
                alloc: [
                    (
                        signer_address,
                        GenesisAccount {
                            balance: U256::from(ETH_TO_WEI) * U256::from(1000),
                            ..Default::default()
                        },
                    ),
                    (
                        selfdestruct_contract,
                        GenesisAccount {
                            code: Some(WRITE_RESTORE_OR_SELFDESTRUCT_RUNTIME_CODE),
                            storage: Some(BTreeMap::from([(slot, original_value)])),
                            ..Default::default()
                        },
                    ),
                ]
                .into(),
                ..MAINNET.genesis.clone()
            })
            .shanghai_activated()
            .with_fork(EthereumHardfork::Cancun, ForkCondition::Timestamp(30))
            .build(),
    );

    let blocks = {
        let provider_factory = create_test_provider_factory_with_chain_spec(chain_spec.clone());
        init_genesis(&provider_factory).expect("init genesis");

        let genesis = provider_factory.sealed_header(0)?.expect("genesis should exist");
        let evm_config = EthEvmConfig::new(chain_spec.clone());
        let mut blocks = Vec::new();
        let mut parent_hash = genesis.hash();
        let gas_price = INITIAL_BASE_FEE as u128;

        for (block_num, timestamp, nonce, value) in [
            (1_u64, 12_u64, 0_u64, U256::ZERO),
            (2_u64, 18_u64, 1_u64, U256::from(1_u64)),
            (3_u64, 24_u64, 2_u64, U256::from(2_u64)),
        ] {
            let tx = sign_tx_with_key_pair(
                key_pair,
                Transaction::Eip1559(TxEip1559 {
                    chain_id: chain_spec.chain.id(),
                    nonce,
                    gas_limit: 120_000,
                    max_fee_per_gas: gas_price,
                    max_priority_fee_per_gas: 0,
                    to: TxKind::Call(selfdestruct_contract),
                    value,
                    input: Bytes::new(),
                    ..Default::default()
                }),
            );
            let block = execute_and_commit_block(
                &provider_factory,
                &evm_config,
                signer_address,
                parent_hash,
                block_num,
                timestamp,
                vec![tx],
            )?;
            parent_hash = block.hash();
            blocks.push(block);
        }

        blocks
    };

    Ok(RevertedSlotSelfdestructScenario {
        chain_spec,
        blocks,
        selfdestruct_contract,
        expected_slot: (slot, U256::from(0x07)),
    })
}

fn setup_same_address_double_wipe_scenario() -> eyre::Result<SameAddressDoubleWipeScenario> {
    let mut rng = generators::rng();
    let key_pair = generate_key(&mut rng);
    let signer_address = public_key_to_address(key_pair.public_key());
    let factory_contract = Address::new([0xaa; 20]);
    let child_init = init_code_for_runtime(&WRITE_OR_SELFDESTRUCT_RUNTIME_CODE);
    let child_contract = create2_address(factory_contract, TEST_CREATE2_SALT, &child_init);

    let chain_spec = Arc::new(
        ChainSpecBuilder::default()
            .chain(MAINNET.chain)
            .genesis(Genesis {
                alloc: [
                    (
                        signer_address,
                        GenesisAccount {
                            balance: U256::from(ETH_TO_WEI) * U256::from(1000),
                            ..Default::default()
                        },
                    ),
                    (
                        factory_contract,
                        GenesisAccount {
                            code: Some(CREATE2_FACTORY_RUNTIME_CODE),
                            ..Default::default()
                        },
                    ),
                ]
                .into(),
                ..MAINNET.genesis.clone()
            })
            .shanghai_activated()
            .with_fork(EthereumHardfork::Cancun, ForkCondition::Timestamp(30))
            .build(),
    );

    let blocks = {
        let provider_factory = create_test_provider_factory_with_chain_spec(chain_spec.clone());
        init_genesis(&provider_factory).expect("init genesis");

        let genesis = provider_factory.sealed_header(0)?.expect("genesis should exist");
        let evm_config = EthEvmConfig::new(chain_spec.clone());
        let mut blocks = Vec::new();
        let mut parent_hash = genesis.hash();
        let gas_price = INITIAL_BASE_FEE as u128;

        for (block_num, timestamp, nonce, to, input) in [
            (1_u64, 4_u64, 0_u64, TxKind::Call(factory_contract), child_init.clone()),
            (2_u64, 8_u64, 1_u64, TxKind::Call(child_contract), Bytes::new()),
            (3_u64, 12_u64, 2_u64, TxKind::Call(child_contract), Bytes::from(vec![0x01])),
            (4_u64, 16_u64, 3_u64, TxKind::Call(factory_contract), child_init),
            (5_u64, 20_u64, 4_u64, TxKind::Call(child_contract), Bytes::new()),
            (6_u64, 24_u64, 5_u64, TxKind::Call(child_contract), Bytes::from(vec![0x01])),
        ] {
            let tx = sign_tx_with_key_pair(
                key_pair,
                Transaction::Eip1559(TxEip1559 {
                    chain_id: chain_spec.chain.id(),
                    nonce,
                    gas_limit: 400_000,
                    max_fee_per_gas: gas_price,
                    max_priority_fee_per_gas: 0,
                    to,
                    value: U256::ZERO,
                    input,
                    ..Default::default()
                }),
            );
            let block = execute_and_commit_block(
                &provider_factory,
                &evm_config,
                signer_address,
                parent_hash,
                block_num,
                timestamp,
                vec![tx],
            )?;
            parent_hash = block.hash();
            blocks.push(block);
        }

        blocks
    };

    Ok(SameAddressDoubleWipeScenario {
        chain_spec,
        blocks,
        child_contract,
        expected_slots: [
            (B256::with_last_byte(0x01), U256::from(0x2a)),
            (B256::with_last_byte(0x02), U256::from(0x99)),
        ],
    })
}

fn setup_same_address_recreate_and_write_same_block_then_wipe_scenario(
) -> eyre::Result<SameAddressDifferentSlotsDoubleWipeScenario> {
    let mut rng = generators::rng();
    let key_pair = generate_key(&mut rng);
    let signer_address = public_key_to_address(key_pair.public_key());
    let factory_contract = Address::new([0xaa; 20]);
    let child_init = init_code_for_runtime(&WRITE_TWO_SLOT_SETS_OR_SELFDESTRUCT_RUNTIME_CODE);
    let child_contract = create2_address(factory_contract, TEST_CREATE2_SALT, &child_init);

    let chain_spec = Arc::new(
        ChainSpecBuilder::default()
            .chain(MAINNET.chain)
            .genesis(Genesis {
                alloc: [
                    (
                        signer_address,
                        GenesisAccount {
                            balance: U256::from(ETH_TO_WEI) * U256::from(1000),
                            ..Default::default()
                        },
                    ),
                    (
                        factory_contract,
                        GenesisAccount {
                            code: Some(CREATE2_FACTORY_RUNTIME_CODE),
                            ..Default::default()
                        },
                    ),
                ]
                .into(),
                ..MAINNET.genesis.clone()
            })
            .shanghai_activated()
            .with_fork(EthereumHardfork::Cancun, ForkCondition::Timestamp(30))
            .build(),
    );

    let blocks = {
        let provider_factory = create_test_provider_factory_with_chain_spec(chain_spec.clone());
        init_genesis(&provider_factory).expect("init genesis");

        let genesis = provider_factory.sealed_header(0)?.expect("genesis should exist");
        let evm_config = EthEvmConfig::new(chain_spec.clone());
        let gas_price = INITIAL_BASE_FEE as u128;

        let mut parent_hash = genesis.hash();
        let mk_tx = |nonce: u64, to: TxKind, value: U256, input: Bytes| {
            sign_tx_with_key_pair(
                key_pair,
                Transaction::Eip1559(TxEip1559 {
                    chain_id: chain_spec.chain.id(),
                    nonce,
                    gas_limit: 400_000,
                    max_fee_per_gas: gas_price,
                    max_priority_fee_per_gas: 0,
                    to,
                    value,
                    input,
                    ..Default::default()
                }),
            )
        };

        let block1 = execute_and_commit_block(
            &provider_factory,
            &evm_config,
            signer_address,
            parent_hash,
            1,
            4,
            vec![mk_tx(0, TxKind::Call(factory_contract), U256::ZERO, child_init.clone())],
        )?;
        parent_hash = block1.hash();

        let block2 = execute_and_commit_block(
            &provider_factory,
            &evm_config,
            signer_address,
            parent_hash,
            2,
            8,
            vec![mk_tx(1, TxKind::Call(child_contract), U256::ZERO, Bytes::new())],
        )?;
        parent_hash = block2.hash();

        let block3 = execute_and_commit_block(
            &provider_factory,
            &evm_config,
            signer_address,
            parent_hash,
            3,
            12,
            vec![mk_tx(2, TxKind::Call(child_contract), U256::from(1), Bytes::new())],
        )?;
        parent_hash = block3.hash();

        // Recreate and write new slots in the same block (N+1).
        let block4 = execute_and_commit_block(
            &provider_factory,
            &evm_config,
            signer_address,
            parent_hash,
            4,
            16,
            vec![
                mk_tx(3, TxKind::Call(factory_contract), U256::ZERO, child_init),
                mk_tx(4, TxKind::Call(child_contract), U256::from(2), Bytes::new()),
            ],
        )?;
        parent_hash = block4.hash();

        let block5 = execute_and_commit_block(
            &provider_factory,
            &evm_config,
            signer_address,
            parent_hash,
            5,
            20,
            vec![mk_tx(5, TxKind::Call(child_contract), U256::from(1), Bytes::new())],
        )?;

        vec![block1, block2, block3, block4, block5]
    };

    Ok(SameAddressDifferentSlotsDoubleWipeScenario {
        chain_spec,
        blocks,
        child_contract,
        expected_slots_first_wipe: [
            (B256::with_last_byte(0x01), U256::from(0x2a)),
            (B256::with_last_byte(0x02), U256::from(0x99)),
        ],
        expected_slots_second_wipe: [
            (B256::with_last_byte(0x04), U256::from(0xab)),
            (B256::with_last_byte(0x05), U256::from(0xcd)),
        ],
        expected_preimage_slots: [
            B256::with_last_byte(0x01),
            B256::with_last_byte(0x02),
            B256::with_last_byte(0x04),
            B256::with_last_byte(0x05),
        ],
    })
}

fn setup_intra_block_and_intra_tx_selfdestruct_scenario(
) -> eyre::Result<IntraBlockAndIntraTxSelfdestructScenario> {
    let mut rng = generators::rng();
    let key_pair = generate_key(&mut rng);
    let signer_address = public_key_to_address(key_pair.public_key());
    let multi_tx_contract = Address::new([0x44; 20]);
    let intra_tx_contract = Address::new([0x55; 20]);

    let chain_spec = Arc::new(
        ChainSpecBuilder::default()
            .chain(MAINNET.chain)
            .genesis(Genesis {
                alloc: [
                    (
                        signer_address,
                        GenesisAccount {
                            balance: U256::from(ETH_TO_WEI) * U256::from(1000),
                            ..Default::default()
                        },
                    ),
                    (
                        multi_tx_contract,
                        GenesisAccount {
                            code: Some(WRITE_OR_SELFDESTRUCT_RUNTIME_CODE),
                            ..Default::default()
                        },
                    ),
                    (
                        intra_tx_contract,
                        GenesisAccount {
                            code: Some(WRITE_RESTORE_THEN_SELFDESTRUCT_RUNTIME_CODE),
                            ..Default::default()
                        },
                    ),
                ]
                .into(),
                ..MAINNET.genesis.clone()
            })
            .shanghai_activated()
            .with_fork(EthereumHardfork::Cancun, ForkCondition::Timestamp(30))
            .build(),
    );

    let blocks = {
        let provider_factory = create_test_provider_factory_with_chain_spec(chain_spec.clone());
        init_genesis(&provider_factory).expect("init genesis");

        let genesis = provider_factory.sealed_header(0)?.expect("genesis should exist");
        let evm_config = EthEvmConfig::new(chain_spec.clone());
        let gas_price = INITIAL_BASE_FEE as u128;

        // Single pre-Cancun block with three txs:
        // 1) write slots (contract `multi_tx_contract`)
        // 2) selfdestruct same contract in same block
        // 3) write->restore->selfdestruct in one tx (`intra_tx_contract`)
        let txs = vec![
            sign_tx_with_key_pair(
                key_pair,
                Transaction::Eip1559(TxEip1559 {
                    chain_id: chain_spec.chain.id(),
                    nonce: 0,
                    gas_limit: 120_000,
                    max_fee_per_gas: gas_price,
                    max_priority_fee_per_gas: 0,
                    to: TxKind::Call(multi_tx_contract),
                    value: U256::ZERO,
                    input: Bytes::new(),
                    ..Default::default()
                }),
            ),
            sign_tx_with_key_pair(
                key_pair,
                Transaction::Eip1559(TxEip1559 {
                    chain_id: chain_spec.chain.id(),
                    nonce: 1,
                    gas_limit: 120_000,
                    max_fee_per_gas: gas_price,
                    max_priority_fee_per_gas: 0,
                    to: TxKind::Call(multi_tx_contract),
                    value: U256::ZERO,
                    input: Bytes::from(vec![0x01]),
                    ..Default::default()
                }),
            ),
            sign_tx_with_key_pair(
                key_pair,
                Transaction::Eip1559(TxEip1559 {
                    chain_id: chain_spec.chain.id(),
                    nonce: 2,
                    gas_limit: 120_000,
                    max_fee_per_gas: gas_price,
                    max_priority_fee_per_gas: 0,
                    to: TxKind::Call(intra_tx_contract),
                    value: U256::ZERO,
                    input: Bytes::new(),
                    ..Default::default()
                }),
            ),
        ];

        let block = execute_and_commit_block(
            &provider_factory,
            &evm_config,
            signer_address,
            genesis.hash(),
            1,
            12,
            txs,
        )?;

        vec![block]
    };

    Ok(IntraBlockAndIntraTxSelfdestructScenario {
        chain_spec,
        blocks,
        multi_tx_contract,
        intra_tx_contract,
        expected_multi_tx_slots: Vec::new(),
    })
}

fn init_v2_pipeline_provider_factory(
    chain_spec: Arc<reth_chainspec::ChainSpec>,
) -> eyre::Result<(TestProviderFactory, reth_primitives_traits::SealedHeader<Header>)> {
    let pipeline_provider_factory = create_test_provider_factory_with_chain_spec(chain_spec);
    init_genesis_with_settings(&pipeline_provider_factory, StorageSettings::v2())?;
    pipeline_provider_factory.set_storage_settings_cache(StorageSettings::v2());
    let pipeline_genesis = pipeline_provider_factory
        .sealed_header(0)?
        .ok_or_else(|| eyre::eyre!("genesis should exist"))?;
    Ok((pipeline_provider_factory, pipeline_genesis))
}

fn execute_and_commit_block(
    provider_factory: &TestProviderFactory,
    evm_config: &EthEvmConfig,
    signer_address: Address,
    parent_hash: B256,
    block_num: u64,
    timestamp: u64,
    transactions: Vec<TransactionSigned>,
) -> eyre::Result<SealedBlock<Block>> {
    let tx_root = calculate_transaction_root(&transactions);
    let temp_header = build_execution_header(parent_hash, block_num, timestamp);
    let provider = provider_factory.database_provider_rw()?;
    let block_with_senders = RecoveredBlock::new_unhashed(
        Block::new(
            temp_header,
            BlockBody { transactions: transactions.clone(), ommers: Vec::new(), withdrawals: None },
        ),
        vec![signer_address; transactions.len()],
    );

    let output = {
        let state_provider = provider.latest();
        let db = StateProviderDatabase::new(&*state_provider);
        let executor = evm_config.batch_executor(db);
        executor.execute(&block_with_senders)?
    };

    let gas_used = output.gas_used;
    let hashed_state = provider.latest().hashed_post_state(&output.state)?;
    type TestStateRoot<'a, TX, A> = StateRoot<
        reth_trie_db::DatabaseTrieCursorFactory<&'a TX, A>,
        reth_trie_db::DatabaseHashedCursorFactory<&'a TX>,
    >;
    let (state_root, _trie_updates) = reth_trie_db::with_adapter!(provider, |A| {
        TestStateRoot::<_, A>::overlay_root_with_updates(
            provider.tx_ref(),
            &hashed_state.clone().into_sorted(),
        )
    })?;

    let receipts: Vec<_> = output.receipts.iter().map(|r| r.with_bloom_ref()).collect();
    let receipts_root = calculate_receipt_root(&receipts);

    let header = Header {
        parent_hash,
        number: block_num,
        state_root,
        transactions_root: tx_root,
        receipts_root,
        gas_limit: 30_000_000,
        gas_used,
        base_fee_per_gas: Some(INITIAL_BASE_FEE),
        timestamp,
        parent_beacon_block_root: (timestamp >= 30).then_some(B256::ZERO),
        blob_gas_used: (timestamp >= 30).then_some(0),
        excess_blob_gas: (timestamp >= 30).then_some(0),
        withdrawals_root: Some(EMPTY_WITHDRAWALS),
        ..Default::default()
    };

    let block: SealedBlock<Block> = SealedBlock::seal_parts(
        header,
        BlockBody { transactions, ommers: Vec::new(), withdrawals: None },
    );

    let plain_state = output.state.to_plain_state(OriginalValuesKnown::Yes);
    provider.write_state_changes(plain_state)?;
    provider.write_hashed_state(&hashed_state.into_sorted())?;
    provider.commit()?;

    Ok(block)
}

fn build_selfdestruct_chain_spec(
    signer_address: Address,
    selfdestruct_contract: Address,
) -> Arc<reth_chainspec::ChainSpec> {
    let initial_balance = U256::from(ETH_TO_WEI) * U256::from(1000);

    Arc::new(
        ChainSpecBuilder::default()
            .chain(MAINNET.chain)
            .genesis(Genesis {
                alloc: [
                    (
                        signer_address,
                        GenesisAccount { balance: initial_balance, ..Default::default() },
                    ),
                    (
                        selfdestruct_contract,
                        GenesisAccount {
                            code: Some(WRITE_OR_SELFDESTRUCT_RUNTIME_CODE),
                            ..Default::default()
                        },
                    ),
                ]
                .into(),
                ..MAINNET.genesis.clone()
            })
            .shanghai_activated()
            .with_fork(EthereumHardfork::Cancun, ForkCondition::Timestamp(30))
            .build(),
    )
}

fn build_execution_header(parent_hash: B256, number: u64, timestamp: u64) -> Header {
    Header {
        parent_hash,
        number,
        gas_limit: 30_000_000,
        base_fee_per_gas: Some(INITIAL_BASE_FEE),
        timestamp,
        parent_beacon_block_root: (timestamp >= 30).then_some(B256::ZERO),
        blob_gas_used: (timestamp >= 30).then_some(0),
        excess_blob_gas: (timestamp >= 30).then_some(0),
        ..Default::default()
    }
}

const fn expected_destroyed_slots() -> [B256; 2] {
    [B256::with_last_byte(0x01), B256::with_last_byte(0x02)]
}

fn assert_destroyed_changeset_entries<P>(
    provider: &P,
    selfdestruct_contract: Address,
) -> eyre::Result<()>
where
    P: StorageChangeSetReader,
{
    let expected = [
        (B256::with_last_byte(0x01), U256::from(0x2a)),
        (B256::with_last_byte(0x02), U256::from(0x99)),
    ];
    assert_destroyed_changeset_entries_in_block(provider, 2, selfdestruct_contract, &expected)
}

fn assert_destroyed_changeset_entries_in_block<P>(
    provider: &P,
    block: u64,
    selfdestruct_contract: Address,
    expected: &[(B256, U256)],
) -> eyre::Result<()>
where
    P: StorageChangeSetReader,
{
    let storage_changesets = provider.storage_changesets_range(block..=block)?;
    let destroyed_entries: Vec<_> = storage_changesets
        .into_iter()
        .filter_map(|(key, entry)| {
            (key.address() == selfdestruct_contract).then_some((entry.key, entry.value))
        })
        .collect();

    assert_eq!(
        destroyed_entries.len(),
        expected.len(),
        "expected exactly {} storage changeset entries for destroyed account at block {}",
        expected.len(),
        block
    );

    for (slot, _) in &destroyed_entries {
        assert_ne!(*slot, keccak256(*slot), "storage changeset key should be plain (not hashed)");
    }

    for pair in expected {
        assert!(
            destroyed_entries.contains(pair),
            "missing expected storage changeset entry for destroyed account: {:?}",
            pair
        );
    }

    Ok(())
}

fn assert_hashed_storage_entries<P>(
    provider: &P,
    address: Address,
    expected: &[(B256, U256)],
) -> eyre::Result<()>
where
    P: DBProvider,
{
    let mut cursor = provider.tx_ref().cursor_dup_read::<tables::HashedStorages>()?;
    let mut actual = Vec::new();
    if let Some((_, entry)) = cursor.seek_exact(keccak256(address))? {
        actual.push((entry.key, entry.value));
        while let Some(entry) = cursor.next_dup_val()? {
            actual.push((entry.key, entry.value));
        }
    }
    actual.sort_unstable_by_key(|(slot, _)| *slot);

    let mut expected =
        expected.iter().map(|(slot, value)| (keccak256(*slot), *value)).collect::<Vec<_>>();
    expected.sort_unstable_by_key(|(slot, _)| *slot);

    assert_eq!(actual, expected, "unexpected hashed storage for account {address}");
    Ok(())
}

fn create_file_client_from_blocks(blocks: Vec<SealedBlock<Block>>) -> Arc<FileClient<Block>> {
    Arc::new(FileClient::from_blocks(blocks))
}

fn build_pipeline_without_history<H, B>(
    provider_factory: TestProviderFactory,
    header_downloader: H,
    body_downloader: B,
    max_block: u64,
    tip: B256,
) -> Pipeline<reth_provider::test_utils::MockNodeTypesWithDB>
where
    H: HeaderDownloader<Header = Header> + 'static,
    B: BodyDownloader<Block = Block> + 'static,
{
    let consensus = NoopConsensus::arc();
    let stages_config = StageConfig::default();
    let evm_config = EthEvmConfig::new(provider_factory.chain_spec());

    let (tip_tx, tip_rx) = watch::channel(B256::ZERO);
    let static_file_producer =
        StaticFileProducer::new(provider_factory.clone(), PruneModes::default());

    let stages = OnlineStages::new(
        provider_factory.clone(),
        tip_rx,
        header_downloader,
        body_downloader,
        stages_config.clone(),
        None,
    )
    .builder()
    .add_set(ExecutionStages::new(
        evm_config,
        consensus,
        stages_config,
        PruneModes::default().sender_recovery,
    ))
    .add_set(HashingStages::default())
    .add_stage(FinishStage::default());

    let pipeline = Pipeline::builder()
        .with_tip_sender(tip_tx)
        .with_max_block(max_block)
        .with_fail_on_unwind(true)
        .add_stages(stages)
        .build(provider_factory, static_file_producer);
    pipeline.set_tip(tip);
    pipeline
}

async fn run_pipeline_range(
    provider_factory: TestProviderFactory,
    file_client: Arc<FileClient<Block>>,
    local_head: reth_primitives_traits::SealedHeader<Header>,
    download_range: std::ops::RangeInclusive<u64>,
    max_block: u64,
) -> eyre::Result<()> {
    // Run a narrow range intentionally so the test can assert per-phase behavior.
    let tip = file_client.tip().expect("tip");
    let consensus = NoopConsensus::arc();
    let stages_config = StageConfig::default();
    let runtime = reth_tasks::Runtime::test();

    let mut header_downloader = ReverseHeadersDownloaderBuilder::new(stages_config.headers)
        .build(file_client.clone(), consensus.clone())
        .into_task_with(&runtime);
    header_downloader.update_local_head(local_head);
    header_downloader.update_sync_target(SyncTarget::Tip(tip));

    let mut body_downloader = BodiesDownloaderBuilder::new(stages_config.bodies)
        .build(file_client, consensus, provider_factory.clone())
        .into_task_with(&runtime);
    body_downloader.set_download_range(download_range).expect("set download range");

    let pipeline = build_pipeline_without_history(
        provider_factory,
        header_downloader,
        body_downloader,
        max_block,
        tip,
    );
    let (_pipeline, result) = pipeline.run_as_fut(None).await;
    result?;
    Ok(())
}

/// Builds tiny runtime bytecode that branches on calldata:
/// - empty calldata: writes two known slots and stops
/// - non-empty calldata: selfdestructs to `beneficiary` and stops
///
/// The known slot/value pairs are used for deterministic assertions in changesets and preimages.
const WRITE_OR_SELFDESTRUCT_RUNTIME_CODE: Bytes = bytes!(
    "3615601c57" // CALLDATASIZE; ISZERO; PUSH1 0x1c; JUMPI
    "737777777777777777777777777777777777777777" // PUSH20 beneficiary
    "ff00" // SELFDESTRUCT; STOP
    "5b" // JUMPDEST (0x1c)
    "602a600155" // SSTORE(1, 0x2a)
    "6099600255" // SSTORE(2, 0x99)
    "00" // STOP
);

/// Builds tiny runtime bytecode with three callvalue-based paths:
/// - `msg.value == 0`: write slot-set A (`(1, 0x2a)`, `(2, 0x99)`)
/// - `msg.value == 2`: write slot-set B (`(4, 0xab)`, `(5, 0xcd)`)
/// - `msg.value == 1`: selfdestruct to `beneficiary`
///
/// Used by the recreate-and-write-in-same-block scenario to ensure first and second wipes
/// restore different slot sets for the same address.
const WRITE_TWO_SLOT_SETS_OR_SELFDESTRUCT_RUNTIME_CODE: Bytes = bytes!(
    "34600114602557" // if callvalue == 1 jump selfdestruct
    "34600214601957" // if callvalue == 2 jump write slot-set B
    "602a600155" // write slot-set A: SSTORE(1, 0x2a)
    "6099600255" // write slot-set A: SSTORE(2, 0x99)
    "00" // STOP
    "5b" // JUMPDEST (0x19)
    "60ab600455" // write slot-set B: SSTORE(4, 0xab)
    "60cd600555" // write slot-set B: SSTORE(5, 0xcd)
    "00" // STOP
    "5b" // JUMPDEST (0x25)
    "737777777777777777777777777777777777777777" // PUSH20 beneficiary
    "ff00" // SELFDESTRUCT; STOP
);

/// Builds tiny runtime bytecode with three value-based paths:
/// - `msg.value == 0`: SSTORE(3, 0x2b)
/// - `msg.value == 1`: SSTORE(3, 0x07)
/// - `msg.value == 2`: SELFDESTRUCT to `beneficiary`
const WRITE_RESTORE_OR_SELFDESTRUCT_RUNTIME_CODE: Bytes = bytes!(
    "34600214601b57" // if callvalue == 2 jump selfdestruct
    "34600114601457" // if callvalue == 1 jump restore
    "602b60035500" // default: SSTORE(3, 0x2b); STOP
    "5b" // JUMPDEST (0x14)
    "600760035500" // restore: SSTORE(3, 0x07); STOP
    "5b" // JUMPDEST (0x1b)
    "737777777777777777777777777777777777777777" // PUSH20 beneficiary
    "ff00" // SELFDESTRUCT; STOP
);

/// Builds tiny runtime bytecode that performs all actions in one call:
/// SSTORE(3, 0x2b) -> SSTORE(3, 0x07) -> SELFDESTRUCT.
const WRITE_RESTORE_THEN_SELFDESTRUCT_RUNTIME_CODE: Bytes = bytes!(
    "602b600355" // SSTORE(3, 0x2b)
    "6007600355" // SSTORE(3, 0x07)
    "737777777777777777777777777777777777777777" // PUSH20 beneficiary
    "ff00" // SELFDESTRUCT; STOP
);

/// CREATE2 init code that selfdestructs before returning runtime bytecode.
const CREATE2_SELFDESTRUCT_INIT_CODE: Bytes = bytes!(
    "737777777777777777777777777777777777777777" // PUSH20 beneficiary
    "ff" // SELFDESTRUCT
);

/// Converts contract runtime bytecode into init code that returns the runtime.
fn init_code_for_runtime(runtime: &Bytes) -> Bytes {
    let len = u8::try_from(runtime.len()).expect("runtime too large for PUSH1 init-code helper");
    let mut init = Vec::with_capacity(12 + runtime.len());
    init.extend_from_slice(&[
        0x60, len, // PUSH1 runtime_len
        0x60, 0x0c, // PUSH1 runtime_offset
        0x60, 0x00, // PUSH1 mem_offset
        0x39, // CODECOPY
        0x60, len, // PUSH1 runtime_len
        0x60, 0x00, // PUSH1 mem_offset
        0xf3, // RETURN
    ]);
    init.extend_from_slice(runtime.as_ref());
    init.into()
}

/// Runtime bytecode for a minimal CREATE2 factory:
/// - calldata is treated as init code
/// - deploys with fixed `salt`
const CREATE2_FACTORY_RUNTIME_CODE: Bytes = bytes!(
    "366000600037" // CALLDATACOPY(0, 0, calldatasize)
    "7f0000000000000000000000000000000000000000000000000000000000000042" // PUSH32 salt
    "3660006000f500" // CREATE2(0, 0, calldatasize, salt); STOP
);

fn create2_address(factory: Address, salt: B256, init_code: &Bytes) -> Address {
    let init_hash = keccak256(init_code.as_ref());
    let mut preimage = [0_u8; 85];
    preimage[0] = 0xff;
    preimage[1..21].copy_from_slice(factory.as_slice());
    preimage[21..53].copy_from_slice(salt.as_slice());
    preimage[53..85].copy_from_slice(init_hash.as_slice());

    let hash = keccak256(preimage);
    Address::from_slice(&hash.as_slice()[12..])
}

fn assert_preimage_rows(preimage_path: &Path, slots: &[B256]) -> eyre::Result<()> {
    let mut builder = Environment::builder();
    builder.set_max_dbs(1);
    builder.set_flags(EnvironmentFlags {
        no_sub_dir: false,
        mode: Mode::ReadOnly,
        ..Default::default()
    });

    let env = builder.open(preimage_path)?;
    let tx = env.begin_ro_txn()?;
    let db = tx.open_db(None)?;

    for slot in slots {
        let hashed = keccak256(*slot);
        let found: Option<[u8; 32]> = tx.get(db.dbi(), hashed.as_slice())?;
        assert_eq!(
            found.map(B256::from),
            Some(*slot),
            "missing/invalid preimage row for slot {:?}",
            slot
        );
    }

    Ok(())
}

fn assert_preimage_rows_for_provider<P>(provider: &P, slots: &[B256]) -> eyre::Result<()>
where
    P: StoragePath,
{
    let preimage_path = provider.storage_path().join("preimage");
    assert!(
        preimage_path.exists(),
        "preimage dir should exist for pre-Cancun execution at {}",
        preimage_path.display()
    );
    assert_preimage_rows(&preimage_path, slots)
}
