observer/crates/blackbeard-cli/src/main.rs

//! `blackbeard` — the operator CLI.
//!
//! Three jobs, all of them things an operator needs *before* or *around* a
//! deploy rather than during normal running:
//!
//! - **`probe`** answers "will the observer work against this node?" without
//!   deploying anything. It is the command to run when a chain is added to the
//!   config, and the one to run on launch day for mainnet.
//! - **`backfill`** pulls history into the database, so a newly deployed
//!   observer has a leaderboard immediately instead of one that fills in over
//!   the following week.
//! - **`standings`** prints the board in a terminal, which is what an operator
//!   actually wants when checking whether a deploy is serving real data.

use std::time::Duration;

use anyhow::{Context, bail};
use blackbeard_core::{digest, hashrate};
use blackbeard_data::rpc::RpcClient;
use blackbeard_data::store::{BlockRecord, ChainRecord, Store};
use blackbeard_entities::{ChainId, MinerId};
use chrono::{TimeZone, Utc};
use clap::{Parser, Subcommand};

const RPC_TIMEOUT: Duration = Duration::from_secs(15);

#[derive(Parser)]
#[command(
    name = "blackbeard",
    version,
    about = "Operator tools for blackbeard.observer"
)]
struct Cli {
    #[command(subcommand)]
    command: Command,
}

#[derive(Subcommand)]
enum Command {
    /// Check that a node answers everything the observer needs, and print what
    /// it says.
    ///
    /// Every failure here is a failure the deployed daemon would hit silently:
    /// an RPC without the QPoW runtime API produces a site with no hashrate, and
    /// headers without a `pow_` digest produce an empty leaderboard — both of
    /// which look like a working deployment.
    Probe {
        /// The node's HTTP JSON-RPC endpoint.
        #[arg(long, default_value = "http://127.0.0.1:9944")]
        rpc_url: String,
        /// Target seconds per block, for the hashrate estimate.
        #[arg(long, default_value_t = 6.0)]
        target_block_time: f64,
        /// How many recent headers to decode. More is a better sample of
        /// whether authorship decoding actually works across blocks.
        #[arg(long, default_value_t = 20)]
        blocks: u64,
    },

    /// Read history from a node into the database.
    Backfill {
        /// The node's HTTP JSON-RPC endpoint.
        #[arg(long, default_value = "http://127.0.0.1:9944")]
        rpc_url: String,
        /// Postgres connection string. mTLS in production; this command is run
        /// by an operator, so it takes a URL rather than the daemon's config.
        #[arg(long, env = "DATABASE_URL")]
        database_url: String,
        /// Chain id, matching the daemon's config.
        #[arg(long)]
        chain: String,
        /// Human-facing chain name, used if the chain row does not exist yet.
        #[arg(long)]
        display_name: Option<String>,
        /// First block to read. Defaults to `to` minus `count`.
        #[arg(long)]
        from: Option<u64>,
        /// Last block to read. Defaults to the current tip.
        #[arg(long)]
        to: Option<u64>,
        /// Blocks to read when `from` is not given.
        #[arg(long, default_value_t = 3600)]
        count: u64,
    },

    /// Print the standings from a running observer.
    Standings {
        /// The observer's API base.
        #[arg(long, default_value = "http://127.0.0.1:25864/v1")]
        api: String,
        /// Chain id.
        #[arg(long, default_value = "planck")]
        chain: String,
        /// Window: hour, six_hours, day or week.
        #[arg(long, default_value = "six_hours")]
        window: String,
        /// Rows to print.
        #[arg(long, default_value_t = 20)]
        limit: usize,
    },
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    match Cli::parse().command {
        Command::Probe {
            rpc_url,
            target_block_time,
            blocks,
        } => probe(&rpc_url, target_block_time, blocks).await,
        Command::Backfill {
            rpc_url,
            database_url,
            chain,
            display_name,
            from,
            to,
            count,
        } => {
            backfill(
                &rpc_url,
                &database_url,
                &chain,
                display_name.as_deref(),
                from,
                to,
                count,
            )
            .await
        }
        Command::Standings {
            api,
            chain,
            window,
            limit,
        } => standings(&api, &chain, &window, limit).await,
    }
}

