lyquor/

lyquor.rs

use std::path::PathBuf;
use std::str::FromStr;
use std::sync::Arc;

use actix::prelude::*;
use actix::{Actor, Addr};
use alloy_node_bindings::{Anvil, AnvilInstance};
use lyquor_lib::lyquid::LyquidPool;
use lyquor_oci::oci::{LYQUID_PACK_ASSET_TYPE_VALUE_LYQUID, OCIReference, OCIRegistry, OCIRegistryAuth};
use lyquor_oci::pack::LyquidPackDigest;
use lyquor_oci::registry::{PulledContent, Registry};
use lyquor_primitives::{LyquidNumber, debug_struct_name};
use serde_json::{Map, Value};
use tokio::signal::unix::{SignalKind, signal};
use tokio::sync::Notify;

use lyquor_api::{
    actor::Stop,
    anyhow::{self, Context as _Context},
    config::{DB, NodeConfig},
    kvstore::{KVStore, Key, PrefixedKVStore, ShadowKVStore},
    profile::{Devnet, LyquorProfile, NetworkType},
    subkey_builder,
};
use lyquor_db::{MemDB, RocksDB};
use lyquor_jsonrpc::client::{ClientConfig, ClientHandle};
use lyquor_lib::{api, lyquid, utils};
use lyquor_primitives::{Address, Bytes, LyquidID, LyteLog, RegisterEvent};
use lyquor_seq::{SequenceBackend, eth, fco::FCO, repo};
use lyquor_vm::log::Log;

subkey_builder!(RootSubKey(
    ([0x01])-lyquid() => Key,
    ([0x02])-fco() => Key,
    ([0x03])-log() => Key,
    ([0x04])-event_store() => Key,
));

fn subkey() -> &'static RootSubKey {
    static KEY: std::sync::OnceLock<RootSubKey> = std::sync::OnceLock::new();
    KEY.get_or_init(|| RootSubKey::new(Bytes::new().into()))
}

const ANVIL_STATE_FILENAME: &str = "anvil.state";

static ANVIL_PID: std::sync::OnceLock<u32> = std::sync::OnceLock::new();

#[derive(Debug, Message)]
#[rtype(result = "()")]
struct LoadLyquid {
    id: LyquidID,
    deps: Vec<LyquidID>,
}

#[derive(Debug, Message)]
#[rtype(result = "()")]
struct SetPool {
    pool: Addr<lyquid::LyquidPool>,
}

struct LyquidDiscovery {
    pool: Option<Addr<lyquid::LyquidPool>>,
    register: Arc<Notify>,
    bartender_id: LyquidID,
    bartender_addr: Address,
}

lyquor_primitives::debug_struct_name!(LyquidDiscovery);

impl Handler<lyquor_seq::eth::GetContract> for LyquidDiscovery {
    type Result = ResponseFuture<Address>;
    #[tracing::instrument(level = "trace", skip(self, msg, _ctx))]
    fn handle(&mut self, msg: lyquor_seq::eth::GetContract, _ctx: &mut Context<Self>) -> Self::Result {
        let lyquor_seq::eth::GetContract { id, idx } = msg;
        if id == self.bartender_id && idx == Some(0) {
            return Box::pin(std::future::ready(self.bartender_addr));
        }
        let pool = self.pool.clone().expect("LyquidDiscovery: Pool not set.");
        let reg = self.register.clone();
        Box::pin(async move {
            loop {
                // FIXME: improve the efficiency of this by filtering the reigster event for
                // the specific lyquid in bartender
                // idx=None means latest; else nth deployment
                if let Some(n) = idx {
                    match pool.send(crate::lyquid::GetLyquidDeploymentInfo { id, idx: n }).await {
                        Ok(Some(info)) => return info.contract,
                        Ok(None) => reg.notified().await,
                        Err(e) => {
                            tracing::error!("Failed to get lyquid deployment info: {}", e);
                            // Stall forever for safety
                            std::future::pending::<()>().await;
                        }
                    }
                } else {
                    match pool.send(crate::lyquid::GetLatestLyquidDeploymentInfo { id }).await {
                        Ok(Some(info)) => return info.contract,
                        Ok(None) => reg.notified().await,
                        Err(e) => {
                            tracing::error!("Failed to get latest lyquid deployment info: {}", e);
                            std::future::pending::<()>().await;
                        }
                    }
                }
            }
        })
    }
}

