Contractless/src/records/memory/mempool/selection.rs

use super::*;
use crate::records::record_chain::pending_effects::{BalanceOperand, PendingEffects};
use crate::records::record_chain::wallet_tx_index::index_wallet_transaction;

fn index_selected_wallet_transaction(
    pending_effects: &mut PendingEffects,
    addresses: &[&str],
    block_height: u32,
    entry_index: usize,
    txid: &[u8],
) -> Result<()> {
    let entry_index =
        u32::try_from(entry_index).map_err(|_| anyhow!("Wallet transaction index overflowed"))?;
    index_wallet_transaction(pending_effects, addresses, block_height, entry_index, txid)
        .map_err(|err| anyhow!(err))
}

pub async fn select_transactions_for_block(limit: i64) -> Result<SelectedMempoolBatch> {
    // Pull the highest-priority unprocessed rows across all mempool
    // tables, keeping the original bytes for block-file streaming later.
    let rows =
        pg_query(
            r#"
            SELECT * FROM (
                SELECT
                    'transfer'::TEXT AS kind, id, fee AS priority_fee, time,
                    sender, receiver, coin, value, nft_series,
                    NULL::BIGINT AS number, NULL::TEXT AS creator, NULL::TEXT AS ticker,
                    NULL::TEXT AS nft_name, NULL::SMALLINT AS series, NULL::BIGINT AS count,
                    NULL::TEXT AS advertiser, NULL::BIGINT AS fee1, NULL::BIGINT AS fee2,
                    NULL::TEXT AS ticker1, NULL::INTEGER AS nft_series1, NULL::TEXT AS ticker2, NULL::INTEGER AS nft_series2, NULL::BIGINT AS value1,
                    NULL::BIGINT AS value2, NULL::TEXT AS sender1, NULL::BIGINT AS tip1,
                    NULL::BIGINT AS tip2, NULL::TEXT AS sender2, NULL::TEXT AS loan_coin,
                    NULL::BIGINT AS loan_amount, NULL::TEXT AS lender, NULL::TEXT AS collateral,
                    NULL::BIGINT AS collateral_amount, NULL::TEXT AS borrower,
                    NULL::BIGINT AS payback_amount, NULL::TEXT AS contract_hash, NULL::TEXT AS address,
                    hash, signature AS signature1, NULL::TEXT AS signature2, NULL::TEXT AS stored_txid, NULL::SMALLINT AS hard_limit, original
                FROM transfer
                WHERE processed = false

                UNION ALL

                SELECT
                    'token'::TEXT AS kind, id, fee AS priority_fee, time,
                    NULL::TEXT AS sender, NULL::TEXT AS receiver, NULL::TEXT AS coin, NULL::BIGINT AS value, NULL::INTEGER AS nft_series,
                    number, creator, ticker, NULL::TEXT AS nft_name, NULL::SMALLINT AS series,
                    NULL::BIGINT AS count, NULL::TEXT AS advertiser, NULL::BIGINT AS fee1,
                    NULL::BIGINT AS fee2, NULL::TEXT AS ticker1, NULL::INTEGER AS nft_series1, NULL::TEXT AS ticker2, NULL::INTEGER AS nft_series2,
                    NULL::BIGINT AS value1, NULL::BIGINT AS value2, NULL::TEXT AS sender1,
                    NULL::BIGINT AS tip1, NULL::BIGINT AS tip2, NULL::TEXT AS sender2,
                    NULL::TEXT AS loan_coin, NULL::BIGINT AS loan_amount, NULL::TEXT AS lender,
                    NULL::TEXT AS collateral, NULL::BIGINT AS collateral_amount, NULL::TEXT AS borrower,
                    NULL::BIGINT AS payback_amount, NULL::TEXT AS contract_hash, NULL::TEXT AS address,
                    hash, signature AS signature1, NULL::TEXT AS signature2, NULL::TEXT AS stored_txid, hard_limit, original
                FROM token
                WHERE processed = false

                UNION ALL

                SELECT
                    'issue_token'::TEXT AS kind, id, fee AS priority_fee, time,
                    NULL::TEXT AS sender, NULL::TEXT AS receiver, NULL::TEXT AS coin, NULL::BIGINT AS value, NULL::INTEGER AS nft_series,
                    number, creator, ticker, NULL::TEXT AS nft_name, NULL::SMALLINT AS series,
                    NULL::BIGINT AS count, NULL::TEXT AS advertiser, NULL::BIGINT AS fee1,
                    NULL::BIGINT AS fee2, NULL::TEXT AS ticker1, NULL::INTEGER AS nft_series1, NULL::TEXT AS ticker2, NULL::INTEGER AS nft_series2,
                    NULL::BIGINT AS value1, NULL::BIGINT AS value2, NULL::TEXT AS sender1,
                    NULL::BIGINT AS tip1, NULL::BIGINT AS tip2, NULL::TEXT AS sender2,
                    NULL::TEXT AS loan_coin, NULL::BIGINT AS loan_amount, NULL::TEXT AS lender,
                    NULL::TEXT AS collateral, NULL::BIGINT AS collateral_amount, NULL::TEXT AS borrower,
                    NULL::BIGINT AS payback_amount, NULL::TEXT AS contract_hash, NULL::TEXT AS address,
                    hash, signature AS signature1, NULL::TEXT AS signature2, NULL::TEXT AS stored_txid, NULL::SMALLINT AS hard_limit, original
                FROM issue_token
                WHERE processed = false

                UNION ALL

                SELECT
                    'burn'::TEXT AS kind, id, fee AS priority_fee, time,
                    NULL::TEXT AS sender, NULL::TEXT AS receiver, coin, value, nft_series,
                    NULL::BIGINT AS number, NULL::TEXT AS creator, NULL::TEXT AS ticker,
                    NULL::TEXT AS nft_name, NULL::SMALLINT AS series, NULL::BIGINT AS count,
                    NULL::TEXT AS advertiser, NULL::BIGINT AS fee1, NULL::BIGINT AS fee2,
                    NULL::TEXT AS ticker1, NULL::INTEGER AS nft_series1, NULL::TEXT AS ticker2, NULL::INTEGER AS nft_series2,
                    NULL::BIGINT AS value1, NULL::BIGINT AS value2, NULL::TEXT AS sender1,
                    NULL::BIGINT AS tip1, NULL::BIGINT AS tip2, NULL::TEXT AS sender2,
                    NULL::TEXT AS loan_coin, NULL::BIGINT AS loan_amount, NULL::TEXT AS lender,
                    NULL::TEXT AS collateral, NULL::BIGINT AS collateral_amount, NULL::TEXT AS borrower,
                    NULL::BIGINT AS payback_amount, NULL::TEXT AS contract_hash, address,
                    hash, signature AS signature1, NULL::TEXT AS signature2, NULL::TEXT AS stored_txid, NULL::SMALLINT AS hard_limit, original
                FROM burn
                WHERE processed = false

                UNION ALL

                SELECT
                    'nft'::TEXT AS kind, id, fee AS priority_fee, time,
                    NULL::TEXT AS sender, NULL::TEXT AS receiver, NULL::TEXT AS coin, NULL::BIGINT AS value, NULL::INTEGER AS nft_series,
                    NULL::BIGINT AS number, creator, NULL::TEXT AS ticker, nft_name, series, count,
                    NULL::TEXT AS advertiser, NULL::BIGINT AS fee1, NULL::BIGINT AS fee2,
                    NULL::TEXT AS ticker1, NULL::INTEGER AS nft_series1, NULL::TEXT AS ticker2, NULL::INTEGER AS nft_series2, NULL::BIGINT AS value1,
                    NULL::BIGINT AS value2, NULL::TEXT AS sender1, NULL::BIGINT AS tip1,
                    NULL::BIGINT AS tip2, NULL::TEXT AS sender2, NULL::TEXT AS loan_coin,
                    NULL::BIGINT AS loan_amount, NULL::TEXT AS lender, NULL::TEXT AS collateral,
                    NULL::BIGINT AS collateral_amount, NULL::TEXT AS borrower,
                    NULL::BIGINT AS payback_amount, NULL::TEXT AS contract_hash, NULL::TEXT AS address,
                    hash, signature AS signature1, NULL::TEXT AS signature2, NULL::TEXT AS stored_txid, NULL::SMALLINT AS hard_limit, original
                FROM nft
                WHERE processed = false

                UNION ALL

                SELECT
                    'marketing'::TEXT AS kind, id, fee AS priority_fee, time,
                    NULL::TEXT AS sender, NULL::TEXT AS receiver, NULL::TEXT AS coin, NULL::BIGINT AS value, NULL::INTEGER AS nft_series,
                    NULL::BIGINT AS number, NULL::TEXT AS creator, NULL::TEXT AS ticker,
                    NULL::TEXT AS nft_name, NULL::SMALLINT AS series, NULL::BIGINT AS count,
                    advertiser, NULL::BIGINT AS fee1, NULL::BIGINT AS fee2, NULL::TEXT AS ticker1, NULL::INTEGER AS nft_series1,
                    NULL::TEXT AS ticker2, NULL::INTEGER AS nft_series2, NULL::BIGINT AS value1, NULL::BIGINT AS value2,
                    NULL::TEXT AS sender1, NULL::BIGINT AS tip1, NULL::BIGINT AS tip2,
                    NULL::TEXT AS sender2, NULL::TEXT AS loan_coin, NULL::BIGINT AS loan_amount,
                    NULL::TEXT AS lender, NULL::TEXT AS collateral, NULL::BIGINT AS collateral_amount,
                    NULL::TEXT AS borrower, NULL::BIGINT AS payback_amount, NULL::TEXT AS contract_hash,
                    NULL::TEXT AS address, hash, signature AS signature1, NULL::TEXT AS signature2,
                    NULL::TEXT AS stored_txid, NULL::SMALLINT AS hard_limit, original
                FROM marketing
                WHERE processed = false

                UNION ALL

                SELECT
                    'vanity_address'::TEXT AS kind, id, fee AS priority_fee, time,
                    NULL::TEXT AS sender, NULL::TEXT AS receiver, NULL::TEXT AS coin, NULL::BIGINT AS value, NULL::INTEGER AS nft_series,
                    NULL::BIGINT AS number, NULL::TEXT AS creator, NULL::TEXT AS ticker,
                    NULL::TEXT AS nft_name, NULL::SMALLINT AS series, NULL::BIGINT AS count,
                    NULL::TEXT AS advertiser, NULL::BIGINT AS fee1, NULL::BIGINT AS fee2, NULL::TEXT AS ticker1, NULL::INTEGER AS nft_series1,
                    NULL::TEXT AS ticker2, NULL::INTEGER AS nft_series2, NULL::BIGINT AS value1, NULL::BIGINT AS value2,
                    NULL::TEXT AS sender1, NULL::BIGINT AS tip1, NULL::BIGINT AS tip2,
                    NULL::TEXT AS sender2, NULL::TEXT AS loan_coin, NULL::BIGINT AS loan_amount,
                    NULL::TEXT AS lender, NULL::TEXT AS collateral, NULL::BIGINT AS collateral_amount,
                    NULL::TEXT AS borrower, NULL::BIGINT AS payback_amount, NULL::TEXT AS contract_hash,
                    address, hash, signature AS signature1, NULL::TEXT AS signature2,
                    vanity_address AS stored_txid, NULL::SMALLINT AS hard_limit, original
                FROM vanity_address
                WHERE processed = false

                UNION ALL

                SELECT
                    'swap'::TEXT AS kind, id, GREATEST(fee1, fee2) AS priority_fee, time,
                    NULL::TEXT AS sender, NULL::TEXT AS receiver, NULL::TEXT AS coin, NULL::BIGINT AS value, NULL::INTEGER AS nft_series,
                    NULL::BIGINT AS number, NULL::TEXT AS creator, NULL::TEXT AS ticker,
                    NULL::TEXT AS nft_name, NULL::SMALLINT AS series, NULL::BIGINT AS count,
                    NULL::TEXT AS advertiser, fee1, fee2, ticker1, nft_series1, ticker2, nft_series2, value1, value2,
                    sender1, tip1, tip2, sender2, NULL::TEXT AS loan_coin, NULL::BIGINT AS loan_amount,
                    NULL::TEXT AS lender, NULL::TEXT AS collateral, NULL::BIGINT AS collateral_amount,
                    NULL::TEXT AS borrower, NULL::BIGINT AS payback_amount, NULL::TEXT AS contract_hash,
                    NULL::TEXT AS address, hash, signature1, signature2, NULL::TEXT AS stored_txid, NULL::SMALLINT AS hard_limit, original
                FROM swap
                WHERE processed = false

                UNION ALL

                SELECT
                    'loan_contract'::TEXT AS kind, id, fee AS priority_fee, time,
                    NULL::TEXT AS sender, NULL::TEXT AS receiver, NULL::TEXT AS coin, NULL::BIGINT AS value, NULL::INTEGER AS nft_series,
                    NULL::BIGINT AS number, NULL::TEXT AS creator, NULL::TEXT AS ticker,
                    NULL::TEXT AS nft_name, NULL::SMALLINT AS series, NULL::BIGINT AS count,
                    NULL::TEXT AS advertiser, NULL::BIGINT AS fee1, NULL::BIGINT AS fee2,
                    NULL::TEXT AS ticker1, NULL::INTEGER AS nft_series1, NULL::TEXT AS ticker2, NULL::INTEGER AS nft_series2, NULL::BIGINT AS value1,
                    NULL::BIGINT AS value2, NULL::TEXT AS sender1, NULL::BIGINT AS tip1,
                    NULL::BIGINT AS tip2, NULL::TEXT AS sender2, loan_coin, loan_amount, lender,
                    collateral, collateral_amount, borrower, NULL::BIGINT AS payback_amount,
                    NULL::TEXT AS contract_hash, NULL::TEXT AS address, hash, signature1, signature2,
                    txid AS stored_txid, NULL::SMALLINT AS hard_limit, original
                FROM loan_contract
                WHERE processed = false

                UNION ALL

                SELECT
                    'loan_payment'::TEXT AS kind, id, fee AS priority_fee, time,
                    NULL::TEXT AS sender, NULL::TEXT AS receiver, NULL::TEXT AS coin, NULL::BIGINT AS value, NULL::INTEGER AS nft_series,
                    NULL::BIGINT AS number, NULL::TEXT AS creator, NULL::TEXT AS ticker,
                    NULL::TEXT AS nft_name, NULL::SMALLINT AS series, NULL::BIGINT AS count,
                    NULL::TEXT AS advertiser, NULL::BIGINT AS fee1, NULL::BIGINT AS fee2,
                    NULL::TEXT AS ticker1, NULL::INTEGER AS nft_series1, NULL::TEXT AS ticker2, NULL::INTEGER AS nft_series2, NULL::BIGINT AS value1,
                    NULL::BIGINT AS value2, NULL::TEXT AS sender1, tip AS tip1, NULL::BIGINT AS tip2,
                    NULL::TEXT AS sender2, NULL::TEXT AS loan_coin, NULL::BIGINT AS loan_amount,
                    NULL::TEXT AS lender, NULL::TEXT AS collateral, NULL::BIGINT AS collateral_amount,
                    NULL::TEXT AS borrower, payback_amount, contract_hash, address, hash,
                    signature AS signature1, NULL::TEXT AS signature2, txid AS stored_txid, NULL::SMALLINT AS hard_limit, original
                FROM loan_payment
                WHERE processed = false

                UNION ALL

                SELECT
                    'collateral_claim'::TEXT AS kind, id, fee AS priority_fee, time,
                    NULL::TEXT AS sender, NULL::TEXT AS receiver, NULL::TEXT AS coin, NULL::BIGINT AS value, NULL::INTEGER AS nft_series,
                    NULL::BIGINT AS number, NULL::TEXT AS creator, NULL::TEXT AS ticker,
                    NULL::TEXT AS nft_name, NULL::SMALLINT AS series, NULL::BIGINT AS count,
                    NULL::TEXT AS advertiser, NULL::BIGINT AS fee1, NULL::BIGINT AS fee2,
                    NULL::TEXT AS ticker1, NULL::INTEGER AS nft_series1, NULL::TEXT AS ticker2, NULL::INTEGER AS nft_series2, NULL::BIGINT AS value1,
                    NULL::BIGINT AS value2, NULL::TEXT AS sender1, NULL::BIGINT AS tip1,
                    NULL::BIGINT AS tip2, NULL::TEXT AS sender2, NULL::TEXT AS loan_coin,
                    NULL::BIGINT AS loan_amount, NULL::TEXT AS lender, NULL::TEXT AS collateral,
                    NULL::BIGINT AS collateral_amount, NULL::TEXT AS borrower,
                    NULL::BIGINT AS payback_amount, contract_hash, address, hash,
                    signature AS signature1, NULL::TEXT AS signature2, NULL::TEXT AS stored_txid, NULL::SMALLINT AS hard_limit, original
                FROM collateral_claim
                WHERE processed = false
            ) AS combined
            ORDER BY priority_fee DESC, time ASC, kind ASC, id ASC
            LIMIT $1
            "#,
            &[&limit],
        )
        .await?;

    let mut transactions = Vec::with_capacity(rows.len());
    let mut originals = Vec::with_capacity(rows.len());
    for row in rows {
        // Preserve the original serialized transaction bytes so the miner can
        // write exactly what entered the mempool.
        originals.push(row.get("original"));

        // The UNION query flattens every mempool table into one row shape;
        // this tag restores the correct selected-transaction variant.
        let kind: String = row.get("kind");
        let transaction = match kind.as_str() {
            "transfer" => SelectedMempoolTransaction::Transfer {
                id: row.get("id"),
                fee: row.get("priority_fee"),
                sender: required_string(&row, "sender")?,
                value: row.get("value"),
                coin: required_string(&row, "coin")?,
                nft_series: row.get("nft_series"),
                receiver: required_string(&row, "receiver")?,
                hash: row.get("hash"),
            },
            "token" => SelectedMempoolTransaction::Token {
                id: row.get("id"),
                fee: row.get("priority_fee"),
                creator: required_string(&row, "creator")?,
                number: row.get("number"),
                ticker: required_string(&row, "ticker")?,
                hash: row.get("hash"),
            },
            "issue_token" => SelectedMempoolTransaction::IssueToken {
                id: row.get("id"),
                fee: row.get("priority_fee"),
                creator: required_string(&row, "creator")?,
                number: row.get("number"),
                ticker: required_string(&row, "ticker")?,
                hash: row.get("hash"),
            },
            "burn" => SelectedMempoolTransaction::Burn {
                id: row.get("id"),
                fee: row.get("priority_fee"),
                address: required_string(&row, "address")?,
                coin: required_string(&row, "coin")?,
                nft_series: row.get("nft_series"),
                value: row.get("value"),
                hash: row.get("hash"),
            },
            "nft" => SelectedMempoolTransaction::Nft {
                id: row.get("id"),
                fee: row.get("priority_fee"),
                creator: required_string(&row, "creator")?,
                nft_name: required_string(&row, "nft_name")?,
                series: row.get("series"),
                count: row.get("count"),
                hash: row.get("hash"),
            },
            "marketing" => SelectedMempoolTransaction::Marketing {
                id: row.get("id"),
                fee: row.get("priority_fee"),
                advertiser: required_string(&row, "advertiser")?,
                hash: row.get("hash"),
            },
            "vanity_address" => SelectedMempoolTransaction::Vanity {
                id: row.get("id"),
                fee: row.get("priority_fee"),
                address: required_string(&row, "address")?,
                vanity_address: required_string(&row, "stored_txid")?,
                hash: row.get("hash"),
            },
            "swap" => SelectedMempoolTransaction::Swap {
                id: row.get("id"),
                fee1: row.get("fee1"),
                fee2: row.get("fee2"),
                ticker1: required_string(&row, "ticker1")?,
                nft_series1: row.get("nft_series1"),
                ticker2: required_string(&row, "ticker2")?,
                nft_series2: row.get("nft_series2"),
                value1: row.get("value1"),
                value2: row.get("value2"),
                sender1: required_string(&row, "sender1")?,
                tip1: row.get("tip1"),
                tip2: row.get("tip2"),
                sender2: required_string(&row, "sender2")?,
                hash: row.get("hash"),
            },
            "loan_contract" => SelectedMempoolTransaction::Lender {
                id: row.get("id"),
                fee: row.get("priority_fee"),
                loan_coin: required_string(&row, "loan_coin")?,
                loan_amount: row.get("loan_amount"),
                lender: required_string(&row, "lender")?,
                collateral: required_string(&row, "collateral")?,
                collateral_amount: row.get("collateral_amount"),
                borrower: required_string(&row, "borrower")?,
                txid: required_string(&row, "stored_txid")?,
                hash: row.get("hash"),
            },
            "loan_payment" => SelectedMempoolTransaction::Borrower {
                id: row.get("id"),
                fee: row.get("priority_fee"),
                payback_amount: row.get("payback_amount"),
                contract_hash: required_string(&row, "contract_hash")?,
                address: required_string(&row, "address")?,
                tip: row.get("tip1"),
                hash: row.get("hash"),
            },
            "collateral_claim" => SelectedMempoolTransaction::Collateral {
                id: row.get("id"),
                fee: row.get("priority_fee"),
                address: required_string(&row, "address")?,
                contract_hash: required_string(&row, "contract_hash")?,
                hash: row.get("hash"),
            },
            _ => return Err(anyhow!("Unsupported mempool kind {kind}")),
        };
        transactions.push(transaction);
    }

    Ok(SelectedMempoolBatch {
        transactions,
        originals,
    })
}

