urma-runtime 0.3.4

use crate::config::{P2P_BLOCK_TIMEOUT_SECS, P2P_TIP_REFRESH_SECS};
use crate::error::{Context, Error, ensure};
use crate::light::{LightSync, Progress};
use crate::p2p::HeaderBootstrap;
use crate::p2p::blocks::BlockCache;
use crate::p2p::headers::{Anchor, HeaderChain};
use crate::p2p::source::{VerifiedBlock, decode_verified};
use crate::p2p::worker::{self, Command, Freshness, Shared, Status, locked};
use crate::transport::{BlockEncoding, Evidence, Provider};
use bitcoin::BlockHash;
use bitcoin::block::Header;
use serde_json::{Value, json};
use std::net::SocketAddr;
use std::path::Path;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{self, SyncSender, TrySendError};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use urma_chain::observation::Chain;

enum Engine {
    Cold,
    Running(SyncSender<Command>),
}

/// P2P header view rooted in a trusted checkpoint and existing local cache.
/// Typed results establish the limited verification described by `VerifiedBlock`.
pub struct P2pProvider {
    chain: Chain,
    shared: Arc<Shared>,
    engine: Mutex<Engine>,
}

impl P2pProvider {
    pub fn new(chain: Chain, cache_dir: &Path) -> Result<Self, Error> {
        Self::with_peers(chain, cache_dir, Anchor::embedded(chain)?, Vec::new())
    }

    pub fn with_peers(
        chain: Chain,
        cache_dir: &Path,
        anchor: Anchor,
        pinned: Vec<SocketAddr>,
    ) -> Result<Self, Error> {
        let started = Instant::now();
        let loaded = HeaderChain::open_at(chain, cache_dir, anchor);
        tracing::debug!(
            target: "urma_startup",
            chain = chain.label(),
            stage = "header_chain_open",
            elapsed_ms = started.elapsed().as_secs_f64() * 1000.0,
            result = ?loaded.as_ref().map(HeaderChain::synced_count),
            "light client stage completed"
        );
        let headers = loaded?;
        Ok(Self::from_headers(chain, headers, pinned, cache_dir))
    }

    /// Open a checkpoint-relative view with authenticated preceding context.
    /// The operator still trusts the bootstrap anchor's chain membership and
    /// height; its context does not establish full consensus or earlier scan
    /// coverage. Construction opens the local cache without starting a worker
    /// or network connections. `warm` or the first verified read starts it.
    pub fn with_bootstrap(
        cache_dir: &Path,
        bootstrap: HeaderBootstrap,
        pinned: Vec<SocketAddr>,
    ) -> Result<Self, Error> {
        let chain = bootstrap.chain();
        let headers = HeaderChain::open_bootstrap(cache_dir, bootstrap)?;
        Ok(Self::from_headers(chain, headers, pinned, cache_dir))
    }

    fn from_headers(
        chain: Chain,
        headers: HeaderChain,
        pinned: Vec<SocketAddr>,
        cache_dir: &Path,
    ) -> Self {
        let shared = Arc::new(Shared {
            chain,
            pinned,
            cache_dir: cache_dir.to_owned(),
            headers: Mutex::new(headers),
            blocks: Mutex::new(BlockCache::new()),
            status: Mutex::new(Status {
                peers: 0,
                target: 0,
                synced: false,
                refreshed: Freshness::Never,
            }),
            stop: AtomicBool::new(false),
        });
        Self {
            chain,
            shared,
            engine: Mutex::new(Engine::Cold),
        }
    }

    pub fn warm(&self) -> Result<(), Error> {
        let mut engine = locked(&self.engine, "engine");
        match &*engine {
            Engine::Running(_commands) => return Ok(()),
            Engine::Cold => (),
        }
        let (sender, receiver) = mpsc::sync_channel(16);
        let background = self.shared.clone();
        std::thread::Builder::new()
            .name("urma-p2p".into())
            .spawn(move || worker::run(background, receiver))?;
        *engine = Engine::Running(sender);
        tracing::debug!(chain = self.chain.label(), "light client worker started");
        Ok(())
    }

    fn ignite(&self) {
        match self.warm() {
            Ok(()) => (),
            Err(error) => tracing::warn!(%error, "light client worker could not start"),
        }
    }

    pub fn progress(&self) -> Progress {
        let headers = locked(&self.shared.headers, "headers");
        let status = locked(&self.shared.status, "status");
        Progress {
            headers_synced: headers.synced_count(),
            headers_target: status.target.max(headers.tip_height()) - headers.start_height(),
            peers_connected: status.peers,
            synced: status.synced,
        }
    }

