decdn_cli/commands/

pool.rs

//! `decdn pool` — client-side payment-pool lifecycle.
//!
//! One `PaymentPool` deposit fans out to every provider its owner pays
//! (ADR 003) — there is no per-provider pool. `open` escrows a fresh deposit,
//! `top-up` adds funds to a pool the caller owns, `close` starts the
//! grace-window close, and `reclaim` refunds the residual once that window has
//! elapsed. `list`/`status` is a read-only dump of the tracked buyer pools and
//! their per-lane voucher watermark.

use std::io::Write;
use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use alloy::primitives::{Address, B256, TxHash, U256};
use alloy::signers::local::PrivateKeySigner;
use anyhow::Context as _;
use decdn_client::buyer_pool::{
    ToppedUpPool, UnconfirmedEscrows, ensure_allowance, escrowed_but_untracked,
    grade_deposit_credit, open_pool, top_up, topped_up_effect,
};
use decdn_common::admin::{AdminRpcClient as _, BuyerPoolsResponse};
use decdn_common::cli::{self, common::expand_tilde};
use decdn_common::config::{
    DEFAULT_CHAIN_ID, DEFAULT_REDEEM_INTERVAL_SECS, FileConfig, capability_expiry_margin_secs,
    load_file_config,
};
use decdn_common::redact::{sanitize_err_chain, sanitize_error_sources};
use decdn_incentive::buyer_pool::{BuyerLoad, BuyerPoolState, BuyerPoolStore};
use decdn_incentive::buyer_pool_redb::{ReadOnlyBuyerPoolStore, RedbBuyerPoolStore};
use decdn_incentive::eth_identity::{self, PasswordUse, load_signer};
use decdn_incentive::payment_pool::{PaymentPool, enumerate_owned_pools};
use decdn_incentive::tx::{RECEIPT_TIMEOUT, await_receipt};
use decdn_incentive::{
    Capability, CapabilityGrant, Deployment, PoolId, PoolLifecycleRevert, voucher_domain,
};
use serde::Serialize;

use decdn_client::provider;

use super::buyer_store::{
    BuyerStoreOwner, classify_buyer_store, client_buyer_db, node_buyer_db,
    open_client_store_for_escrow,
};
use super::chain_ctx;
use super::interrupt::{Interrupt, Interrupted, name_held_escrows};

/// Dispatch `decdn pool <subcommand>`.
pub async fn pool_dispatch(args: &cli::PoolArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
    match &args.command {
        cli::PoolCommand::List(a) => list(a, config_path).await,
        cli::PoolCommand::Open(a) => open(a, config_path).await,
        cli::PoolCommand::TopUp(a) => top_up_cmd(a, config_path).await,
        cli::PoolCommand::Close(a) => close(a, config_path).await,
        cli::PoolCommand::Reclaim(a) => reclaim(a, config_path).await,
        cli::PoolCommand::Assign(a) => assign(a, config_path).await,
    }
}

/// Chain coordinates resolved flag > `[blockchain]`/`[identity]` config >
/// default. Pure (parse-only) so the precedence is unit-testable.
#[derive(Debug)]
struct Resolved {
    rpc_url: String,
    payment_pool: Address,
    chain_id: u64,
    data_dir: PathBuf,
    keystore: PathBuf,
    /// File holding the keystore password, consulted after the
    /// `DECDN_KEYSTORE_PASSWORD` env var and before a prompt. CLI/env-only —
    /// passwords do not belong in a config file even by reference.
    keystore_password_file: Option<PathBuf>,
}

impl Resolved {
    /// The chain and `PaymentPool` this command runs against.
    const fn deployment(&self) -> Deployment {
        Deployment {
            chain_id: self.chain_id,
            payment_pool: self.payment_pool,
        }
    }
}

/// Resolve the buyer-store data dir: flag > `[identity]` config > client-scoped
/// `~/.decdn/client` default. Shared by the read-only `list` path (which needs
/// nothing else) and [`resolve_chain`].
fn resolve_data_dir(data_dir: Option<PathBuf>, file: &FileConfig) -> anyhow::Result<PathBuf> {
    data_dir
        .or_else(|| file.identity.as_ref().and_then(|i| i.data_dir.clone()))
        .map(|p| expand_tilde(&p))
        .or_else(cli::default_client_data_dir)
        .ok_or_else(|| {
            anyhow::anyhow!("data_dir not set and no default available (pass --data-dir)")
        })
}

fn resolve_chain(args: &cli::PoolChainArgs, file: &FileConfig) -> anyhow::Result<Resolved> {
    let bc = file.blockchain.as_ref();
    let rpc_url = args
        .rpc_url
        .clone()
        .or_else(|| bc.and_then(|b| b.rpc_url.clone()))
        .ok_or_else(|| anyhow::anyhow!("rpc_url not set (--rpc-url or blockchain.rpc_url)"))?;
    let payment_pool_raw = args
        .payment_pool_address
        .clone()
        .or_else(|| bc.and_then(|b| b.payment_pool_address.clone()))
        .ok_or_else(|| {
            anyhow::anyhow!(
                "payment_pool_address not set (--payment-pool-address or \
                 blockchain.payment_pool_address)"
            )
        })?;
    let payment_pool =
        super::chain_ctx::parse_nonzero_address(&payment_pool_raw, "payment_pool_address")?;
    let chain_id = args
        .chain_id
        .or_else(|| bc.and_then(|b| b.chain_id))
        .unwrap_or(DEFAULT_CHAIN_ID);
    let data_dir = resolve_data_dir(args.data_dir.clone(), file)?;
    let keystore = args
        .keystore
        .clone()
        .or_else(|| bc.and_then(|b| b.eth_keystore.clone()))
        .map_or_else(
            || eth_identity::keystore_path(&data_dir),
            |p| expand_tilde(&p),
        );
    Ok(Resolved {
        rpc_url,
        payment_pool,
        chain_id,
        data_dir,
        keystore,
        keystore_password_file: args.keystore_password_file.as_deref().map(expand_tilde),
    })
}

/// Buyer signer for the on-chain `pool` commands (vouchers +
/// open/top-up/close/reclaim txs). Password from `KEYSTORE_PASSWORD_ENV`, else
/// `--keystore-password-file`, else TTY.
fn load_buyer_signer(chain: &Resolved) -> anyhow::Result<PrivateKeySigner> {
    let password = super::chain_ctx::read_keystore_password(
        &super::chain_ctx::password_sources(
            chain.keystore_password_file.as_deref(),
            PasswordUse::Unlock,
        ),
        "eth keystore password",
    )?
    .into_secret();
    load_signer(&chain.keystore, &password)
}

/// Parse a user-supplied `--pool`: 64 hex chars, optionally `0x`-prefixed (the
/// on-chain `poolId` is `keccak256(owner, ownerPoolNonce)`, a `bytes32`).
fn parse_pool_id(s: &str) -> anyhow::Result<PoolId> {
    B256::from_str(s)
        .map_err(|e| anyhow::anyhow!("invalid --pool {s:?}: expected a 32-byte hex hash: {e}"))
}

/// Reject a `getPool` read whose on-chain `owner` isn't this keystore — a wrong
/// keystore/contract or a stale `--pool` id (a zero `owner` means the pool was
/// never opened on this contract).
fn ensure_owned(on_chain_owner: Address, ours: Address, pool_id: PoolId) -> anyhow::Result<()> {
    anyhow::ensure!(
        on_chain_owner == ours,
        "pool {pool_id} on-chain owner {on_chain_owner} is not this keystore's address {ours} \
         — wrong keystore/contract, or a mistyped --pool"
    );
    Ok(())
}

/// `decdn pool open`: escrow a fresh `PaymentPool` deposit (`openPool`). The
/// resulting pool fans out to every provider the owner later pays — there is no
/// per-provider open.
async fn open(args: &cli::PoolOpenArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
    let file = load_file_config(config_path)?;
    // Before chain coordinates are even resolved, and so before the keystore
    // prompt: escrowing into a store the daemon never reads is the failure,
    // not a degraded outcome, and the answer depends only on the data dir.
    let data_dir = resolve_data_dir(args.chain.data_dir.clone(), &file)?;
    let store = open_client_store_for_escrow(&data_dir, "open a pool")?;
    let chain = resolve_chain(&args.chain, &file)?;
    let signer = Arc::new(load_buyer_signer(&chain)?);
    let owner = signer.address();
    let rpc = provider::build_provider(&chain.rpc_url, &signer)?;
    let contract = PaymentPool::new(chain.payment_pool, rpc.clone());
    let deployment = chain.deployment();

    let token = contract
        .usdc()
        .call()
        .await
        .map_err(|e| anyhow::Error::new(e).context("read PaymentPool.usdc()"))?;
    let deposit = U256::from(args.deposit_micro_usdc);

    ensure_allowance(&rpc, token, owner, chain.payment_pool, Some(deposit)).await?;

    // Watched from the open on: a Ctrl-C does not cancel a broadcast
    // `openPool`, so the command names it rather than dying with no hash.
    let escrows = UnconfirmedEscrows::default();
    let mut interrupt = Interrupt::watch(escrows.clone());
    let opened = open_and_record(
        &contract, &store, deployment, token, owner, deposit, &escrows,
    );
    let result = tokio::select! {
        result = opened => result,
        () = interrupt.wait() => Err(Interrupted.into()),
    };
    let pool_id = name_held_escrows(result, &escrows)?;

    let out = PoolOpenJson {
        pool_id: format!("{pool_id:#x}"),
    };
    serde_json::to_writer_pretty(std::io::stdout(), &out)?;
    println!();
    Ok(())
}

/// Escrow `deposit` in a fresh pool (`openPool`) from `owner`, record it in
/// `store`, and return its id. `escrows` names the open from before its submit
/// until the record resolves, so a second Ctrl-C during the write still names
/// it.
///
/// # Errors
///
/// The `openPool` fails or is not confirmed, or the record write fails after
/// it mined: the deposit is then escrowed and untracked, and `escrows` holds
/// it.
async fn open_and_record<P>(
    contract: &PaymentPool::PaymentPoolInstance<P>,
    store: &dyn BuyerPoolStore,
    deployment: Deployment,
    token: Address,
    owner: Address,
    deposit: U256,
    escrows: &UnconfirmedEscrows,
) -> anyhow::Result<PoolId>
where
    P: alloy::providers::Provider + Clone,
{
    let (opened, unrecorded) = open_pool(
        contract,
        deployment,
        token,
        owner,
        deposit,
        Some(escrows),
        Some(RECEIPT_TIMEOUT),
    )
    .await?;
    // The deposit is escrowed on-chain; a failed local record leaves it
    // untracked (reconcile against the tx).
    match store.record(&opened.state) {
        Ok(()) => unrecorded.recorded(),
        Err(e) => {
            return Err(unrecorded.untracked(escrowed_but_untracked(
                "buyer pool opened",
                opened.tx,
                e,
            )));
        }
    }
    Ok(opened.state.pool_id)
}

/// `decdn pool open` machine-readable output.
#[derive(Serialize)]
struct PoolOpenJson {
    #[serde(rename = "poolId")]
    pool_id: String,
}

/// `decdn pool top-up`: add funds to a pool the caller owns (`topUp`).
async fn top_up_cmd(args: &cli::PoolTopUpArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
    let file = load_file_config(config_path)?;
    let data_dir = resolve_data_dir(args.chain.data_dir.clone(), &file)?;
    let store = open_client_store_for_escrow(&data_dir, "top up a pool")?;
    let chain = resolve_chain(&args.chain, &file)?;
    let pool_id = parse_pool_id(&args.pool)?;
    let signer = Arc::new(load_buyer_signer(&chain)?);
    let owner = signer.address();
    let rpc = provider::build_provider(&chain.rpc_url, &signer)?;
    let contract = PaymentPool::new(chain.payment_pool, rpc.clone());

    let pool = contract
        .getPool(pool_id)
        .call()
        .await
        .map_err(|e| anyhow::Error::new(e).context("getPool failed"))?;
    ensure_owned(pool.owner, owner, pool_id)?;

    let additional = U256::from(args.amount_micro_usdc);
    // The pool does not store its token — it is the contract's immutable
    // `usdc()`, the same address the open path reads.
    let token = contract
        .usdc()
        .call()
        .await
        .map_err(|e| anyhow::Error::new(e).context("read PaymentPool.usdc()"))?;
    ensure_allowance(&rpc, token, owner, chain.payment_pool, Some(additional)).await?;

    // Watched from the `topUp` on: a Ctrl-C does not cancel a broadcast
    // `topUp`, so the command names it rather than dying with no hash.
    let escrows = UnconfirmedEscrows::default();
    let mut interrupt = Interrupt::watch(escrows.clone());
    let topped_up = top_up_and_credit(&contract, &store, owner, pool_id, additional, &escrows);
    let result = tokio::select! {
        result = topped_up => result,
        () = interrupt.wait() => Err(Interrupted.into()),
    };
    let (credited, new_deposit) = name_held_escrows(result, &escrows)?;
    println!("topped up pool {pool_id} by {credited} µUSDC; deposit now {new_deposit} µUSDC");
    Ok(())
}

