main.rs

//! Evolve Node Daemon (evd)
//!
//! A full node implementation that exposes the Evolve execution layer via gRPC
//! for external consensus integration, with JSON-RPC for queries and mempool
//! for transaction collection.
//!
//! ## Architecture
//!
//! ```text
//! ┌─────────────────┐     gRPC      ┌──────────────────────────────────┐
//! │  External       │◄─────────────►│              evd                 │
//! │  Consensus      │               │                                  │
//! │  (ev-node)      │               │  ┌────────┐  ┌────────┐          │
//! └─────────────────┘               │  │  STF   │  │Mempool │          │
//!                                   │  └────────┘  └────────┘          │
//!        ┌──────────┐  JSON-RPC     │  ┌────────┐  ┌────────┐          │
//!        │  Client  │◄─────────────►│  │  RPC   │  │  QMDB  │          │
//!        └──────────┘               │  │ Server │  │Storage │          │
//!                                   │  └────────┘  └────────┘          │
//!                                   └──────────────────────────────────┘
//! ```
//!
//! ## Transaction Formats
//!
//! - **ETH**: Standard Ethereum RLP-encoded transactions (type 0x02)
//! - **Micro**: Minimal fixed-layout transactions (type 0x83) for high throughput
//!
//! ## Usage
//!
//! ```bash
//! # Start with default testapp genesis
//! evd run
//!
//! # Start with a custom genesis file
//! evd run --genesis-file path/to/genesis.json
//!
//! # Custom addresses
//! evd run --grpc-addr 0.0.0.0:50051 --rpc-addr 0.0.0.0:8545
//!
//! # Initialize genesis only
//! evd init
//! ```
//!
//! ## Custom Genesis JSON Format
//!
//! When `--genesis-file` is provided, accounts are pre-registered as ETH EOAs
//! with deterministic IDs derived from their addresses.
//!
//! ```json
//! {
//!   "token": {
//!     "name": "evolve",
//!     "symbol": "ev",
//!     "decimals": 6,
//!     "icon_url": "https://lol.wtf",
//!     "description": "The evolve coin"
//!   },
//!   "minter_id": 100002,
//!   "accounts": [
//!     { "eth_address": "0xf39Fd6e51aad88F6F4ce6aB8827279cffFb92266", "balance": 1000000000 }
//!   ]
//! }
//! ```

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;

use alloy_primitives::{keccak256, Address, B256, U256};
use clap::{Args, Parser, Subcommand};
use commonware_runtime::tokio::{Config as TokioConfig, Runner};
use commonware_runtime::{Runner as RunnerTrait, Spawner};
use evolve_chain_index::{
    build_index_data, BlockMetadata, ChainIndex, ChainStateProvider, ChainStateProviderConfig,
    PersistentChainIndex,
};
use evolve_core::{AccountId, ReadonlyKV};
use evolve_eth_jsonrpc::{start_server_with_subscriptions, RpcServerConfig, SubscriptionManager};
use evolve_evnode::{EvnodeServer, EvnodeServerConfig, ExecutorServiceConfig, OnBlockExecuted};
use evolve_mempool::{new_shared_mempool, Mempool, SharedMempool};
use evolve_node::{
    init_tracing as init_node_tracing, resolve_node_config, resolve_node_config_init,
    GenesisOutput, InitArgs, NodeConfig, RunArgs,
};
use evolve_rpc_types::SyncStatus;
use evolve_server::{
    load_chain_state, save_chain_state, BlockBuilder, ChainState, CHAIN_STATE_KEY,
};
use evolve_stf_traits::{AccountsCodeStorage, StateChange};
use evolve_storage::{Operation, QmdbStorage, Storage, StorageConfig};
use evolve_testapp::genesis_config::{load_genesis_config, EvdGenesisConfig, EvdGenesisResult};
use evolve_testapp::{
    build_mempool_stf, default_gas_config, do_genesis_inner, initialize_custom_genesis_resources,
    install_account_codes, PLACEHOLDER_ACCOUNT,
};
use evolve_testing::server_mocks::AccountStorageMock;
use evolve_tx_eth::TxContext;

