bal-server/src/bin/bal-pusher.rs

extern crate bitcoincore_rpc;
extern crate zmq;
use bitcoin::Network;

use bitcoincore_rpc::{Auth, Client, Error, RpcApi, bitcoin};
use bitcoincore_rpc_json::GetBlockchainInfoResult;

use ed25519_dalek::{
    Signer as _, SigningKey,
    pkcs8::{DecodePrivateKey, EncodePublicKey},
};
use log::{debug, error, info, trace, warn};
use serde::Deserialize;
use serde::Serialize;
use serde_json::json;
use std::collections::{BTreeMap, HashMap};
use std::env;
use std::error::Error as StdError;
use std::str;
use std::thread;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use zmq::{Context, Socket};

use bal_server::db::{calculate_and_upsert_stats, get_pending_txs, open_database};
use bal_server::validation::is_valid_welist_url;
use base64::{Engine as _, engine::general_purpose};
use reqwest::Client as rClient;
use std::net::SocketAddr;
use url::Url;

// BIP-65: locktime values below this are block heights, at or above are UNIX timestamps.
const LOCKTIME_THRESHOLD: i64 = 500_000_000;
const VERSION: &str = env!("CARGO_PKG_VERSION");

/// Legacy ping signature: concatenation of URL, chain, height, median time and
/// block hash, verified against `/.pub_key.pem`.
const PING_SIG_VERSION_LEGACY: u8 = 1;
/// Canonical JSON ping signature (`WELIST-PING-V2\n` + sorted compact JSON).
const PING_SIG_VERSION_CANONICAL: u8 = 2;
/// Domain-separation prefix signed before a v2 canonical JSON payload.
const PING_SIGNATURE_PREFIX: &[u8] = b"WELIST-PING-V2\n";
/// Maximum accepted `info` length, matching the WeTime `/ping` validation.
const MAX_INFO_BYTES: usize = 4096;
/// Maximum accepted `version` length, matching the WeTime `/ping` validation.
const MAX_VERSION_BYTES: usize = 64;

/// Onion delegation certificate issued offline by the onion identity key.
///
/// The field names and types match the WeTime service's `OnionDelegation`
/// wire format. The certificate is public data; only the onion identity key
/// and the delegated ping key are secret.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct OnionDelegation {
    /// Onion host the certificate is bound to.
    onion: String,
    /// Base64 SubjectPublicKeyInfo DER of the delegated Ed25519 key.
    key: String,
    /// Unix timestamp in seconds after which the certificate is invalid.
    valid_until: u64,
    /// Base64 Ed25519 signature by the onion identity key.
    signature: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct MyConfig {
    db_backend: String,
    db_file: String,
    pg_dsn: String,
    bitcoin_dir: String,
    regtest: NetworkParams,
    testnet: NetworkParams,
    testnet4: NetworkParams,
    signet: NetworkParams,
    mainnet: NetworkParams,
    send_stats: bool,
    url: String,
    ssl_key_path: String,
    /// WeTime ping signature version. Defaults to the legacy scheme so an
    /// unmodified deployment keeps working.
    ping_sig_version: u8,
    /// Optional onion delegation certificate loaded from
    /// `WELIST_PING_DELEGATION_FILE`. When present and valid, onion pings are
    /// signed with `SSL_KEY_PATH` as the delegated key instead of the onion
    /// identity key. The certificate is checked at startup against the
    /// configured onion host, its expiry, and the signing key.
    onion_delegation: Option<OnionDelegation>,
}

impl Default for MyConfig {
    fn default() -> Self {
        MyConfig {
            db_backend: env::var("BAL_SERVER_DB_BACKEND").unwrap_or("sqlite".to_string()),
            db_file: env::var("BAL_SERVER_DB_FILE").unwrap_or("bal.db".to_string()),
            pg_dsn: env::var("BAL_SERVER_PG_DSN").unwrap_or_default(),
            bitcoin_dir: env::var("BAL_PUSHER_BITCOIN_DIR").unwrap_or("".to_string()),
            regtest: get_network_params_default(Network::Regtest),
            testnet: get_network_params_default(Network::Testnet),
            testnet4: get_network_params_default(Network::Testnet4),
            signet: get_network_params_default(Network::Signet),
            mainnet: get_network_params_default(Network::Bitcoin),
            send_stats: env::var("BAL_PUSHER_SEND_STATS")
                .unwrap_or("false".to_string())
                .parse::<bool>()
                .unwrap_or(false),
            url: env::var("BAL_SERVER_URL").unwrap_or("http://127.0.0.1:9133".to_string()),
            ssl_key_path: env::var("SSL_KEY_PATH").unwrap_or("privkey.pem".to_string()),
            ping_sig_version: parse_ping_sig_version(),
            onion_delegation: None,
        }
    }
}