pub async fn apply_selected_transaction_math(
    batch: &SelectedMempoolBatch,
    db: &Db,
    miner: String,
    block_number: u32,
    start_index: usize,
    pending_effects: &mut PendingEffects,
) -> Result<()> {
    // Net balance deltas are collected first so multiple changes to the same
    // address/asset pair become one balance-sheet write.
    let mut balance_changes: HashMap<BalanceKey, i64> = HashMap::new();
    let miner_address = address_key_bytes(db, &miner);
    let base_coin = BASECOIN.as_bytes().to_vec();
    let mut tx_index = start_index;
    for tx in &batch.transactions {
        // Transaction index is stored with the block number for later txid
        // lookups inside the saved block payload.
        tx_index = tx_index.wrapping_add(1);
        match tx {
            SelectedMempoolTransaction::Transfer {
                fee,
                sender,
                value,
                coin,
                nft_series,
                receiver,
                hash,
                ..
            } => {
                // Transfers move the asset, charge the base fee, and credit
                // that fee to the miner including this transaction.
                let transfer_asset = nft_asset_name(coin, *nft_series as u32);
                add_balance_change(db, &mut balance_changes, sender, &transfer_asset, -*value);
                add_balance_change(db, &mut balance_changes, receiver, &transfer_asset, *value);
                add_balance_change(db, &mut balance_changes, sender, &BASECOIN, -*fee);
                add_balance_change_bytes(
                    &mut balance_changes,
                    miner_address.clone(),
                    base_coin.clone(),
                    *fee,
                );
                let nft_tree = db.open_tree("nfts")?;
                if nft_tree.contains_key(transfer_asset.as_bytes())? {
                    pending_effects.append_tree(
                        "nft_history",
                        transfer_asset.as_bytes().to_vec(),
                        decode(hash)?,
                    );
                }
                pending_effects.set_tree(
                    "txid",
                    decode(hash)?,
                    format!("{block_number}:{tx_index}").into_bytes(),
                );
                index_selected_wallet_transaction(
                    pending_effects,
                    &[sender, receiver],
                    block_number,
                    tx_index,
                    &decode(hash)?,
                )?;
            }
            SelectedMempoolTransaction::Token {
                fee,
                creator,
                number,
                ticker,
                hash,
                ..
            } => {
                // Token creation mints the initial supply and records metadata
                // used by later issue and burn validation.
                add_balance_change(db, &mut balance_changes, creator, ticker, *number);
                add_balance_change(db, &mut balance_changes, creator, &BASECOIN, -*fee);
                add_balance_change_bytes(
                    &mut balance_changes,
                    miner_address.clone(),
                    base_coin.clone(),
                    *fee,
                );

                let number_bytes = (*number as u64).to_le_bytes();
                pending_effects.set_tree(
                    "tokens",
                    ticker.as_bytes().to_vec(),
                    number_bytes.to_vec(),
                );

                // Local mined blocks need to persist the token hard-limit
                // metadata just like the downloaded-block save path does.
                let client_handle = db_client().await?;
                let client = client_handle.as_ref();
                let token_row = client
                    .query_opt(
                        "SELECT hard_limit FROM token WHERE hash = $1 LIMIT 1",
                        &[hash],
                    )
                    .await?;
                let hard_limit = token_row
                    .and_then(|row| row.try_get::<_, Option<i16>>("hard_limit").ok().flatten())
                    .unwrap_or(1) as u8;
                pending_effects.set_tree(
                    "token_limits",
                    ticker.as_bytes().to_vec(),
                    vec![hard_limit],
                );
                pending_effects.set_tree(
                    "token_origins",
                    ticker.as_bytes().to_vec(),
                    hash.as_bytes().to_vec(),
                );
                pending_effects.append_tree(
                    "token_history",
                    ticker.as_bytes().to_vec(),
                    decode(hash)?,
                );
                pending_effects.set_tree(
                    "txid",
                    decode(hash)?,
                    format!("{block_number}:{tx_index}").into_bytes(),
                );
                index_selected_wallet_transaction(
                    pending_effects,
                    &[creator],
                    block_number,
                    tx_index,
                    &decode(hash)?,
                )?;
            }
            SelectedMempoolTransaction::IssueToken {
                fee,
                creator,
                number,
                ticker,
                hash,
                ..
            } => {
                add_balance_change(db, &mut balance_changes, creator, ticker, *number);
                add_balance_change(db, &mut balance_changes, creator, &BASECOIN, -*fee);
                add_balance_change_bytes(
                    &mut balance_changes,
                    miner_address.clone(),
                    base_coin.clone(),
                    *fee,
                );

                // Additional issuance increases the live fungible supply without
                // changing the original token metadata entries.
                pending_effects.add_token_supply(ticker, *number as u64);
                pending_effects.append_tree(
                    "token_history",
                    ticker.as_bytes().to_vec(),
                    decode(hash)?,
                );
                pending_effects.set_tree(
                    "txid",
                    decode(hash)?,
                    format!("{block_number}:{tx_index}").into_bytes(),
                );
                index_selected_wallet_transaction(
                    pending_effects,
                    &[creator],
                    block_number,
                    tx_index,
                    &decode(hash)?,
                )?;
            }
            SelectedMempoolTransaction::Burn {
                fee,
                address,
                coin,
                nft_series,
                value,
                hash,
                ..
            } => {
                let burned_asset = nft_asset_name(coin, *nft_series as u32);
                add_balance_change(db, &mut balance_changes, address, &burned_asset, -*value);
                add_balance_change(db, &mut balance_changes, address, &BASECOIN, -*fee);
                add_balance_change_bytes(
                    &mut balance_changes,
                    miner_address.clone(),
                    base_coin.clone(),
                    *fee,
                );

                // Live NFT state removes burned assets, while fungible token
                // state reduces the stored supply by the burned amount.
                pending_effects.burn_asset_state(&burned_asset, coin, decode(hash)?, *value as u64);
                pending_effects.set_tree(
                    "txid",
                    decode(hash)?,
                    format!("{block_number}:{tx_index}").into_bytes(),
                );
                index_selected_wallet_transaction(
                    pending_effects,
                    &[address],
                    block_number,
                    tx_index,
                    &decode(hash)?,
                )?;
            }
            SelectedMempoolTransaction::Nft {
                fee,
                creator,
                nft_name,
                series,
                count,
                hash,
                ..
            } => {
                // Series-one NFT creation expands into numbered assets; later
                // series use the given NFT name as the asset key.
                if *series == 1 {
                    for item_number in 1..=*count {
                        let nft_save_name = nft_asset_name(nft_name, item_number as u32);
                        add_balance_change(
                            db,
                            &mut balance_changes,
                            creator,
                            &nft_save_name,
                            NFT_UNIT,
                        );

                        pending_effects.set_tree(
                            "nfts",
                            nft_save_name.as_bytes().to_vec(),
                            b"1".to_vec(),
                        );
                        pending_effects.set_tree(
                            "nft_origins",
                            nft_save_name.as_bytes().to_vec(),
                            decode(hash)?,
                        );
                        pending_effects.append_tree(
                            "nft_history",
                            nft_save_name.into_bytes(),
                            decode(hash)?,
                        );
                    }
                } else {
                    add_balance_change(db, &mut balance_changes, creator, nft_name, NFT_UNIT);

                    pending_effects.set_tree("nfts", nft_name.as_bytes().to_vec(), b"1".to_vec());
                    pending_effects.set_tree(
                        "nft_origins",
                        nft_name.as_bytes().to_vec(),
                        decode(hash)?,
                    );
                    pending_effects.append_tree(
                        "nft_history",
                        nft_name.as_bytes().to_vec(),
                        decode(hash)?,
                    );
                }
                add_balance_change(db, &mut balance_changes, creator, &BASECOIN, -*fee);
                add_balance_change_bytes(
                    &mut balance_changes,
                    miner_address.clone(),
                    base_coin.clone(),
                    *fee,
                );
                pending_effects.set_tree(
                    "txid",
                    decode(hash)?,
                    format!("{block_number}:{tx_index}").into_bytes(),
                );
                index_selected_wallet_transaction(
                    pending_effects,
                    &[creator],
                    block_number,
                    tx_index,
                    &decode(hash)?,
                )?;
            }
            SelectedMempoolTransaction::Marketing {
                fee,
                advertiser,
                hash,
                ..
            } => {
                add_balance_change(db, &mut balance_changes, advertiser, &BASECOIN, -*fee);
                add_balance_change_bytes(
                    &mut balance_changes,
                    miner_address.clone(),
                    base_coin.clone(),
                    *fee,
                );
                pending_effects.set_tree(
                    "txid",
                    decode(hash)?,
                    format!("{block_number}:{tx_index}").into_bytes(),
                );
                index_selected_wallet_transaction(
                    pending_effects,
                    &[advertiser],
                    block_number,
                    tx_index,
                    &decode(hash)?,
                )?;
            }
            SelectedMempoolTransaction::Vanity {
                fee,
                address,
                vanity_address,
                hash,
                ..
            } => {
                // Vanity transactions update the alias registry during the
                // final block-effect commit so rollback can restore it.
                pending_effects.update_vanity(address, vanity_address, hash);
                add_balance_change(db, &mut balance_changes, address, &BASECOIN, -*fee);
                add_balance_change_bytes(
                    &mut balance_changes,
                    miner_address.clone(),
                    base_coin.clone(),
                    *fee,
                );
                pending_effects.set_tree(
                    "txid",
                    decode(hash)?,
                    format!("{block_number}:{tx_index}").into_bytes(),
                );
                index_selected_wallet_transaction(
                    pending_effects,
                    &[address],
                    block_number,
                    tx_index,
                    &decode(hash)?,
                )?;
            }
            SelectedMempoolTransaction::Swap {
                fee1,
                fee2,
                ticker1,
                nft_series1,
                ticker2,
                nft_series2,
                value1,
                value2,
                sender1,
                tip1,
                tip2,
                sender2,
                hash,
                ..
            } => {
                // Swaps debit both sides first, then credit received assets,
                // miner fees, and optional asset tips.
                let asset1 = nft_asset_name(ticker1, *nft_series1 as u32);
                let asset2 = nft_asset_name(ticker2, *nft_series2 as u32);
                add_balance_change(db, &mut balance_changes, sender1, &asset1, -*value1);
                add_balance_change(db, &mut balance_changes, sender1, &asset1, -*tip1);
                add_balance_change(db, &mut balance_changes, sender2, &asset2, -*value2);
                add_balance_change(db, &mut balance_changes, sender2, &asset2, -*tip2);
                add_balance_change(db, &mut balance_changes, sender2, &asset1, *value1);
                add_balance_change(db, &mut balance_changes, sender1, &asset2, *value2);
                add_balance_change(db, &mut balance_changes, sender1, &BASECOIN, -*fee1);
                add_balance_change(db, &mut balance_changes, sender2, &BASECOIN, -*fee2);
                add_balance_change_bytes(
                    &mut balance_changes,
                    miner_address.clone(),
                    base_coin.clone(),
                    *fee1,
                );
                add_balance_change_bytes(
                    &mut balance_changes,
                    miner_address.clone(),
                    base_coin.clone(),
                    *fee2,
                );
                add_balance_change_bytes(
                    &mut balance_changes,
                    miner_address.clone(),
                    asset1.as_bytes().to_vec(),
                    *tip1,
                );
                add_balance_change_bytes(
                    &mut balance_changes,
                    miner_address.clone(),
                    asset2.as_bytes().to_vec(),
                    *tip2,
                );
                let nft_tree = db.open_tree("nfts")?;
                if nft_tree.contains_key(asset1.as_bytes())? {
                    pending_effects.append_tree(
                        "nft_history",
                        asset1.as_bytes().to_vec(),
                        decode(hash)?,
                    );
                }
                if nft_tree.contains_key(asset2.as_bytes())? {
                    pending_effects.append_tree(
                        "nft_history",
                        asset2.as_bytes().to_vec(),
                        decode(hash)?,
                    );
                }
                pending_effects.set_tree(
                    "txid",
                    decode(hash)?,
                    format!("{block_number}:{tx_index}").into_bytes(),
                );
                index_selected_wallet_transaction(
                    pending_effects,
                    &[sender1, sender2],
                    block_number,
                    tx_index,
                    &decode(hash)?,
                )?;
            }
            SelectedMempoolTransaction::Lender {
                fee,
                loan_coin,
                loan_amount,
                lender,
                collateral,
                collateral_amount,
                borrower,
                txid,
                hash,
                ..
            } => {
                // Loan creation moves the loan asset to the borrower and locks
                // collateral under a contract-specific holding key.
                add_balance_change(db, &mut balance_changes, lender, &BASECOIN, -*fee);
                add_balance_change_bytes(
                    &mut balance_changes,
                    miner_address.clone(),
                    base_coin.clone(),
                    *fee,
                );
                add_balance_change(db, &mut balance_changes, lender, loan_coin, -*loan_amount);
                add_balance_change(db, &mut balance_changes, borrower, loan_coin, *loan_amount);
                add_balance_change(
                    db,
                    &mut balance_changes,
                    borrower,
                    collateral,
                    -*collateral_amount,
                );

                let collateral_holding = format!("collateral_{txid}");
                add_balance_change(
                    db,
                    &mut balance_changes,
                    &collateral_holding,
                    collateral,
                    *collateral_amount,
                );

                pending_effects.set_tree("loan", decode(txid)?, b"true".to_vec());
                let nft_tree = db.open_tree("nfts")?;
                if nft_tree.contains_key(collateral.as_bytes())? {
                    pending_effects.append_tree(
                        "nft_history",
                        collateral.as_bytes().to_vec(),
                        decode(hash)?,
                    );
                }
                if nft_tree.contains_key(loan_coin.as_bytes())? {
                    pending_effects.append_tree(
                        "nft_history",
                        loan_coin.as_bytes().to_vec(),
                        decode(hash)?,
                    );
                }
                pending_effects.set_tree(
                    "txid",
                    decode(hash)?,
                    format!("{block_number}:{tx_index}").into_bytes(),
                );
                index_selected_wallet_transaction(
                    pending_effects,
                    &[lender, borrower],
                    block_number,
                    tx_index,
                    &decode(hash)?,
                )?;
            }
            SelectedMempoolTransaction::Borrower {
                fee,
                payback_amount,
                contract_hash,
                address,
                tip,
                hash,
                ..
            } => {
                // Loan payments resolve the contract, then move repayment from
                // borrower to lender while paying any tip to the miner.
                let (loan_coin, lender) = resolve_loan_details(db, contract_hash).await?;
                add_balance_change(db, &mut balance_changes, address, &BASECOIN, -*fee);
                add_balance_change_bytes(
                    &mut balance_changes,
                    miner_address.clone(),
                    base_coin.clone(),
                    *fee,
                );
                add_balance_change_bytes(
                    &mut balance_changes,
                    address_key_bytes(db, address),
                    loan_coin.clone(),
                    -*payback_amount,
                );
                add_balance_change_bytes(
                    &mut balance_changes,
                    lender.clone(),
                    loan_coin.clone(),
                    *payback_amount,
                );
                add_balance_change_bytes(
                    &mut balance_changes,
                    address_key_bytes(db, address),
                    loan_coin.clone(),
                    -*tip,
                );
                add_balance_change_bytes(
                    &mut balance_changes,
                    miner_address.clone(),
                    loan_coin.clone(),
                    *tip,
                );

                pending_effects
                    .append_contract_payment(decode(contract_hash)?, *payback_amount as u64);
                let loan_coin_name = binary_to_string(loan_coin.clone());
                let nft_tree = db.open_tree("nfts")?;
                if nft_tree.contains_key(loan_coin_name.as_bytes())? {
                    pending_effects.append_tree(
                        "nft_history",
                        loan_coin_name.as_bytes().to_vec(),
                        decode(hash)?,
                    );
                }
                pending_effects.set_tree(
                    "txid",
                    decode(hash)?,
                    format!("{block_number}:{tx_index}").into_bytes(),
                );
                let lender_address = binary_to_string(lender.clone());
                index_selected_wallet_transaction(
                    pending_effects,
                    &[address, &lender_address],
                    block_number,
                    tx_index,
                    &decode(hash)?,
                )?;
            }
            SelectedMempoolTransaction::Collateral {
                fee,
                address,
                contract_hash,
                hash,
                ..
            } => {
                // Collateral claims release the contract holding to the claimant
                // and mark the loan as closed.
                let (collateral, collateral_amount) =
                    resolve_collateral_details(db, contract_hash).await?;
                let collateral_holding = format!("collateral_{contract_hash}");
                add_balance_change(db, &mut balance_changes, address, &BASECOIN, -*fee);
                add_balance_change_bytes(
                    &mut balance_changes,
                    miner_address.clone(),
                    base_coin.clone(),
                    *fee,
                );
                add_balance_change_bytes(
                    &mut balance_changes,
                    collateral_holding.as_bytes().to_vec(),
                    collateral.clone(),
                    -collateral_amount,
                );
                add_balance_change_bytes(
                    &mut balance_changes,
                    address_key_bytes(db, address),
                    collateral.clone(),
                    collateral_amount,
                );

                pending_effects.set_tree("loan", decode(contract_hash)?, b"false".to_vec());
                let collateral_name = binary_to_string(collateral.clone());
                let nft_tree = db.open_tree("nfts")?;
                if nft_tree.contains_key(collateral_name.as_bytes())? {
                    pending_effects.append_tree(
                        "nft_history",
                        collateral_name.as_bytes().to_vec(),
                        decode(hash)?,
                    );
                }
                pending_effects.set_tree(
                    "txid",
                    decode(hash)?,
                    format!("{block_number}:{tx_index}").into_bytes(),
                );
                index_selected_wallet_transaction(
                    pending_effects,
                    &[address],
                    block_number,
                    tx_index,
                    &decode(hash)?,
                )?;
            }
        }
    }

    for (key, change) in balance_changes {
        // The balance sheet stores string keys, so convert the internal byte
        // keys back into wallet/asset strings before applying the delta.
        let address = if key.address.len() == Wallet::SHORT_ADDRESS_BYTES_LENGTH {
            Wallet::bytes_to_short_address(&key.address).unwrap_or_default()
        } else {
            binary_to_string(key.address)
        };
        let coin = binary_to_string(key.coin);
        let operand = if change >= 0 {
            BalanceOperand::Addition
        } else {
            BalanceOperand::Subtraction
        };
        pending_effects.add_balance(&address, change.unsigned_abs(), &coin, operand);
    }

    Ok(())
}