// latest variant handled in GetContract via idx=None

impl Handler<lyquor_vm::log::LogItem> for LyquidDiscovery {
    type Result = ();

    #[tracing::instrument(level = "trace", skip(msg))]
    fn handle(&mut self, msg: lyquor_vm::log::LogItem, ctx: &mut Context<Self>) -> Self::Result {
        let event: Option<RegisterEvent> = lyquor_primitives::decode_object(&msg.0.data);
        if let Some(event) = event {
            ctx.spawn(self.load(event.id, event.deps).into_actor(self));
        }
    }
}

impl Handler<LoadLyquid> for LyquidDiscovery {
    type Result = ResponseFuture<()>;

    #[tracing::instrument(level = "trace")]
    fn handle(&mut self, msg: LoadLyquid, _ctx: &mut Context<Self>) -> Self::Result {
        Box::pin(self.load(msg.id, msg.deps))
    }
}

impl Handler<SetPool> for LyquidDiscovery {
    type Result = ();

    #[tracing::instrument(level = "trace")]
    fn handle(&mut self, msg: SetPool, ctx: &mut Context<Self>) -> Self::Result {
        if self.pool.is_some() {
            return;
        }
        let pool = msg.pool;
        let me = ctx.address();
        self.pool = Some(pool.clone());
        ctx.spawn(
            async move {
                // TODO: right now we just subscribe to every lyquid known in history, need to change this
                // to user-configured set (or groups)
                let lyquids_with_deps = match pool.send(crate::lyquid::GetRegisteredLyquidList).await.ok().flatten() {
                    None => {
                        // FIXME: this could be triggered when bartender has only executed one slot
                        // upon recovery. There is a race that before bartender repeat the
                        // execution of the first slot, its state is without image so this call
                        // will fail.
                        tracing::error!("Failed to obtain the registered Lyquid list.");
                        return;
                    }
                    Some(lyquids_with_deps) => lyquids_with_deps,
                };
                // Now we have the actual dependency list from the bartender
                for (id, deps) in lyquids_with_deps.into_iter() {
                    //
                    //  --- NB: We assume we already have all dependency binaries. ---
                    //
                    if let Err(e) = me.send(LoadLyquid { id, deps }).await {
                        tracing::error!("Failed to send LoadLyquid message: {}", e);
                    }
                }
            }
            .into_actor(self),
        );
    }
}