#[derive(Parser)]
#[command(name = "evd")]
#[command(about = "Evolve node daemon with gRPC execution layer")]
#[command(version)]
struct Cli {
    #[command(subcommand)]
    command: Commands,
}

#[derive(Subcommand)]
enum Commands {
    /// Run the node with gRPC and JSON-RPC servers
    Run(EvdRunArgs),
    /// Initialize genesis state without running
    Init(EvdInitArgs),
}

type EvdRunArgs = RunArgs<EvdRunCustom>;
type EvdInitArgs = InitArgs<EvdInitCustom>;

#[derive(Args)]
struct EvdRunCustom {
    /// Path to a genesis JSON file with ETH accounts (uses default testapp genesis if omitted)
    #[arg(long)]
    genesis_file: Option<String>,
}

#[derive(Args)]
struct EvdInitCustom {
    /// Path to a genesis JSON file with ETH accounts (uses default testapp genesis if omitted)
    #[arg(long)]
    genesis_file: Option<String>,
}

fn main() {
    let cli = Cli::parse();

    match cli.command {
        Commands::Run(args) => {
            let config = resolve_node_config(&args.common, &args.native);
            init_node_tracing(&config.observability.log_level);
            let genesis_config = match load_genesis_config(args.custom.genesis_file.as_deref()) {
                Ok(genesis_config) => genesis_config,
                Err(err) => {
                    tracing::error!("{err}");
                    std::process::exit(2);
                }
            };
            run_node(config, genesis_config);
        }
        Commands::Init(args) => {
            let config = resolve_node_config_init(&args.common);
            init_node_tracing(&config.observability.log_level);
            let genesis_config = match load_genesis_config(args.custom.genesis_file.as_deref()) {
                Ok(genesis_config) => genesis_config,
                Err(err) => {
                    tracing::error!("{err}");
                    std::process::exit(2);
                }
            };
            init_genesis(&config.storage.path, genesis_config);
        }
    }
}