/// Load the onion delegation certificate from `WELIST_PING_DELEGATION_FILE`.
///
/// The file is a JSON object with `onion`, `key`, `valid_until` and
/// `signature`, produced offline with the onion identity key. A missing or
/// malformed file disables delegation, so the pusher falls back to signing with
/// `SSL_KEY_PATH` directly.
fn load_onion_delegation() -> Option<OnionDelegation> {
    let path = env::var("WELIST_PING_DELEGATION_FILE").ok()?;
    let path = path.trim();
    if path.is_empty() {
        return None;
    }
    let text = match std::fs::read_to_string(path) {
        Ok(text) => text,
        Err(err) => {
            warn!(
                "Cannot read WELIST_PING_DELEGATION_FILE={} ({}), onion delegation disabled",
                path, err
            );
            return None;
        }
    };
    match serde_json::from_str::<OnionDelegation>(&text) {
        Ok(delegation) => Some(delegation),
        Err(err) => {
            warn!(
                "Cannot parse WELIST_PING_DELEGATION_FILE={} ({}), onion delegation disabled",
                path, err
            );
            None
        }
    }
}

/// Parse `WELIST_PING_SIG_VERSION`, falling back to the legacy scheme.
///
/// Unsupported or malformed values are reported and treated as `1` so a typo
/// cannot silently break stats reporting.
fn parse_ping_sig_version() -> u8 {
    match env::var("WELIST_PING_SIG_VERSION") {
        Ok(value) => match value.trim().parse::<u8>() {
            Ok(version)
                if version == PING_SIG_VERSION_LEGACY || version == PING_SIG_VERSION_CANONICAL =>
            {
                version
            }
            Ok(version) => {
                warn!(
                    "Unsupported WELIST_PING_SIG_VERSION={}, using legacy scheme",
                    version
                );
                PING_SIG_VERSION_LEGACY
            }
            Err(err) => {
                warn!(
                    "Invalid WELIST_PING_SIG_VERSION={:?} ({}), using legacy scheme",
                    value, err
                );
                PING_SIG_VERSION_LEGACY
            }
        },
        Err(_) => PING_SIG_VERSION_LEGACY,
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct NetworkParams {
    host: String,
    port: u16,
    dir_path: String,
    db_field: String,
    cookie_file: String,
    rpc_user: String,
    rpc_pass: String,
    zmq_listener: String,
}
fn get_network_params(cfg: &MyConfig, network: Network) -> &NetworkParams {
    match network {
        Network::Testnet => &cfg.testnet,
        Network::Testnet4 => &cfg.testnet4,
        Network::Signet => &cfg.signet,
        Network::Regtest => &cfg.regtest,
        _ => &cfg.mainnet,
    }
}
fn get_network_params_default(network: Network) -> NetworkParams {
    match network {
        Network::Testnet => NetworkParams {
            host: "http://127.0.0.1".to_string(),
            port: 18332,
            dir_path: "testnet3/".to_string(),
            db_field: "testnet".to_string(),
            cookie_file: "".to_string(),
            rpc_user: "".to_string(),
            rpc_pass: "".to_string(),
            zmq_listener: "tcp://127.0.0.1:23332".to_string(),
        },
        Network::Testnet4 => NetworkParams {
            host: "http://127.0.0.1".to_string(),
            port: 48332,
            dir_path: "testnet4/".to_string(),
            db_field: "testnet4".to_string(),
            cookie_file: "".to_string(),
            rpc_user: "".to_string(),
            rpc_pass: "".to_string(),
            zmq_listener: "tcp://127.0.0.1:24332".to_string(),
        },
        Network::Signet => NetworkParams {
            host: "http://127.0.0.1".to_string(),
            port: 18332,
            dir_path: "signet/".to_string(),
            db_field: "signet".to_string(),
            cookie_file: "".to_string(),
            rpc_user: "".to_string(),
            rpc_pass: "".to_string(),
            zmq_listener: "tcp://127.0.0.1:22332".to_string(),
        },
        Network::Regtest => NetworkParams {
            host: "http://127.0.0.1".to_string(),
            port: 18443,
            dir_path: "regtest/".to_string(),
            db_field: "regtest".to_string(),
            cookie_file: "".to_string(),
            rpc_user: "".to_string(),
            rpc_pass: "".to_string(),
            zmq_listener: "tcp://127.0.0.1:21332".to_string(),
        },
        _ => NetworkParams {
            host: "http://127.0.0.1".to_string(),
            port: 8332,
            dir_path: "".to_string(),
            db_field: "bitcoin".to_string(),
            cookie_file: "".to_string(),
            rpc_user: "".to_string(),
            rpc_pass: "".to_string(),
            zmq_listener: "tcp://127.0.0.1:28332".to_string(),
        },
    }
}

fn get_cookie_filename(network: &NetworkParams) -> Result<String, Box<dyn StdError>> {
    if !network.cookie_file.is_empty() {
        Ok(network.cookie_file.clone())
    } else {
        match env::var_os("HOME") {
            Some(home) => match home.to_str() {
                Some(home_str) => {
                    let cookie_file_path =
                        format!("{}/.bitcoin/{}.cookie", home_str, network.dir_path);

                    Ok(cookie_file_path)
                }
                None => Err("wrong HOME value".into()),
            },
            None => Err("Please Set HOME environment variable".into()),
        }
    }
}
fn get_client_from_username(
    url: &str,
    network: &NetworkParams,
) -> Result<(Client, GetBlockchainInfoResult), Box<dyn StdError>> {
    if !network.rpc_user.is_empty() {
        match Client::new(
            url,
            Auth::UserPass(network.rpc_user.to_string(), network.rpc_pass.to_string()),
        ) {
            Ok(client) => match client.get_blockchain_info() {
                Ok(bcinfo) => Ok((client, bcinfo)),
                Err(err) => Err(err.into()),
            },
            Err(err) => Err(err.into()),
        }
    } else {
        Err("Failed".into())
    }
}
fn get_client_from_cookie(
    url: &str,
    network: &NetworkParams,
) -> Result<(Client, GetBlockchainInfoResult), Box<dyn StdError>> {
    match get_cookie_filename(network) {
        Ok(cookie) => match Client::new(url, Auth::CookieFile(cookie.into())) {
            Ok(client) => match client.get_blockchain_info() {
                Ok(bcinfo) => Ok((client, bcinfo)),
                Err(err) => Err(err.into()),
            },
            Err(err) => Err(err.into()),
        },
        Err(err) => Err(err),
    }
}
fn get_client(
    network: &NetworkParams,
) -> Result<(Client, GetBlockchainInfoResult), Box<dyn StdError>> {
    let url = format!("{}:{}/", network.host, network.port);
    debug!("trying to connect to bitcoin daemon:{url}");
    match get_client_from_username(&url, network) {
        Ok(client) => Ok(client),
        Err(_) => match get_client_from_cookie(&url, network) {
            Ok(client) => Ok(client),
            Err(err) => Err(err),
        },
    }
}
async fn main_result(cfg: &MyConfig, network_params: &NetworkParams) -> Result<(), Error> {
    match get_client(network_params) {
        Ok((rpc, bcinfo)) => {
            info!("connected");
            info!("median time: {}", bcinfo.median_time);
            info!("blocks: {}", bcinfo.blocks);
            debug!("best block hash: {}", bcinfo.best_block_hash);

            let average_time = bcinfo.median_time;

            let connection_string = match cfg.db_backend.as_str() {
                "sqlite" => cfg.db_file.clone(),
                "postgresql" => cfg.pg_dsn.clone(),
                other => {
                    error!("Unknown DB backend: {}", other);
                    return Ok(());
                }
            };

            let db = match open_database(&cfg.db_backend, &connection_string).await {
                Ok(pool) => pool,
                Err(e) => {
                    error!("Fatal: {}", e);
                    std::process::exit(1);
                }
            };
            info!("db open {}", &connection_string);

            let pending_txs = match get_pending_txs(
                &db,
                &network_params.db_field,
                LOCKTIME_THRESHOLD,
                bcinfo.blocks as i64,
                average_time as i64,
            )
            .await
            {
                Ok(txs) => txs,
                Err(e) => {
                    warn!("Failed to query pending transactions: {}", e);
                    return Ok(());
                }
            };

            let mut pushed_txs: Vec<String> = Vec::new();
            let mut invalid_txs: HashMap<String, String> = HashMap::new();

            for row in &pending_txs {
                let txid = &row.txid;
                let tx = &row.tx;
                let locktime = row.locktime;
                info!("to be pushed: {}: {}", txid, locktime);
                match rpc.send_raw_transaction(tx.as_str()) {
                    Ok(o) => {
                        info!("tx: {} pusshata PUSHED\n{}", txid, o);
                        pushed_txs.push(txid.clone());
                    }
                    Err(err) => {
                        warn!("Error: {}\n{}", err, txid);
                        invalid_txs.insert(txid.clone(), err.to_string());
                    }
                };
            }

            for txid in &pushed_txs {
                if let Err(e) = bal_server::db::update_tx_status(&db, txid, 1, None).await {
                    error!("Failed to update tx status: {}", e);
                }
            }
            for (txid, txerr) in &invalid_txs {
                if let Err(e) = bal_server::db::update_tx_status(&db, txid, 2, Some(txerr)).await {
                    error!("Failed to update tx status: {}", e);
                }
            }
            if let Err(e) = send_stats_report(cfg, bcinfo).await {
                error!("send_stats_report failed: {}", e);
            }
            if let Err(e) = calculate_and_upsert_stats(&db, &network_params.db_field).await {
                warn!("calculate_stats failed: {e}");
            }
        }
        Err(erx) => {
            error!("impossible to get client: {}, retrying on next block", erx);
            thread::sleep(Duration::from_secs(5));
            return Ok(());
        }
    }
    Ok(())
}

/// Parse the `(host, port)` pair from a base URL like `https://host[:port]`.
fn parse_host_port(base_url: &str) -> Option<(String, u16)> {
    let url = Url::parse(base_url).ok()?;
    let host = url
        .host_str()?
        .trim_start_matches('[')
        .trim_end_matches(']')
        .to_string();
    let port = url.port_or_known_default().unwrap_or(443);
    Some((host, port))
}

/// Resolve `host:port` and return the first IPv6 (AAAA) address, if any.
async fn resolve_first_ipv6(host: &str, port: u16) -> Option<SocketAddr> {
    use std::net::ToSocketAddrs;
    let host = host.to_string();
    tokio::task::spawn_blocking(move || {
        format!("{}:{}", host, port)
            .to_socket_addrs()
            .ok()
            .and_then(|mut addrs| addrs.find(|a| a.is_ipv6()))
    })
    .await
    .ok()
    .flatten()
}

async fn welist_http_client(welist_url: &str) -> rClient {
    let prefer_ipv6 = env::var("BAL_PUSHER_PREFER_IPV6")
        .unwrap_or("false".to_string())
        .parse::<bool>()
        .unwrap_or(false);
    if !prefer_ipv6 {
        return new_welist_client();
    }
    let (host, port) = match parse_host_port(welist_url) {
        Some(hp) => hp,
        None => {
            warn!("BAL_PUSHER_PREFER_IPV6: cannot parse '{welist_url}', using default resolver");
            return new_welist_client();
        }
    };
    match resolve_first_ipv6(&host, port).await {
        Some(addr) => {
            debug!("BAL_PUSHER_PREFER_IPV6: pinning {host} to {addr}");
            rClient::builder()
                .timeout(Duration::from_secs(10))
                .resolve(&host, addr)
                .build()
                .unwrap_or_else(|_| new_welist_client())
        }
        None => {
            debug!("BAL_PUSHER_PREFER_IPV6: no IPv6 address for {host}, using default resolver");
            new_welist_client()
        }
    }
}

fn new_welist_client() -> rClient {
    rClient::builder()
        .timeout(Duration::from_secs(10))
        .build()
        .unwrap_or_else(|_| rClient::new())
}

/// Metadata advertised by the local bal-server `/{chain}/info` endpoint.
///
/// These values are included in a v2 ping so the executor can publish signed
/// `info`, `base_fee` and `version` values to the WeTime service.
#[derive(Debug, Clone, Default, Deserialize)]
struct ExecutorMetadata {
    #[serde(default)]
    info: Option<String>,
    #[serde(default)]
    base_fee: Option<i64>,
    #[serde(default)]
    version: Option<String>,
}

/// Normalize an executor URL exactly like the WeTime `/ping` handler.
fn normalize_executor_url(url: &str) -> String {
    url.trim().trim_end_matches('/').to_string()
}

/// Drop metadata that the WeTime service would reject.
fn sanitize_metadata(metadata: ExecutorMetadata) -> ExecutorMetadata {
    ExecutorMetadata {
        info: metadata
            .info
            .filter(|value| !value.is_empty() && value.len() <= MAX_INFO_BYTES),
        base_fee: metadata.base_fee.filter(|value| *value >= 0),
        version: metadata
            .version
            .filter(|value| !value.is_empty() && value.len() <= MAX_VERSION_BYTES),
    }
}

/// Return the URL host exactly as the WeTime service extracts it from the
/// normalized executor URL.
fn executor_host(url: &str) -> Option<String> {
    Url::parse(url)
        .ok()
        .and_then(|parsed| parsed.host_str().map(str::to_string))
}

/// Return `true` when the URL host is a v3 `.onion` address.
///
/// This mirrors the service-side check (`host.ends_with(".onion")`) so both
/// sides agree on whether delegation may be sent. The pusher cannot reach its
/// own onion address without a Tor proxy, so v2 pings for onion URLs are
/// signed with `null` metadata instead.
fn is_onion_url(url: &str) -> bool {
    executor_host(url).is_some_and(|host| host.ends_with(".onion"))
}

/// Return the current Unix time in seconds, or `0` if the clock is before the
/// epoch.
fn unix_now() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|duration| duration.as_secs())
        .unwrap_or(0)
}

/// Return whether `delegation.key` matches the public key derived from
/// `signing_key`.
///
/// `None` means the public key could not be encoded as SPKI DER. Only the
/// public half is compared; the private key is never logged or transmitted.
fn delegation_key_matches(signing_key: &SigningKey, delegation: &OnionDelegation) -> Option<bool> {
    let public_der = signing_key.verifying_key().to_public_key_der().ok()?;
    let expected = general_purpose::STANDARD.encode(public_der.as_bytes());
    Some(delegation.key.trim() == expected)
}

/// Validate the loaded onion delegation certificate against the configured
/// executor URL and signing key.
///
/// The WeTime service rejects a present-but-invalid certificate instead of
/// falling back to the onion identity key, so a stale certificate would make
/// every onion ping fail. Check the parts the pusher can verify locally at
/// startup: the certificate is bound to the configured onion host, it has not
/// expired, and its delegated key matches `SSL_KEY_PATH`. On any mismatch,
/// delegation is disabled with a warning and the pusher signs with
/// `SSL_KEY_PATH` directly.
fn validate_onion_delegation(cfg: &mut MyConfig) {
    let Some(delegation) = cfg.onion_delegation.as_ref() else {
        return;
    };

    let normalized_url = normalize_executor_url(&cfg.url);
    let Some(host) = executor_host(&normalized_url) else {
        warn!(
            "WELIST_PING_DELEGATION_FILE is set but BAL_SERVER_URL={} is not a valid URL; disabling onion delegation",
            cfg.url
        );
        cfg.onion_delegation = None;
        return;
    };

    if !host.ends_with(".onion") {
        warn!(
            "WELIST_PING_DELEGATION_FILE is set but BAL_SERVER_URL host {} is not an onion address; disabling onion delegation",
            host
        );
        cfg.onion_delegation = None;
        return;
    }

    if delegation.onion != host {
        warn!(
            "Onion delegation is bound to {} but the executor URL host is {}; disabling onion delegation. Re-issue the certificate for the exact host.",
            delegation.onion, host
        );
        cfg.onion_delegation = None;
        return;
    }

    let now = unix_now();
    if delegation.valid_until < now {
        warn!(
            "Onion delegation expired at {} (now {}); disabling onion delegation. Re-issue the certificate.",
            delegation.valid_until, now
        );
        cfg.onion_delegation = None;
        return;
    }

    let signing_key = match SigningKey::read_pkcs8_pem_file(&cfg.ssl_key_path) {
        Ok(key) => key,
        Err(err) => {
            warn!(
                "Cannot read SSL_KEY_PATH={} to validate onion delegation ({}); disabling onion delegation",
                cfg.ssl_key_path, err
            );
            cfg.onion_delegation = None;
            return;
        }
    };

    match delegation_key_matches(&signing_key, delegation) {
        Some(true) => {}
        Some(false) => {
            warn!(
                "Onion delegation delegates to a different key than SSL_KEY_PATH={}; disabling onion delegation. Re-issue the certificate for the current ping key.",
                cfg.ssl_key_path
            );
            cfg.onion_delegation = None;
        }
        None => {
            warn!(
                "Cannot encode the public key from SSL_KEY_PATH={}; disabling onion delegation",
                cfg.ssl_key_path
            );
            cfg.onion_delegation = None;
        }
    }
}

/// Return the delegation certificate to send for `url`, if any.
///
/// Delegation is onion-only: sending it for a DNS host would be rejected by
/// the WeTime service.
fn delegation_for_url<'a>(cfg: &'a MyConfig, url: &str) -> Option<&'a OnionDelegation> {
    if is_onion_url(url) {
        cfg.onion_delegation.as_ref()
    } else {
        None
    }
}

/// Fetch the signed metadata published by the local bal-server.
///
/// Failure is not fatal: a v2 ping is still valid with `null` metadata, it just
/// cannot refresh those columns on the WeTime service.
async fn fetch_executor_metadata(cfg: &MyConfig, chain: &str) -> ExecutorMetadata {
    let base_url = normalize_executor_url(&cfg.url);
    if is_onion_url(&base_url) {
        debug!("skipping executor metadata fetch for onion URL");
        return ExecutorMetadata::default();
    }
    let endpoint = format!("{}/{}/info", base_url, chain);
    let client = new_welist_client();
    match client.get(&endpoint).send().await {
        Ok(response) if response.status().is_success() => {
            match response.json::<ExecutorMetadata>().await {
                Ok(metadata) => {
                    debug!("executor metadata: {:?}", metadata);
                    sanitize_metadata(metadata)
                }
                Err(err) => {
                    warn!("cannot parse executor metadata from {}: {}", endpoint, err);
                    ExecutorMetadata::default()
                }
            }
        }
        Ok(response) => {
            warn!(
                "executor metadata endpoint {} returned {}",
                endpoint,
                response.status()
            );
            ExecutorMetadata::default()
        }
        Err(err) => {
            warn!("cannot fetch executor metadata from {}: {}", endpoint, err);
            ExecutorMetadata::default()
        }
    }
}