async fn probe(rpc_url: &str, target_block_time: f64, blocks: u64) -> anyhow::Result<()> {
    let rpc = RpcClient::new(rpc_url, RPC_TIMEOUT)?;

    let health = rpc
        .health()
        .await
        .with_context(|| format!("{rpc_url} did not answer system_health"))?;
    println!(
        "node        peers {}, syncing {}",
        health.peers, health.is_syncing
    );

    let genesis = rpc.genesis().await?.unwrap_or_default();
    println!("genesis     {genesis}");

    let props = rpc.properties().await.unwrap_or_default();
    println!(
        "token       {} ({} decimals), ss58 prefix {}",
        props.token_symbol.as_deref().unwrap_or("?"),
        props.token_decimals.map_or("?".into(), |d| d.to_string()),
        props.ss58_format.map_or("?".into(), |p| p.to_string()),
    );

    let head = rpc.header(None).await?.context("node returned no head")?;
    let height = head
        .height()
        .context("head carried an undecodable number")?;
    println!("height      {height}");

    let difficulty = rpc.difficulty().await.context(
        "QPoWApi_get_difficulty failed — without it the site has no hashrate and no difficulty",
    )?;
    println!("difficulty  {difficulty} expected hashes per block");
    println!(
        "hashrate    {:.3} GH/s at the {target_block_time}s target",
        hashrate::network_hashrate(difficulty, hashrate::Interval::Target(target_block_time)) / 1e9
    );

    // The load-bearing check. Everything above can pass on a node whose headers
    // this observer cannot read, and that node produces an empty leaderboard
    // with no error anywhere.
    println!("\ndecoding the last {blocks} headers…");
    let mut decoded = 0u64;
    let mut authors = std::collections::BTreeMap::<MinerId, u32>::new();
    let mut timestamps = 0u64;
    for h in height.saturating_sub(blocks)..height {
        let Some(hash) = rpc.block_hash(h).await? else {
            continue;
        };
        let Some(header) = rpc.header(Some(&hash)).await? else {
            continue;
        };
        if let Some(miner) = digest::author_preimage(&header.digest.logs) {
            decoded += 1;
            *authors.entry(miner).or_default() += 1;
        }
        if rpc.block_timestamp_ms(&hash).await?.is_some() {
            timestamps += 1;
        }
    }

    println!("authors     {decoded}/{blocks} headers carried a pow_ reward preimage");
    println!("timestamps  {timestamps}/{blocks} blocks carried a decodable timestamp inherent");
    println!("distinct    {} miners in that sample", authors.len());
    for (miner, n) in authors.iter().take(10) {
        println!("            {n:>3}  {miner}");
    }

    if decoded == 0 {
        bail!(
            "no header carried a pow_ PreRuntime digest. The observer would run, report healthy, \
             and show an empty leaderboard — this chain's headers are not in a shape it can read."
        );
    }
    if decoded < blocks {
        println!(
            "\nnote: {} of {blocks} headers had no author digest. Genesis and any non-PoW block \
             are expected; a large fraction is not.",
            blocks - decoded
        );
    }
    println!("\nOK — this node is usable.");
    Ok(())
}