/// Add `additional` to `owner`'s pool `pool_id` (`topUp`), credit it in
/// `store`, and return the credited amount with the new local deposit.
/// `escrows` names the `topUp` from before its submit until the credit
/// resolves, so a second Ctrl-C during the write still names it.
///
/// # Errors
///
/// The `topUp` fails or is not confirmed, or the credit does not land after
/// it mined: the USDC is then escrowed and untracked, and `escrows` holds it.
async fn top_up_and_credit<P>(
    contract: &PaymentPool::PaymentPoolInstance<P>,
    store: &dyn BuyerPoolStore,
    owner: Address,
    pool_id: PoolId,
    additional: U256,
    escrows: &UnconfirmedEscrows,
) -> anyhow::Result<(U256, U256)>
where
    P: alloy::providers::Provider + Clone,
{
    let (ToppedUpPool { credited, tx, .. }, unrecorded) = top_up(
        contract,
        owner,
        pool_id,
        additional,
        Some(escrows),
        Some(RECEIPT_TIMEOUT),
    )
    .await?;
    // `add_deposit` reports a backend fault as `Err` and a committed-row
    // mismatch as a non-`Added` `Ok` (the row vanished or now tracks a
    // different pool). Both mean the same thing here: the USDC is escrowed
    // and the local deposit is short by `credited`. Exiting 0 on either
    // would let a wrapper record the pool as funded — after which the
    // low-water check re-fires on every later fetch, and re-running this
    // command escrows again.
    let effect = topped_up_effect(pool_id, credited);
    let new_deposit =
        match grade_deposit_credit(store.add_deposit(owner, pool_id, credited), &effect, tx) {
            Ok(deposit) => {
                unrecorded.recorded();
                deposit
            }
            Err(err) => return Err(unrecorded.untracked(err)),
        };
    Ok((credited, new_deposit))
}

/// Where `pool close` / `pool reclaim` record that a pool is gone: after a
/// landed `closePool` / `reclaim`, and on a refusal or `--all` skip that reads
/// the pool as `Closing` or `Closed`.
///
/// The on-chain leg is identical either way; only the local bookkeeping
/// differs, and on a daemon's data dir there is none the CLI can legitimately
/// do (#2078).
#[derive(Clone, Copy)]
enum LocalBookkeeping<'a> {
    /// Clear the row in the client store this CLI owns. Only a row on
    /// `deployment` is cleared: a pool id repeats across deployments, so the
    /// same id on another `PaymentPool` names a different pool.
    Store {
        /// The client store.
        store: &'a dyn BuyerPoolStore,
        /// The chain and `PaymentPool` this command runs against.
        deployment: Deployment,
    },
    /// A `decdn-node` daemon owns this data dir. `decdn pool` has no code path
    /// to that daemon's `buyer.redb`, so its row stays as it is and the
    /// operator is told so — the alternative, a no-op `forget` graded as a
    /// clean close, is the silent lie this variant exists to remove.
    DaemonOwned(&'a Path),
}

/// What [`LocalBookkeeping::forget_not_open`] did with the row.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum NotOpenClear {
    /// A row for the pool was deleted.
    Cleared,
    /// Nothing to do: no row for the pool on this deployment, a pool that is
    /// still `Open` (or has an unrecognized status), or a daemon's data dir.
    Untouched,
    /// The store faulted, so the row may survive. A warning is already on
    /// stderr.
    Faulted,
}

impl NotOpenClear {
    /// The suffix for the refusal or skip line that reports this clear.
    const fn suffix(self) -> &'static str {
        match self {
            Self::Cleared => "; cleared its row from the local store",
            Self::Untouched => "",
            Self::Faulted => "; its local row could not be cleared (see the warning above)",
        }
    }
}

impl LocalBookkeeping<'_> {
    /// Build from the store the command opened: `None` means the data dir is a
    /// daemon's, so `buyer_db` names the file that was left untouched.
    const fn new<'a>(
        store: Option<&'a RedbBuyerPoolStore>,
        buyer_db: &'a Path,
        deployment: Deployment,
    ) -> LocalBookkeeping<'a> {
        match store {
            Some(store) => LocalBookkeeping::Store { store, deployment },
            None => LocalBookkeeping::DaemonOwned(buyer_db),
        }
    }

    /// Say, on stderr, that the daemon's row survives and what to do about it.
    /// The on-chain state has already changed, so this is not a warning the
    /// operator may ignore: the daemon still believes it owns a live pool.
    fn report_daemon_row_untouched(buyer_db: &Path, pool_id: PoolId, verb: &str) {
        eprintln!(
            "warning: pool {pool_id} {verb} on-chain, but the daemon's row in {} was NOT \
             cleared — `decdn pool` writes only the client store, and a running decdn-node holds \
             its own exclusively. Run `decdn node pools`: if that pool is listed, the daemon is \
             still using it — restart decdn-node so its bootstrap reconciles against the chain.",
            buyer_db.display()
        );
    }

    /// Clear the row after a mined `closePool`.
    ///
    /// # Errors
    ///
    /// Errors when the client store faulted — see [`grade_local_forget`].
    fn forget_after_close(
        self,
        owner: Address,
        pool_id: PoolId,
        tx: TxHash,
        reclaim_note: &str,
    ) -> anyhow::Result<()> {
        match self {
            Self::Store { store, deployment } => grade_local_forget(
                forget_on(store, deployment, owner, pool_id),
                pool_id,
                tx,
                reclaim_note,
            ),
            Self::DaemonOwned(buyer_db) => {
                Self::report_daemon_row_untouched(buyer_db, pool_id, "closed");
                Ok(())
            }
        }
    }

    /// Clear the row after a mined `reclaim`. Best-effort on both arms: the
    /// refund itself already landed.
    fn forget_after_reclaim(self, owner: Address, pool_id: PoolId) {
        match self {
            Self::Store { store, deployment } => {
                // `Ok(false)` costs nothing: reclaim is permissionless and the
                // row is legitimately absent after a `pool close` or a prior
                // reclaim. An `Err` is not free — the row may survive pointing
                // at a pool that is now `Closed`, whose `deposit` field still
                // reads healthy, so a later fetch reuses it and signs vouchers
                // `redeemMany` rejects outright. Warn rather than fail: the
                // refund itself landed, and a re-run of `reclaim --pool` reads
                // the pool as `Closed` and clears the row on that refusal (see
                // `forget_not_open`).
                if let Err(e) = forget_on(store, deployment, owner, pool_id) {
                    eprintln!(
                        "warning: pool {pool_id} reclaimed on-chain but clearing it from the \
                         local store failed: {e}; if the row survived, a later fetch reuses this \
                         closed pool and its vouchers are rejected — re-run `decdn pool reclaim \
                         --pool {pool_id}` to clear it"
                    );
                }
            }
            Self::DaemonOwned(buyer_db) => {
                Self::report_daemon_row_untouched(buyer_db, pool_id, "reclaimed");
            }
        }
    }

    /// Clear the row for a pool the chain shows as `Closing` or `Closed`, on a
    /// refusal or a `--all` skip that sends no transaction. Any other status
    /// leaves the store untouched.
    ///
    /// A row like this survives when the run that closed or reclaimed the pool
    /// could not clear it: the row-clear faulted, or the tx's outcome was
    /// unknown (an unread receipt, or a send whose broadcast is unknown)
    /// but it mined. It also survives a close or reclaim run from another
    /// machine or data dir. A later fetch then reuses a pool that is winding
    /// down or done. Seeing the status on-chain is enough to drop the row, so a
    /// later `close` / `reclaim` on this data dir that reads such a status
    /// repairs it. The clear is compare-and-delete on this deployment, so a row
    /// for a newer pool, or for the same id on another deployment, survives.
    ///
    /// Best-effort: a store fault warns on stderr and returns
    /// [`NotOpenClear::Faulted`], and the next run that reads the status
    /// retries the clear.
    fn forget_not_open(
        self,
        owner: Address,
        pool_id: PoolId,
        status: PaymentPool::Status,
    ) -> NotOpenClear {
        if !no_longer_open(status) {
            return NotOpenClear::Untouched;
        }
        match self {
            Self::Store { store, deployment } => {
                match forget_on(store, deployment, owner, pool_id) {
                    Ok(true) => NotOpenClear::Cleared,
                    Ok(false) => NotOpenClear::Untouched,
                    Err(e) => {
                        eprintln!(
                            "warning: pool {pool_id} is {} on-chain but clearing it from the \
                             local store failed: {e}; if the row survived, a later fetch may \
                             reuse this pool, whose vouchers stop being redeemable at its \
                             dispute deadline — re-run this command to retry the clear",
                            status_label(status)
                        );
                        NotOpenClear::Faulted
                    }
                }
            }
            Self::DaemonOwned(buyer_db) => {
                let verb = format!("is already {}", status_label(status));
                Self::report_daemon_row_untouched(buyer_db, pool_id, &verb);
                NotOpenClear::Untouched
            }
        }
    }
}

/// Compare-and-delete `owner`'s row for `pool_id`, but only when the store's
/// row for that id is on `deployment`. A pool id is `keccak256(owner, nonce)`,
/// so it repeats on every `PaymentPool` deployment, and `forget_if_pool` alone
/// cannot tell the pools apart. The read and the delete are two transactions;
/// the client store is held exclusively by this process, so nothing writes in
/// between. Returns whether a row was deleted.
fn forget_on(
    store: &dyn BuyerPoolStore,
    deployment: Deployment,
    owner: Address,
    pool_id: PoolId,
) -> Result<bool, decdn_incentive::StoreError> {
    match store.get_by_pool_id(pool_id)? {
        Some(state) if state.is_on(deployment) => store.forget_if_pool(owner, pool_id),
        _ => Ok(false),
    }
}

/// True for the statuses that mean a pool is winding down or done: `Closing`
/// and `Closed`. A local row for such a pool is stale. Returns false for
/// `Open` and for the hidden invalid `sol!` variant, so an unrecognized status
/// never touches the store.
const fn no_longer_open(status: PaymentPool::Status) -> bool {
    matches!(
        status,
        PaymentPool::Status::Closing | PaymentPool::Status::Closed
    )
}

/// Grade the local row-clear that follows a mined `closePool`.
///
/// `pool close` drops the local record *so that* a later `decdn get` opens a
/// fresh pool rather than reusing one that is winding down. Only the
/// `Err` leaves that in doubt.
///
/// The clear is compare-and-delete on this deployment (see [`forget_on`]), and
/// its `Ok(false)` means the compare found nothing to delete: the owner index
/// holds no row, it already points at a newer pool, or the row for this id is on
/// another deployment. In each state nothing maps this owner to the closed pool,
/// which is the guarantee `close` wanted, and deleting anything further would
/// destroy a live row — the lost update the compare-and-delete exists to prevent.
/// Closing a pool this store never tracked lands there routinely: a second
/// machine, a fresh `--data-dir`, or an older pool superseded by a later
/// `pool open`. That is a clean close.
///
/// An `Err` is different. The write faulted, so the row may survive with a live
/// owner index pointing at the closed pool. A later fetch then reuses it and
/// signs vouchers that stop being redeemable at the dispute deadline, against a
/// deposit the owner is about to reclaim.
///
/// The close itself landed, so the error is a [`CloseUnrecorded`]: a `--all`
/// sweep counts the pool as closed and still exits nonzero, because the money
/// moved and the local record did not follow. Its message names the re-run of
/// the close (which reads the pool as `Closing` and clears the row) and the
/// reclaim (the deadline the success line would have printed, which also
/// clears the row).
///
/// # Errors
///
/// Errors when the store faulted, leaving the row's fate unknown.
fn grade_local_forget(
    outcome: Result<bool, decdn_incentive::StoreError>,
    pool_id: PoolId,
    tx: TxHash,
    reclaim_note: &str,
) -> anyhow::Result<()> {
    match outcome {
        Ok(_) => Ok(()),
        Err(cause) => Err(anyhow::Error::new(CloseUnrecorded {
            pool_id,
            tx,
            reclaim_note: reclaim_note.to_string(),
            cause,
        })),
    }
}

/// A `closePool` error whose close landed on-chain but whose local row-clear
/// faulted. Recover it with `err.downcast_ref::<CloseUnrecorded>()`.
///
/// The store fault is rendered inline, next to "failed", and is not a
/// `source()`: a chain-rendered source would trail the remedy sentence and
/// read as the reclaim's fault.
#[derive(Debug)]
struct CloseUnrecorded {
    /// The pool that closed.
    pool_id: PoolId,
    /// The landed `closePool` transaction.
    tx: TxHash,
    /// The reclaim instruction the success line would have printed.
    reclaim_note: String,
    /// The store fault that left the row's fate unknown.
    cause: decdn_incentive::StoreError,
}

impl std::fmt::Display for CloseUnrecorded {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let Self {
            pool_id,
            tx,
            reclaim_note,
            cause,
        } = self;
        write!(
            f,
            "pool {pool_id} closed on-chain (tx {tx}) but clearing it from the local store \
             failed: {cause}; the row may survive and send a later fetch back to this closing \
             pool, whose vouchers stop being redeemable at the dispute deadline — re-run `decdn \
             pool close --pool {pool_id}` to clear it, or {reclaim_note}, which also clears the \
             row"
        )
    }
}