pub async fn clear_selected_transaction_sql(
    db: &Db,
    miner: String,
    limit: i64,
    block_number: u32,
    start_index: usize,
) -> Result<SelectedMempoolBatch> {
    // Selection and ledger math stay paired so the caller can stream this exact
    // selected batch into the block file afterward.
    let batch = select_transactions_for_block(limit).await?;
    let mut pending_effects = PendingEffects::default();
    apply_selected_transaction_math(
        &batch,
        db,
        miner,
        block_number,
        start_index,
        &mut pending_effects,
    )
    .await?;
    pending_effects.apply(db).map_err(|err| anyhow!(err))?;
    Ok(batch)
}

pub async fn stream_selected_transaction_originals(
    file: &mut File,
    batch: &SelectedMempoolBatch,
) -> Result<()> {
    if batch.originals.is_empty() {
        file.flush()?;
        return Ok(());
    }

    // Block payloads store the original mempool bytes back-to-back in selected
    // order rather than rebuilding each transaction from parsed fields.
    let total_len: usize = batch.originals.iter().map(Vec::len).sum();
    let mut payload = Vec::with_capacity(total_len);
    for original in &batch.originals {
        payload.extend_from_slice(original);
    }
    file.write_all(&payload)?;
    file.flush()?;
    Ok(())
}