impl LyquidDiscovery {
    fn load(&mut self, id: LyquidID, deps: Vec<LyquidID>) -> impl Future<Output = ()> + 'static {
        let pool = self.pool.clone().expect("LyquidDiscovery: Pool not set.");
        let register = self.register.clone();
        async move {
            // NOTE: locator is only a retrieval hint; image_hash in slots remains authoritative.
            let image_locator = match pool
                .send(crate::lyquid::GetLatestLyquidDeploymentInfo { id })
                .await
                .ok()
                .flatten()
            {
                Some(info) => info.image_locator,
                None => None,
            };

            if let Some(image_locator) = image_locator {
                // FIXME: only pull by locator when the slot image hash is not available in the local ImageRepo.
                tracing::info!("LyquidDiscovery: downloading image from locator: {}", image_locator);
                Self::pull_image_from_locator(pool.clone(), image_locator)
                    .await
                    .unwrap_or_else(|e| {
                        tracing::error!("Failed to pull lyquid image from locator: {e}");
                    });
            }

            match pool.send(crate::lyquid::LoadLyquid { id, deps }).await {
                Ok(result) => match result {
                    Ok(already_exists) => {
                        if !already_exists {
                            tracing::info!("Discovered {}", id);
                        }
                    }
                    Err(e) => tracing::error!("Failed to register discovered Lyquid: {e}"),
                },
                Err(e) => tracing::error!("Failed to send LoadLyquid message: {}", e),
            }
            register.notify_waiters();
        }
    }

    async fn pull_image_from_locator(pool: Addr<LyquidPool>, locator: String) -> Result<(), String> {
        let oci_reference = OCIReference::from_str(&locator)
            .map_err(|e| format!("Unsupported image locator format (expected OCI reference): {e}"))?;
        let host = match oci_reference.registry().split_once('/').map(|(host, _)| host) {
            Some(h) => h,
            None => oci_reference.registry(),
        };
        let auth = match OCIRegistry::get_docker_credential(host) {
            Ok(auth) => auth,
            Err(e) => {
                tracing::trace!(
                    "No docker credential found (reason: {}), falling back to anonymous auth.",
                    e
                );
                OCIRegistryAuth::Anonymous
            }
        };
        let oci = OCIRegistry::new(auth, oci_reference.registry());
        let digest = oci_reference
            .digest()
            .ok_or_else(|| "OCI reference must include a digest.".to_string())
            .and_then(|d| LyquidPackDigest::from_oci_digest(d).map_err(|e| e.to_string()))?;

        // Get image
        let PulledContent::Layer(wasm) = oci
            .pull(
                oci_reference.repository(),
                None,
                Some(&digest),
                Some(LYQUID_PACK_ASSET_TYPE_VALUE_LYQUID),
                None,
                None,
            )
            .await
            .map_err(|e| e.to_string())?
        else {
            return Err("Failed to pull wasm image layer from oci registry.".to_string());
        };

        // Push image into repo
        let repo_hash = match pool
            .send(crate::lyquid::ImageRepo)
            .await
            .unwrap()
            .0
            .push(wasm.into())
            .await
        {
            Ok(hash) => hash,
            Err(e) => return Err(e.to_string()),
        };

        tracing::info!("Image hash: {:?}", repo_hash);

        // Is this necessary to check the hash here?

        Ok(())
    }
}

impl Handler<Stop> for LyquidDiscovery {
    type Result = ();

    #[tracing::instrument(level = "trace")]
    fn handle(&mut self, _: Stop, ctx: &mut Context<Self>) -> Self::Result {
        ctx.stop();
    }
}

impl Actor for LyquidDiscovery {
    type Context = actix::Context<Self>;

    #[tracing::instrument(level = "trace", skip(_ctx))]
    fn started(&mut self, _ctx: &mut Self::Context) {}

    #[tracing::instrument(level = "trace", skip(_ctx))]
    fn stopped(&mut self, _ctx: &mut Self::Context) {}
}

struct DBFlusher {
    last_bn: u64,
    db_dir: PathBuf,
    store: Arc<ShadowKVStore<Box<dyn KVStore>>>,
    fco: Addr<FCO>,
    anvil_chain: Option<ClientHandle>,
    flush_interval_secs: u64,
}

lyquor_primitives::debug_struct_name!(DBFlusher);

impl DBFlusher {
    fn prepare_flush(&self) -> impl Future<Output = anyhow::Result<()>> + 'static {
        let anvil_chain = self.anvil_chain.clone();
        let fco = self.fco.clone();
        let db_dir = self.db_dir.clone();
        async move {
            fco.send(lyquor_seq::fco::Commit)
                .await
                .map_err(anyhow::Error::from)
                .and_then(|e| e.map_err(anyhow::Error::from))?;

            if let Some(client) = &anvil_chain {
                // If running devnet (anvil), dump chain state.
                //
                // We dump the anvil chain state *after* the commit of FCO, so FCO's progress will
                // not outpace the dumped chain state (i.e., includes execution caused by the
                // blocks later). Of course, this somehow assume Anvil's API handling has an
                // ordering (the dump chain state is more recent than that handled previous API
                // responses).
                use lyquor_jsonrpc::types::{AnvilDumpState, AnvilDumpStateResp};
                let dump = client
                    .request::<AnvilDumpState, AnvilDumpStateResp>(AnvilDumpState)
                    .await
                    .map(|r| r.0);

                match dump {
                    Ok(state) => {
                        // Write the anvil state to file
                        let devnet_path = db_dir.join("devnet");
                        std::fs::create_dir_all(&devnet_path)?;
                        std::fs::write(devnet_path.join(ANVIL_STATE_FILENAME), &state)?;
                    }
                    Err(_) => {
                        tracing::error!(
                            "DBFlusher: Failed to dump state from anvil. This could cause issue due to the loss of local chain state upon next startup."
                        );
                    }
                }
            }

            Ok(())
        }
    }

    fn flush(&mut self, res: anyhow::Result<()>) {
        if let Err(e) = res.and_then(|_| self.store.flush().map_err(|e| e.into())) {
            tracing::error!("Failed to write: {e}");
        } else {
            tracing::info!("Written to the storage backend (block={}).", self.last_bn);
        }
    }
}