impl std::error::Error for CloseUnrecorded {}

/// Terminal outcome of a single close/reclaim tx.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum TxOutcome {
    /// Mined with a success receipt.
    Landed,
    /// Deterministically reverted (caught at estimation or a failing receipt).
    Reverted,
}

/// A mined-but-reverted tx returns `Ok(receipt)` with `status() == false`, so map
/// the receipt status to the outcome.
const fn receipt_outcome(landed: bool) -> TxOutcome {
    if landed {
        TxOutcome::Landed
    } else {
        TxOutcome::Reverted
    }
}

/// What `close --all` does with one enumerated pool, decided from its on-chain
/// status alone. Only `Open` pools have anything to close.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ClosePlan {
    /// `Open` — send `closePool`.
    Close,
    /// Not `Open` — nothing to do; the string is the human-readable reason.
    Skip(&'static str),
}

/// What `reclaim` does with one pool, decided from its status and dispute
/// deadline against the chain's head time. It drives the `--all` sweep, the
/// `--pool` pre-check (`single_reclaim_gate`), and the `pool list` Reclaimable
/// column.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ReclaimPlan {
    /// `Closing` and its dispute window has elapsed — send `reclaim`.
    Reclaim,
    /// `Open` — it must be closed first, so there is nothing to reclaim yet.
    SkipOpen,
    /// `Closing` but still inside the dispute window; carries the absolute
    /// Unix deadline so the operator learns when it becomes reclaimable.
    SkipInWindow(u64),
    /// `Closed` — the reclaim already happened (that is what sets `Closed`).
    SkipClosed,
    /// An unrecognized on-chain status (the hidden invalid `sol!` variant) —
    /// nothing safe to reclaim, and not asserted to have been reclaimed.
    SkipUnknown,
}

/// Running counts for a `--all` sweep. Each pool read lands in exactly one of
/// `acted`, `skipped`, `failed`, or `unconfirmed`; `unrecorded` counts a subset
/// of `acted` and `skipped`. The exit code keys off `failed`, `unconfirmed`, and
/// `unrecorded`; `acted` and `skipped` are reported for the operator.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
struct BatchTally {
    /// Pools whose close or reclaim landed on-chain, including the
    /// `unrecorded` ones.
    acted: usize,
    /// Pools intentionally left untouched (wrong status / window not elapsed).
    skipped: usize,
    /// Pools whose read or transaction errored with a known outcome.
    failed: usize,
    /// Pools whose transaction may have broadcast and may still mine (the
    /// error carries [`TxUnconfirmed`]). Disjoint from `failed`.
    unconfirmed: usize,
    /// Pools that are no longer `Open` on-chain but whose local row-clear
    /// faulted, so the row may survive: a landed close (the error is a
    /// [`CloseUnrecorded`]; also counted in `acted`), or a skipped `Closing` /
    /// `Closed` pool (also counted in `skipped`).
    unrecorded: usize,
}

/// How [`BatchTally::record_error`] counted one pool's error.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ErrorClass {
    /// A definite failure.
    Failed,
    /// The transaction may still mine ([`TxUnconfirmed`]).
    Unconfirmed,
    /// The transaction landed, but the local row-clear faulted
    /// ([`CloseUnrecorded`]).
    LandedUnrecorded,
}

impl BatchTally {
    /// Count one skipped pool, and as `unrecorded` when its row-clear faulted.
    const fn record_skip(&mut self, cleared: NotOpenClear) {
        self.skipped += 1;
        if matches!(cleared, NotOpenClear::Faulted) {
            self.unrecorded += 1;
        }
    }

    /// Count one pool's `closePool` / `reclaim` error: as `acted` and
    /// `unrecorded` when it carries [`CloseUnrecorded`], as `unconfirmed` when
    /// it carries [`TxUnconfirmed`], else as `failed`.
    fn record_error(&mut self, err: &anyhow::Error) -> ErrorClass {
        if err.downcast_ref::<CloseUnrecorded>().is_some() {
            self.acted += 1;
            self.unrecorded += 1;
            ErrorClass::LandedUnrecorded
        } else if err.downcast_ref::<TxUnconfirmed>().is_some() {
            self.unconfirmed += 1;
            ErrorClass::Unconfirmed
        } else {
            self.failed += 1;
            ErrorClass::Failed
        }
    }
}

/// The context on a `closePool` / `reclaim` error whose transaction may have
/// broadcast and may still mine: its send failed in transport, or it was
/// broadcast and its receipt was not read (a failed read, or none within
/// [`RECEIPT_TIMEOUT`]). A re-run may double-submit, so a `--all` sweep counts
/// it apart from a definite failure. Recover it with
/// `err.downcast_ref::<TxUnconfirmed>()`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct TxUnconfirmed {
    /// The contract call, `closePool` or `reclaim`.
    label: &'static str,
    /// The broadcast transaction. `None` when the send failed in transport:
    /// the RPC node may have broadcast the transaction without returning its
    /// hash.
    tx: Option<TxHash>,
}

impl std::fmt::Display for TxUnconfirmed {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let label = self.label;
        match self.tx {
            Some(tx) => write!(
                f,
                "{label} sent (tx {tx:#x}) but its receipt was not read; it may still mine — \
                 check the tx before re-running"
            ),
            None => write!(
                f,
                "{label} send failed: the request may have reached the node and broadcast the \
                 transaction — check this signer's pending and mined transactions before \
                 re-running"
            ),
        }
    }
}

impl std::error::Error for TxUnconfirmed {}

/// Classify one enumerated pool for `close --all`.
const fn plan_close(status: PaymentPool::Status) -> ClosePlan {
    match status {
        PaymentPool::Status::Open => ClosePlan::Close,
        PaymentPool::Status::Closing => ClosePlan::Skip("already closing"),
        PaymentPool::Status::Closed => ClosePlan::Skip("already closed"),
        // `sol!` enums carry a hidden invalid variant, so a wildcard is
        // required; an unrecognized status has nothing safe to close.
        _ => ClosePlan::Skip("unknown on-chain status"),
    }
}

/// Classify one pool for `reclaim`. `reclaim` reverts unless
/// the pool is `Closing` past its deadline, so the plan pre-filters to exactly
/// that case and names why each other pool is skipped. `now == deadline` counts
/// as elapsed, matching the on-chain `block.timestamp >= disputeDeadline` gate.
const fn plan_reclaim(status: PaymentPool::Status, dispute_deadline: u64, now: u64) -> ReclaimPlan {
    match status {
        PaymentPool::Status::Open => ReclaimPlan::SkipOpen,
        PaymentPool::Status::Closing if now >= dispute_deadline => ReclaimPlan::Reclaim,
        PaymentPool::Status::Closing => ReclaimPlan::SkipInWindow(dispute_deadline),
        PaymentPool::Status::Closed => ReclaimPlan::SkipClosed,
        // `sol!` enums carry a hidden invalid variant, so a wildcard is
        // required; an unrecognized status is not asserted to be reclaimed.
        _ => ReclaimPlan::SkipUnknown,
    }
}

/// Decide from one `getPool` read and the chain's head time whether `reclaim
/// --pool` may send. Each status or deadline condition that makes `reclaim`
/// revert is refused here with its own cause, so the operator learns what to
/// do (and, in the dispute window, when) without a transaction or a keystore
/// unlock.
///
/// # Errors
///
/// Errors when the pool does not exist, is still `Open`, is `Closing` inside
/// its dispute window, is already `Closed`, or has an unrecognized status. The
/// zero-`owner` check runs first: a never-opened pool reads as status `0`,
/// which is `Open`, and would otherwise be reported as still Open.
fn single_reclaim_gate(
    pool_id: PoolId,
    owner: Address,
    status: PaymentPool::Status,
    dispute_deadline: u64,
    now: u64,
) -> anyhow::Result<()> {
    anyhow::ensure!(
        owner != Address::ZERO,
        "pool {pool_id} does not exist on this PaymentPool contract — a mistyped --pool, or \
         the wrong payment_pool address"
    );
    match plan_reclaim(status, dispute_deadline, now) {
        ReclaimPlan::Reclaim => Ok(()),
        ReclaimPlan::SkipOpen => anyhow::bail!(
            "pool {pool_id} is still Open — run `decdn pool close --pool {pool_id}` first"
        ),
        ReclaimPlan::SkipInWindow(deadline) => anyhow::bail!(
            "pool {pool_id} is still in its dispute window; reclaimable after {}",
            format_expiry(deadline, now)
        ),
        ReclaimPlan::SkipClosed => {
            anyhow::bail!("pool {pool_id} is already Closed — its residual was already reclaimed")
        }
        ReclaimPlan::SkipUnknown => {
            anyhow::bail!("pool {pool_id} has an unrecognized on-chain status — nothing to reclaim")
        }
    }
}

/// Turn a finished `--all` sweep into a process result. Any failed,
/// unconfirmed, or unrecorded pool is a nonzero exit; "nothing eligible" is
/// success — the summary line already says so.
///
/// # Errors
///
/// Errors when at least one pool failed to `verb`, its `verb` transaction may
/// have broadcast without a confirmed outcome, or it is no longer `Open` but
/// its local row-clear faulted.
fn batch_result(tally: &BatchTally, verb: &str) -> anyhow::Result<()> {
    let failed = (tally.failed > 0).then(|| format!("{} pool(s) failed to {verb}", tally.failed));
    let unconfirmed = (tally.unconfirmed > 0).then(|| {
        format!(
            "{} pool(s) may have broadcast a {verb} transaction — check this signer's pending \
             and mined transactions before re-running",
            tally.unconfirmed
        )
    });
    let unrecorded = (tally.unrecorded > 0).then(|| {
        format!(
            "{} pool(s) are no longer Open on-chain but kept a local row that may be stale — \
             re-run the same command to clear it",
            tally.unrecorded
        )
    });
    let causes: Vec<String> = [failed, unconfirmed, unrecorded]
        .into_iter()
        .flatten()
        .collect();
    if causes.is_empty() {
        return Ok(());
    }
    anyhow::bail!("{}; see the summary above", causes.join(", and "))
}

/// Send `closePool(pool_id)`, wait for the receipt, and on success clear the
/// local buyer row so a later fetch opens a fresh pool. Ownership and `Open`
/// status are the caller's responsibility. Returns the reclaim note shared by
/// the success line and the row-clear error.
///
/// # Errors
///
/// Errors when gas estimation rejects the call (see `send_reverted`), the send
/// fails (an error that leaves the broadcast unknown says so; see
/// `send_failed`), the receipt is not read within [`RECEIPT_TIMEOUT`], the tx
/// reverted, or the local row-clear faulted after a landed close (the error
/// carries [`CloseUnrecorded`]).
async fn close_and_forget<P>(
    contract: &PaymentPool::PaymentPoolInstance<P>,
    store: LocalBookkeeping<'_>,
    owner: Address,
    pool_id: PoolId,
) -> anyhow::Result<String>
where
    P: alloy::providers::Provider + Clone,
{
    let pending = match contract.closePool(pool_id).send().await {
        Ok(pending) => pending,
        Err(e) if e.as_revert_data().is_some() => {
            return Err(send_reverted(e, "closePool", pool_id, None));
        }
        Err(e) => return Err(send_failed(e, "closePool")),
    };
    // Read before the wait consumes `pending`: an unread receipt leaves the
    // tx's fate unknown, and the hash is the operator's only handle on it. The
    // wait is bounded, so a dropped tx cannot stall a `--all` sweep (#2433).
    let tx = *pending.tx_hash();
    let receipt = await_receipt(pending, Some(RECEIPT_TIMEOUT))
        .await
        .map_err(|e| {
            e.context(TxUnconfirmed {
                label: "closePool",
                tx: Some(tx),
            })
        })?;

    match receipt_outcome(receipt.status()) {
        TxOutcome::Landed => {
            // The dispute deadline is only known post-close; read it best-effort
            // and never print the pre-close `0` if that read fails. The local
            // record is dropped so a later `decdn get` opens a fresh
            // pool rather than reusing one that is winding down. Any transport
            // error is redacted here, since the caller may print it directly
            // rather than through `main`'s error boundary.
            let deadline_note = match contract.getPool(pool_id).call().await {
                Ok(p) => format!("after Unix {}", p.disputeDeadline),
                Err(e) => format!(
                    "after the dispute window (couldn't read the exact deadline: {})",
                    sanitize_error_sources(&e)
                ),
            };
            // One note for both the success line and the failure, so the two
            // cannot drift into different instructions for the same deadline.
            let reclaim_note = format!("run `decdn pool reclaim --pool {pool_id}` {deadline_note}");
            store.forget_after_close(owner, pool_id, tx, &reclaim_note)?;
            Ok(reclaim_note)
        }
        TxOutcome::Reverted => anyhow::bail!(
            "closePool reverted on-chain for pool {pool_id} (tx {tx:#x}; it may have raced a \
             concurrent close)"
        ),
    }
}