fn run_node(config: NodeConfig, genesis_config: Option<EvdGenesisConfig>) {
    tracing::info!("=== Evolve Node Daemon (evd) ===");

    std::fs::create_dir_all(&config.storage.path).expect("failed to create data directory");

    let storage_config = StorageConfig {
        path: config.storage.path.clone().into(),
        ..Default::default()
    };

    let runtime_config = TokioConfig::default()
        .with_storage_directory(&config.storage.path)
        .with_worker_threads(4);

    let runner = Runner::new(runtime_config);

    runner.start(move |context| {
        async move {
            let context_for_shutdown = context.clone();

            // Initialize QMDB storage
            let storage = QmdbStorage::new(context, storage_config)
                .await
                .expect("failed to create storage");

            // Set up account codes
            let codes = build_codes();
            tracing::info!("Installed account codes: {:?}", codes.list_identifiers());

            // Load or run genesis
            let (genesis_result, initial_height) =
                match load_chain_state::<EvdGenesisResult, _>(&storage) {
                    Some(state) => {
                        tracing::info!("Resuming from existing state at height {}", state.height);
                        (state.genesis_result, state.height)
                    }
                    None => {
                        tracing::info!("No existing state found, running genesis...");
                        let output = run_genesis(&storage, &codes, genesis_config.as_ref());
                        commit_genesis(&storage, output.changes, &output.genesis_result)
                            .await
                            .expect("genesis commit failed");
                        tracing::info!("Genesis complete: {:?}", output.genesis_result);
                        (output.genesis_result, 1)
                    }
                };

            // Build STF with scheduler from genesis
            let gas_config = default_gas_config();
            let stf = build_mempool_stf(gas_config, genesis_result.scheduler);

            // Create shared mempool
            let mempool: SharedMempool<Mempool<TxContext>> = new_shared_mempool();
            // Create chain index backed by SQLite (only when needed)
            let chain_index = if config.rpc.enabled || config.rpc.enable_block_indexing {
                let chain_index_db_path =
                    std::path::PathBuf::from(&config.storage.path).join("chain-index.sqlite");
                let index = Arc::new(
                    PersistentChainIndex::new(&chain_index_db_path)
                        .expect("failed to open chain index database"),
                );
                if let Err(e) = index.initialize() {
                    tracing::warn!("Failed to initialize chain index: {:?}", e);
                }
                Some(index)
            } else {
                None
            };

            // Set up JSON-RPC server if enabled
            let rpc_handle = if config.rpc.enabled {
                let subscriptions = Arc::new(SubscriptionManager::new());
                let codes_for_rpc = Arc::new(build_codes());

                let state_provider_config = ChainStateProviderConfig {
                    chain_id: config.chain.chain_id,
                    protocol_version: "0x1".to_string(),
                    gas_price: U256::ZERO,
                    sync_status: SyncStatus::NotSyncing(false),
                };

                let state_provider = ChainStateProvider::with_mempool(
                    Arc::clone(chain_index.as_ref().expect("chain index required for RPC")),
                    state_provider_config,
                    codes_for_rpc,
                    mempool.clone(),
                );

                let rpc_addr = config.parsed_rpc_addr();
                let server_config = RpcServerConfig {
                    http_addr: rpc_addr,
                    chain_id: config.chain.chain_id,
                };

                tracing::info!("Starting JSON-RPC server on {}", rpc_addr);
                let handle = start_server_with_subscriptions(
                    server_config,
                    state_provider,
                    Arc::clone(&subscriptions),
                )
                .await
                .expect("failed to start RPC server");

                Some(handle)
            } else {
                None
            };

            // Shared state for the block callback
            let parent_hash = Arc::new(std::sync::RwLock::new(B256::ZERO));
            let current_height = Arc::new(AtomicU64::new(initial_height));

            // Build the OnBlockExecuted callback: commits state to storage + indexes blocks
            let storage_for_callback = storage.clone();
            let chain_index_for_callback = chain_index.clone();
            let parent_hash_for_callback = Arc::clone(&parent_hash);
            let current_height_for_callback = Arc::clone(&current_height);
            let callback_chain_id = config.chain.chain_id;
            let executor_config = ExecutorServiceConfig::default();
            let callback_max_gas = executor_config.max_gas;
            let callback_indexing_enabled = config.rpc.enable_block_indexing;

            let on_block_executed: OnBlockExecuted = Arc::new(move |info| {
                // 1. Commit state changes to QmdbStorage
                let operations = state_changes_to_operations(info.state_changes);

                let commit_hash = futures::executor::block_on(async {
                    storage_for_callback
                        .batch(operations)
                        .await
                        .expect("storage batch failed");
                    storage_for_callback
                        .commit()
                        .await
                        .expect("storage commit failed")
                });
                let state_root = B256::from_slice(commit_hash.as_bytes());

                // 2. Compute block hash and build metadata
                let prev_parent = *parent_hash_for_callback.read().unwrap();
                let block_hash = compute_block_hash(info.height, info.timestamp, prev_parent);

                let metadata = BlockMetadata::new(
                    block_hash,
                    prev_parent,
                    state_root,
                    info.timestamp,
                    callback_max_gas,
                    Address::ZERO,
                    callback_chain_id,
                );

                // 3. Reconstruct block and index it
                let block = BlockBuilder::<TxContext>::new()
                    .number(info.height)
                    .timestamp(info.timestamp)
                    .transactions(info.transactions)
                    .build();

                let (stored_block, stored_txs, stored_receipts) =
                    build_index_data(&block, &info.block_result, &metadata);

                if let Some(ref chain_index) = chain_index_for_callback {
                    if callback_indexing_enabled {
                        if let Err(e) =
                            chain_index.store_block(stored_block, stored_txs, stored_receipts)
                        {
                            tracing::warn!("Failed to index block {}: {:?}", info.height, e);
                        } else {
                            tracing::debug!(
                                "Indexed block {} (hash={}, state_root={})",
                                info.height,
                                block_hash,
                                state_root
                            );
                        }
                    }
                }

                // 4. Update parent hash and height for next block
                *parent_hash_for_callback.write().unwrap() = block_hash;
                current_height_for_callback.store(info.height, Ordering::SeqCst);
            });

            // Configure gRPC server
            let grpc_config = EvnodeServerConfig {
                addr: config.parsed_grpc_addr(),
                enable_gzip: config.grpc.enable_gzip,
                max_message_size: config.grpc_max_message_size_usize(),
                executor_config,
            };

            let grpc_addr = config.parsed_grpc_addr();
            tracing::info!("Starting gRPC server on {}", grpc_addr);
            tracing::info!("Configuration:");
            tracing::info!("  - Chain ID: {}", config.chain.chain_id);
            tracing::info!("  - gRPC compression: {}", config.grpc.enable_gzip);
            tracing::info!("  - JSON-RPC: {}", config.rpc.enabled);
            tracing::info!("  - Block indexing: {}", config.rpc.enable_block_indexing);
            tracing::info!("  - Initial height: {}", initial_height);

            // Create gRPC server with mempool and block callback
            let server = EvnodeServer::with_mempool(
                grpc_config,
                stf,
                storage.clone(),
                build_codes(),
                mempool,
            )
            .with_on_block_executed(on_block_executed);

            tracing::info!("Server ready. Press Ctrl+C to stop.");

            // Run gRPC server with shutdown handling
            tokio::select! {
                result = server.serve() => {
                    if let Err(e) = result {
                        tracing::error!("gRPC server error: {}", e);
                    }
                }
                _ = tokio::signal::ctrl_c() => {
                    tracing::info!("Received Ctrl+C, shutting down...");
                    context_for_shutdown
                        .stop(0, Some(Duration::from_secs(config.operations.shutdown_timeout_secs)))
                        .await
                        .expect("shutdown failed");
                }
            }

            // Save chain state with actual committed height
            let chain_state = ChainState {
                height: current_height.load(Ordering::SeqCst),
                genesis_result,
            };
            if let Err(e) = save_chain_state(&storage, &chain_state).await {
                tracing::error!("Failed to save chain state: {}", e);
            }

            // Stop RPC server
            if let Some(handle) = rpc_handle {
                tracing::info!("Stopping JSON-RPC server...");
                handle.stop().expect("failed to stop RPC server");
            }

            tracing::info!("Shutdown complete.");
        }
    });
}

fn init_genesis(data_dir: &str, genesis_config: Option<EvdGenesisConfig>) {
    tracing::info!("=== Evolve Node Daemon - Genesis Init ===");

    std::fs::create_dir_all(data_dir).expect("failed to create data directory");

    let storage_config = StorageConfig {
        path: data_dir.into(),
        ..Default::default()
    };

    let runtime_config = TokioConfig::default()
        .with_storage_directory(data_dir)
        .with_worker_threads(1);

    let runner = Runner::new(runtime_config);

    runner.start(move |context| async move {
        let storage = QmdbStorage::new(context, storage_config)
            .await
            .expect("failed to create storage");

        if load_chain_state::<EvdGenesisResult, _>(&storage).is_some() {
            tracing::error!("State already initialized; refusing to re-run genesis");
            return;
        }

        let codes = build_codes();
        let output = run_genesis(&storage, &codes, genesis_config.as_ref());

        commit_genesis(&storage, output.changes, &output.genesis_result)
            .await
            .expect("genesis commit failed");

        tracing::info!("Genesis complete!");
        tracing::info!("  Token: {:?}", output.genesis_result.token);
        tracing::info!("  Scheduler: {:?}", output.genesis_result.scheduler);
    });
}

fn build_codes() -> AccountStorageMock {
    let mut codes = AccountStorageMock::default();
    install_account_codes(&mut codes);
    codes
}