pub async fn delete_selected_transactions(batch: &SelectedMempoolBatch) -> Result<()> {
    // Each transaction kind still lives in a separate SQL table, so deletion
    // groups the selected IDs by table after the block has been written.
    let transfer_ids = ids_for_table(batch, "transfer");
    let token_ids = ids_for_table(batch, "token");
    let issue_token_ids = ids_for_table(batch, "issue_token");
    let burn_ids = ids_for_table(batch, "burn");
    let nft_ids = ids_for_table(batch, "nft");
    let marketing_ids = ids_for_table(batch, "marketing");
    let vanity_ids = ids_for_table(batch, "vanity_address");
    let swap_ids = ids_for_table(batch, "swap");
    let lender_ids = ids_for_table(batch, "loan_contract");
    let borrower_ids = ids_for_table(batch, "loan_payment");
    let collateral_ids = ids_for_table(batch, "collateral_claim");

    delete_rows("transfer", &transfer_ids).await?;
    delete_rows("token", &token_ids).await?;
    delete_rows("issue_token", &issue_token_ids).await?;
    delete_rows("burn", &burn_ids).await?;
    delete_rows("nft", &nft_ids).await?;
    delete_rows("marketing", &marketing_ids).await?;
    delete_rows("vanity_address", &vanity_ids).await?;
    delete_rows("swap", &swap_ids).await?;
    delete_rows("loan_contract", &lender_ids).await?;
    delete_rows("loan_payment", &borrower_ids).await?;
    delete_rows("collateral_claim", &collateral_ids).await?;

    Ok(())
}