/// Send `reclaim(pool_id)`, wait for the receipt, and on success best-effort
/// clear the local row. `reclaim` is permissionless, so no ownership check is
/// needed. Callers pre-check the pool with `plan_reclaim`, so a revert here
/// means the chain disagreed with that pre-check. Returns `Ok(())` once the
/// refund lands.
///
/// # Errors
///
/// Errors when gas estimation rejects the call (see `send_reverted`), the send
/// fails (an error that leaves the broadcast unknown says so; see
/// `send_failed`), the receipt is not read within [`RECEIPT_TIMEOUT`], or the
/// tx reverted after the pre-check passed (see `reclaim_reverted`). A failed
/// local row-clear only warns — the refund itself landed.
async fn reclaim_and_forget<P>(
    contract: &PaymentPool::PaymentPoolInstance<P>,
    store: LocalBookkeeping<'_>,
    owner: Address,
    pool_id: PoolId,
) -> anyhow::Result<()>
where
    P: alloy::providers::Provider + Clone,
{
    let pending = match contract.reclaim(pool_id).send().await {
        Ok(pending) => pending,
        Err(e) if e.as_revert_data().is_some() => {
            return Err(send_reverted(
                e,
                "reclaim",
                pool_id,
                Some(RECLAIM_UNKNOWN_REVERT),
            ));
        }
        Err(e) => return Err(send_failed(e, "reclaim")),
    };
    // Read before the wait consumes `pending`: an unread receipt leaves the
    // tx's fate unknown, and the hash is the operator's only handle on it. The
    // wait is bounded, so a dropped tx cannot stall a `--all` sweep (#2433).
    let tx = *pending.tx_hash();
    let receipt = await_receipt(pending, Some(RECEIPT_TIMEOUT))
        .await
        .map_err(|e| {
            e.context(TxUnconfirmed {
                label: "reclaim",
                tx: Some(tx),
            })
        })?;

    match receipt_outcome(receipt.status()) {
        TxOutcome::Landed => {
            store.forget_after_reclaim(owner, pool_id);
            Ok(())
        }
        TxOutcome::Reverted => anyhow::bail!(reclaim_reverted(pool_id, tx)),
    }
}

/// The guessed cause of a mined `reclaim` revert. A receipt carries no revert
/// data, so nothing narrows the cause.
const RECLAIM_REVERT_GUESS: &str =
    "most likely a concurrent reclaim; the USDC refund transfer to the owner can also revert";

/// The cause of a `reclaim` that gas estimation rejects with no known
/// `PaymentPool` error. A concurrent reclaim reverts `PoolNotClosing`, which
/// decodes, so the remaining revert in `reclaim` is the refund transfer.
const RECLAIM_UNKNOWN_REVERT: &str = "not a PaymentPool lifecycle error; most likely the USDC \
                                      refund transfer to the owner reverted";

/// The error for a mined `reclaim` (tx `tx`) that reverted after the pre-check
/// passed. A receipt carries no revert data, so the cause is a guess.
fn reclaim_reverted(pool_id: PoolId, tx: TxHash) -> String {
    format!(
        "reclaim reverted on-chain for pool {pool_id} (tx {tx:#x}) after the pre-check passed \
         ({RECLAIM_REVERT_GUESS})"
    )
}

/// The error for a `closePool` / `reclaim` send that carries revert data. Gas
/// estimation rejected the call, so nothing was broadcast. A known
/// `PaymentPool` selector names the cause; any other selector falls back to
/// `unknown_cause`, when the caller has one. The source error keeps the raw
/// revert data and any `Error(string)` reason either way.
fn send_reverted(
    err: alloy::contract::Error,
    label: &str,
    pool_id: PoolId,
    unknown_cause: Option<&str>,
) -> anyhow::Error {
    let cause = match PoolLifecycleRevert::decode(err.as_revert_data().as_ref()) {
        Some(revert) => {
            let hint = match revert {
                PoolLifecycleRevert::PoolNotOpen => {
                    "the pool is no longer Open; it may have raced a concurrent close"
                }
                PoolLifecycleRevert::NotPoolOwner => "this keystore does not own the pool",
                PoolLifecycleRevert::PoolNotClosing => {
                    "the pool is not Closing; most likely a concurrent reclaim already closed it"
                }
                PoolLifecycleRevert::GraceWindowActive => "the dispute window has not elapsed yet",
            };
            format!(": {} — {hint}", revert.signature())
        }
        None => unknown_cause.map_or_else(String::new, |cause| format!(" ({cause})")),
    };
    anyhow::Error::new(err).context(format!(
        "{label} would revert for pool {pool_id}{cause} (rejected at gas estimation; nothing was \
         sent)"
    ))
}

/// The error for a `closePool` / `reclaim` send that failed with no revert
/// data. When `send_broadcast_unknown` holds (a timeout, a reset, an HTTP 5xx,
/// an unreadable response), the tx may have broadcast before its response was
/// lost and no hash came back. The error then carries [`TxUnconfirmed`], whose
/// message sends the operator to this signer's own transactions. Any other
/// failure (a JSON-RPC rejection, an HTTP 4xx, a local fault) broadcast nothing
/// and gets no such note (#1577).
fn send_failed(err: alloy::contract::Error, label: &'static str) -> anyhow::Error {
    if decdn_incentive::tx::send_broadcast_unknown(&err) {
        anyhow::Error::new(err).context(TxUnconfirmed { label, tx: None })
    } else {
        anyhow::Error::new(err).context(format!("{label} send failed"))
    }
}

/// `decdn pool close`: start the grace-window close on a pool the caller owns
/// (`closePool`). Redemptions stay valid until the dispute deadline; `pool
/// reclaim` refunds the residual after it elapses. `--all` closes every `Open`
/// pool this keystore owns; `--pool` closes exactly one.
async fn close(args: &cli::PoolCloseArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
    let file = load_file_config(config_path)?;
    // Parse the target id before any store/keystore/provider work, so a
    // malformed `--pool` fails fast without touching the chain. `None` is the
    // `--all` sweep (clap's arg group guarantees exactly one of the two).
    let target = args.pool.as_deref().map(parse_pool_id).transpose()?;

    let store_owner = classify_buyer_store(&resolve_data_dir(args.chain.data_dir.clone(), &file)?)?;
    if target.is_none() {
        // `--all` enumerates from chain, not from the store, so on a node data
        // dir it would close the pool the daemon is paying from right now.
        store_owner.refuse_sweep("close")?;
    }
    let chain = resolve_chain(&args.chain, &file)?;
    let store = store_owner.open_for_write()?;
    let buyer_db = node_buyer_db(&chain.data_dir);
    let books = LocalBookkeeping::new(store.as_ref(), &buyer_db, chain.deployment());
    let signer = Arc::new(load_buyer_signer(&chain)?);
    let owner = signer.address();
    let rpc = provider::build_provider(&chain.rpc_url, &signer)?;
    let contract = PaymentPool::new(chain.payment_pool, rpc);

    let Some(pool_id) = target else {
        return close_all(&contract, books, owner).await;
    };

    let pool = contract
        .getPool(pool_id)
        .call()
        .await
        .map_err(|e| anyhow::Error::new(e).context("getPool failed"))?;
    ensure_owned(pool.owner, owner, pool_id)?;
    if !matches!(pool.status, PaymentPool::Status::Open) {
        return Err(refuse_close(books, owner, pool_id, pool.status));
    }

    let reclaim_note = close_and_forget(&contract, books, owner, pool_id).await?;
    println!("closed pool {pool_id}; dispute window open — {reclaim_note}");
    Ok(())
}

/// The refusal for `close --pool` on a pool that is not `Open`. A `Closing` or
/// `Closed` pool first has its local row cleared (see
/// `LocalBookkeeping::forget_not_open`), so a re-run repairs a row that a
/// landed close or reclaim could not clear.
fn refuse_close(
    books: LocalBookkeeping<'_>,
    owner: Address,
    pool_id: PoolId,
    status: PaymentPool::Status,
) -> anyhow::Error {
    let cleared = books.forget_not_open(owner, pool_id, status).suffix();
    anyhow::anyhow!(
        "pool {pool_id} is not Open — nothing to close (already closing or closed){cleared}"
    )
}

/// `close --all`: enumerate every pool this keystore owns and close the `Open`
/// ones; a skipped `Closing` or `Closed` pool has its local row cleared. One
/// pool's read or close failure is recorded and the sweep continues.
/// A close whose tx may have broadcast without a confirmed outcome counts as
/// unconfirmed, not failed. A close that landed, or a skipped pool, whose
/// row-clear faulted counts as "row not cleared" too. The exit code is nonzero
/// if any pool failed, is unconfirmed, or kept its row.
async fn close_all<P>(
    contract: &PaymentPool::PaymentPoolInstance<P>,
    store: LocalBookkeeping<'_>,
    owner: Address,
) -> anyhow::Result<()>
where
    P: alloy::providers::Provider + Clone,
{
    let ids = enumerate_owned_pools(contract, owner).await?;
    if ids.is_empty() {
        println!("no pools found for this keystore — nothing to close");
        return Ok(());
    }

    let mut tally = BatchTally::default();
    for pool_id in ids {
        let status = match contract.getPool(pool_id).call().await {
            Ok(pool) => pool.status,
            Err(e) => {
                eprintln!(
                    "failed to read pool {pool_id}: {}",
                    sanitize_error_sources(&e)
                );
                tally.failed += 1;
                continue;
            }
        };
        match plan_close(status) {
            ClosePlan::Close => match close_and_forget(contract, store, owner, pool_id).await {
                Ok(note) => {
                    println!("closed pool {pool_id}; dispute window open — {note}");
                    tally.acted += 1;
                }
                Err(e) => {
                    let msg = sanitize_err_chain(&e);
                    match tally.record_error(&e) {
                        ErrorClass::LandedUnrecorded => eprintln!("warning: {msg}"),
                        ErrorClass::Unconfirmed => {
                            eprintln!("close of pool {pool_id} unconfirmed: {msg}");
                        }
                        ErrorClass::Failed => eprintln!("failed to close pool {pool_id}: {msg}"),
                    }
                }
            },
            ClosePlan::Skip(reason) => {
                let cleared = store.forget_not_open(owner, pool_id, status);
                println!("skipped pool {pool_id}: {reason}{}", cleared.suffix());
                tally.record_skip(cleared);
            }
        }
    }
    println!(
        "close --all: closed {}, skipped {}, failed {}, unconfirmed {}, row not cleared {}",
        tally.acted, tally.skipped, tally.failed, tally.unconfirmed, tally.unrecorded
    );
    batch_result(&tally, "close")
}

/// `decdn pool reclaim`: refund the residual deposit of a pool once its grace
/// window has elapsed (`reclaim`; permissionless — callable by anyone, but only
/// the owner receives funds). `--all` reclaims every pool this keystore owns
/// whose window has elapsed; `--pool` reclaims exactly one, after a chain-read
/// pre-check of its status and deadline that runs before the keystore unlock.
/// A refusal for a `Closing` or `Closed` pool also clears its local row.
async fn reclaim(args: &cli::PoolReclaimArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
    let file = load_file_config(config_path)?;
    // Parse the target id before any store/keystore/provider work, so a
    // malformed `--pool` fails fast without touching the chain. `None` is the
    // `--all` sweep (clap's arg group guarantees exactly one of the two).
    let target = args.pool.as_deref().map(parse_pool_id).transpose()?;

    let store_owner = classify_buyer_store(&resolve_data_dir(args.chain.data_dir.clone(), &file)?)?;
    if target.is_none() {
        store_owner.refuse_sweep("reclaim")?;
    }
    let chain = resolve_chain(&args.chain, &file)?;
    let deployment = chain.deployment();
    if let Some(pool_id) = target {
        // `getPool` and the head block are reads, so they run on a wallet-free
        // provider before the keystore unlock: a pool that `reclaim` would
        // refuse fails here with its cause and asks for no password.
        let reader = PaymentPool::new(
            chain.payment_pool,
            provider::build_read_provider(&chain.rpc_url)?,
        );
        let pool = reader
            .getPool(pool_id)
            .call()
            .await
            .map_err(|e| anyhow::Error::new(e).context("getPool failed"))?;
        let now = chain_ctx::head_timestamp(reader.provider()).await?;
        if let Err(refusal) =
            single_reclaim_gate(pool_id, pool.owner, pool.status, pool.disputeDeadline, now)
        {
            // Only a `Closing` or `Closed` pool has a row to clear; any
            // other refusal opens no store.
            if !no_longer_open(pool.status) {
                return Err(refusal);
            }
            let opened = store_owner.open_for_write();
            let buyer_db = node_buyer_db(&chain.data_dir);
            let books = opened
                .as_ref()
                .map(|store| LocalBookkeeping::new(store.as_ref(), &buyer_db, deployment));
            return Err(refuse_reclaim(
                &refusal,
                books,
                pool.owner,
                pool_id,
                pool.status,
            ));
        }
    }
    let store = store_owner.open_for_write()?;
    let buyer_db = node_buyer_db(&chain.data_dir);
    let books = LocalBookkeeping::new(store.as_ref(), &buyer_db, deployment);
    let signer = Arc::new(load_buyer_signer(&chain)?);
    let owner = signer.address();
    let rpc = provider::build_provider(&chain.rpc_url, &signer)?;
    let contract = PaymentPool::new(chain.payment_pool, rpc);

    let Some(pool_id) = target else {
        return reclaim_all(&contract, books, owner).await;
    };

    reclaim_and_forget(&contract, books, owner, pool_id).await?;
    println!("reclaimed pool {pool_id}; residual deposit refunded to its owner");
    Ok(())
}