/// Fields covered by a v2 WeTime ping signature.
#[derive(Clone, Copy, Debug)]
struct PingPayload<'a> {
    /// Normalized executor URL.
    url: &'a str,
    /// Canonical lowercase chain name.
    chain: &'a str,
    /// Self-reported block height.
    height: u64,
    /// Self-reported median time.
    median_time: u64,
    /// Self-reported block hash.
    last_block_hash: &'a str,
    /// Optional executor description.
    info: Option<&'a str>,
    /// Optional executor base fee.
    base_fee: Option<i64>,
    /// Optional executor version.
    version: Option<&'a str>,
}

impl PingPayload<'_> {
    /// Build the exact bytes covered by a v2 ping signature.
    ///
    /// The output is `WELIST-PING-V2\n` followed by compact JSON with sorted
    /// keys, every field present, and `null` for absent optional values. This
    /// must match the canonicalization performed by the WeTime `/ping` handler
    /// byte for byte.
    fn canonical_bytes(&self) -> Result<Vec<u8>, serde_json::Error> {
        let mut fields: BTreeMap<&str, serde_json::Value> = BTreeMap::new();
        fields.insert("base_fee", json!(self.base_fee));
        fields.insert("chain", json!(self.chain));
        fields.insert("height", json!(self.height));
        fields.insert("info", json!(self.info));
        fields.insert("last_block_hash", json!(self.last_block_hash));
        fields.insert("median_time", json!(self.median_time));
        fields.insert("sig_version", json!(PING_SIG_VERSION_CANONICAL));
        fields.insert("url", json!(self.url));
        fields.insert("version", json!(self.version));

        let body = serde_json::to_vec(&fields)?;
        let mut message = Vec::with_capacity(PING_SIGNATURE_PREFIX.len() + body.len());
        message.extend_from_slice(PING_SIGNATURE_PREFIX);
        message.extend_from_slice(&body);
        Ok(message)
    }
}

async fn send_stats_report(
    cfg: &MyConfig,
    bcinfo: GetBlockchainInfoResult,
) -> Result<(), reqwest::Error> {
    if cfg.send_stats {
        debug!("sending report to welist");
        let welist_url = env::var("WELIST_SERVER_URL")
            .unwrap_or("https://welist.bitcoin-after.life".to_string());
        let skip_validation = env::var("WELIST_SKIP_URL_VALIDATION")
            .unwrap_or("false".to_string())
            .parse::<bool>()
            .unwrap_or(false);
        if !skip_validation && !is_valid_welist_url(&welist_url) {
            warn!(
                "Invalid or unsafe WELIST_SERVER_URL: {}. Skipping stats report.",
                welist_url
            );
            return Ok(());
        }
        let client = welist_http_client(&welist_url).await;
        let url = format!("{}/ping", welist_url);
        debug!("welist url: {}", url);
        let chain = bcinfo.chain.to_string().to_lowercase();
        let last_block_hash = bcinfo.best_block_hash.to_string();

        let request = client
            .post(url)
            .header("User-Agent", format!("bal-pusher/{}", VERSION));

        let request = if cfg.ping_sig_version == PING_SIG_VERSION_CANONICAL {
            let executor_url = normalize_executor_url(&cfg.url);
            let metadata = fetch_executor_metadata(cfg, &chain).await;
            let payload = PingPayload {
                url: &executor_url,
                chain: &chain,
                height: bcinfo.blocks,
                median_time: bcinfo.median_time,
                last_block_hash: &last_block_hash,
                info: metadata.info.as_deref(),
                base_fee: metadata.base_fee,
                version: metadata.version.as_deref(),
            };
            let message = match payload.canonical_bytes() {
                Ok(message) => message,
                Err(err) => {
                    error!("Cannot build canonical ping payload, skipping: {}", err);
                    return Ok(());
                }
            };
            trace!(
                "canonical message to be sent: {}",
                String::from_utf8_lossy(&message)
            );
            let sign = match sign_bytes(cfg.ssl_key_path.as_str(), &message) {
                Ok(signature) => signature,
                Err(e) => {
                    error!("Cannot sign stats report, skipping: {}", e);
                    return Ok(());
                }
            };
            let mut body = json!({
                "url":              executor_url,
                "chain":            chain,
                "height":           bcinfo.blocks,
                "median_time":      bcinfo.median_time,
                "last_block_hash":  last_block_hash,
                "signature":        sign,
                "sig_version":      PING_SIG_VERSION_CANONICAL,
                "info":             metadata.info,
                "base_fee":         metadata.base_fee,
                "version":          metadata.version,
            });
            if let Some(delegation) = delegation_for_url(cfg, &executor_url) {
                body["delegation"] = json!(delegation);
            }
            request.json(&body)
        } else {
            // The v1 signature authenticates only the legacy liveness
            // concatenation. `info`, `base_fee` and `version` are intentionally
            // not sent: they are outside the signature, and the WeTime service
            // ignores them for v1 pings. Use v2 when metadata must be published
            // and authenticated.
            let message = format!(
                "{0}{1}{2}{3}{4}",
                cfg.url, chain, bcinfo.blocks, bcinfo.median_time, last_block_hash
            );
            trace!("message to be sent: {}", message);
            let sign = match sign_message(cfg.ssl_key_path.as_str(), message.as_str()) {
                Ok(signature) => signature,
                Err(e) => {
                    error!("Cannot sign stats report, skipping: {}", e);
                    return Ok(());
                }
            };
            request.json(&json!(
            {
                "url":              cfg.url,
                "chain":            chain,
                "height":           bcinfo.blocks,
                "median_time":      bcinfo.median_time,
                "last_block_hash":  last_block_hash,
                "signature":        sign,
            }))
        };

        let response = request.send().await?;
        let status = response.status();
        let body = response.text().await?;
        info!(
            "Report to welist({}) status={} body={}",
            welist_url, status, body
        );
    } else {
        debug!("Not sending stats");
    }
    Ok(())
}

fn sign_bytes(private_key_path: &str, message: &[u8]) -> Result<String, Box<dyn StdError>> {
    let signing_key = SigningKey::read_pkcs8_pem_file(private_key_path)
        .map_err(|e| format!("failed to parse private key PEM at {private_key_path}: {e}"))?;
    let signature = signing_key.sign(message);

    Ok(general_purpose::STANDARD.encode(signature.to_bytes()))
}

fn sign_message(private_key_path: &str, message: &str) -> Result<String, Box<dyn StdError>> {
    sign_bytes(private_key_path, message.as_bytes())
}

fn parse_env(cfg: &mut MyConfig) {
    if let Ok(value) = env::var("BAL_SERVER_DB_BACKEND") {
        cfg.db_backend = value;
    }
    if let Ok(value) = env::var("BAL_SERVER_DB_FILE") {
        cfg.db_file = value;
    }
    if let Ok(value) = env::var("BAL_SERVER_PG_DSN") {
        cfg.pg_dsn = value;
    }
    cfg.regtest = parse_env_netconfig(cfg, "regtest");
    cfg.signet = parse_env_netconfig(cfg, "signet");
    cfg.testnet = parse_env_netconfig(cfg, "testnet");
    cfg.testnet4 = parse_env_netconfig(cfg, "testnet4");
    drop(parse_env_netconfig(cfg, "bitcoin"));
}
fn parse_env_netconfig(cfg_lock: &mut MyConfig, chain: &str) -> NetworkParams {
    let cfg = match chain {
        "regtest" => &mut cfg_lock.regtest,
        "signet" => &mut cfg_lock.signet,
        "testnet" => &mut cfg_lock.testnet,
        "testnet4" => &mut cfg_lock.testnet4,
        &_ => &mut cfg_lock.mainnet,
    };
    if let Ok(value) = env::var(format!("BAL_PUSHER_{}_HOST", chain.to_uppercase())) {
        cfg.host = value;
    }
    if let Ok(value) = env::var(format!("BAL_PUSHER_{}_PORT", chain.to_uppercase()))
        && let Ok(port_num) = value.parse::<u64>()
    {
        if let Ok(port) = u16::try_from(port_num) {
            cfg.port = port;
        } else {
            error!(
                "Port value {} exceeds u16 range for chain {}",
                port_num, chain
            );
        }
    }
    if let Ok(value) = env::var(format!("BAL_PUSHER_{}_DIR_PATH", chain.to_uppercase())) {
        cfg.dir_path = value;
    }
    if let Ok(value) = env::var(format!("BAL_PUSHER_{}_DB_FIELD", chain.to_uppercase())) {
        cfg.db_field = value;
    }
    if let Ok(value) = env::var(format!("BAL_PUSHER_{}_COOKIE_FILE", chain.to_uppercase())) {
        cfg.cookie_file = value;
    }
    if let Ok(value) = env::var(format!("BAL_PUSHER_{}_RPC_USER", chain.to_uppercase())) {
        cfg.rpc_user = value;
    }
    if let Ok(value) = env::var(format!("BAL_PUSHER_{}_RPC_PASSWORD", chain.to_uppercase())) {
        cfg.rpc_pass = value;
    }
    if let Ok(value) = env::var(format!("BAL_PUSHER_{}_ZMQ_HASHBLOCK", chain.to_uppercase())) {
        cfg.zmq_listener = value;
    }
    cfg.clone()
}

#[tokio::main]
async fn main() -> std::io::Result<()> {
    env_logger::init();
    let mut cfg = MyConfig::default();

    parse_env(&mut cfg);
    cfg.onion_delegation = load_onion_delegation();
    validate_onion_delegation(&mut cfg);
    let mut args = std::env::args();
    let _exe_name = args.next().unwrap();
    let arg_network = match args.next() {
        Some(nargs) => nargs,
        None => "bitcoin".to_string(),
    };
    let network = match arg_network.as_str() {
        "testnet" => Network::Testnet,
        "testnet4" => Network::Testnet4,
        "signet" => Network::Signet,
        "regtest" => Network::Regtest,
        _ => Network::Bitcoin,
    };

    info!("Network: {}", arg_network);
    let network_params = get_network_params(&cfg, network);

    let context = Context::new();
    let socket: Socket = context.socket(zmq::SUB).unwrap();

    let zmq_address = network_params.zmq_listener.clone();
    info!("zmq listening on: {}", zmq_address);
    loop {
        match socket.connect(&zmq_address) {
            Ok(_) => break,
            Err(e) => {
                error!("ZMQ connect failed: {}, retrying in 5s...", e);
                thread::sleep(Duration::from_secs(5));
            }
        }
    }

    match socket.set_subscribe(b"") {
        Ok(_) => {
            info!("ZMQ subscribed to all topics on {}", zmq_address);
        }
        Err(e) => {
            error!("ZMQ subscribe failed: {}, exiting", e);
            return Ok(());
        }
    }

    if let Err(e) = main_result(&cfg, network_params).await {
        error!("main_result failed on startup: {}", e);
    }
    info!("waiting new blocks..");
    socket.set_rcvtimeo(5000).unwrap(); // 5 seconds timeout
    let mut consecutive_timeouts: u32 = 0;
    loop {
        let message = match socket.recv_multipart(0) {
            Ok(m) => m,
            Err(e) => {
                consecutive_timeouts += 1;
                if consecutive_timeouts.is_multiple_of(720) {
                    error!(
                        "No ZMQ messages for {}s ({} consecutive timeouts), is bitcoind ZMQ active on {}?",
                        consecutive_timeouts * 5,
                        consecutive_timeouts,
                        zmq_address
                    );
                } else {
                    trace!("ZMQ recv timeout or error: {}, retrying...", e);
                }
                continue;
            }
        };
        if consecutive_timeouts > 0 {
            info!(
                "ZMQ connection restored after {} consecutive timeouts",
                consecutive_timeouts
            );
        }
        consecutive_timeouts = 0;
        let Some(topic) = message.first().cloned() else {
            warn!("ZMQ: received empty multipart message, skipping");
            continue;
        };
        let body = message.get(1).cloned().unwrap_or_default();
        debug!("ZMQ:GET TOPIC: {}", String::from_utf8_lossy(&topic));
        trace!("ZMQ:GET BODY: {}", hex::encode(&body));
        if topic == b"hashblock" {
            info!("NEW BLOCK: {}", hex::encode(&body));
            if let Err(e) = main_result(&cfg, network_params).await {
                error!("main_result failed on new block: {}", e);
            }
        }
        thread::sleep(Duration::from_millis(100)); // Sleep for 100ms
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn parse_host_port_https_default_port() {
        assert_eq!(
            parse_host_port("https://welist.bitcoin-after.life"),
            Some(("welist.bitcoin-after.life".to_string(), 443))
        );
    }

    #[test]
    fn parse_host_port_explicit_port_and_path() {
        assert_eq!(
            parse_host_port("https://example.com:8443/ping"),
            Some(("example.com".to_string(), 8443))
        );
    }

    #[test]
    fn parse_host_port_http_default_port() {
        assert_eq!(
            parse_host_port("http://example.com"),
            Some(("example.com".to_string(), 80))
        );
    }

    #[test]
    fn parse_host_port_ipv6_literal_brackets_stripped() {
        assert_eq!(
            parse_host_port("https://[2a13:2c0::1]:443"),
            Some(("2a13:2c0::1".to_string(), 443))
        );
    }

    #[test]
    fn parse_host_port_invalid_url() {
        assert_eq!(parse_host_port("not a url"), None);
    }

    #[test]
    fn normalize_executor_url_trims_and_strips_trailing_slashes() {
        assert_eq!(
            normalize_executor_url("  https://we.example.com/  "),
            "https://we.example.com"
        );
        assert_eq!(
            normalize_executor_url("https://we.example.com"),
            "https://we.example.com"
        );
    }

    #[test]
    fn onion_urls_skip_metadata_fetch() {
        assert!(is_onion_url(
            "http://abcdefghijklmnopqrstuvwxyz234567abcdefghijklmnopqrstuv.onion"
        ));
        assert!(!is_onion_url("https://we.example.com"));
        assert!(!is_onion_url("not a url"));
    }

    fn test_delegation(onion: &str) -> OnionDelegation {
        OnionDelegation {
            onion: onion.to_string(),
            key: "KEY".to_string(),
            valid_until: unix_now() + 3600,
            signature: "SIG".to_string(),
        }
    }

    #[test]
    fn delegation_is_only_returned_for_onion_urls() {
        let cfg = MyConfig {
            onion_delegation: Some(test_delegation("example.onion")),
            ..MyConfig::default()
        };

        assert!(delegation_for_url(&cfg, "http://example.onion").is_some());
        assert!(delegation_for_url(&cfg, "https://we.example.com").is_none());

        let cfg = MyConfig::default();
        assert!(delegation_for_url(&cfg, "http://example.onion").is_none());
    }

    #[test]
    fn delegation_bound_to_another_onion_is_disabled() {
        let mut cfg = MyConfig {
            url: "http://expected.onion".to_string(),
            onion_delegation: Some(test_delegation("other.onion")),
            ..MyConfig::default()
        };

        validate_onion_delegation(&mut cfg);

        assert!(cfg.onion_delegation.is_none());
    }

    #[test]
    fn delegation_on_non_onion_url_is_disabled() {
        let mut cfg = MyConfig {
            url: "https://we.example.com".to_string(),
            onion_delegation: Some(test_delegation("example.onion")),
            ..MyConfig::default()
        };

        validate_onion_delegation(&mut cfg);

        assert!(cfg.onion_delegation.is_none());
    }

    #[test]
    fn delegation_key_must_match_the_signing_key() {
        let signing_key = SigningKey::from_bytes(&[7u8; 32]);
        let public_der = signing_key
            .verifying_key()
            .to_public_key_der()
            .expect("public key encodes as SPKI DER");
        let matching = OnionDelegation {
            key: general_purpose::STANDARD.encode(public_der.as_bytes()),
            ..test_delegation("example.onion")
        };

        assert_eq!(delegation_key_matches(&signing_key, &matching), Some(true));

        let mismatching = OnionDelegation {
            key: "different".to_string(),
            ..matching
        };
        assert_eq!(
            delegation_key_matches(&signing_key, &mismatching),
            Some(false)
        );
    }

    #[test]
    fn canonical_ping_payload_is_sorted_compact_and_complete() {
        let payload = PingPayload {
            url: "https://example.com",
            chain: "bitcoin",
            height: 100,
            median_time: 1000,
            last_block_hash: "abc",
            info: None,
            base_fee: None,
            version: None,
        };
        let payload = payload.canonical_bytes().unwrap();
        assert_eq!(
            String::from_utf8(payload).unwrap(),
            "WELIST-PING-V2\n{\"base_fee\":null,\"chain\":\"bitcoin\",\"height\":100,\"info\":null,\"last_block_hash\":\"abc\",\"median_time\":1000,\"sig_version\":2,\"url\":\"https://example.com\",\"version\":null}"
        );
    }

    #[test]
    fn canonical_ping_payload_includes_metadata_and_escapes_strings() {
        let payload = PingPayload {
            url: "https://example.com",
            chain: "bitcoin",
            height: 1,
            median_time: 2,
            last_block_hash: "abc",
            info: Some("line\n\"quoted\""),
            base_fee: Some(1500),
            version: Some("0.3.0"),
        };
        let payload = payload.canonical_bytes().unwrap();
        let message = String::from_utf8(payload).unwrap();
        assert!(message.contains("\"base_fee\":1500"));
        assert!(message.contains("\"info\":\"line\\n\\\"quoted\\\"\""));
        assert!(message.contains("\"version\":\"0.3.0\""));
    }

    #[test]
    fn sanitize_metadata_drops_values_the_service_rejects() {
        let metadata = sanitize_metadata(ExecutorMetadata {
            info: Some("x".repeat(MAX_INFO_BYTES + 1)),
            base_fee: Some(-1),
            version: Some(String::new()),
        });
        assert_eq!(metadata.info, None);
        assert_eq!(metadata.base_fee, None);
        assert_eq!(metadata.version, None);

        let metadata = sanitize_metadata(ExecutorMetadata {
            info: Some("ok".to_string()),
            base_fee: Some(0),
            version: Some(VERSION.to_string()),
        });
        assert_eq!(metadata.info.as_deref(), Some("ok"));
        assert_eq!(metadata.base_fee, Some(0));
        assert_eq!(metadata.version.as_deref(), Some(VERSION));
    }
}