#[derive(Message, Debug)]
#[rtype(result = "()")]
struct FlushDB;

impl Handler<FlushDB> for DBFlusher {
    type Result = ResponseActFuture<Self, ()>;

    #[tracing::instrument(level = "trace", skip(self, _ctx))]
    fn handle(&mut self, _: FlushDB, _ctx: &mut actix::Context<Self>) -> Self::Result {
        Box::pin(self.prepare_flush().into_actor(self).map(move |res, actor, _ctx| {
            actor.flush(res);
        }))
    }
}

impl Handler<Stop> for DBFlusher {
    type Result = ResponseActFuture<Self, ()>;

    #[tracing::instrument(level = "trace")]
    fn handle(&mut self, _: Stop, _ctx: &mut actix::Context<Self>) -> Self::Result {
        Box::pin(self.prepare_flush().into_actor(self).map(move |res, actor, _ctx| {
            actor.flush(res);
        }))
    }
}

impl Actor for DBFlusher {
    type Context = actix::Context<Self>;

    #[tracing::instrument(level = "trace", skip_all)]
    fn started(&mut self, ctx: &mut Self::Context) {
        if self.flush_interval_secs > 0 {
            let periodic_write = tokio::time::Duration::from_secs(self.flush_interval_secs);
            ctx.run_interval(periodic_write, |_act, ctx| {
                ctx.address().do_send(FlushDB);
            });
        }
    }

    #[tracing::instrument(level = "trace", skip_all)]
    fn stopped(&mut self, _ctx: &mut Self::Context) {}
}

async fn setup_eth_backend(
    profile: &LyquorProfile, reg: lyquor_seq::eth::ContractRegistry, jsonrpc_client: ClientHandle,
) -> anyhow::Result<eth::Backend> {
    let finality_tag: lyquor_jsonrpc::types::BlockNumber =
        serde_json::from_str(profile.finality()).context("Invalid finality tag.")?;

    Ok(eth::Backend::new(
        eth::Config::builder()
            .finality_tag(finality_tag)
            .registry(reg)
            .blocks_per_request(std::num::NonZeroUsize::new(10).context("Block per request should be >0.")?)
            .build(),
        jsonrpc_client,
    ))
}

#[derive(Default)]
struct Subsystems {
    fco: Option<Addr<FCO>>,
    log: Option<Addr<Log>>,
    pool: Option<Addr<lyquid::LyquidPool>>,
    api: Option<Addr<api::EthAPIServer>>,
    db: Option<Addr<DBFlusher>>,
    dis: Option<Addr<LyquidDiscovery>>,
    client: Option<ClientHandle>,
    eth_backend: Option<lyquor_seq::eth::Backend>,
}
debug_struct_name!(Subsystems);

impl Subsystems {
    #[tracing::instrument(level = "trace")]
    async fn stop(&mut self) {
        if let Some(api) = self.api.take() {
            api.send(Stop).await.ok();
        }
        if let Some(eth_backend) = self.eth_backend.take() {
            eth_backend.stop().await;
        }
        if let Some(log) = self.log.take() {
            log.send(Stop).await.ok();
        }
        if let Some(dis) = self.dis.take() {
            dis.send(Stop).await.ok();
        }
        if let Some(pool) = self.pool.take() {
            pool.send(Stop).await.ok();
        }
        if let Some(db) = self.db.take() {
            db.send(Stop).await.ok();
        }
        if let Some(dis) = self.dis.take() {
            dis.send(Stop).await.ok();
        }
        if let Some(fco) = self.fco.take() {
            fco.send(Stop).await.ok();
        }
    }
}

/// TODO: refactor this with actix's Actor model.
struct Node {
    sys: Subsystems,
    config: Arc<NodeConfig>,
    profile: Arc<LyquorProfile>,
    seq_eth_url: String,
    started_anvil: bool,
    network: Option<lyquor_net::hub::Hub>,
}

lyquor_primitives::debug_struct_name!(Node);

impl Handler<Stop> for Node {
    type Result = AtomicResponse<Self, ()>;

    #[tracing::instrument(level = "trace")]
    fn handle(&mut self, _: Stop, _ctx: &mut Context<Self>) -> Self::Result {
        let mut sys = std::mem::replace(&mut self.sys, Subsystems::default());
        AtomicResponse::new(Box::pin(
            async move {
                sys.stop().await;
            }
            .into_actor(self)
            .map(|_, _act, ctx| {
                ctx.stop();
            }),
        ))
    }
}

impl Actor for Node {
    type Context = Context<Self>;

    #[tracing::instrument(level = "trace", skip(ctx))]
    fn started(&mut self, ctx: &mut Self::Context) {
        // NOTE: use spawn here instead of wait so Stop can be handled
        ctx.spawn(
            Self::init(
                self.config.clone(),
                self.profile.clone(),
                self.seq_eth_url.clone(),
                self.started_anvil,
                self.network.take().expect("Node: Network not found."),
            )
            .into_actor(self)
            .map(|sys, act, ctx| {
                match sys {
                    Ok(sys) => {
                        act.sys = sys;
                    }
                    Err(e) => {
                        tracing::error!("Failed to initialize node: {e:?}");
                        ctx.stop();
                    }
                };
            }),
        );
    }

    #[tracing::instrument(level = "trace", skip(_ctx))]
    fn stopped(&mut self, _ctx: &mut Self::Context) {}
}

impl Node {
    #[tracing::instrument(level = "trace", skip_all)]
    async fn init(
        config: Arc<NodeConfig>, profile: Arc<LyquorProfile>, seq_eth_url: String, started_anvil: bool,
        network: lyquor_net::hub::Hub,
    ) -> anyhow::Result<Subsystems> {
        let mut sys = Subsystems::default();

        // Set up Ethereum sequence backend.
        let jsonrpc_client = ClientConfig::builder().url(seq_eth_url.parse()?).build().into_client();
        sys.client = Some(jsonrpc_client.clone());

        //// 1. DB & Storage
        let (store, lyq_store, anvil_chain): (
            Arc<ShadowKVStore<Box<dyn KVStore>>>,
            Arc<dyn utils::LVMStoreFactory>,
            Option<ClientHandle>,
        ) = match &config.storage.db {
            DB::MemDb => {
                tracing::warn!("using MemDB, all data will be lost upon exit");
                let store = Arc::new(ShadowKVStore::new(Box::new(MemDB::new()) as Box<dyn KVStore>));
                (store, Arc::new(utils::LVMMemStore), None)
            }
            DB::RocksDb { db_dir } => {
                let base_dir = db_dir.clone();
                let store = Arc::new(ShadowKVStore::new(
                    Box::new(RocksDB::new(&base_dir.join(profile.id()))?) as Box<dyn KVStore>,
                ));
                let lyq_store = Arc::new(utils::LVMDBStore(Arc::new(PrefixedKVStore::new(
                    store.clone(),
                    subkey().lyquid(),
                ))));

                // Load anvil state if we started our own anvil and state file exists
                let anvil_chain = if started_anvil {
                    let state_file = db_dir.join(profile.id()).join(ANVIL_STATE_FILENAME);
                    if state_file.exists() {
                        use lyquor_jsonrpc::types::{AnvilLoadState, AnvilLoadStateResp};
                        let state_data = std::fs::read(&state_file)?;
                        let load_result = jsonrpc_client
                            .request::<AnvilLoadState, AnvilLoadStateResp>(AnvilLoadState(state_data.into()))
                            .await?;
                        tracing::info!("Loaded anvil state from {}: {}.", state_file.display(), load_result.0);
                    } else {
                        tracing::info!(
                            "No existing anvil state file found at {}, starting with fresh state.",
                            state_file.display()
                        );
                    }
                    Some(jsonrpc_client.clone())
                } else {
                    None
                };

                (store, lyq_store, anvil_chain)
            }
        };
        ////

        //// 2. Fate-Constrainted Ordering Subsystem.
        // Start discovering Lyquid and sequencing contracts.
        let dis = LyquidDiscovery {
            pool: None,
            register: Arc::new(Notify::new()),
            bartender_id: *profile.bartender_id(),
            bartender_addr: Address::from_str(profile.bartender_addr())
                .context("Invalid bartender address in profile.")?
                .into(),
        }
        .start();
        sys.dis = Some(dis.clone());

        let eth_backend = setup_eth_backend(&profile, dis.clone().recipient(), jsonrpc_client.clone()).await?;

        let fco_store = PrefixedKVStore::new(store.clone(), subkey().fco());
        let fco = FCO::new(
            eth_backend.clone(),
            Arc::new(fco_store),
            Some(profile.init_chain_position()),
        )?
        .start();
        sys.fco = Some(fco.clone());
        // Keep a separate reference for shutdown management
        sys.eth_backend = Some(eth_backend);
        ////

        //// 3. Lyquor Virtual Machine Subsystem.
        // Configure Log subsystem.
        let log_store = PrefixedKVStore::new(store.clone(), subkey().log());
        let log = Log::new(Arc::new(lyquor_state::DBSimpleStateStore::new(log_store))).start();
        sys.log = Some(log.clone());

        let mut lvm = lyquid::LVM::new(
            log.clone(),
            network,
            10,
            Some(lyquor_vm::InterOps {
                inter: fco.clone().recipient(),
                submit: fco.clone().recipient(),
            }),
        );
        let api_subs = api::Subscriptions::new().start();
        let event_store = Arc::new(lyquor_seq::event_store::EventStore::new(
            Arc::new(PrefixedKVStore::new(store.clone(), subkey().event_store())),
            1000,
        ));
        ////

        //// 4. UPC peers
        for peer in &config.peers {
            lvm.upc_mut()
                .add_remote(peer.node_id, Some(peer.upc_addr.clone()))
                .await
                .with_context(|| format!("Failed to add UPC peer {}", peer.node_id))?;
        }
        ////

        //// 5. Pool of all Lyquids
        let image_repo_store = RocksDB::new(&config.storage.image_repo_dir)?;
        let image_repo: Arc<dyn repo::LyquidImageRepo> = Arc::new(repo::KVStoreRepo::new(image_repo_store));
        let bartender_id = *profile.bartender_id();
        let pool = lyquid::LyquidPool::new(
            bartender_id,
            lvm,
            lyquid::LyquidPoolSetup {
                fco: fco.clone(),
                store_factory: lyq_store,
                image_repo,
                event_store,
                api_subs: api_subs.clone(),
                config: lyquid::LyquidConfig::builder().build(),
            },
        )
        .await?
        .start();
        sys.pool = Some(pool.clone());
        ////

        // Configure bartender.
        let bartender_setup = {
            let register_topic = LyteLog::tagged_value_topic("Register");
            match log
                .send(lyquor_vm::log::GetNumberOfRecords {
                    id: bartender_id,
                    topic: register_topic,
                })
                .await
            {
                Ok(count) if count > 0 => None,
                Ok(_) => Some(log.send(lyquor_vm::log::WaitNext {
                    id: bartender_id,
                    topic: register_topic,
                })),
                Err(e) => return Err(e).context("Failed to get number of records during bartender setup."),
            }
        };

        // Start the API server - use the same jsonrpc client as sequencer
        let api = lyquor_lib::api::EthAPIServer::new(
            &config.network.api_addr,
            jsonrpc_client.clone(),
            api_subs,
            fco.clone(),
            pool.clone(),
        )
        .await?;
        sys.api = Some(api.start());

        // Wait for bartender to be registered.
        if let Some(reg) = bartender_setup {
            tracing::info!("Waiting for bartender to be registered.");
            reg.await.context("Failed to wait for bartender registration.")?;
        }
        // Bartender, as a normal Lyquid, may have some re-execution, let's make sure it is done.
        let bartender_ins = pool
            .send(lyquid::GetBartender)
            .await
            .expect("Failed to obtain bartender.");
        let bartender_ln = bartender_ins.instance().max_number().unwrap_or(LyquidNumber::ZERO);
        tracing::debug!("Waiting for bartender to reach {bartender_ln}.");
        bartender_ins
            .wait_until(bartender_ln)
            .await
            .context("Failed to wait for the initial re-execution of bartender.")?;

        // Finish up LyquidDiscovery.
        log.send(lyquor_vm::log::Subscribe {
            id: bartender_id,
            topic: LyteLog::tagged_value_topic("Register"),
            subscriber: dis.clone().recipient(),
        })
        .await
        .context("Failed to subscribe to log system for LyquidDiscovery.")?;
        dis.send(SetPool { pool: pool.clone() })
            .await
            .context("Failed to set pool in LyquidDiscovery.")?;

        tracing::info!("INIT COMPLETE (bartender={bartender_id})");

        // Start a DB writer that persist when node is idle.
        let db = DBFlusher {
            last_bn: 0,
            db_dir: config.resolved_db_dir(),
            store,
            fco,
            anvil_chain,
            flush_interval_secs: config.flush_interval_secs,
        }
        .start();
        sys.db = Some(db);

        Ok(sys)
    }
}

fn generate_unused_port() -> anyhow::Result<u16> {
    let listener = std::net::TcpListener::bind("0.0.0.0:0").context("Failed to bind to local address.")?;
    let port = listener.local_addr().context("Failed to get local address.")?.port();
    Ok(port)
}

fn start_devnet_anvil() -> anyhow::Result<AnvilInstance> {
    let anvil_port = generate_unused_port()?;
    // Mask the SIGINT signal so anvil process won't terminate on SIGINT (it'll only
    // terminate when dropped)
    use nix::sys::signal::{self, SigSet, SigmaskHow, Signal};
    let mut sigset = SigSet::empty();
    sigset.add(Signal::SIGINT);
    let mut old_mask = SigSet::empty();
    signal::sigprocmask(SigmaskHow::SIG_BLOCK, Some(&sigset), Some(&mut old_mask))
        .context("Failed to mask SIGINT for anvil.")?;

    let mut anvil = Anvil::new().port(anvil_port).keep_stdout().try_spawn()?;

    let stdout_reader = std::io::BufReader::new(
        anvil
            .child_mut()
            .stdout
            .take()
            .context("Failed to read from anvil stdout.")?,
    );

    tokio::task::spawn_blocking(move || {
        use std::io::BufRead;
        let mut stdout_lines = stdout_reader.lines();
        while let Some(line) = stdout_lines.next() {
            tracing::debug!(target: "lyquor_anvil", "{}", line.unwrap_or_else(|_| "".to_string()));
        }
    });

    tracing::info!("Anvil started at port {}.", anvil_port);
    signal::sigprocmask(SigmaskHow::SIG_SETMASK, Some(&old_mask), None)
        .context("Failed to restore old signal mask for anvil.")?;

    Ok(anvil)
}

fn parse_override_value(raw: &str) -> Value {
    if let Ok(v) = raw.parse::<bool>() {
        return Value::from(v);
    }
    if let Ok(v) = raw.parse::<u64>() {
        return Value::from(v);
    }
    if let Ok(v) = raw.parse::<i64>() {
        return Value::from(v);
    }
    if let Ok(v) = raw.parse::<f64>() {
        return Value::from(v);
    }
    Value::from(raw.to_string())
}

fn insert_override_path(overrides: &mut Map<String, Value>, key: &str, value: Value) -> anyhow::Result<()> {
    let mut segments = key.split('.').peekable();
    let mut current = overrides;

    while let Some(segment) = segments.next() {
        if segments.peek().is_none() {
            current.insert(segment.to_string(), value);
            return Ok(());
        }

        let entry = current
            .entry(segment.to_string())
            .or_insert_with(|| Value::Object(Map::<String, Value>::new()));

        match entry {
            Value::Object(dict) => {
                current = dict;
            }
            _ => anyhow::bail!("Override path '{key}' conflicts with non-table value."),
        }
    }

    Ok(())
}

fn build_config_overrides(matches: &clap::ArgMatches) -> anyhow::Result<Map<String, Value>> {
    let mut overrides: Map<String, Value> = Map::new();

    if let Some(entries) = matches.get_many::<String>("config-override") {
        for entry in entries {
            let (key, value) = entry
                .split_once('=')
                .ok_or_else(|| anyhow::anyhow!("Invalid --config-override '{entry}', expected key=value"))?;
            insert_override_path(&mut overrides, key, parse_override_value(value))?;
        }
    }

    Ok(overrides)
}

#[actix::main]
async fn main() -> anyhow::Result<()> {
    // Install panic hook to cleanup anvil on panic
    std::panic::set_hook(Box::new(|_| {
        if let Some(&pid) = ANVIL_PID.get() {
            tracing::warn!("Panic detected, killing anvil process {}", pid);
            use nix::sys::signal::{Signal, kill};
            let _ = kill(nix::unistd::Pid::from_raw(pid as i32), Signal::SIGTERM);
        }
    }));

    lyquor_cli::setup_tracing()?;

    println!("{}", lyquor_cli::format_logo_banner());

    let matches = clap::command!()
        .propagate_version(true)
        .arg(clap::arg!(--config <PATH> "Path to the Lyquor config file (toml, yaml, json).").required(false))
        .arg(
            clap::arg!(--"config-override" <KEY_VALUE> "Override config with key=value (repeatable).")
                .required(false)
                .action(clap::ArgAction::Append)
                .value_parser(clap::builder::NonEmptyStringValueParser::new()),
        )
        .get_matches();

    let config_path = matches.get_one::<String>("config").map(PathBuf::from);
    let overrides = build_config_overrides(&matches)?;
    let config = NodeConfig::load(config_path, overrides)?;

    let node_seed = config.node_key.seed_bytes()?;
    let (_, tls_config) = lyquor_tls::generator::single_node_config_with_seed(&node_seed);
    let mut network = lyquor_net::hub::HubBuilder::new(tls_config)
        .listen_addr(config.network.upc_addr.clone())
        .build()
        .context("Failed to setup network.")?;
    network.start().await.context("Failed to start network.")?;

    let mut _anvil: Option<AnvilInstance> = None;
    let profile: Arc<LyquorProfile> = Arc::new(match config.profile {
        NetworkType::Devnet => {
            let ws_url = match &config.force_seq_eth {
                None => {
                    let anvil = start_devnet_anvil()?;
                    // Store anvil PID for panic hook
                    ANVIL_PID.set(anvil.child().id()).ok();
                    let ep = anvil.ws_endpoint();
                    _anvil = Some(anvil);
                    ep
                }
                Some(url) => url.to_string(),
            };

            Devnet::new(ws_url).into()
        }
        _ => return Err(anyhow::anyhow!("Network is not supported.")),
    });
    let seq_eth_url = config
        .force_seq_eth
        .clone()
        .unwrap_or_else(|| profile.sequencer().to_string());

    // Signal handlers should be registered before starting any actors, so they'll be stopped
    // gracefully.
    let mut sigint = signal(SignalKind::interrupt()).context("Failed to register SIGINT handler")?;
    let mut sigterm = signal(SignalKind::terminate()).context("Failed to register SIGTERM handler")?;

    let node = Node {
        sys: Subsystems::default(),
        config: Arc::new(config),
        profile: profile.clone(),
        seq_eth_url,
        started_anvil: _anvil.is_some(),
        network: Some(network),
    }
    .start();

    tokio::select! {
        _ = sigint.recv() => {
            tracing::info!("received SIGINT");
        }
        _ = sigterm.recv() => {
            tracing::info!("received SIGTERM");
        }
    }

    Ok(node.send(Stop).await?)
}