/// Finish a `reclaim --pool` refusal for a `Closing` or `Closed` pool: clear its
/// local row (see `LocalBookkeeping::forget_not_open`) and name the clear in the
/// refusal. `owner` is the on-chain owner, so the clear needs no keystore
/// unlock; compare-and-delete leaves a row for any other pool alone.
///
/// The refusal is the answer the operator asked for, and the clear is a
/// repair, so a store that will not open (`books` is `Err`) only warns: the
/// refusal still reaches the operator with its cause.
fn refuse_reclaim(
    refusal: &anyhow::Error,
    books: Result<LocalBookkeeping<'_>, &anyhow::Error>,
    owner: Address,
    pool_id: PoolId,
    status: PaymentPool::Status,
) -> anyhow::Error {
    let cleared = match books {
        Ok(books) => books.forget_not_open(owner, pool_id, status),
        Err(e) => {
            eprintln!(
                "warning: pool {pool_id} is {} on-chain but the local store could not be opened \
                 to clear its row: {}; re-run this command once the store opens",
                status_label(status),
                sanitize_err_chain(e)
            );
            NotOpenClear::Faulted
        }
    };
    anyhow::anyhow!("{refusal:#}{}", cleared.suffix())
}

/// `reclaim --all`: enumerate every pool this keystore owns and reclaim the ones
/// whose dispute window has elapsed. Pools still `Open`, still inside the
/// window, or already reclaimed are skipped (in-window pools report when they
/// become reclaimable), and a skipped `Closing` or `Closed` pool has its local
/// row cleared. One pool's failure is recorded and the sweep continues.
/// A reclaim whose tx may have broadcast without a confirmed outcome counts as
/// unconfirmed, not failed. A skipped pool whose row-clear faulted counts as
/// "row not cleared". The exit code is nonzero if any pool failed, is
/// unconfirmed, or kept its row.
async fn reclaim_all<P>(
    contract: &PaymentPool::PaymentPoolInstance<P>,
    store: LocalBookkeeping<'_>,
    owner: Address,
) -> anyhow::Result<()>
where
    P: alloy::providers::Provider + Clone,
{
    let ids = enumerate_owned_pools(contract, owner).await?;
    if ids.is_empty() {
        println!("no pools found for this keystore — nothing to reclaim");
        return Ok(());
    }
    let now = chain_ctx::head_timestamp(contract.provider()).await?;

    let mut tally = BatchTally::default();
    for pool_id in ids {
        let pool = match contract.getPool(pool_id).call().await {
            Ok(pool) => pool,
            Err(e) => {
                eprintln!(
                    "failed to read pool {pool_id}: {}",
                    sanitize_error_sources(&e)
                );
                tally.failed += 1;
                continue;
            }
        };
        match plan_reclaim(pool.status, pool.disputeDeadline, now) {
            ReclaimPlan::Reclaim => {
                match reclaim_and_forget(contract, store, owner, pool_id).await {
                    Ok(()) => {
                        println!(
                            "reclaimed pool {pool_id}; residual deposit refunded to its owner"
                        );
                        tally.acted += 1;
                    }
                    Err(e) => {
                        let msg = sanitize_err_chain(&e);
                        match tally.record_error(&e) {
                            ErrorClass::LandedUnrecorded => eprintln!("warning: {msg}"),
                            ErrorClass::Unconfirmed => {
                                eprintln!("reclaim of pool {pool_id} unconfirmed: {msg}");
                            }
                            ErrorClass::Failed => {
                                eprintln!("failed to reclaim pool {pool_id}: {msg}");
                            }
                        }
                    }
                }
            }
            ReclaimPlan::SkipOpen => {
                println!("skipped pool {pool_id}: still Open — run `pool close` first");
                tally.skipped += 1;
            }
            ReclaimPlan::SkipInWindow(deadline) => {
                let cleared = store.forget_not_open(owner, pool_id, pool.status);
                println!(
                    "skipped pool {pool_id}: dispute window open — reclaimable after {}{}",
                    format_expiry(deadline, now),
                    cleared.suffix()
                );
                tally.record_skip(cleared);
            }
            ReclaimPlan::SkipClosed => {
                let cleared = store.forget_not_open(owner, pool_id, pool.status);
                println!(
                    "skipped pool {pool_id}: already reclaimed{}",
                    cleared.suffix()
                );
                tally.record_skip(cleared);
            }
            ReclaimPlan::SkipUnknown => {
                println!("skipped pool {pool_id}: unrecognized on-chain status");
                tally.skipped += 1;
            }
        }
    }
    println!(
        "reclaim --all: reclaimed {}, skipped {}, failed {}, unconfirmed {}, row not cleared {}",
        tally.acted, tally.skipped, tally.failed, tally.unconfirmed, tally.unrecorded
    );
    batch_result(&tally, "reclaim")
}

/// Resolve the capability's absolute Unix-seconds expiry from the mutually
/// exclusive `--expiry-secs` (relative to now) / `--expiry-at` (absolute)
/// flags. Exactly one is required, and the result must lie in the future — a
/// capability that expires at or before now is dead on arrival (a node stops
/// accepting its vouchers a margin before expiry, ADR 003 §Revocation).
fn resolve_expiry(
    now: u64,
    expiry_secs: Option<u64>,
    expiry_at: Option<u64>,
) -> anyhow::Result<u64> {
    let expiry = match (expiry_secs, expiry_at) {
        (Some(secs), None) => now.checked_add(secs).ok_or_else(|| {
            anyhow::anyhow!("--expiry-secs {secs} overflows the Unix epoch from now ({now})")
        })?,
        (None, Some(at)) => at,
        // clap `conflicts_with` rules out `(Some, Some)`; this leaves `(None, None)`.
        _ => anyhow::bail!("exactly one of --expiry-secs or --expiry-at is required"),
    };
    anyhow::ensure!(
        expiry > now,
        "capability expiry {expiry} is not in the future (now is {now}); it would be dead on \
         arrival — pick a later --expiry-at or a positive --expiry-secs"
    );
    Ok(expiry)
}

/// The capability-expiry margin of a node at the default redeem interval.
const DEFAULT_NODE_EXPIRY_MARGIN_SECS: u64 =
    capability_expiry_margin_secs(DEFAULT_REDEEM_INTERVAL_SECS);

/// Whether a capability's remaining lifetime at `now` lies within the
/// capability-expiry margin of a node at the default redeem interval. Such a node
/// refuses every voucher under it from the start (ADR 003 §Revocation). A node's
/// actual interval is unknown here, so `assign` only warns.
const fn expires_within_default_node_margin(expiry: u64, now: u64) -> bool {
    expiry.saturating_sub(now) <= DEFAULT_NODE_EXPIRY_MARGIN_SECS
}

/// Current Unix time in whole seconds.
///
/// # Errors
///
/// Errors if the system clock is before the Unix epoch. Defaulting to `0` there
/// would let `assign` pass its `expiry > now` guard trivially and print a token
/// that expired decades ago.
fn unix_now() -> anyhow::Result<u64> {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .map_err(|e| anyhow::anyhow!("the system clock is before the Unix epoch: {e}"))
}

/// How long `assign` waits for the chain head before it falls back to the
/// local clock.
const HEAD_READ_TIMEOUT: Duration = Duration::from_secs(5);

/// The clock `assign` checks a capability's expiry against: the chain head's
/// timestamp, which is the clock the contract applies at redemption. A local
/// clock that drifts would otherwise let `assign` print a token the chain
/// already treats as expired.
///
/// The read needs no keystore, so it runs before the unlock. Offline issuance
/// stays valid: when the head cannot be read within `timeout`, `assign` warns
/// and falls back to the local clock, so an endpoint that accepts and never
/// answers cannot stall it.
///
/// # Errors
///
/// Errors only on that fallback, when the system clock is before the Unix epoch.
async fn issuance_now(rpc_url: &str, timeout: Duration) -> anyhow::Result<u64> {
    let head = match provider::build_read_provider(rpc_url) {
        Ok(reader) => tokio::time::timeout(timeout, chain_ctx::head_timestamp(&reader))
            .await
            .unwrap_or_else(|_| Err(anyhow::anyhow!("no answer within {timeout:?}"))),
        Err(e) => Err(e),
    };
    match head {
        Ok(now) => Ok(now),
        Err(e) => {
            eprintln!(
                "warning: could not read the chain head time ({}); checking the capability \
                 expiry against the local clock instead",
                sanitize_err_chain(&e)
            );
            unix_now()
        }
    }
}

/// Render an absolute Unix-seconds expiry as `"<ts> (in Nd Nh Nm)"` relative to
/// `now`. The relative tail is what an operator actually reasons about; the raw
/// timestamp is kept for an exact, timezone-free reference.
fn format_expiry(expiry: u64, now: u64) -> String {
    let remaining = expiry.saturating_sub(now);
    let days = remaining / 86_400;
    let hours = (remaining % 86_400) / 3_600;
    let mins = (remaining % 3_600) / 60;
    format!("Unix {expiry} (in {days}d {hours}h {mins}m)")
}

/// `decdn pool assign`: issue an owner-signed spending capability delegating a
/// bounded spend on a pool the caller owns to a delegate `--signer` key, and
/// print the `dcap1:` token to hand to that delegated client (ADR 003
/// §Capability delegation).
///
/// The capability is signed offline with the owner keystore against the pool's
/// EIP-712 domain (`PaymentPool` address + chain id). Offline issuance is valid,
/// so an unreachable RPC only warns — the node is the one that enforces the
/// owner signature against the pool's on-chain owner at redemption time. An
/// owner read that *succeeds* is different evidence: a pool's owner is fixed at
/// open, so a different owner — or no such pool — proves the token is already
/// dead, and no token is printed.
async fn assign(args: &cli::PoolAssignArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
    let file = load_file_config(config_path)?;
    let chain = resolve_chain(&args.chain, &file)?;
    let pool_id = parse_pool_id(&args.pool)?;
    let signer_addr = super::chain_ctx::parse_nonzero_address(&args.signer, "--signer")?;

    let now = issuance_now(&chain.rpc_url, HEAD_READ_TIMEOUT).await?;
    let expiry = resolve_expiry(now, args.expiry_secs, args.expiry_at)?;
    if expires_within_default_node_margin(expiry, now) {
        eprintln!(
            "warning: the capability expires in {}s; a node with the default redeem interval \
             refuses vouchers within {DEFAULT_NODE_EXPIRY_MARGIN_SECS}s of a capability's \
             expiry, so such a node serves this delegate nothing",
            expiry.saturating_sub(now)
        );
    }

    let owner_signer = load_buyer_signer(&chain)?;
    let owner = owner_signer.address();
    let domain = voucher_domain(chain.chain_id, chain.payment_pool);
    let spending_cap = args.cap_micro_usdc;

    let capability = Capability {
        signer: signer_addr,
        spending_cap,
        pool_id,
        expiry,
    };
    let signed = capability.sign(&owner_signer, &domain)?;
    let grant = CapabilityGrant::from_signed_capability(&signed);
    let token = grant.to_token();

    // The owner the node will recover from the token — by construction this is
    // the keystore we just signed with, so it doubles as a codec round-trip
    // check before the token ever leaves the machine.
    let recovered = grant
        .owner(&domain)
        .map_err(|e| anyhow::anyhow!("re-recovering the owner from the fresh token failed: {e}"))?;

    // On-chain owner check. A read we could not perform only warns — offline
    // issuance is valid, and the serving node enforces the owner signature
    // against the pool's real owner at redemption. A read that *succeeded* and
    // disagrees is different evidence entirely, and fails: see
    // `ensure_on_chain_owner`.
    match provider::build_provider(&chain.rpc_url, &owner_signer) {
        Ok(rpc) => {
            let contract = PaymentPool::new(chain.payment_pool, rpc);
            ensure_on_chain_owner(&contract, pool_id, owner).await?;
        }
        Err(e) => eprintln!(
            "warning: could not build an RPC provider to check the on-chain owner of pool \
             {pool_id} ({}); issuing anyway — the node verifies the owner signature at redemption",
            sanitize_err_chain(&e)
        ),
    }

    println!("pool:         {pool_id}");
    println!("delegate:     {signer_addr} (the voucher-signing key this authorizes)");
    println!(
        "spending cap: {} USDC ({} µUSDC)",
        format_usdc_u256(U256::from(spending_cap)),
        args.cap_micro_usdc
    );
    println!("expiry:       {}", format_expiry(expiry, now));
    println!("owner:        {recovered} (recovered from the signature)");
    println!("Token (give this to the delegated client):");
    println!("{token}");
    Ok(())
}