    /// Absolute height header sync heads for: the best height connected peers
    /// announced, never below the local tip. Announced, not proven. `None`
    /// while no peer is connected or none has announced a height.
    pub fn target_height(&self) -> Option<u64> {
        let headers = locked(&self.shared.headers, "headers");
        let status = locked(&self.shared.status, "status");
        (status.peers > 0 && status.target > 0).then(|| status.target.max(headers.tip_height()))
    }

    pub fn state(&self) -> LightSync {
        let progress = self.progress();
        match &*locked(&self.engine, "engine") {
            Engine::Cold => LightSync::Cold(progress),
            Engine::Running(_commands) => LightSync::Running(progress),
        }
    }

    pub fn synced(&self) -> bool {
        locked(&self.shared.status, "status").synced
    }

    pub fn ready(&self) -> bool {
        let status = locked(&self.shared.status, "status");
        status.synced && status.peers > 0
    }

    fn command(&self, command: Command) -> Result<(), Error> {
        let sender = match &*locked(&self.engine, "engine") {
            Engine::Cold => {
                return Err(Error::Unsupported(
                    "light client worker has not been started".into(),
                ));
            }
            Engine::Running(sender) => sender.clone(),
        };
        match sender.try_send(command) {
            Ok(()) => Ok(()),
            Err(TrySendError::Full(_command)) => {
                Err(Error::Capacity("light client command queue is full".into()))
            }
            Err(TrySendError::Disconnected(_command)) => {
                Err(Error::Unsupported("light client worker stopped".into()))
            }
        }
    }

    fn source_ready(&self) -> Result<(), Error> {
        self.warm()?;
        if !self.ready() {
            return Err(Error::Unsupported(
                "verified P2P source is not ready; retry after header sync and peer connection"
                    .into(),
            ));
        }
        Ok(())
    }

    /// Snapshot of the active header tip, relative to the configured trusted
    /// checkpoint and local header cache, not full consensus validation.
    /// Starts the worker if cold and fails immediately while it is not ready.
    /// Stale-tip refresh waits at most `P2P_BLOCK_TIMEOUT_SECS` for its reply;
    /// a full command queue is refused immediately. No other source is used.
    pub fn verified_tip(&self) -> Result<(u64, Header), Error> {
        self.source_ready()?;
        self.refresh_if_stale()?;
        self.source_ready()?;
        let headers = locked(&self.shared.headers, "headers");
        let height = headers.tip_height();
        Ok((height, headers.header_at(height)?))
    }

    /// Active header at `height`, under the same trust and readiness contract
    /// as `verified_tip()`. Heights before the checkpoint are unavailable.
    pub fn verified_header(&self, height: u64) -> Result<Header, Error> {
        self.source_ready()?;
        self.refresh_if_stale()?;
        self.source_ready()?;
        locked(&self.shared.headers, "headers").header_at(height)
    }

    /// Fetch by the captured active header's hash, check block integrity, and
    /// refuse the result if that header was replaced or removed during fetch.
    /// Retains original bytes; see `VerifiedBlock` for verification limits.
    /// Readiness and queue admission fail fast. Refresh and block replies each
    /// wait at most `P2P_BLOCK_TIMEOUT_SECS`; no router or fallback is involved.
    pub fn verified_block(&self, height: u64) -> Result<VerifiedBlock, Error> {
        let header = self.verified_header(height)?;
        let raw = self.fetch_raw(header.block_hash())?;
        let block = decode_verified(self.chain, height, header, raw)?;
        let headers = locked(&self.shared.headers, "headers");
        ensure!(
            headers.header_at(height)? == header,
            "active verified header changed during block fetch at height {height}"
        );
        Ok(block)
    }

    fn fetch_raw(&self, hash: BlockHash) -> Result<Vec<u8>, Error> {
        let (reply, answer) = mpsc::sync_channel(1);
        self.command(Command::Block { hash, reply })?;
        answer
            .recv_timeout(Duration::from_secs(P2P_BLOCK_TIMEOUT_SECS))
            .map_err(|error| Error::Unsupported(format!("block fetch did not complete: {error}")))?
    }

    fn refresh_if_stale(&self) -> Result<(), Error> {
        let stale = match locked(&self.shared.status, "status").refreshed {
            Freshness::Never => true,
            Freshness::At(when) => when.elapsed() > Duration::from_secs(P2P_TIP_REFRESH_SECS),
        };
        if !stale {
            return Ok(());
        }
        let (reply, answer) = mpsc::sync_channel(1);
        self.command(Command::Refresh { reply })?;
        answer
            .recv_timeout(Duration::from_secs(P2P_BLOCK_TIMEOUT_SECS))
            .map_err(|error| {
                Error::Unsupported(format!("header refresh did not complete: {error}"))
            })?
    }

    fn chain_info(&self) -> Result<Value, Error> {
        self.refresh_if_stale()?;
        let headers = locked(&self.shared.headers, "headers");
        let name = match self.chain {
            Chain::BitcoinMainnet | Chain::LitecoinMainnet => "main",
            Chain::LitecoinTestnet | Chain::BitcoinTestnet4 => "test",
            Chain::BitcoinRegtest => "regtest",
        };
        Ok(json!({
            "chain": name,
            "blocks": headers.tip_height(),
            "headers": headers.tip_height(),
            "bestblockhash": headers.tip_hash().to_string(),
            "initialblockdownload": false,
        }))
    }

    fn block_hash(&self, args: &[Value]) -> Result<Value, Error> {
        let height = args
            .first()
            .context("missing block height")?
            .as_u64()
            .context("block height must be a non-negative integer")?;
        self.refresh_if_stale()?;
        let headers = locked(&self.shared.headers, "headers");
        Ok(json!(headers.hash_at(height)?.to_string()))
    }

    fn block_header(&self, args: &[Value]) -> Result<Value, Error> {
        let hash = requested_hash(args)?;
        let headers = locked(&self.shared.headers, "headers");
        let height = headers.height_of(hash)?;
        let header = headers.header_at(height)?;
        Ok(json!({
            "hash": hash.to_string(),
            "confirmations": headers.tip_height() - height + 1,
            "height": height,
            "version": header.version.to_consensus(),
            "merkleroot": header.merkle_root.to_string(),
            "time": header.time,
            "nonce": header.nonce,
            "bits": format!("{:08x}", header.bits.to_consensus()),
            "previousblockhash": header.prev_blockhash.to_string(),
        }))
    }

    fn block(&self, args: &[Value]) -> Result<Value, Error> {
        let hash = requested_hash(args)?;
        ensure!(
            args.get(1).context("missing block verbosity")? == &json!(0),
            "the light client serves raw blocks only (verbosity 0)"
        );
        let raw = self.fetch_raw(hash)?;
        Ok(json!(hex::encode(raw)))
    }
}

fn requested_hash(args: &[Value]) -> Result<BlockHash, Error> {
    Ok(args
        .first()
        .context("missing block hash")?
        .as_str()
        .context("block hash must be a string")?
        .parse()?)
}

impl Provider for P2pProvider {
    fn label(&self) -> String {
        format!("p2p light client ({})", self.chain.label())
    }

    fn evidence(&self) -> Evidence {
        Evidence::LightClientInclusion
    }

    fn block_encoding(&self) -> BlockEncoding {
        BlockEncoding::Esplora
    }

    fn supports(&self, method: &str) -> bool {
        match method {
            "getblock" | "getblockheader" => {
                self.ignite();
                self.ready()
            }
            "getblockchaininfo" | "getblockhash" => self.ready(),
            _other => false,
        }
    }

    fn call(&self, chain: Chain, method: &str, args: &[Value]) -> Result<Value, Error> {
        ensure!(
            chain.genesis()? == self.chain.genesis()?,
            "light client serves {} only",
            self.chain.label()
        );
        match method {
            "getblockchaininfo" => self.chain_info(),
            "getblockhash" => self.block_hash(args),
            "getblockheader" => self.block_header(args),
            "getblock" => self.block(args),
            other => Err(Error::Unsupported(format!(
                "the peer-to-peer light client does not serve {other}"
            ))),
        }
    }
}

impl Provider for Arc<P2pProvider> {
    fn label(&self) -> String {
        P2pProvider::label(self)
    }

    fn evidence(&self) -> Evidence {
        P2pProvider::evidence(self)
    }

    fn block_encoding(&self) -> BlockEncoding {
        P2pProvider::block_encoding(self)
    }

    fn supports(&self, method: &str) -> bool {
        P2pProvider::supports(self, method)
    }

    fn call(&self, chain: Chain, method: &str, args: &[Value]) -> Result<Value, Error> {
        P2pProvider::call(self, chain, method, args)
    }
}

impl Drop for P2pProvider {
    fn drop(&mut self) {
        self.shared.stop.store(true, Ordering::Release);
    }
}

#[cfg(test)]
#[path = "source_tests.rs"]
mod source_tests;