#[allow(clippy::too_many_arguments)]
async fn backfill(
    rpc_url: &str,
    database_url: &str,
    chain: &str,
    display_name: Option<&str>,
    from: Option<u64>,
    to: Option<u64>,
    count: u64,
) -> anyhow::Result<()> {
    let rpc = RpcClient::new(rpc_url, RPC_TIMEOUT)?;
    let pool = sqlx::postgres::PgPoolOptions::new()
        .max_connections(4)
        .connect(database_url)
        .await
        .context("connecting to postgres")?;
    let store = Store::from_pool(pool);
    store.migrate().await?;

    let tip = rpc
        .header(None)
        .await?
        .and_then(|h| h.height())
        .context("node returned no usable head")?;
    let to = to.unwrap_or(tip);
    let from = from.unwrap_or_else(|| to.saturating_sub(count));
    if from > to {
        bail!("--from {from} is above --to {to}");
    }

    let chain_id = ChainId(chain.to_owned());
    // The chain row has to exist before any block can reference it.
    store
        .upsert_chain(&ChainRecord {
            chain: chain_id.clone(),
            display_name: display_name.unwrap_or(chain).to_owned(),
            mainnet: false,
            genesis: rpc.genesis().await.ok().flatten(),
            token_symbol: None,
            token_decimals: None,
            target_block_time: 6.0,
        })
        .await?;

    println!("reading {from}..={to} from {rpc_url}");

    // Written in batches rather than one statement: a week of blocks is a
    // hundred thousand rows, and a single array parameter that large is a
    // needlessly large allocation on both ends.
    const BATCH: usize = 500;
    let mut batch: Vec<BlockRecord> = Vec::with_capacity(BATCH);
    let mut written = 0u64;
    let mut skipped = 0u64;

    for height in from..=to {
        let Some(hash) = rpc.block_hash(height).await? else {
            skipped += 1;
            continue;
        };
        let Some(header) = rpc.header(Some(&hash)).await? else {
            skipped += 1;
            continue;
        };
        let Some(miner) = digest::author_preimage(&header.digest.logs) else {
            skipped += 1;
            continue;
        };
        let authored_at = rpc
            .block_timestamp_ms(&hash)
            .await?
            .and_then(|ms| i64::try_from(ms).ok())
            .and_then(|ms| Utc.timestamp_millis_opt(ms).single());

        batch.push(BlockRecord {
            chain: chain_id.clone(),
            height,
            hash,
            miner,
            authored_at,
            // Backfilled blocks are stamped with the author's own timestamp
            // where there is one. Stamping them "now" would make a week of
            // history look like it all arrived in the last minute, and every
            // per-miner chart would collapse into one bucket.
            observed_at: authored_at.unwrap_or_else(Utc::now),
            // Asked at the parent, per block: `pallet_qpow` retargets in
            // `on_finalize`, so the difficulty stored against block h lives in
            // the state of h-1. One current value copied across the whole range
            // would draw a flat difficulty history over a period that in fact
            // retargeted on every block. `None` where a pruned node cannot say.
            difficulty: rpc
                .difficulty_for_child_of(&header.parent_hash)
                .await
                .ok()
                .map(|d| hashrate::u512_to_dec(d).0),
            // Backfill is history by definition: no observation here timed
            // anything, and `RollingWindow` must not treat it as if it had.
            at_tip: false,
        });

        if batch.len() >= BATCH {
            written += store.record_blocks(&batch).await?;
            batch.clear();
            println!("  {written} written…");
        }
    }
    written += store.record_blocks(&batch).await?;

    println!(
        "done: {written} blocks written, {skipped} skipped (no author digest or not on chain)"
    );
    Ok(())
}

async fn standings(api: &str, chain: &str, window: &str, limit: usize) -> anyhow::Result<()> {
    let url = format!("{api}/chains/{chain}/leaderboard?window={window}&limit={limit}");
    let body: serde_json::Value = reqwest::Client::new()
        .get(&url)
        .timeout(RPC_TIMEOUT)
        .send()
        .await
        .with_context(|| format!("GET {url}"))?
        .error_for_status()?
        .json()
        .await?;

    println!(
        "{} · window {} ({} blocks observed)",
        chain,
        body["window"].as_str().unwrap_or(window),
        body["observed_blocks"].as_u64().unwrap_or(0)
    );
    println!(
        "{:>3}  {:<34} {:>6} {:>7} {:>12}",
        "#", "MINER", "BLOCKS", "SHARE", "EST. RATE"
    );
    for row in body["rows"].as_array().into_iter().flatten() {
        let rate = row["hashrate_estimate"].as_f64().unwrap_or(0.0);
        println!(
            "{:>3}  {:<34} {:>6} {:>6.1}% {:>9.2} MH/s",
            row["rank"].as_u64().unwrap_or(0),
            row["display"].as_str().unwrap_or("?"),
            row["blocks"].as_u64().unwrap_or(0),
            row["share"].as_f64().unwrap_or(0.0) * 100.0,
            rate / 1e6,
        );
    }
    Ok(())
}