/// Refuse to issue a capability the command has already proven dead, and warn
/// when it could not find out either way.
///
/// The two outcomes are different evidence, and it matters that they do not
/// share an exit code. A `getPool` that **succeeded** proves what the token is
/// worth: a pool's owner is fixed at open, so an owner that is not this keystore
/// means the token is rejected at redemption. Printing it and exiting 0 means
/// `decdn pool assign … > delegate.token` writes a file that looks valid and is
/// not, and nothing downstream finds out until a delegate's first voucher
/// bounces. A read that could not be performed proves nothing: offline issuance
/// is valid by design (`Assign` is signed entirely from the owner keystore), and
/// the serving node enforces the owner signature at redemption regardless — so
/// that leg degrades to a warning.
///
/// # Errors
///
/// Errors when the read succeeded and the pool is not owned by `expected` —
/// including a pool that does not exist on this contract.
async fn ensure_on_chain_owner<P>(
    contract: &PaymentPool::PaymentPoolInstance<P>,
    pool_id: PoolId,
    expected: Address,
) -> anyhow::Result<()>
where
    P: alloy::providers::Provider + Clone,
{
    match contract.getPool(pool_id).call().await {
        Ok(pool) => grade_on_chain_owner(pool.owner, expected).into_result(pool_id, expected),
        Err(e) => {
            eprintln!(
                "warning: could not read pool {pool_id} on-chain to confirm ownership ({}); \
                 issuing anyway — the node verifies the owner signature at redemption",
                sanitize_error_sources(&e)
            );
            Ok(())
        }
    }
}

/// The verdict [`ensure_on_chain_owner`] reaches on a read that succeeded.
///
/// Separated from the RPC call so the decision is unit-testable without a
/// provider, and so the two arms — "the chain says no" and "the chain did not
/// answer" — stay distinct.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum OwnerVerdict {
    /// The pool exists and this keystore owns it.
    Owned,
    /// No pool exists at this id on this contract. `getPool` zero-fills an
    /// unknown key instead of reverting, so this arrives as a successful read.
    NoSuchPool,
    /// The pool exists and this address owns it instead.
    OwnedByOther(Address),
}

/// Grade the `owner` a successful `getPool` returned against the issuing
/// keystore.
fn grade_on_chain_owner(on_chain: Address, expected: Address) -> OwnerVerdict {
    if on_chain.is_zero() {
        OwnerVerdict::NoSuchPool
    } else if on_chain == expected {
        OwnerVerdict::Owned
    } else {
        OwnerVerdict::OwnedByOther(on_chain)
    }
}

impl OwnerVerdict {
    /// Refuse to issue a capability this verdict has proven dead.
    ///
    /// Both failing arms name what the operator has to change, and they name
    /// different things: a wrong keystore is not a wrong `--pool`.
    ///
    /// # Errors
    ///
    /// Errors on every verdict except [`OwnerVerdict::Owned`].
    fn into_result(self, pool_id: PoolId, expected: Address) -> anyhow::Result<()> {
        match self {
            Self::Owned => Ok(()),
            Self::NoSuchPool => anyhow::bail!(
                "pool {pool_id} does not exist on this PaymentPool contract; a capability for a \
                 pool that was never opened is rejected at redemption, so no token was issued — \
                 check --pool, --payment-pool-address, and --chain-id"
            ),
            Self::OwnedByOther(on_chain) => anyhow::bail!(
                "pool {pool_id} on-chain owner {on_chain} is not this keystore's address \
                 {expected}; a capability signed by a non-owner is rejected at redemption, so no \
                 token was issued — sign with the owner keystore, or check --pool"
            ),
        }
    }
}

/// `decdn pool list` / `status`: read-only dump of the tracked buyer pools and
/// their per-lane voucher watermark.
///
/// Names the store it read, always. There are two, in the same directory and
/// with the same table format but no other relation: the CLI's own
/// `buyer-pools.redb` and a daemon's `buyer.redb`. Reading the first and
/// reporting it as the second makes `pools=0` meaningless on a node host
/// (#2078), so on a daemon's data dir this asks the daemon instead — redb holds
/// that file exclusively for the daemon's lifetime, so there is no disk path to
/// it while the node is up.
async fn list(args: &cli::PoolListArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
    let data_dir = match args.chain.data_dir.clone() {
        Some(dir) => expand_tilde(&dir),
        None => resolve_data_dir(None, &load_file_config(config_path)?)?,
    };

    if args.all {
        return list_all(args, config_path, &data_dir).await;
    }

    match classify_buyer_store(&data_dir)? {
        BuyerStoreOwner::Node { data_dir, .. } => {
            list_from_daemon(args, config_path, &data_dir).await
        }
        BuyerStoreOwner::Client { data_dir } => list_from_client_store(args, &data_dir),
    }
}

/// `decdn pool list --all`: every pool this keystore owns on chain.
///
/// The local listings answer "what does this store remember?". This answers
/// "what do I own?", and the two are different questions precisely when it
/// matters: a reset data dir loses the only local record of a funded deposit,
/// and the store-backed listing then prints nothing because the file it reads
/// is the file that went missing (#2072). The chain still has the pool.
///
/// Read-only, so unlike `close --all` / `reclaim --all` it is not refused on a
/// node's data dir. Those two write — they would close the pool the daemon is
/// paying from. This one looks.
///
/// The local view is consulted best-effort, and feeds three things: the
/// `TRACKED` column, the `local_store=` / `local_store_read` flag, and the
/// stderr warning naming rows that will not decode. An unreadable store leaves
/// the column unknown rather than failing the listing, because an unreadable
/// store is the case this exists for.
async fn list_all(
    args: &cli::PoolListArgs,
    config_path: Option<&Path>,
    data_dir: &Path,
) -> anyhow::Result<()> {
    let file = load_file_config(config_path)?;
    let chain = resolve_chain(&args.chain, &file)?;
    let signer = Arc::new(load_buyer_signer(&chain)?);
    let owner = signer.address();
    let rpc = provider::build_provider(&chain.rpc_url, &signer)?;
    let contract = PaymentPool::new(chain.payment_pool, rpc);

    let ids = enumerate_owned_pools(&contract, owner).await?;
    let tracked = tracked_pool_ids(args, config_path, data_dir).await;
    let now = chain_ctx::head_timestamp(contract.provider()).await?;

    let mut rows = Vec::with_capacity(ids.len());
    for pool_id in ids {
        let pool = contract.getPool(pool_id).call().await.map_err(|e| {
            anyhow::anyhow!("getPool({pool_id}) failed: {}", sanitize_error_sources(&e))
        })?;
        rows.push(ChainPoolRow {
            pool_id,
            status: pool.status,
            deposit: U256::from(pool.deposit),
            total_redeemed: U256::from(pool.totalRedeemed),
            dispute_deadline: pool.disputeDeadline,
            tracked: tracked.as_ref().and_then(|local| local.verdict(pool_id)),
        });
    }

    // The corrupt rows are named on stderr here too, so `--all` is not the one
    // listing that stays silent about a deposit stranded by an unreadable
    // record. stdout keeps only the table, for scripts.
    if let Some(local) = tracked.as_ref() {
        write_skipped_pools(&mut std::io::stderr().lock(), &local.undecodable())?;
    }

    let mut out = std::io::stdout().lock();
    if args.json {
        let view = PoolListAllJson {
            source: SOURCE_CHAIN,
            owner: format!("{owner:#x}"),
            chain_id: chain.chain_id,
            payment_pool: format!("{:#x}", chain.payment_pool),
            local_store_read: tracked.is_some(),
            pools: rows.iter().map(|r| ChainPoolJson::at(r, now)).collect(),
        };
        serde_json::to_writer_pretty(&mut out, &view)?;
        writeln!(out)?;
    } else {
        write_chain_pools(
            &mut out,
            owner,
            chain.chain_id,
            &rows,
            now,
            tracked.is_some(),
        )?;
    }
    Ok(())
}

/// The state of one pool's row in the local record.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum RowState {
    /// The row decoded.
    Decoded,
    /// A row exists and will not decode. The escrowed deposit is untracked
    /// until the record is repaired — a different remedy from an absent row, so
    /// it must not render as one.
    Undecodable,
}

/// What the local record says about the pools the chain reports.
///
/// One entry per pool the record mentions, because "the store has no row for
/// this pool" and "the store has a row it cannot decode" are different answers
/// and only the first means the deposit is untracked. An undecodable row is
/// still a row: `pool_id` is the table's primary key, so it survives whatever
/// corrupted the value bytes, and the store demonstrably holds that pool.
///
/// A map rather than two sets, so a pool cannot be both at once.
#[derive(Debug, Default)]
struct TrackedPools(std::collections::BTreeMap<PoolId, RowState>);

impl TrackedPools {
    /// Whether the local record tracks `pool_id`: `None` when a row exists but
    /// could not be decoded, so the honest answer for that one pool is unknown
    /// while every other pool's answer stays exact.
    fn verdict(&self, pool_id: PoolId) -> Option<bool> {
        match self.0.get(&pool_id) {
            Some(RowState::Decoded) => Some(true),
            Some(RowState::Undecodable) => None,
            None => Some(false),
        }
    }

    /// The pools with a row that will not decode, for the operator warning.
    fn undecodable(&self) -> Vec<PoolId> {
        self.0
            .iter()
            .filter(|(_, state)| **state == RowState::Undecodable)
            .map(|(id, _)| *id)
            .collect()
    }

    /// Project a running daemon's `admin_v1_pools` answer.
    ///
    /// `None` when any id on the wire will not parse. That is a format drift
    /// between the daemon's spelling and this binary's, and the failure mode of
    /// guessing is the worst one available: a dropped id renders as `no`, which
    /// tells an operator to close a pool the daemon is paying from right now.
    /// One unparseable id makes the whole answer untrustworthy, so it is
    /// reported as unknown rather than partially believed.
    fn from_wire(resp: &BuyerPoolsResponse) -> Option<Self> {
        let mut rows = std::collections::BTreeMap::new();
        for (id, state) in resp
            .pools
            .iter()
            .map(|p| (p.pool_id.as_str(), RowState::Decoded))
            .chain(
                resp.skipped
                    .iter()
                    .map(|p| (p.as_str(), RowState::Undecodable)),
            )
        {
            rows.insert(PoolId::from_str(id).ok()?, state);
        }
        Some(Self(rows))
    }
}

impl From<BuyerLoad> for TrackedPools {
    fn from(load: BuyerLoad) -> Self {
        Self(
            load.pools
                .iter()
                .map(|p| (p.pool_id, RowState::Decoded))
                .chain(
                    load.skipped
                        .into_iter()
                        .map(|id| (id, RowState::Undecodable)),
                )
                .collect(),
        )
    }
}

/// What the local record tracks, or `None` when it gave no usable answer.
///
/// `None` is a real answer, not a failure: it is what an operator sees when the
/// store is the thing that was lost, and rendering it as "unknown" beside the
/// chain's rows is more honest than reporting every pool untracked. It covers
/// the whole store — one undecodable row makes that one pool unknown, never the
/// others.
///
/// **Every read here is read-only.** `--all` sends no transaction and writes no
/// pool record, and it must not manufacture a store either: creating an empty
/// one and reporting its emptiness is the #2078 defect, and it would also make
/// a lost store — the condition this command exists to surface — indistinguish-
/// able from a store that tracks nothing.
///
/// A store that will not open is named on stderr rather than folded into the
/// column, because `?` alone cannot say whether to restore a backup, wait for
/// another process, or repair a record.
async fn tracked_pool_ids(
    args: &cli::PoolListArgs,
    config_path: Option<&Path>,
    data_dir: &Path,
) -> Option<TrackedPools> {
    let owner = match classify_buyer_store(data_dir) {
        Ok(owner) => owner,
        Err(err) => {
            let mut w = std::io::stderr().lock();
            let _ = writeln!(
                w,
                "warning: {err:#}; every TRACKED value below is unknown, which is not the same \
                 as untracked"
            );
            return None;
        }
    };
    match owner {
        BuyerStoreOwner::Client { data_dir } => read_local_pools(&client_buyer_db(&data_dir)),
        BuyerStoreOwner::Node { data_dir, .. } => {
            let buyer_db = node_buyer_db(&data_dir);
            match ReadOnlyBuyerPoolStore::open_file(&buyer_db) {
                Ok(reader) => load_local_pools(&reader, &buyer_db),
                // Only the lock means "a daemon holds this file"; that is the
                // one failure the admin RPC can answer instead. Every other
                // failure is about the file, and asking the daemon would report
                // it as an unknown column with no reason attached.
                Err(decdn_incentive::StoreError::AlreadyOpen { .. }) => {
                    daemon_pool_ids(args, config_path).await
                }
                Err(err) => {
                    warn_unreadable_store(&buyer_db, &err);
                    None
                }
            }
        }
    }
}