/// Run genesis using the default testapp genesis or a custom genesis config.
fn run_genesis<S: ReadonlyKV + Storage>(
    storage: &S,
    codes: &AccountStorageMock,
    genesis_config: Option<&EvdGenesisConfig>,
) -> GenesisOutput<EvdGenesisResult> {
    match genesis_config {
        Some(config) => run_custom_genesis(storage, codes, config),
        None => run_default_genesis(storage, codes),
    }
}

/// Default genesis using testapp's `do_genesis_inner` (sequential account IDs).
fn run_default_genesis<S: ReadonlyKV + Storage>(
    storage: &S,
    codes: &AccountStorageMock,
) -> GenesisOutput<EvdGenesisResult> {
    use evolve_core::BlockContext;

    tracing::info!("Running default testapp genesis...");

    let gas_config = default_gas_config();
    let stf = build_mempool_stf(gas_config, PLACEHOLDER_ACCOUNT);
    let genesis_block = BlockContext::new(0, 0);

    let (accounts, state) = stf
        .system_exec(storage, codes, genesis_block, |env| do_genesis_inner(env))
        .expect("genesis failed");

    let changes = state.into_changes().expect("failed to get state changes");

    let genesis_result = EvdGenesisResult {
        token: accounts.atom,
        scheduler: accounts.scheduler,
    };

    GenesisOutput {
        genesis_result,
        changes,
    }
}

/// Custom genesis with ETH EOA accounts from a genesis JSON file.
///
/// Registers funded EOA accounts via `EthEoaAccountRef::initialize` inside
/// `system_exec`, then initializes the token with their balances.
fn run_custom_genesis<S: ReadonlyKV + Storage>(
    storage: &S,
    codes: &AccountStorageMock,
    genesis_config: &EvdGenesisConfig,
) -> GenesisOutput<EvdGenesisResult> {
    use evolve_core::BlockContext;

    let funded_accounts = genesis_config
        .funded_accounts()
        .expect("invalid address in genesis config");

    let minter = AccountId::new(genesis_config.minter_id);
    let metadata = genesis_config.token.to_metadata();

    let gas_config = default_gas_config();
    let stf = build_mempool_stf(gas_config, PLACEHOLDER_ACCOUNT);
    let genesis_block = BlockContext::new(0, 0);

    let (genesis_result, state) = stf
        .system_exec(storage, codes, genesis_block, |env| {
            let resources = initialize_custom_genesis_resources(
                &funded_accounts,
                metadata.clone(),
                minter,
                env,
            )?;

            Ok(EvdGenesisResult {
                token: resources.token,
                scheduler: resources.scheduler,
            })
        })
        .expect("genesis failed");

    let changes = state.into_changes().expect("failed to get state changes");

    GenesisOutput {
        genesis_result,
        changes,
    }
}

fn compute_block_hash(height: u64, timestamp: u64, parent_hash: B256) -> B256 {
    let mut data = Vec::with_capacity(48);
    data.extend_from_slice(&height.to_le_bytes());
    data.extend_from_slice(&timestamp.to_le_bytes());
    data.extend_from_slice(parent_hash.as_slice());
    keccak256(&data)
}

async fn commit_genesis<S: Storage>(
    storage: &S,
    changes: Vec<StateChange>,
    genesis_result: &EvdGenesisResult,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let mut operations = state_changes_to_operations(changes);

    let chain_state = ChainState {
        height: 1,
        genesis_result: *genesis_result,
    };
    operations.push(Operation::Set {
        key: CHAIN_STATE_KEY.to_vec(),
        value: borsh::to_vec(&chain_state).map_err(|e| format!("serialize: {e}"))?,
    });

    storage
        .batch(operations)
        .await
        .map_err(|e| format!("batch failed: {:?}", e))?;

    storage
        .commit()
        .await
        .map_err(|e| format!("commit failed: {:?}", e))?;

    Ok(())
}

fn state_changes_to_operations(changes: Vec<StateChange>) -> Vec<Operation> {
    changes
        .into_iter()
        .map(|change| match change {
            StateChange::Set { key, value } => Operation::Set { key, value },
            StateChange::Remove { key } => Operation::Remove { key },
        })
        .collect()
}