/// Read a buyer store at `path` without creating or locking it.
///
/// An absent store is an empty [`TrackedPools`], not an unknown: nothing is
/// tracked, which is exactly what a reset data dir should report against the
/// pools the chain still holds.
fn read_local_pools(path: &Path) -> Option<TrackedPools> {
    match ReadOnlyBuyerPoolStore::open_file(path) {
        Ok(reader) => load_local_pools(&reader, path),
        Err(decdn_incentive::StoreError::Absent { .. }) => Some(TrackedPools::default()),
        Err(err) => {
            warn_unreadable_store(path, &err);
            None
        }
    }
}

/// Hydrate an opened store, naming a read failure rather than silently
/// blanking the column.
fn load_local_pools(reader: &ReadOnlyBuyerPoolStore, path: &Path) -> Option<TrackedPools> {
    match reader.load_all() {
        Ok(load) => Some(TrackedPools::from(load)),
        Err(err) => {
            warn_unreadable_store(path, &err);
            None
        }
    }
}

/// Say on stderr why the `TRACKED` column is unknown.
///
/// The column has one glyph for every reason, so without this an operator
/// cannot tell a store another process holds from one that needs restoring
/// from backup. stdout keeps only the table, for scripts.
fn warn_unreadable_store(path: &Path, err: &decdn_incentive::StoreError) {
    let mut w = std::io::stderr().lock();
    let _ = writeln!(
        w,
        "warning: the local buyer store at {} could not be read ({err}); every TRACKED value \
         below is unknown, which is not the same as untracked",
        path.display()
    );
}

/// Ask a running daemon what it tracks. `None` on any failure — this feeds one
/// column, never the listing itself.
///
/// `skipped` crosses the admin wire as hex strings for exactly this reason: the
/// daemon cannot decode those rows either, and dropping them here would report
/// the pools as untracked.
async fn daemon_pool_ids(
    args: &cli::PoolListArgs,
    config_path: Option<&Path>,
) -> Option<TrackedPools> {
    if args.timeout_ms == 0 {
        // `list` rejects this outright; here the listing still stands, so say
        // what was skipped rather than leaving an unexplained column.
        let mut w = std::io::stderr().lock();
        let _ = writeln!(
            w,
            "warning: --timeout-ms 0 skips the daemon lookup (jsonrpsee reads a zero duration \
             as 'never'), so every TRACKED value below is unknown"
        );
        return None;
    }
    let url =
        crate::commands::node::resolve_admin_url(args.admin_url.as_deref(), config_path).ok()?;
    let client = jsonrpsee::http_client::HttpClientBuilder::default()
        .request_timeout(std::time::Duration::from_millis(args.timeout_ms))
        .build(&url)
        .ok()?;
    let resp: BuyerPoolsResponse = client.pools().await.ok()?;
    let tracked = TrackedPools::from_wire(&resp);
    if tracked.is_none() {
        let mut w = std::io::stderr().lock();
        let _ = writeln!(
            w,
            "warning: the daemon reported a pool id this binary cannot parse, so every TRACKED \
             value below is unknown. The daemon and this CLI disagree on the wire format — \
             check that both are the same build."
        );
    }
    tracked
}

/// Read the CLI's own `buyer-pools.redb` under `data_dir`.
fn list_from_client_store(args: &cli::PoolListArgs, data_dir: &Path) -> anyhow::Result<()> {
    let store_path = client_buyer_db(data_dir);
    let store = RedbBuyerPoolStore::open(data_dir)?;
    let load = store.load_all()?;
    write_store_listing(args, &store_path, SOURCE_CLIENT_STORE, "", load)
}

/// Render a [`BuyerLoad`] this process read out of a redb file itself.
///
/// Shared by the client store and the offline read of a stopped daemon's
/// store. `provenance` is appended to the `store=` line: the two reads produce
/// the same row shape from different files, so which file — and whether a
/// daemon was running — must be on the output, not inferred.
fn write_store_listing(
    args: &cli::PoolListArgs,
    store_path: &Path,
    source: &'static str,
    provenance: &str,
    load: BuyerLoad,
) -> anyhow::Result<()> {
    let BuyerLoad {
        mut pools,
        mut skipped,
    } = load;
    // Stable output regardless of the store's internal key order.
    pools.sort_by_key(|p| p.pool_id);
    skipped.sort_unstable();
    write_skipped_pools(&mut std::io::stderr().lock(), &skipped)?;

    let mut out = std::io::stdout().lock();
    if args.json {
        let view = PoolListJson {
            store: store_path.display().to_string(),
            source,
            pools: pools.iter().map(PoolJson::from).collect(),
            skipped: skipped.iter().map(|p| format!("{p:#x}")).collect(),
        };
        serde_json::to_writer_pretty(&mut out, &view)?;
        writeln!(out)?;
    } else {
        writeln!(out, "store={}{provenance}", store_path.display())?;
        write_pools(&mut out, &pools, &skipped)?;
    }
    Ok(())
}

/// Read a node's `buyer.redb` — through `admin_v1_pools` while the daemon runs,
/// and off disk when it does not.
///
/// The fallback is never the *client* store: that is a different file with
/// unrelated contents, and printing its `pools=0` here is precisely the failure
/// this path exists to prevent. It is the daemon's own file, read-only.
///
/// Only a refused connection takes the offline route, because it is the one
/// failure that positively means nothing is listening. A `Call` error means the
/// daemon ANSWERED and refused, so it is up. Every other transport error and
/// every timeout is ambiguous — a dropped packet, a firewall, a blackholing
/// port — and an ambiguous signal must not be resolved by opening a file a
/// running daemon may own.
///
/// On the refused path the lock settles it: `redb` holds its process-exclusive
/// lock for the lifetime of an open `Database`, so a read-only open succeeding
/// proves no daemon holds the file, and `AlreadyOpen` proves one does — which
/// makes a refused admin port an admin-URL problem rather than a stopped node.
async fn list_from_daemon(
    args: &cli::PoolListArgs,
    config_path: Option<&Path>,
    data_dir: &Path,
) -> anyhow::Result<()> {
    let buyer_db = node_buyer_db(data_dir);
    anyhow::ensure!(
        args.timeout_ms > 0,
        "--timeout-ms must be > 0 (jsonrpsee treats Duration::ZERO as \
         'never' rather than 'sub-millisecond deadline')"
    );
    let url = crate::commands::node::resolve_admin_url(args.admin_url.as_deref(), config_path)?;
    let client = jsonrpsee::http_client::HttpClientBuilder::default()
        .request_timeout(std::time::Duration::from_millis(args.timeout_ms))
        .build(&url)
        .with_context(|| format!("failed to build admin JSON-RPC client for {url}"))?;

    let resp: BuyerPoolsResponse = match client.pools().await {
        Ok(resp) => resp,
        // Connection refused: nothing is listening, so the daemon is very
        // likely down and its store readable. Try it before reporting.
        Err(jsonrpsee::core::client::Error::Transport(inner))
            if crate::commands::node::is_connection_refused(inner.as_ref()) =>
        {
            return list_from_stopped_daemon(args, data_dir, &buyer_db, &url, inner.as_ref());
        }
        Err(err) => {
            return Err(unreachable_daemon_error(
                args, data_dir, &buyer_db, &url, err,
            ));
        }
    };

    let mut out = std::io::stdout().lock();
    if args.json {
        let view = DaemonPoolListJson {
            store: buyer_db.display().to_string(),
            source: SOURCE_DAEMON,
            pools: &resp.pools,
            skipped: &resp.skipped,
        };
        serde_json::to_writer_pretty(&mut out, &view)?;
        writeln!(out)?;
    } else {
        write_skipped_buyer_pools(&mut std::io::stderr().lock(), &resp.skipped)?;
        writeln!(
            out,
            "store={} (read from the running daemon)",
            buyer_db.display()
        )?;
        write_buyer_pools(&mut out, &resp)?;
    }
    Ok(())
}

/// The error for an admin call that produced no pools, explaining which file
/// the answer would have come from.
fn unreachable_daemon_error(
    args: &cli::PoolListArgs,
    data_dir: &Path,
    buyer_db: &Path,
    url: &str,
    err: jsonrpsee::core::client::Error,
) -> anyhow::Error {
    {
        // A `Call` error means the daemon ANSWERED and refused — it is up, so
        // telling the operator to start it would send them the wrong way. Only
        // the transport and timeout arms warrant that advice.
        let answered = matches!(err, jsonrpsee::core::client::Error::Call(_));
        let classified = crate::commands::node::classify_client_error(url, args.timeout_ms, err);
        if answered {
            classified.context(format!(
                "{} is a decdn-node data dir, so its buyer pools live in {}, which this command \
                 reads through the daemon — a running daemon holds that file exclusively. The \
                 daemon answered and refused the read.",
                data_dir.display(),
                buyer_db.display(),
            ))
        } else {
            classified.context(format!(
                "{} is a decdn-node data dir, so the pools that matter are in {} — which a \
                 running daemon holds exclusively and no other process can open. This command \
                 therefore asks the daemon, and the daemon did not answer. Start decdn-node, or \
                 pass --data-dir <client dir> to inspect a client store instead.",
                data_dir.display(),
                buyer_db.display(),
            ))
        }
    }
}

/// Read a stopped daemon's `buyer.redb` off disk, or say why that is not what
/// is happening.
///
/// Three outcomes, and the third is why this is not simply a fallback:
///
/// - the file opens read-only, so no daemon holds it: render it, labelled as a
///   disk read, so provenance stays on the output.
/// - the file is write-locked, so a daemon IS running: the admin URL is wrong,
///   not the node down. Say that instead — it is the answer the operator needs.
/// - anything else: the original unreachable-daemon error, with what the disk
///   read found appended, because "no daemon answered AND the store will not
///   open" is two facts, not one.
fn list_from_stopped_daemon(
    args: &cli::PoolListArgs,
    data_dir: &Path,
    buyer_db: &Path,
    url: &str,
    transport: &(dyn std::error::Error + Send + Sync + 'static),
) -> anyhow::Result<()> {
    match ReadOnlyBuyerPoolStore::open_file(buyer_db) {
        Ok(reader) => {
            let load = reader.load_all()?;
            write_store_listing(
                args,
                buyer_db,
                SOURCE_NODE_STORE_OFFLINE,
                " (read from disk; no daemon running)",
                load,
            )
        }
        Err(decdn_incentive::StoreError::AlreadyOpen { .. }) => anyhow::bail!(
            "admin at {url} refused the connection ({transport}), but a process holds {} \
             exclusively — so a decdn-node IS running against {}. The admin URL or admin_port \
             is wrong, not the node down. Pass --admin-url, or check admin_port in the node's \
             config.",
            buyer_db.display(),
            data_dir.display(),
        ),
        // A crashed daemon is both unreachable AND unrepaired, so this is the
        // common post-mortem case, not a footnote on the generic failure. The
        // remedy names the process that owns the file, which the store cannot.
        Err(decdn_incentive::StoreError::NeedsRepair { .. }) => anyhow::bail!(
            "admin at {url} refused the connection ({transport}), and {} was not shut down \
             cleanly — it needs a repair pass that only a writable open can run. Start \
             decdn-node once against {} to repair it, then read it again.",
            buyer_db.display(),
            data_dir.display(),
        ),
        // No store at all is an answer, not a failure to report: this node has
        // never opened a pool, or the store was deleted to force re-adoption.
        Err(decdn_incentive::StoreError::Absent { .. }) => anyhow::bail!(
            "admin at {url} refused the connection ({transport}), and {} does not exist — this \
             node holds no buyer pools on disk. It has not opened one, or the store was deleted \
             to force re-adoption. Start decdn-node and read it again, or run `decdn pool list \
             --all` to see what this keystore owns on chain.",
            buyer_db.display(),
        ),
        Err(store_err) => Err(anyhow::anyhow!(
            "admin at {url} refused the connection ({transport}); is the node running, and is \
             admin_port configured correctly? Reading {} from disk instead did not work either: \
             {store_err}",
            buyer_db.display(),
        )
        .context(format!(
            "{} is a decdn-node data dir, so the pools that matter are in {} — not in a client \
             store. Pass --data-dir <client dir> to inspect a client store instead.",
            data_dir.display(),
            buyer_db.display(),
        ))),
    }
}

/// One pool as the chain describes it, plus whether the local record has it.
///
/// No `Debug`: the `sol!`-generated `Status` has none. [`status_label`] is the
/// spelling anything user-facing wants anyway.
struct ChainPoolRow {
    /// On-chain `poolId`.
    pool_id: PoolId,
    /// Lifecycle state from `getPool`.
    status: PaymentPool::Status,
    /// Escrowed deposit, in micro-USDC.
    deposit: U256,
    /// Cumulative amount redeemed against the pool, in micro-USDC.
    total_redeemed: U256,
    /// Absolute Unix deadline after which a `Closing` pool is reclaimable. `0`
    /// while the pool is still `Open`.
    dispute_deadline: u64,
    /// Whether the local record tracks this pool; `None` when it gave no usable
    /// answer for this pool. See [`TrackedPools::verdict`].
    tracked: Option<bool>,
}

/// Wire spelling of a `getPool` status, matching the lowercase vocabulary the
/// rest of the `pool` output uses.
pub(crate) const fn status_label(status: PaymentPool::Status) -> &'static str {
    match status {
        PaymentPool::Status::Open => "open",
        PaymentPool::Status::Closing => "closing",
        PaymentPool::Status::Closed => "closed",
        // `sol!` enums carry a hidden invalid variant.
        _ => "?",
    }
}

/// When the pool's residual can be reclaimed, in the operator's terms.
fn reclaimable_label(row: &ChainPoolRow, now: u64) -> String {
    match plan_reclaim(row.status, row.dispute_deadline, now) {
        ReclaimPlan::Reclaim => "now".to_string(),
        ReclaimPlan::SkipOpen => "close it first".to_string(),
        ReclaimPlan::SkipInWindow(deadline) => format_expiry(deadline, now),
        ReclaimPlan::SkipClosed => "already reclaimed".to_string(),
        ReclaimPlan::SkipUnknown => "?".to_string(),
    }
}

/// `yes` / `no` / `?`, where `?` means the local record gave no usable answer
/// for this pool — the whole store was unreadable, or it holds a row for this
/// pool that will not decode. Neither is the claim "this pool is untracked",
/// and the remedies differ: one is a lost store, the other a record to repair.
const fn tracked_label(tracked: Option<bool>) -> &'static str {
    match tracked {
        Some(true) => "yes",
        Some(false) => "no",
        None => "?",
    }
}

/// Render `--all` as the aligned table.
///
/// Same conventions as [`write_pools`]: a `key=value` summary line first so
/// scripts can grep without `--json`, a sentinel when there is nothing, and the
/// one variable-width column last and unpadded.
fn write_chain_pools(
    w: &mut impl Write,
    owner: Address,
    chain_id: u64,
    rows: &[ChainPoolRow],
    now: u64,
    local_store_read: bool,
) -> std::io::Result<()> {
    // `local_store=` is what separates a column of `?` that means "no local
    // answer at all" from one that means "these rows will not decode". Without
    // it the table is silently lossy about the one condition `--all` exists to
    // surface, and only `--json` carries the distinction.
    writeln!(
        w,
        "owner={owner:#x} chain_id={chain_id} pools={} local_store={}",
        rows.len(),
        if local_store_read {
            "read"
        } else {
            "unreadable"
        },
    )?;
    if rows.is_empty() {
        writeln!(w, "(this keystore owns no pools on this contract)")?;
        return Ok(());
    }
    writeln!(
        w,
        "{:<14} {:<8} {:>14} {:>14} {:>7} RECLAIMABLE",
        "POOL", "STATUS", "DEPOSIT", "REDEEMED", "TRACKED"
    )?;
    for row in rows {
        writeln!(
            w,
            "{:<14} {:<8} {:>14} {:>14} {:>7} {}",
            short_hex(&format!("{:#x}", row.pool_id)),
            status_label(row.status),
            format_usdc_u256(row.deposit),
            format_usdc_u256(row.total_redeemed),
            tracked_label(row.tracked),
            reclaimable_label(row, now),
        )?;
    }
    Ok(())
}

/// `--all` as JSON.
///
/// A third shape beside [`PoolListJson`] and [`DaemonPoolListJson`], and
/// deliberately so: these rows come from the chain, carry lifecycle fields the
/// stores do not hold, and carry no lanes. Read `source` before `pools`.
#[derive(Serialize)]
struct PoolListAllJson {
    /// Always [`SOURCE_CHAIN`].
    source: &'static str,
    /// The keystore address the pools were enumerated by.
    owner: String,
    /// Chain id the contract was read on.
    chain_id: u64,
    /// `PaymentPool` address the pools live in.
    payment_pool: String,
    /// Whether the local record could be read at all. When `false`, every
    /// pool's `tracked` is `null` and says nothing about the pool. When `true`,
    /// a `null` `tracked` is specific to that pool: its row will not decode.
    local_store_read: bool,
    /// Every pool the owner holds, in the order the contract enumerates them.
    pools: Vec<ChainPoolJson>,
}

/// One `--all` pool.
#[derive(Serialize)]
struct ChainPoolJson {
    /// On-chain `poolId`, `0x`-prefixed hex.
    pool_id: String,
    /// `open`, `closing`, `closed`, or `?`.
    status: &'static str,
    /// Escrowed deposit, in micro-USDC.
    deposit_micro_usdc: String,
    /// Cumulative redeemed amount, in micro-USDC.
    total_redeemed_micro_usdc: String,
    /// Absolute Unix deadline after which the residual is reclaimable; `0`
    /// while the pool is `Open`.
    dispute_deadline: u64,
    /// True once the dispute window has elapsed and the pool is `Closing`.
    reclaimable_now: bool,
    /// Whether the local record tracks this pool; `null` when the record gave
    /// no usable answer — an unreadable store, or a row for this pool that will
    /// not decode. Read `local_store_read` to tell the two apart.
    tracked: Option<bool>,
}

impl ChainPoolJson {
    /// Project a row as of `now`, so the time-dependent field is computed once
    /// against the same clock the table uses.
    fn at(row: &ChainPoolRow, now: u64) -> Self {
        Self {
            pool_id: format!("{:#x}", row.pool_id),
            status: status_label(row.status),
            deposit_micro_usdc: row.deposit.to_string(),
            total_redeemed_micro_usdc: row.total_redeemed.to_string(),
            dispute_deadline: row.dispute_deadline,
            reclaimable_now: matches!(
                plan_reclaim(row.status, row.dispute_deadline, now),
                ReclaimPlan::Reclaim
            ),
            tracked: row.tracked,
        }
    }
}

/// `source` value for a listing read directly from the CLI's own store.
const SOURCE_CLIENT_STORE: &str = "client_store";
/// `source` value for a listing the running daemon answered.
const SOURCE_DAEMON: &str = "daemon";
/// `source` value for a listing read off a stopped daemon's own store file.
///
/// Its own value, never [`SOURCE_CLIENT_STORE`], even though the row shape is
/// identical: the two come from different files with unrelated contents, and
/// conflating them is the whole defect this area exists to prevent.
const SOURCE_NODE_STORE_OFFLINE: &str = "node_store_offline";
/// `source` value for the chain-authoritative `--all` listing.
const SOURCE_CHAIN: &str = "chain";

/// Warn about every undecodable buyer row on stderr. The `pool_id` (the store's
/// primary key) is the only available repair handle.
fn write_skipped_pools(w: &mut impl Write, skipped: &[PoolId]) -> std::io::Result<()> {
    for pool_id in skipped {
        writeln!(
            w,
            "warning: buyer pool {pool_id:#x} could not be decoded; its deposit remains \
             escrowed but untracked until the record is repaired"
        )?;
    }
    Ok(())
}

/// Render the tracked buyer pools as an aligned table. Pure (writes to any
/// sink) so the layout is unit-testable without a store.
fn write_pools(
    w: &mut impl Write,
    pools: &[BuyerPoolState],
    skipped: &[PoolId],
) -> std::io::Result<()> {
    writeln!(w, "pools={}", pools.len())?;
    if pools.is_empty() {
        if skipped.is_empty() {
            writeln!(w, "(no tracked pools)")?;
        }
        return Ok(());
    }
    writeln!(
        w,
        "{:<14} {:<14} {:<14} {:>14} {:>8}",
        "POOL", "OWNER", "TOKEN", "DEPOSIT", "LANES"
    )?;
    for p in pools {
        writeln!(
            w,
            "{:<14} {:<14} {:<14} {:>14} {:>8}",
            short_hex(&format!("{:#x}", p.pool_id)),
            short_hex(&format!("{:#x}", p.owner)),
            short_hex(&format!("{:#x}", p.token)),
            format_usdc_u256(p.deposit),
            p.lane_count(),
        )?;
    }
    Ok(())
}

/// Render a daemon's buyer-pool snapshot as an aligned table, in the same
/// columns [`write_pools`] uses for the client store, so an operator reading
/// both sees the same shape. Pure (writes to any sink) so the layout is
/// unit-testable without an admin hop.
///
/// Addresses arrive EIP-55 checksummed from the wire, where the client path
/// renders them lowercase — the columns match, the casing does not.
///
/// Writes the table only. Undecodable rows go to stderr through
/// [`write_skipped_buyer_pools`], as they do on the client path, so a script
/// capturing stdout does not find warning lines ahead of `pools=`.
///
/// Shared by `decdn node pools` and by `decdn pool list` when it routes to a
/// running daemon.
pub(crate) fn write_buyer_pools(
    w: &mut impl Write,
    resp: &BuyerPoolsResponse,
) -> std::io::Result<()> {
    writeln!(w, "pools={}", resp.pools.len())?;
    if resp.pools.is_empty() {
        if resp.skipped.is_empty() {
            writeln!(w, "(no tracked pools)")?;
        }
        return Ok(());
    }
    writeln!(
        w,
        "{:<14} {:<14} {:<14} {:>14} {:>8}",
        "POOL", "OWNER", "TOKEN", "DEPOSIT", "LANES"
    )?;
    for p in &resp.pools {
        writeln!(
            w,
            "{:<14} {:<14} {:<14} {:>14} {:>8}",
            short_hex(&p.pool_id),
            short_hex(&p.owner),
            short_hex(&p.token),
            format_usdc_u256(U256::from(p.deposit_micro_usdc)),
            p.lanes.len(),
        )?;
    }
    Ok(())
}

/// Warn about every undecodable row in a daemon's answer, on stderr — the
/// stream [`write_skipped_pools`] uses for the client store's equivalent.
pub(crate) fn write_skipped_buyer_pools(
    w: &mut impl Write,
    skipped: &[String],
) -> std::io::Result<()> {
    for pool_id in skipped {
        writeln!(
            w,
            "warning: buyer pool {pool_id} could not be decoded; its deposit remains escrowed but \
             untracked until the record is repaired"
        )?;
    }
    Ok(())
}

/// Top-level `--json` document for a listing of the CLI's own store.
///
/// `store` and `source` are the half a machine consumer needs to know *which*
/// buyer store answered: the same command on the same host can read either
/// file, and they are unrelated (#2078).
#[derive(Serialize)]
struct PoolListJson {
    store: String,
    source: &'static str,
    pools: Vec<PoolJson>,
    skipped: Vec<String>,
}

/// Top-level `--json` document for a listing the daemon answered.
///
/// Borrows the admin DTOs rather than re-shaping them, so the wire types stay
/// the single definition of the daemon's answer. The consequence is that the
/// two sources do NOT emit the same pool objects: this one carries
/// `deposit_micro_usdc` as a number and EIP-55 addresses, where a client-store
/// listing carries `deposit_usdc` as a decimal string and lowercase addresses.
/// Read `source` before parsing `pools`.
#[derive(Serialize)]
struct DaemonPoolListJson<'a> {
    store: String,
    source: &'static str,
    pools: &'a [decdn_common::admin::BuyerPoolSnapshot],
    skipped: &'a [String],
}

/// Serializable view for `--json`, including the per-lane watermark so a
/// programmatic consumer sees the same fan-out the table's `LANES` count only
/// summarizes.
#[derive(Serialize)]
struct PoolJson {
    pool_id: String,
    chain_id: u64,
    payment_pool: String,
    owner: String,
    token: String,
    deposit_usdc: String,
    lanes: Vec<LaneJson>,
}

#[derive(Serialize)]
struct LaneJson {
    signer: String,
    provider: String,
    last_amount_usdc: String,
    last_bytes_delivered: String,
}

impl From<&BuyerPoolState> for PoolJson {
    fn from(p: &BuyerPoolState) -> Self {
        Self {
            pool_id: format!("{:#x}", p.pool_id),
            chain_id: p.deployment.chain_id,
            payment_pool: format!("{:#x}", p.deployment.payment_pool),
            owner: format!("{:#x}", p.owner),
            token: format!("{:#x}", p.token),
            deposit_usdc: format_usdc_u256(p.deposit),
            lanes: p
                .lanes()
                .map(|(lane, progress)| LaneJson {
                    signer: format!("{:#x}", lane.signer),
                    provider: format!("{:#x}", lane.provider),
                    last_amount_usdc: format_usdc_u256(progress.last_amount),
                    last_bytes_delivered: progress.last_bytes.to_string(),
                })
                .collect(),
        }
    }
}

/// Format a `U256` micro-USDC amount as `"N.NNNNNN"`. USDC has 6 decimals
/// (ADR 003), so 1 USDC == `1_000_000` micro-USDC.
fn format_usdc_u256(micro: U256) -> String {
    let scale = U256::from(1_000_000u64);
    let whole = micro / scale;
    // `micro % scale < 1_000_000`, so the `u64` narrowing is always exact.
    let frac = (micro % scale).to::<u64>();
    format!("{whole}.{frac:06}")
}

/// Abbreviate a `0x`-prefixed hex string to `0x` + the first 10 nibbles + `…`
/// for table display; short inputs pass through unchanged. Uses `get(..)` to
/// stay within the workspace `indexing_slicing` clippy denial.
fn short_hex(hex: &str) -> String {
    match hex.get(..12) {
        Some(prefix) if hex.len() > 12 => format!("{prefix}…"),
        _ => hex.to_string(),
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests;