bsv-wallet-cli 0.13.0

//! The CLI as a host of the transaction tracker (`bsv-tracker`).
//!
//! The tracker holds one word per transaction, the evidence behind it, the
//! hints it has heard and the next re-ask. Only headers and proofs change a
//! word; every broadcaster word is a hint that schedules a re-ask. The crate
//! fetches nothing, reads no clock and stores nothing: the host supplies the
//! four things below, runs the re-asks, and keeps the state.
//!
//! # The four host things
//!
//! | the tracker's trait | here | what it is |
//! |---|---|---|
//! | `Headers` | [`HeaderSnapshot`] | the header service's tip and the headers at the heights a pass needs, read once per pass through the wallet's services; a height not read is "no answer", a fault is a fault |
//! | `ProofFetcher` | [`Courier`] | `get_merkle_path` for one named transaction; what it carries is a raw path, checked by the tracker against the snapshot |
//! | `HintSource` | [`MemoryHints`] | the broadcast memory rows of the wallet's storage (what a broadcaster answered, what Arcade pushed over SSE or the webhook), each word fed once |
//! | `Clock` | [`SystemClock`] | seconds since the epoch |
//!
//! The tracker's traits are synchronous and the wallet's transports are
//! not, so a pass reads first (headers, memory rows, proofs for the re-asks
//! that are due) and steps second.
//!
//! # The state, in the CLI's storage
//!
//! One row per tracked transaction in `tracker_states`, in the wallet's own
//! database. A row holds inputs, never a trusted word: whether the host
//! built it, the hints in the order heard, and the raw merkle path. Loading
//! a row replays those inputs, so a stored path is checked again against
//! the active header every time it is read (the tracker has no
//! deserializer for a checked proof, by design). The `word`, `height` and
//! `reask` columns are what the tracker said at the last step, kept for
//! reading and for the indexes; nothing is decided from them.
//!
//! # What a pass asks, and what it never asks
//!
//! A pass ([`tick`]) names transactions; it never walks the chain and
//! never asks a chain index. It loads the rows that are not yet mined, feeds
//! the new hints, ticks the clock, and runs the re-asks that are due: one
//! proof request for one transaction, with a doubling pause between two
//! requests for the same transaction. A mined row is read again only when
//! the header at or below its height moved (the last twelve tips are kept
//! to see that) or when a spend names it ([`spend_guard`]).
//!
//! # What the header service pushes
//!
//! The daemon's chain-event feed ([`crate::chain_feed`]) hands each
//! envelope to [`deliver_chain_event`]: the rows the event names by index
//! (the mined rows at or above a fork, reorg or split height, the rows
//! marked by an earlier event for a tip, the rows on an invalidated
//! block's hash) are loaded into a [`Tracker`], and `Tracker::on_chain`
//! moves their words. An event that left a row anything but a plain
//! checked `mined` is kept on the row as one more input to replay (the
//! `chain` column), so a `stale` or a suspect mark survives the next load;
//! a new proof, or a stored path dropped because it no longer checks,
//! clears it. The cursor is stored in the same transaction.
//!
//! One re-ask is an announce, not a proof request: a transaction a
//! broadcaster called not final (a forced `gift-claim` before its lock
//! time) is kept with its BEEF and handed to the broadcasters again at the
//! lock time ([`heard`], and the re-announce step of [`tick`]).

use std::collections::{BTreeMap, HashMap, VecDeque};
use std::convert::Infallible;

use anyhow::Result;
use bsv_sdk::transaction::MerklePath;
use bsv_sdk::wallet::{InternalizeActionArgs, WalletInterface};
use bsv_tracker::{
    decode_envelope, ChainEnvelope, ChainEvent, CheckError, Clock, Evidence, Header, HeaderError,
    Headers, Height, Hint, HintSource, HintStatus, HostAction, Input, Params, Proof, ProofFetcher,
    Reask, State, Timestamp, Tracker, TxId, Word,
};
use bsv_wallet_toolbox::services::PostBeefResult;
use bsv_wallet_toolbox::{BroadcastStatus, StorageSqlx, WalletServices};
use serde::{Deserialize, Serialize};
use sqlx::Row as _;

/// Default seconds without a new hint before an announced or seen
/// transaction is asked for its proof (`TRACKER_AGE_SECS`): one block
/// interval.
pub const DEFAULT_AGE_SECS: u64 = 600;
/// Default re-asks one pass runs (`TRACKER_MAX_ASKS`).
pub const DEFAULT_MAX_ASKS: usize = 20;
/// The pause between two re-asks of one transaction doubles up to this
/// many times (600 s, then 1200 s, up to 64 times the age threshold).
const MAX_BACKOFF_DOUBLINGS: u32 = 6;
/// A transaction asked this many times in a row with no proof is not asked
/// again until a new hint names it (about five days at the default age).
const MAX_ASKS_PER_TX: u32 = 16;
/// Recent tips kept to see a moved header.
const TIP_RING: usize = 12;
/// Default spend depth (`SPEND_DEPTH`), and its floor: a coin is offered
/// when the block holding its transaction is at or below the tip.
pub const DEFAULT_SPEND_DEPTH: Height = 1;

// ---------------------------------------------------------------------------
// Headers
// ---------------------------------------------------------------------------

/// One pass's view of the active chain: the tip and the headers at the
/// heights the pass named, read from the header service through the
/// wallet's services. Fails closed: a height that was not read has no
/// answer, a read that failed is a fault, and with no tip nothing checks.
#[derive(Debug, Clone)]
pub struct HeaderSnapshot {
    tip: std::result::Result<(Height, String), String>,
    headers: BTreeMap<Height, std::result::Result<Header, String>>,
}

impl HeaderSnapshot {
    /// Read the tip and the headers at `heights`.
    pub async fn take<V: WalletServices + ?Sized>(
        services: &V,
        heights: impl IntoIterator<Item = Height>,
    ) -> Self {
        let tip = services
            .get_chain_tip_header()
            .await
            .map(|h| (h.height, h.hash.to_ascii_lowercase()))
            .map_err(|e| e.to_string());
        let mut snapshot = Self {
            tip,
            headers: BTreeMap::new(),
        };
        snapshot.extend(services, heights).await;
        snapshot
    }

    /// A snapshot of nothing: no tip, no header. Every check fails closed.
    pub fn unread() -> Self {
        Self {
            tip: Err("the header service was not read".to_string()),
            headers: BTreeMap::new(),
        }
    }

    /// Read the headers at `heights` that this snapshot does not hold yet.
    pub async fn extend<V: WalletServices + ?Sized>(
        &mut self,
        services: &V,
        heights: impl IntoIterator<Item = Height>,
    ) {
        for height in heights {
            if self.headers.contains_key(&height) {
                continue;
            }
            let answer = match services.get_header_for_height(height).await {
                Ok(bytes) => header_of(&bytes),
                Err(e) => Err(e.to_string()),
            };
            self.headers.insert(height, answer);
        }
    }

    /// The tip's height and hash, when the header service gave one.
    pub fn tip(&self) -> Option<(Height, &str)> {
        self.tip.as_ref().ok().map(|(h, hash)| (*h, hash.as_str()))
    }
}

impl Headers for HeaderSnapshot {
    fn header_at(&self, height: Height) -> std::result::Result<Option<Header>, HeaderError> {
        match self.headers.get(&height) {
            Some(Ok(header)) => Ok(Some(header.clone())),
            Some(Err(fault)) => Err(HeaderError(fault.clone())),
            None => Ok(None),
        }
    }

    fn tip_height(&self) -> std::result::Result<Height, HeaderError> {
        self.tip
            .as_ref()
            .map(|(height, _)| *height)
            .map_err(|fault| HeaderError(fault.clone()))
    }
}

/// The tracker's header projection of an 80-byte block header: its hash
/// and its merkle root, both as displayed.
fn header_of(bytes: &[u8]) -> std::result::Result<Header, String> {
    if bytes.len() != 80 {
        return Err(format!("a header of {} bytes", bytes.len()));
    }
    let display = |raw: &[u8]| {
        let mut reversed = raw.to_vec();
        reversed.reverse();
        hex::encode(reversed)
    };
    Ok(Header {
        hash: display(&bsv_sdk::primitives::hash::sha256d(bytes)),
        merkle_root: display(&bytes[36..68]),
    })
}

// ---------------------------------------------------------------------------
// Clock
// ---------------------------------------------------------------------------

/// The host's clock: seconds since the epoch.
#[derive(Debug, Clone, Copy, Default)]
pub struct SystemClock;

impl Clock for SystemClock {
    fn now(&self) -> Timestamp {
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0)
    }
}

// ---------------------------------------------------------------------------
// Hints
// ---------------------------------------------------------------------------

/// The hints of one pass: the broadcast memory rows of the tracked
/// transactions whose `(provider, word)` the row has not heard yet. A word
/// a broadcaster repeats every minute is one hint, so a refreshed `seen`
/// never resets the age clock.
#[derive(Debug, Default)]
pub struct MemoryHints {
    queue: VecDeque<(TxId, Hint)>,
}

impl MemoryHints {
    async fn read(storage: &StorageSqlx, rows: &[TrackedRow]) -> Result<Self> {
        let txids: Vec<String> = rows.iter().map(|r| r.txid.clone()).collect();
        if txids.is_empty() {
            return Ok(Self::default());
        }
        let mut records = storage.broadcast_records(None, &txids).await?;
        records.sort_by_key(|r| r.seen_at);
        let mut queue = VecDeque::new();
        for record in records {
            let Some(status) = record.ladder_status() else {
                continue;
            };
            let txid = record.txid.to_ascii_lowercase();
            let Some(row) = rows.iter().find(|r| r.txid == txid) else {
                continue;
            };
            // A broadcaster's "mined" carries no height here and is no
            // proof: it reads as a word that disagrees and asks for one.
            let (label, status) = match status {
                BroadcastStatus::Accepted => ("accepted", HintStatus::Accepted),
                BroadcastStatus::Seen => ("seen", HintStatus::Seen),
                BroadcastStatus::Mined => ("mined", HintStatus::Unknown),
                BroadcastStatus::Rejected => (
                    "rejected",
                    HintStatus::Rejected {
                        reason: record.status.clone(),
                    },
                ),
                BroadcastStatus::Unknown => ("unknown", HintStatus::Unknown),
            };
            let source = format!("{}|{}", record.provider, label);
            if row.hints.iter().any(|h| h.source == source) {
                continue;
            }
            let observed = record.seen_at.timestamp().max(0) as Timestamp;
            queue.push_back((txid, Hint::new(source, status, observed)));
        }
        Ok(Self { queue })
    }
}

impl HintSource for MemoryHints {
    type Error = Infallible;

    fn next_hint(&mut self) -> std::result::Result<Option<(TxId, Hint)>, Self::Error> {
        Ok(self.queue.pop_front())
    }
}

// ---------------------------------------------------------------------------
// Proofs
// ---------------------------------------------------------------------------

/// The host's proof transport: `get_merkle_path` for the transactions whose
/// re-ask is due, read before the tracker steps. What it hands over is a
/// raw path bound to its txid; the tracker checks it.
#[derive(Debug, Default)]
pub struct Courier {
    carried: HashMap<TxId, std::result::Result<Option<Proof>, String>>,
}

impl Courier {
    /// Ask for the proof of each of `txids`, once.
    pub async fn carry<V: WalletServices + ?Sized>(services: &V, txids: &[TxId]) -> Self {
        let mut carried = HashMap::new();
        for txid in txids {
            let answer = match services.get_merkle_path(txid, false).await {
                Ok(found) => match found.merkle_path {
                    Some(hex) => MerklePath::from_hex(&hex)
                        .map_err(|e| format!("not a merkle path: {e}"))
                        .and_then(|path| Proof::new(txid.clone(), path).map_err(|e| e.to_string()))
                        .map(Some),
                    None => Ok(None),
                },
                Err(e) => Err(e.to_string()),
            };
            carried.insert(txid.clone(), answer);
        }
        Self { carried }
    }

    fn heights(&self) -> Vec<Height> {
        self.carried
            .values()
            .filter_map(|answer| answer.as_ref().ok()?.as_ref().map(Proof::height))
            .collect()
    }
}

impl ProofFetcher for Courier {
    type Error = String;

    fn fetch(
        &mut self,
        txid: &str,
        _reask: &Reask,
    ) -> std::result::Result<Option<Proof>, Self::Error> {
        self.carried.remove(txid).unwrap_or(Ok(None))
    }
}

// ---------------------------------------------------------------------------
// The stored row
// ---------------------------------------------------------------------------

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
struct StoredHint {
    source: String,
    status: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    height: Option<Height>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    reason: Option<String>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    competitors: Vec<TxId>,
    /// The courier of a `mempool_present` hint (tracker 0.5.0).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    courier: Option<String>,
    observed: Timestamp,
}

impl StoredHint {
    fn of(hint: &Hint) -> Self {
        let mut courier = None;
        let (status, height, reason, competitors) = match &hint.status {
            HintStatus::Accepted => ("accepted", None, None, vec![]),
            HintStatus::Seen => ("seen", None, None, vec![]),
            HintStatus::Mined { height } => ("mined", Some(*height), None, vec![]),
            HintStatus::StaleBlock => ("stale_block", None, None, vec![]),
            HintStatus::Rejected { reason } => ("rejected", None, Some(reason.clone()), vec![]),
            HintStatus::DoubleSpend { competitors } => {
                ("double_spend", None, None, competitors.clone())
            }
            HintStatus::OrphanMempool => ("orphan_mempool", None, None, vec![]),
            HintStatus::Unknown => ("unknown", None, None, vec![]),
            HintStatus::MempoolPresent { source } => {
                courier = Some(source.clone());
                ("mempool_present", None, None, vec![])
            }
        };
        Self {
            source: hint.source.clone(),
            status: status.to_string(),
            height,
            reason,
            competitors,
            courier,
            observed: hint.observed,
        }
    }

    fn hint(&self) -> Hint {
        let status = match self.status.as_str() {
            "accepted" => HintStatus::Accepted,
            "seen" => HintStatus::Seen,
            "mined" => self
                .height
                .map_or(HintStatus::Unknown, |height| HintStatus::Mined { height }),
            "stale_block" => HintStatus::StaleBlock,
            "rejected" => HintStatus::Rejected {
                reason: self.reason.clone().unwrap_or_default(),
            },
            "double_spend" => HintStatus::DoubleSpend {
                competitors: self.competitors.clone(),
            },
            "orphan_mempool" => HintStatus::OrphanMempool,
            "mempool_present" => self.courier.clone().map_or(HintStatus::Unknown, |source| {
                HintStatus::MempoolPresent { source }
            }),
            _ => HintStatus::Unknown,
        };
        Hint::new(self.source.clone(), status, self.observed)
    }
}

/// One tracked transaction as the CLI's storage holds it: inputs to replay.
#[derive(Debug, Clone, PartialEq, Eq)]
struct TrackedRow {
    txid: TxId,
    built: bool,
    /// Oldest first.
    hints: Vec<StoredHint>,
    /// The raw merkle path, hex. Checked again on every load.
    bump: Option<String>,
    asks: u32,
    next_ask_at: Option<Timestamp>,
    /// The BEEF to hand to the broadcasters again, and when.
    reannounce_beef: Option<Vec<u8>>,
    reannounce_at: Option<Timestamp>,
    /// What the wallet records once a broadcaster takes it: the
    /// `internalizeAction` arguments, JSON.
    on_accept: Option<String>,
    /// Chain events that left the checked word other than a plain
    /// `mined` (stale, or suspect), oldest first, as their envelopes.
    /// Replayed after the proof on every load.
    chain: Vec<String>,
}

impl TrackedRow {
    fn new(txid: &str) -> Self {
        Self {
            txid: txid.to_ascii_lowercase(),
            built: true,
            hints: vec![],
            bump: None,
            asks: 0,
            next_ask_at: None,
            reannounce_beef: None,
            reannounce_at: None,
            on_accept: None,
            chain: vec![],
        }
    }

    /// The stored chain events, decoded. They were decoded once before
    /// they were stored; one that no longer decodes is reported and not
    /// replayed.
    fn chain_events(&self) -> Vec<ChainEvent> {
        self.chain
            .iter()
            .filter_map(|json| match decode_envelope(json.as_bytes()) {
                Ok(event) => Some(event),
                Err(fault) => {
                    tracing::warn!(txid = %self.txid, %fault, "a stored chain event does not decode");
                    None
                }
            })
            .collect()
    }

    /// Keep `event` on the row when it left the state anything but a
    /// plain checked `mined`; forget every kept event when it did. One
    /// `tip` is kept at most (each rechecks the same suspect mark).
    fn keep_chain(&mut self, state: &State, event: &ChainEvent) -> Result<()> {
        let plain = matches!(state.word(), Word::Mined(_)) && !state.suspect();
        if plain && state.reask().is_none() {
            self.chain.clear();
            return Ok(());
        }
        if matches!(event, ChainEvent::Tip { .. } | ChainEvent::TipAge { .. }) {
            self.chain.retain(|json| {
                !matches!(
                    decode_envelope(json.as_bytes()),
                    Ok(ChainEvent::Tip { .. } | ChainEvent::TipAge { .. })
                )
            });
        }
        self.chain
            .push(serde_json::to_string(&ChainEnvelope::new(event.clone()))?);
        Ok(())
    }

    /// Feed the stored inputs to `tracker`, as [`TrackedRow::replay`] feeds
    /// them to one state.
    fn replay_into<H: Headers>(&self, tracker: &mut Tracker, headers: &H) {
        tracker.track(self.txid.clone());
        if self.built {
            let _ = tracker.apply(&self.txid, headers, Input::Host(HostAction::Build));
        }
        for hint in &self.hints {
            let _ = tracker.apply(&self.txid, headers, Input::Hint(hint.hint()));
        }
        if let Some(Ok(proof)) = self.proof() {
            let _ = tracker.apply(&self.txid, headers, Input::Evidence(Evidence::Proof(proof)));
        }
        for event in self.chain_events() {
            let _ = tracker.apply(&self.txid, headers, Input::Evidence(Evidence::Chain(event)));
        }
    }

    fn proof(&self) -> Option<std::result::Result<Proof, CheckError>> {
        let hex = self.bump.as_deref()?;
        Some(
            MerklePath::from_hex(hex)
                .map_err(|e| CheckError::InvalidProof(e.to_string()))
                .and_then(|path| Proof::new(self.txid.clone(), path)),
        )
    }

    fn proof_height(&self) -> Option<Height> {
        self.proof()?.ok().map(|p| p.height())
    }

    /// Replay the stored inputs. The stored path is checked against
    /// `headers` like any other proof; a path that no longer checks leaves
    /// the hint-tier word and a re-ask, and is returned as the fault.
    fn replay<H: Headers>(&self, params: &Params, headers: &H) -> (State, Option<CheckError>) {
        let mut state = State::new(self.txid.clone());
        if self.built {
            let _ = state.step(params, headers, Input::Host(HostAction::Build));
        }
        for hint in &self.hints {
            let _ = state.step(params, headers, Input::Hint(hint.hint()));
        }
        let fault = match self.proof() {
            Some(Ok(proof)) => state
                .step(params, headers, Input::Evidence(Evidence::Proof(proof)))
                .err(),
            Some(Err(fault)) => Some(fault),
            None => None,
        };
        for event in self.chain_events() {
            let _ = state.step(params, headers, Input::Evidence(Evidence::Chain(event)));
        }
        (state, fault)
    }
}

/// The tracker's word, as one lowercase label.
pub fn word_label(word: &Word) -> &'static str {
    match word {
        Word::Unknown => "unknown",
        Word::Built => "built",
        Word::Announced => "announced",
        Word::Seen => "seen",
        Word::Mined(_) => "mined",
        Word::Stale => "stale",
        Word::Rejected { .. } => "rejected",
        Word::Conflicted { .. } => "conflicted",
        Word::Abandoned => "abandoned",
    }
}

fn reask_label(reask: &Reask) -> String {
    match reask {
        Reask::Reorg { fork_height } => format!("reorg@{fork_height}"),
        Reask::Fork { height } => format!("fork@{height}"),
        Reask::Hint(hint) => format!("hint:{}", hint.source),
        Reask::ProofFailed { height } => format!("proof_failed@{height}"),
        Reask::Recheck => "recheck".to_string(),
        Reask::Age => "age".to_string(),
        Reask::Spend => "spend".to_string(),
    }
}

async fn ensure_tables(storage: &StorageSqlx) -> Result<()> {
    for sql in [
        "CREATE TABLE IF NOT EXISTS tracker_states (\
            txid TEXT PRIMARY KEY, \
            built INTEGER NOT NULL DEFAULT 1, \
            hints TEXT NOT NULL DEFAULT '[]', \
            bump TEXT, \
            word TEXT NOT NULL DEFAULT 'unknown', \
            height INTEGER, \
            header_hash TEXT, \
            reask TEXT, \
            asks INTEGER NOT NULL DEFAULT 0, \
            next_ask_at INTEGER, \
            reannounce_beef BLOB, \
            reannounce_at INTEGER, \
            on_accept TEXT, \
            updated_at INTEGER NOT NULL DEFAULT 0)",
        "CREATE INDEX IF NOT EXISTS tracker_states_word ON tracker_states (word)",
        "CREATE INDEX IF NOT EXISTS tracker_states_height ON tracker_states (height)",
        "CREATE TABLE IF NOT EXISTS tracker_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)",
    ] {
        sqlx::query(sql).execute(storage.pool()).await?;
    }
    // Added with the chain-event feed (0.9.0); a table made before it
    // gains the column empty.
    let has_chain: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM pragma_table_info('tracker_states') WHERE name = 'chain'",
    )
    .fetch_one(storage.pool())
    .await?;
    if has_chain == 0 {
        sqlx::query("ALTER TABLE tracker_states ADD COLUMN chain TEXT NOT NULL DEFAULT '[]'")
            .execute(storage.pool())
            .await?;
    }
    Ok(())
}

const ROW_COLUMNS: &str =
    "txid, built, hints, bump, asks, next_ask_at, reannounce_beef, reannounce_at, on_accept, chain";

fn row_of(row: &sqlx::sqlite::SqliteRow) -> TrackedRow {
    let hints: String = row.get("hints");
    let chain: String = row.get("chain");
    TrackedRow {
        txid: row.get("txid"),
        built: row.get::<i64, _>("built") != 0,
        hints: serde_json::from_str(&hints).unwrap_or_default(),
        bump: row.get("bump"),
        asks: row.get::<i64, _>("asks").max(0) as u32,
        next_ask_at: row
            .get::<Option<i64>, _>("next_ask_at")
            .map(|t| t.max(0) as Timestamp),
        reannounce_beef: row.get("reannounce_beef"),
        reannounce_at: row
            .get::<Option<i64>, _>("reannounce_at")
            .map(|t| t.max(0) as Timestamp),
        on_accept: row.get("on_accept"),
        chain: serde_json::from_str(&chain).unwrap_or_default(),
    }
}

async fn load_row(storage: &StorageSqlx, txid: &str) -> Result<Option<TrackedRow>> {
    let row = sqlx::query(&format!(
        "SELECT {ROW_COLUMNS} FROM tracker_states WHERE txid = ?"
    ))
    .bind(txid.to_ascii_lowercase())
    .fetch_optional(storage.pool())
    .await?;
    Ok(row.as_ref().map(row_of))
}

async fn store_row(
    storage: &StorageSqlx,
    row: &TrackedRow,
    state: &State,
    now: Timestamp,
) -> Result<()> {
    store_row_in(storage.pool(), row, state, now).await
}

async fn store_row_in<'e, E: sqlx::Executor<'e, Database = sqlx::Sqlite>>(
    executor: E,
    row: &TrackedRow,
    state: &State,
    now: Timestamp,
) -> Result<()> {
    let (height, header_hash) = match state.word() {
        Word::Mined(mined) => (
            Some(mined.height() as i64),
            Some(mined.checked().header().hash.clone()),
        ),
        _ => (None, None),
    };
    sqlx::query(
        "INSERT INTO tracker_states \
            (txid, built, hints, bump, word, height, header_hash, reask, asks, next_ask_at, \
             reannounce_beef, reannounce_at, on_accept, chain, updated_at) \
         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \
         ON CONFLICT(txid) DO UPDATE SET built = excluded.built, hints = excluded.hints, \
            bump = excluded.bump, word = excluded.word, height = excluded.height, \
            header_hash = excluded.header_hash, reask = excluded.reask, asks = excluded.asks, \
            next_ask_at = excluded.next_ask_at, reannounce_beef = excluded.reannounce_beef, \
            reannounce_at = excluded.reannounce_at, on_accept = excluded.on_accept, \
            chain = excluded.chain, updated_at = excluded.updated_at",
    )
    .bind(&row.txid)
    .bind(row.built as i64)
    .bind(serde_json::to_string(&row.hints)?)
    .bind(&row.bump)
    .bind(word_label(state.word()))
    .bind(height)
    .bind(header_hash)
    .bind(state.reask().map(reask_label))
    .bind(row.asks as i64)
    .bind(row.next_ask_at.map(|t| t as i64))
    .bind(&row.reannounce_beef)
    .bind(row.reannounce_at.map(|t| t as i64))
    .bind(&row.on_accept)
    .bind(serde_json::to_string(&row.chain)?)
    .bind(now as i64)
    .execute(executor)
    .await?;
    Ok(())
}

/// What the storage holds for `txid`: the word, the mined height and the
/// pending re-ask the tracker gave at its last step. A view for a person;
/// nothing is decided from it.
pub async fn stored_word(
    storage: &StorageSqlx,
    txid: &str,
) -> Result<Option<(String, Option<Height>, Option<String>)>> {
    ensure_tables(storage).await?;
    let row = sqlx::query("SELECT word, height, reask FROM tracker_states WHERE txid = ?")
        .bind(txid.to_ascii_lowercase())
        .fetch_optional(storage.pool())
        .await?;
    Ok(row.map(|r| {
        (
            r.get("word"),
            r.get::<Option<i64>, _>("height").map(|h| h as Height),
            r.get("reask"),
        )
    }))
}

// ---------------------------------------------------------------------------
// What a broadcaster answered to a transaction the host just handed over
// ---------------------------------------------------------------------------

/// The broadcasters' answer to one post, read as a hint.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PostAnswer {
    /// A broadcaster took it.
    Accepted { provider: String },
    /// None took it and one called it not final: its lock time has not
    /// passed for the node behind that broadcaster. Transient.
    NotFinal { provider: String, said: String },
    /// None took it, for some other reason (or none answered).
    Refused { said: String },
}

/// Does a broadcaster's refusal say "not final"? The node's word is
/// `non-final`; ARC and Arcade carry it under code 476, which the toolbox
/// reads as retryable (bsv-wallet-toolbox-rs `ARCADE_CODE_NON_FINAL`).
pub fn says_not_final(text: &str) -> bool {
    let lower = text.to_ascii_lowercase();
    ["non-final", "nonfinal", "non final", "not final"]
        .iter()
        .any(|word| lower.contains(word))
        || lower
            .split(|c: char| !c.is_ascii_alphanumeric())
            .any(|token| token == "476")
}

/// Read the results of one `post_beef`.
pub fn read_post(results: &[PostBeefResult]) -> PostAnswer {
    if let Some(accepted) = results.iter().find(|r| r.is_success()) {
        return PostAnswer::Accepted {
            provider: accepted.name.clone(),
        };
    }
    let said: Vec<(String, String)> = results
        .iter()
        .map(|r| {
            let detail = r
                .error
                .clone()
                .or_else(|| r.txid_results.iter().find_map(|t| t.data.clone()))
                .unwrap_or_else(|| r.status.clone());
            (r.name.clone(), detail)
        })
        .collect();
    if let Some((provider, detail)) = said.iter().find(|(_, detail)| says_not_final(detail)) {
        return PostAnswer::NotFinal {
            provider: provider.clone(),
            said: detail.clone(),
        };
    }
    PostAnswer::Refused {
        said: if said.is_empty() {
            "none answered".to_string()
        } else {
            said.iter()
                .map(|(name, detail)| format!("{name}: {detail}"))
                .collect::<Vec<_>>()
                .join("; ")
        },
    }
}

/// A BEEF to hand to the broadcasters again, and when.
#[derive(Debug, Clone)]
pub struct Reannounce {
    /// The BEEF, as it was posted.
    pub beef: Vec<u8>,
    /// Not before this time (seconds since the epoch).
    pub at: Timestamp,
    /// What the wallet records when a broadcaster takes it then.
    pub on_accept: Option<InternalizeActionArgs>,
}

/// Track a transaction this wallet built and just handed to a broadcaster,
/// and feed what the broadcaster said, as a hint. `not_before` holds every
/// re-ask until that time (a transaction that cannot be mined before its
/// lock time has no proof to ask for before it); `reannounce` keeps the
/// BEEF to hand over again then. Returns the tracker's word and its
/// pending re-ask.
pub async fn heard<C: Clock>(
    storage: &StorageSqlx,
    clock: &C,
    opts: &TickOptions,
    txid: &str,
    hint: Hint,
    not_before: Option<Timestamp>,
    reannounce: Option<Reannounce>,
) -> Result<(String, Option<String>)> {
    ensure_tables(storage).await?;
    let mut row = match load_row(storage, txid).await? {
        // A row with a stored path is past this: its word is read, never
        // rewritten from a hint.
        Some(row) if row.bump.is_some() => {
            let stored = stored_word(storage, txid).await?.unwrap_or_default();
            return Ok((stored.0, stored.2));
        }
        Some(row) => row,
        None => TrackedRow::new(txid),
    };
    let params = opts.params();
    let headers = HeaderSnapshot::unread();
    let (mut state, _) = row.replay(&params, &headers);
    if !row.hints.iter().any(|h| h.source == hint.source) {
        row.hints.push(StoredHint::of(&hint));
        let _ = state.step(&params, &headers, Input::Hint(hint));
    }
    row.asks = 0;
    row.next_ask_at = not_before;
    match reannounce {
        Some(again) => {
            row.reannounce_beef = Some(again.beef);
            row.reannounce_at = Some(again.at);
            row.on_accept = match again.on_accept {
                Some(args) => Some(serde_json::to_string(&args)?),
                None => None,
            };
        }
        None => {
            row.reannounce_beef = None;
            row.reannounce_at = None;
            row.on_accept = None;
        }
    }
    store_row(storage, &row, &state, clock.now()).await?;
    Ok((
        word_label(state.word()).to_string(),
        state.reask().map(reask_label),
    ))
}

// ---------------------------------------------------------------------------
// The pass
// ---------------------------------------------------------------------------

/// The host's parameters of a pass.
#[derive(Debug, Clone)]
pub struct TickOptions {
    /// Seconds without a new hint before an announced or seen transaction
    /// is asked for its proof (the tracker's `Params::age_threshold`).
    pub age_threshold: Timestamp,
    /// Re-asks one pass runs.
    pub max_asks: usize,
    /// How deep the block holding a coin's transaction must be before the
    /// coin is offered (`SPEND_DEPTH`): the tip is depth 1. The stack's
    /// wallet charter R3 (b), ruled 2026-10-12 (bsv-stack-lean
    /// `lean/Wallet.lean`, `guardDepth`): the spend depth is the host's
    /// parameter, named in its config, with 1 the floor; a number below 1
    /// reads as 1. Read by [`spend_guard`], where the host decides whether
    /// a coin may be spent.
    pub spend_depth: Height,
}

impl TickOptions {
    /// `TRACKER_AGE_SECS`, `TRACKER_MAX_ASKS` and `SPEND_DEPTH`, with their
    /// defaults.
    pub fn from_env() -> Self {
        fn env<T: std::str::FromStr>(key: &str, default: T) -> T {
            std::env::var(key)
                .ok()
                .and_then(|v| v.trim().parse().ok())
                .unwrap_or(default)
        }
        Self {
            age_threshold: env("TRACKER_AGE_SECS", DEFAULT_AGE_SECS).max(1),
            max_asks: env("TRACKER_MAX_ASKS", DEFAULT_MAX_ASKS),
            spend_depth: env("SPEND_DEPTH", DEFAULT_SPEND_DEPTH),
        }
        .floored()
    }

    /// The options with the spend depth at its floor of 1 or above (R3).
    #[must_use]
    pub fn floored(mut self) -> Self {
        self.spend_depth = self.spend_depth.max(DEFAULT_SPEND_DEPTH);
        self
    }

    fn params(&self) -> Params {
        Params {
            age_threshold: self.age_threshold,
        }
    }
}

/// What one pass did.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
pub struct TickReport {
    /// Rows read this pass (never the mined ones, unless a header moved).
    pub tracked: usize,
    /// The wallet's own unproven transactions tracked for the first time.
    pub adopted: usize,
    /// Proven creators of coins we hold, tracked for the first time from
    /// the toolbox's stored proof replayed against the headers of the
    /// moment: `mined` when its root checks.
    pub adopted_mined: Vec<String>,
    /// The same, whose stored root does not meet the header at its height:
    /// tracked with the re-ask `proofFailed`, no word written.
    pub adopted_proof_failed: Vec<String>,
    /// New hints fed.
    pub hints: usize,
    /// Re-asks run: one proof request each.
    pub asked: usize,
    /// Transactions whose proof checked this pass: the word is `mined`.
    pub mined: Vec<String>,
    /// Re-asks that brought no proof; each is asked again after its pause.
    pub no_proof: Vec<String>,
    /// Mined rows whose stored path no longer meets the active header.
    pub moved: Vec<String>,
    /// The lowest height whose header moved since the last pass, if any.
    pub moved_from: Option<Height>,
    /// Transactions handed to the broadcasters again and accepted.
    pub reannounced: Vec<String>,
    /// Transactions handed over again and still called not final; each is
    /// handed over again after a pause.
    pub not_final: Vec<String>,
    /// What the wallet is to record for each transaction accepted on a
    /// re-announce (`txid`, the `internalizeAction` arguments): the caller
    /// that holds the wallet runs [`record_accepted`].
    #[serde(skip)]
    pub on_accept: Vec<(String, String)>,
    /// Faults the host reports: `txid: what failed`.
    pub faults: Vec<String>,
    /// What the wallet's storage did with each checked proof.
    pub stored: Vec<String>,
}

impl TickReport {
    /// Nothing moved and nothing was asked.
    pub fn is_quiet(&self) -> bool {
        self.adopted == 0
            && self.adopted_mined.is_empty()
            && self.adopted_proof_failed.is_empty()
            && self.hints == 0
            && self.asked == 0
            && self.reannounced.is_empty()
            && self.not_final.is_empty()
            && self.moved.is_empty()
            && self.faults.is_empty()
    }

    /// One line.
    pub fn summary(&self) -> String {
        format!(
            "tracker tick: {} tracked ({} new), {} proven creator(s) adopted ({} mined, {} proof failed), {} hint(s), {} announced again ({} still not final), {} re-ask(s): {} mined, {} without a proof, {} moved, {} fault(s)",
            self.tracked,
            self.adopted,
            self.adopted_mined.len() + self.adopted_proof_failed.len(),
            self.adopted_mined.len(),
            self.adopted_proof_failed.len(),
            self.hints,
            self.reannounced.len(),
            self.not_final.len(),
            self.asked,
            self.mined.len(),
            self.no_proof.len(),
            self.moved.len(),
            self.faults.len(),
        )
    }
}

/// The recent tips, newest last, read from and written to `tracker_meta`.
async fn read_ring(storage: &StorageSqlx) -> Vec<(Height, String)> {
    sqlx::query("SELECT value FROM tracker_meta WHERE key = 'tips'")
        .fetch_optional(storage.pool())
        .await
        .ok()
        .flatten()
        .and_then(|r| serde_json::from_str(&r.get::<String, _>("value")).ok())
        .unwrap_or_default()
}

async fn write_ring(storage: &StorageSqlx, ring: &[(Height, String)]) -> Result<()> {
    sqlx::query(
        "INSERT INTO tracker_meta (key, value) VALUES ('tips', ?) \
         ON CONFLICT(key) DO UPDATE SET value = excluded.value",
    )
    .bind(serde_json::to_string(ring)?)
    .execute(storage.pool())
    .await?;
    Ok(())
}

/// The lowest height whose header is no longer the one a past pass saw,
/// and the ring with the moved entries dropped and the tip appended. One
/// header read when the tip advanced, none when it did not, and one more
/// per replaced block. Deeper than the ring, the ring's lowest height is
/// the answer and the host says so.
async fn moved_since<V: WalletServices + ?Sized>(
    services: &V,
    snapshot: &mut HeaderSnapshot,
    mut ring: Vec<(Height, String)>,
) -> (Option<Height>, Vec<(Height, String)>) {
    let Some((tip_height, tip_hash)) = snapshot.tip().map(|(h, hash)| (h, hash.to_string())) else {
        return (None, ring);
    };
    let mut moved = None;
    while let Some((height, hash)) = ring.last().cloned() {
        let active = if height == tip_height {
            Some(tip_hash.clone())
        } else if height > tip_height {
            None
        } else {
            snapshot.extend(services, [height]).await;
            match snapshot.header_at(height) {
                Ok(Some(header)) => Some(header.hash),
                // Could not look: nothing is concluded this pass.
                _ => return (moved, ring),
            }
        };
        if active.as_deref() == Some(hash.as_str()) {
            break;
        }
        moved = Some(height);
        ring.pop();
    }
    if ring.last().map(|(h, _)| *h) != Some(tip_height) {
        ring.push((tip_height, tip_hash));
    }
    if ring.len() > TIP_RING {
        let extra = ring.len() - TIP_RING;
        ring.drain(..extra);
    }
    (moved, ring)
}

/// One pass of the host. See the module docs.
pub async fn tick<V: WalletServices + ?Sized, C: Clock>(
    storage: &StorageSqlx,
    services: &V,
    clock: &C,
    opts: &TickOptions,
) -> Result<TickReport> {
    ensure_tables(storage).await?;
    let params = opts.params();
    let now = clock.now();
    let mut report = TickReport::default();

    // The wallet's own transactions that are out and unproven, tracked
    // from the first pass that sees them. Never anyone else's.
    let adopted = sqlx::query(
        "INSERT OR IGNORE INTO tracker_states (txid, word, updated_at) \
         SELECT DISTINCT lower(txid), 'built', ? FROM transactions \
         WHERE txid IS NOT NULL AND status IN ('unproven', 'sending')",
    )
    .bind(now as i64)
    .execute(storage.pool())
    .await?;
    report.adopted = adopted.rows_affected() as usize;

    // Did a header we stood on move? Then the mined rows at or above it
    // are read again, and no other mined row is.
    let mut snapshot = HeaderSnapshot::take(services, []).await;
    let (moved_from, ring) = moved_since(services, &mut snapshot, read_ring(storage).await).await;
    report.moved_from = moved_from;

    let mut rows: Vec<TrackedRow> = sqlx::query(&format!(
        "SELECT {ROW_COLUMNS} FROM tracker_states \
         WHERE word IN ('unknown', 'built', 'announced', 'seen', 'stale') \
            OR (word = 'mined' AND ? IS NOT NULL AND height >= ?) \
         ORDER BY txid"
    ))
    .bind(moved_from.map(|h| h as i64))
    .bind(moved_from.map(|h| h as i64))
    .fetch_all(storage.pool())
    .await?
    .iter()
    .map(row_of)
    .collect();
    report.tracked = rows.len();

    // Read: the headers the stored paths name, and the new hints.
    snapshot
        .extend(services, rows.iter().filter_map(TrackedRow::proof_height))
        .await;
    let mut hints = MemoryHints::read(storage, &rows).await?;

    // Step: replay each row, feed its hints, tick its clock.
    let mut states: HashMap<TxId, State> = HashMap::new();
    for row in rows.iter_mut() {
        let (state, fault) = row.replay(&params, &snapshot);
        if let Some(fault) = fault {
            // The stored path does not check against the active header:
            // the row is back at its hint-tier word with a re-ask, due now.
            if matches!(fault, CheckError::RootMismatch(_)) {
                // The path proves inclusion in a block that is no longer
                // active: it is dropped, and a new one is asked for.
                row.bump = None;
                row.chain.clear();
                report.moved.push(row.txid.clone());
                tracing::warn!(
                    marker = "tracker_stored_proof_moved",
                    txid = %row.txid,
                    "a stored merkle path no longer meets the active header; asking again"
                );
            } else {
                report.faults.push(format!("{}: {fault}", row.txid));
            }
            row.asks = 0;
            row.next_ask_at = None;
        }
        states.insert(row.txid.clone(), state);
    }
    while let Ok(Some((txid, hint))) = hints.next_hint() {
        let (Some(state), Some(row)) = (
            states.get_mut(&txid),
            rows.iter_mut().find(|r| r.txid == txid),
        ) else {
            continue;
        };
        row.hints.push(StoredHint::of(&hint));
        let _ = state.step(&params, &snapshot, Input::Hint(hint));
        report.hints += 1;
        // A word that disagrees is asked about now, not after the pause
        // an earlier re-ask earned.
        if matches!(state.reask(), Some(Reask::Hint(_))) {
            row.asks = 0;
            row.next_ask_at = None;
        }
    }
    // Announce again what a broadcaster called not final, at its time.
    // The lock time passing on our clock is not the node's rule passing
    // (the node reads the median time past, about an hour behind; the rule
    // is the sibling's, bsv-script-lean `docs/BLOCK-CONSENSUS.md` section
    // 2.2), so "not final" again is expected and waits one more pause.
    for row in rows.iter_mut() {
        let (Some(beef), Some(at)) = (row.reannounce_beef.clone(), row.reannounce_at) else {
            continue;
        };
        let Some(state) = states.get_mut(&row.txid) else {
            continue;
        };
        if at > now || matches!(state.word(), Word::Mined(_)) {
            continue;
        }
        let pause = Some(now.saturating_add(opts.age_threshold));
        let answer = services
            .post_beef(&beef, std::slice::from_ref(&row.txid))
            .await
            .map(|results| read_post(&results));
        match answer {
            Ok(PostAnswer::Accepted { provider }) => {
                let hint = Hint::new(format!("{provider}|accepted"), HintStatus::Accepted, now);
                row.hints.push(StoredHint::of(&hint));
                let _ = state.step(&params, &snapshot, Input::Hint(hint));
                row.reannounce_beef = None;
                row.reannounce_at = None;
                row.asks = 0;
                row.next_ask_at = pause;
                report.reannounced.push(row.txid.clone());
                if let Some(args) = row.on_accept.take() {
                    report.on_accept.push((row.txid.clone(), args));
                }
            }
            Ok(PostAnswer::NotFinal { .. }) => {
                row.asks = row.asks.saturating_add(1);
                row.next_ask_at = pause;
                if row.asks >= MAX_ASKS_PER_TX {
                    row.reannounce_at = None;
                    report.faults.push(format!(
                        "{}: still not final after {MAX_ASKS_PER_TX} announces; not announced again",
                        row.txid
                    ));
                } else {
                    row.reannounce_at = pause;
                    report.not_final.push(row.txid.clone());
                }
            }
            Ok(PostAnswer::Refused { said }) => {
                row.reannounce_at = None;
                row.next_ask_at = pause;
                report.faults.push(format!(
                    "{}: refused when announced again: {said}",
                    row.txid
                ));
            }
            Err(e) => {
                // Could not look: the same time again, one pause on.
                row.reannounce_at = pause;
                report
                    .faults
                    .push(format!("{}: could not announce again: {e}", row.txid));
            }
        }
    }
    for state in states.values_mut() {
        let _ = state.step(&params, &snapshot, Input::Host(HostAction::Tick(now)));
    }

    // Read again: a proof for each re-ask that is due, and its header.
    let due: Vec<TxId> = rows
        .iter()
        .filter(|row| {
            states[&row.txid].reask().is_some()
                && row.asks < MAX_ASKS_PER_TX
                && row.next_ask_at.is_none_or(|at| at <= now)
        })
        .take(opts.max_asks)
        .map(|row| row.txid.clone())
        .collect();
    let mut courier = Courier::carry(services, &due).await;
    snapshot.extend(services, courier.heights()).await;

    // Step again: each proof through the tracker's check.
    for txid in &due {
        let (Some(state), Some(row)) = (
            states.get_mut(txid),
            rows.iter_mut().find(|r| &r.txid == txid),
        ) else {
            continue;
        };
        report.asked += 1;
        let reask = state.reask().cloned().unwrap_or(Reask::Age);
        let fetched = courier.fetch(txid, &reask);
        let proven = match fetched {
            Ok(Some(proof)) => {
                let hex = proof.path().to_hex();
                match state.step(&params, &snapshot, Input::Evidence(Evidence::Proof(proof))) {
                    Ok(()) => {
                        row.bump = Some(hex);
                        row.chain.clear();
                        true
                    }
                    Err(fault) => {
                        report.faults.push(format!("{txid}: {fault}"));
                        false
                    }
                }
            }
            Ok(None) => {
                report.no_proof.push(txid.clone());
                false
            }
            Err(fault) => {
                report
                    .faults
                    .push(format!("{txid}: could not ask for a proof: {fault}"));
                false
            }
        };
        if proven {
            row.asks = 0;
            row.next_ask_at = None;
            report.mined.push(txid.clone());
        } else {
            let pause = opts
                .age_threshold
                .saturating_mul(1 << row.asks.min(MAX_BACKOFF_DOUBLINGS));
            row.asks = row.asks.saturating_add(1);
            row.next_ask_at = Some(now.saturating_add(pause));
        }
    }

    // Keep the state, and hand each checked proof to the wallet's own
    // storage through its one proof funnel.
    for row in &rows {
        let state = &states[&row.txid];
        store_row(storage, row, state, now).await?;
        if let (true, Word::Mined(mined)) = (report.mined.contains(&row.txid), state.word()) {
            let checked = mined.checked();
            let outcome = storage
                .ingest_merkle_proof(
                    &row.txid,
                    &checked.proof().path().to_binary(),
                    checked.height(),
                    &checked.header().hash,
                    Some(checked.root()),
                )
                .await;
            report.stored.push(match outcome {
                Ok(outcome) => format!("{}: {outcome:?}", row.txid),
                Err(e) => format!("{}: not stored: {e}", row.txid),
            });
        }
    }
    adopt_proven(storage, services, &mut snapshot, &params, now, &mut report).await?;
    write_ring(storage, &ring).await?;
    Ok(report)
}

/// Proven creators adopted per pass at most ([`adopt_proven`]).
pub const MAX_ADOPT_PER_PASS: usize = 100;

/// Track the creators the toolbox proved before the tracker held a row for
/// them (its monitor's backstop, `check_for_proofs`, moves a transaction to
/// `completed` with `proven_tx_id` set): every `completed` transaction with
/// a stored proof that creates a coin we hold and has no tracker row. The
/// stored path is replayed as the row's input against the headers of the
/// moment, as the guard's [`read_creators`] replays it: a root that checks
/// gives the row the word `mined` at its height with that path as its
/// evidence; a root that does not meet the header at its height gives a row
/// with the pending re-ask `proofFailed`, which the next pass runs (the
/// path dropped, a fresh one asked for). A header the service could not
/// give, or a stored path that does not parse, adopts nothing this pass.
/// The toolbox's own rows are never written. The stack's wallet charter,
/// section 1: about a transaction the wallet believes only the tracker's
/// word, and each action hosts one tracker state.
///
/// Load: one store query per pass, and one header read per height not
/// already read this pass, for at most [`MAX_ADOPT_PER_PASS`] creators.
async fn adopt_proven<V: WalletServices + ?Sized>(
    storage: &StorageSqlx,
    services: &V,
    snapshot: &mut HeaderSnapshot,
    params: &Params,
    now: Timestamp,
    report: &mut TickReport,
) -> Result<()> {
    let found = sqlx::query(
        "SELECT lower(t.txid) AS txid, MAX(t.is_outgoing) AS made, p.merkle_path \
         FROM transactions t \
         JOIN proven_txs p ON p.proven_tx_id = t.proven_tx_id \
         WHERE t.txid IS NOT NULL AND t.status = 'completed' \
           AND NOT EXISTS (SELECT 1 FROM tracker_states s WHERE s.txid = lower(t.txid)) \
           AND EXISTS (SELECT 1 FROM outputs o \
                       WHERE o.transaction_id = t.transaction_id AND o.spent_by IS NULL) \
         GROUP BY lower(t.txid) ORDER BY MIN(t.transaction_id) LIMIT ?",
    )
    .bind(MAX_ADOPT_PER_PASS as i64)
    .fetch_all(storage.pool())
    .await?;
    let mut rows = Vec::with_capacity(found.len());
    for r in &found {
        let txid: String = r.get("txid");
        let path: Option<Vec<u8>> = r.get("merkle_path");
        let Some(path) = path
            .as_deref()
            .and_then(|bytes| MerklePath::from_binary(bytes).ok())
        else {
            report.faults.push(format!(
                "{txid}: the stored proof does not parse; not adopted"
            ));
            continue;
        };
        rows.push(TrackedRow {
            built: r.get::<i64, _>("made") != 0,
            bump: Some(path.to_hex()),
            ..TrackedRow::new(&txid)
        });
    }
    snapshot
        .extend(services, rows.iter().filter_map(TrackedRow::proof_height))
        .await;
    for row in rows {
        let (state, fault) = row.replay(params, &*snapshot);
        match fault {
            None if matches!(state.word(), Word::Mined(_)) => {
                report.adopted_mined.push(row.txid.clone());
            }
            Some(CheckError::RootMismatch(_)) => {
                tracing::warn!(
                    marker = "tracker_adopted_proof_failed",
                    txid = %row.txid,
                    reask = ?state.reask(),
                    "a proven creator's stored merkle path does not meet the active header; tracked with its re-ask"
                );
                report.adopted_proof_failed.push(row.txid.clone());
            }
            // Could not look: nothing is adopted, the next pass looks again.
            Some(fault) => {
                report
                    .faults
                    .push(format!("{}: not adopted: {fault}", row.txid));
                continue;
            }
            None => {
                report.faults.push(format!(
                    "{}: not adopted: the stored proof gave the word {}",
                    row.txid,
                    word_label(state.word())
                ));
                continue;
            }
        }
        store_row(storage, &row, &state, now).await?;
    }
    Ok(())
}

/// Record in the wallet what a pass's re-announces got accepted (E4): the
/// wallet's storage is the verdict for its own actions, so a claim the
/// wallet made is internalized from the BEEF it already holds, and nothing
/// scans the chain for it. Each outcome is added to `report.stored`.
pub async fn record_accepted<W: WalletInterface>(wallet: &W, report: &mut TickReport) {
    for (txid, args) in std::mem::take(&mut report.on_accept) {
        let line = match serde_json::from_str::<InternalizeActionArgs>(&args) {
            Ok(args) => match wallet.internalize_action(args, "bsv-wallet-cli").await {
                Ok(result) if result.accepted => format!("{txid}: recorded in the wallet"),
                Ok(_) => format!("{txid}: the wallet did not accept its own transaction"),
                Err(e) => format!(
                    "{txid}: not recorded in the wallet ({e}); `bsv-wallet receive {txid}` records it once it is mined"
                ),
            },
            Err(e) => format!("{txid}: not recorded in the wallet (stored arguments: {e})"),
        };
        if !line.ends_with("recorded in the wallet") {
            tracing::warn!(marker = "own_transaction_not_recorded", "{line}");
        }
        report.stored.push(line);
    }
}

// ---------------------------------------------------------------------------
// The spend guard
// ---------------------------------------------------------------------------

/// The wallet spend guard, as the tracker's README gives it (bsv-tracker
/// `README.md`, "Wallet: spend guard"), unchanged: the scheduled recheck
/// runs before the coin is used, and an unavailable header is an error that
/// prevents the spend.
fn readme_spend_guard<H: Headers>(
    state: &mut State,
    params: &Params,
    headers: &H,
) -> std::result::Result<bool, CheckError> {
    state.step(params, headers, Input::Host(HostAction::SpendAttempt))?;
    state.step(params, headers, Input::Evidence(Evidence::Recheck))?;
    Ok(matches!(state.word(), Word::Mined(_)) && !state.suspect())
}

/// Run the spend guard for a coin of the tracked transaction `txid`, as
/// the host: the row is loaded (its stored path checked against a fresh
/// header), the README's guard runs on it, and the row is stored again.
/// `Ok(true)` only for a checked, non-suspect `mined`; `Err` when the
/// header service could not answer. A transaction the host does not track
/// is not mined.
pub async fn spend_guard<V: WalletServices + ?Sized, C: Clock>(
    storage: &StorageSqlx,
    services: &V,
    clock: &C,
    opts: &TickOptions,
    txid: &str,
) -> Result<std::result::Result<bool, CheckError>> {
    ensure_tables(storage).await?;
    let Some(row) = load_row(storage, txid).await? else {
        return Ok(Ok(false));
    };
    let params = opts.params();
    let snapshot = HeaderSnapshot::take(services, row.proof_height()).await;
    let (mut state, fault) = row.replay(&params, &snapshot);
    match fault {
        // Could not look (no tip, no header at the height, a path that
        // does not parse): an error, no spend, and the row is left as it
        // was. "Could not look" changes no word.
        Some(fault) if !matches!(fault, CheckError::RootMismatch(_)) => return Ok(Err(fault)),
        _ => {}
    }
    let split = under_split(state.word(), open_split_floor(storage).await?);
    let verdict = readme_spend_guard(&mut state, &params, &snapshot).and_then(|offered| {
        Ok(offered && !split && deep_enough(&state, &snapshot, opts.spend_depth)?)
    });
    store_row(storage, &row, &state, clock.now()).await?;
    Ok(verdict)
}

/// The guard's reading of one coin's creator, for the coin gate
/// ([`crate::spend_gate`]).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CreatorReading {
    /// The tracker's word after the guard's recheck, as its label.
    pub word: &'static str,
    /// How deep the creator's block is under the tip (the tip is 1); 0 for
    /// a word with no block.
    pub depth: u64,
    /// A fork in progress, or a split not yet resolved, at or below its
    /// height marks it suspect.
    pub suspect: bool,
    /// `mined`, checked against the headers of the moment, not suspect, at
    /// the spend depth (floor 1): the guard offers it with no rule.
    pub offered: bool,
    /// Could not look (no tip, no header at the height): the fault. Nothing
    /// is offered on it.
    pub fault: Option<String>,
}

/// Run the spend guard over many creators against one read of the headers:
/// the tip once, and one header per height a stored path names. Each
/// creator is the tracked row when the tracker holds one, else the
/// toolbox's stored merkle path (`proven_txs`, given here as its bytes)
/// replayed as a row's input: a stored proof is checked again against the
/// headers of the moment, never trusted by age. A creator with neither
/// reads `unknown`. Nothing is written: the tracker's rows are not stored
/// back and no row is adopted.
pub async fn read_creators<V: WalletServices + ?Sized>(
    storage: &StorageSqlx,
    services: &V,
    opts: &TickOptions,
    creators: &[(String, Option<Vec<u8>>)],
) -> Result<HashMap<String, CreatorReading>> {
    ensure_tables(storage).await?;
    let params = opts.params();
    let mut rows = Vec::with_capacity(creators.len());
    for (txid, stored_path) in creators {
        let mut row = match load_row(storage, txid).await? {
            Some(row) => row,
            None => TrackedRow {
                built: false,
                ..TrackedRow::new(txid)
            },
        };
        if row.bump.is_none() {
            row.bump = stored_path
                .as_deref()
                .and_then(|bytes| MerklePath::from_binary(bytes).ok())
                .map(|path| path.to_hex());
        }
        rows.push(row);
    }
    let snapshot =
        HeaderSnapshot::take(services, rows.iter().filter_map(TrackedRow::proof_height)).await;
    let split_floor = open_split_floor(storage).await?;
    let mut readings = HashMap::with_capacity(rows.len());
    for row in rows {
        let (mut state, fault) = row.replay(&params, &snapshot);
        let fault = match fault {
            Some(fault) if !matches!(fault, CheckError::RootMismatch(_)) => Some(fault.to_string()),
            _ => None,
        };
        let verdict = match fault {
            Some(_) => Ok(false),
            None => readme_spend_guard(&mut state, &params, &snapshot).and_then(|offered| {
                Ok(offered && deep_enough(&state, &snapshot, opts.spend_depth)?)
            }),
        };
        let depth = match (state.word().height(), snapshot.tip()) {
            (Some(height), Some((tip, _))) => {
                (u64::from(tip) + 1).saturating_sub(u64::from(height))
            }
            _ => 0,
        };
        let (offered, fault) = match verdict {
            Ok(offered) => (offered, fault),
            Err(e) => (false, fault.or(Some(e.to_string()))),
        };
        let split = under_split(state.word(), split_floor);
        readings.insert(
            row.txid.clone(),
            CreatorReading {
                word: word_label(state.word()),
                depth,
                suspect: state.suspect() || split,
                offered: offered && !split,
                fault,
            },
        );
    }
    Ok(readings)
}

/// Is the block holding a `mined` word's transaction at least
/// `spend_depth` deep under `headers`' tip (the tip is depth 1; the stack's
/// `Tracker.depth hd h = hd.tip + 1 - h`)? The depth reads as 1 at least
/// (R3's floor). An unreadable tip is an error, as for the guard.
fn deep_enough<H: Headers>(
    state: &State,
    headers: &H,
    spend_depth: Height,
) -> std::result::Result<bool, CheckError> {
    let Some(height) = state.word().height() else {
        return Ok(false);
    };
    let tip = headers.tip_height().map_err(CheckError::Headers)?;
    let depth = (u64::from(tip) + 1).saturating_sub(u64::from(height));
    Ok(depth >= u64::from(spend_depth.max(DEFAULT_SPEND_DEPTH)))
}

// ---------------------------------------------------------------------------
// Chain events from the header service's feed
// ---------------------------------------------------------------------------

/// The splits the feed has delivered and not yet seen resolved: each
/// split's key and its `forkHeight` (`tracker_meta`, key `open_splits`),
/// written in the delivery's transaction.
async fn read_open_splits<'e, E: sqlx::Executor<'e, Database = sqlx::Sqlite>>(
    executor: E,
) -> Result<BTreeMap<String, Height>> {
    let value: Option<String> =
        sqlx::query_scalar("SELECT value FROM tracker_meta WHERE key = 'open_splits'")
            .fetch_optional(executor)
            .await?;
    Ok(match value {
        Some(v) => serde_json::from_str(&v)?,
        None => BTreeMap::new(),
    })
}

/// The lowest `forkHeight` of the splits still open, if any. The guard
/// holds every coin whose creator is mined at or above it until its
/// `splitResolved`: the header service answers one branch, and which
/// branch stands is open (depth is the host's; the ruling of 2026-10-16,
/// the scenario `tracker-split-observation-writes-no-word.md` column 4).
pub async fn open_split_floor(storage: &StorageSqlx) -> Result<Option<Height>> {
    ensure_tables(storage).await?;
    Ok(read_open_splits(storage.pool()).await?.into_values().min())
}

/// Is a `mined` word at `height` above an open split?
fn under_split(word: &Word, floor: Option<Height>) -> bool {
    matches!((word, floor), (Word::Mined(m), Some(floor)) if floor <= m.height())
}

/// The feed's stored cursor: the last envelope delivered, `0` at birth.
/// One row of `tracker_meta`, beside the tracker's rows.
pub async fn feed_cursor(storage: &StorageSqlx) -> Result<u64> {
    ensure_tables(storage).await?;
    let value: Option<String> =
        sqlx::query_scalar("SELECT value FROM tracker_meta WHERE key = 'feed_cursor'")
            .fetch_optional(storage.pool())
            .await?;
    Ok(match value {
        Some(v) => v
            .parse()
            .map_err(|e| anyhow::anyhow!("the stored feed cursor {v:?}: {e}"))?,
        None => 0,
    })
}

/// The frozen record: an outpoint the feed's `frozen` event named, with
/// the envelope's cursor and when it was delivered. The spend gate reads it
/// and never offers such a coin (R6: nothing here clears a row).
async fn ensure_frozen(pool: &sqlx::SqlitePool) -> Result<()> {
    sqlx::query(
        "CREATE TABLE IF NOT EXISTS frozen_outpoints (\
            txid TEXT NOT NULL, \
            vout INTEGER NOT NULL, \
            cursor INTEGER NOT NULL, \
            frozen_at INTEGER NOT NULL, \
            PRIMARY KEY (txid, vout))",
    )
    .execute(pool)
    .await?;
    Ok(())
}

/// Every frozen outpoint: (lowercase txid, vout).
pub async fn frozen_outpoints(
    pool: &sqlx::SqlitePool,
) -> Result<std::collections::HashSet<(String, i64)>> {
    ensure_frozen(pool).await?;
    let rows = sqlx::query("SELECT txid, vout FROM frozen_outpoints")
        .fetch_all(pool)
        .await?;
    Ok(rows
        .iter()
        .map(|r| (r.get::<String, _>("txid"), r.get::<i64, _>("vout")))
        .collect())
}

/// How many outpoints the frozen record holds.
pub async fn frozen_count(pool: &sqlx::SqlitePool) -> Result<u64> {
    ensure_frozen(pool).await?;
    let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM frozen_outpoints")
        .fetch_one(pool)
        .await?;
    Ok(n.max(0) as u64)
}

/// What one delivered envelope did to one tracked transaction.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ChainMove {
    /// The transaction the event named, by index.
    pub txid: String,
    /// The tracker's word after the event.
    pub word: String,
    /// Its pending re-ask, if any.
    pub reask: Option<String>,
    /// Marked by a fork or a split, awaiting a recheck.
    pub suspect: bool,
    /// The tracker's fault for this transaction, reported, if any.
    pub fault: Option<String>,
}

/// One delivered envelope.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ChainDelivery {
    /// The envelope's cursor, now stored.
    pub cursor: u64,
    /// The event's kind.
    pub kind: &'static str,
    /// The transactions it named.
    pub moved: Vec<ChainMove>,
}

/// The event's wire kind.
pub fn event_kind(event: &ChainEvent) -> &'static str {
    match event {
        ChainEvent::Tip { .. } => "tip",
        ChainEvent::Fork { .. } => "fork",
        ChainEvent::Reorg { .. } => "reorg",
        ChainEvent::Invalidated { .. } => "invalidated",
        ChainEvent::Frozen { .. } => "frozen",
        ChainEvent::TipAge { .. } => "tipAge",
        ChainEvent::Split { .. } => "split",
        ChainEvent::SplitResolved { .. } => "splitResolved",
    }
}

/// The headers a reorg replaced, over the daemon's snapshot: a stored
/// proof is replayed against the header it was checked against (the
/// event's deactivated header at its height, the prior tip as the tip),
/// so the tracker sees the `mined` word the reorg takes away. The event
/// writes no `mined` word through this view; `Tracker::on_chain` turns
/// the words it names `stale`.
struct Replaced<'a> {
    snapshot: &'a HeaderSnapshot,
    replaced: BTreeMap<Height, Header>,
    prior_tip: Option<Height>,
}

impl Headers for Replaced<'_> {
    fn header_at(&self, height: Height) -> std::result::Result<Option<Header>, HeaderError> {
        match self.replaced.get(&height) {
            Some(header) => Ok(Some(header.clone())),
            None => self.snapshot.header_at(height),
        }
    }

    fn tip_height(&self) -> std::result::Result<Height, HeaderError> {
        match self.prior_tip {
            Some(tip) => Ok(tip),
            None => self.snapshot.tip_height(),
        }
    }
}

/// Deliver one chain envelope (already decoded, at `cursor`) to the
/// tracked transactions it names, through `Tracker::on_chain` with the
/// daemon's headers, and store the moved rows and the cursor in one
/// transaction. A cursor at or below the stored one was delivered before
/// (delivery is at least once) and changes nothing.
///
/// The rows are named by index, never by a scan of the chain: a fork, a
/// reorg or a split names the mined rows at or above its height; a tip,
/// the rows an earlier event marked; an invalidated block, the rows on its
/// hash; a freeze or a tip age names none here (they are the spend and
/// freshness guards' inputs). A freeze writes its outpoint to the frozen
/// record ([`frozen_outpoints`]) in the same transaction as the cursor. The header service is read only when a row
/// is named: the tip and the heights of the named rows' proofs.
pub async fn deliver_chain_event<V: WalletServices + ?Sized, C: Clock>(
    storage: &StorageSqlx,
    services: &V,
    clock: &C,
    opts: &TickOptions,
    cursor: u64,
    event: ChainEvent,
) -> Result<ChainDelivery> {
    let kind = event_kind(&event);
    if cursor <= feed_cursor(storage).await? {
        return Ok(ChainDelivery {
            cursor,
            kind,
            moved: vec![],
        });
    }
    let named = format!("SELECT {ROW_COLUMNS} FROM tracker_states");
    let query = match &event {
        ChainEvent::Fork { height: from, .. }
        | ChainEvent::Reorg {
            fork_height: from, ..
        }
        | ChainEvent::Split {
            fork_height: from, ..
        }
        | ChainEvent::SplitResolved {
            fork_height: from, ..
        } => Some(
            sqlx::query(&format!(
                "{named} WHERE word = 'mined' AND height >= ? ORDER BY txid"
            ))
            .bind(i64::from(*from))
            .fetch_all(storage.pool())
            .await?,
        ),
        ChainEvent::Tip { .. } => Some(
            sqlx::query(&format!(
                "{named} WHERE word = 'mined' AND chain <> '[]' ORDER BY txid"
            ))
            .fetch_all(storage.pool())
            .await?,
        ),
        ChainEvent::Invalidated { block_hash } => Some(
            sqlx::query(&format!("{named} WHERE header_hash = ? ORDER BY txid"))
                .bind(block_hash.to_ascii_lowercase())
                .fetch_all(storage.pool())
                .await?,
        ),
        ChainEvent::Frozen { .. } | ChainEvent::TipAge { .. } => None,
    };
    let mut rows: Vec<TrackedRow> = query.iter().flatten().map(row_of).collect();

    let now = clock.now();
    let mut moved = Vec::new();
    if matches!(event, ChainEvent::Frozen { .. }) {
        ensure_frozen(storage.pool()).await?;
    }
    let mut tx = storage.pool().begin().await?;
    if let ChainEvent::Frozen { outpoint } = &event {
        sqlx::query(
            "INSERT INTO frozen_outpoints (txid, vout, cursor, frozen_at) VALUES (?, ?, ?, ?) \
             ON CONFLICT(txid, vout) DO NOTHING",
        )
        .bind(outpoint.txid.to_ascii_lowercase())
        .bind(i64::from(outpoint.vout))
        .bind(cursor as i64)
        .bind(now as i64)
        .execute(&mut *tx)
        .await?;
    }
    if !rows.is_empty() {
        let snapshot =
            HeaderSnapshot::take(services, rows.iter().filter_map(TrackedRow::proof_height)).await;
        let (replaced, prior_tip) = match &event {
            ChainEvent::Reorg {
                deactivated_headers,
                ..
            } => (
                deactivated_headers
                    .iter()
                    .map(|h| {
                        (
                            h.height,
                            Header {
                                hash: h.hash.to_ascii_lowercase(),
                                merkle_root: h.merkle_root.to_ascii_lowercase(),
                            },
                        )
                    })
                    .collect(),
                deactivated_headers.first().map(|h| h.height),
            ),
            _ => (BTreeMap::new(), None),
        };
        let replay_view = Replaced {
            snapshot: &snapshot,
            replaced,
            prior_tip,
        };
        let mut tracker = Tracker::new(opts.params());
        for row in &rows {
            row.replay_into(&mut tracker, &replay_view);
        }
        for update in tracker.on_chain(&snapshot, event.clone()) {
            let (Some(state), Some(row)) = (
                tracker.get(&update.txid),
                rows.iter_mut().find(|r| r.txid == update.txid),
            ) else {
                continue;
            };
            row.keep_chain(state, &event)?;
            if matches!(update.reask, Some(Reask::Reorg { .. })) {
                // Asked now, not after the pause an earlier re-ask earned.
                row.asks = 0;
                row.next_ask_at = None;
            }
            store_row_in(&mut *tx, row, state, now).await?;
            let fault = update.result.err().map(|e| e.to_string());
            if let Some(fault) = &fault {
                tracing::warn!(txid = %update.txid, kind, cursor, %fault, "a chain event's recheck could not look");
            }
            moved.push(ChainMove {
                txid: update.txid,
                word: word_label(state.word()).to_string(),
                reask: state.reask().map(reask_label),
                suspect: state.suspect(),
                fault,
            });
        }
    }
    if let ChainEvent::Split {
        key, fork_height, ..
    }
    | ChainEvent::SplitResolved {
        key, fork_height, ..
    } = &event
    {
        let mut open = read_open_splits(&mut *tx).await?;
        if matches!(event, ChainEvent::Split { .. }) {
            open.insert(key.clone(), *fork_height);
        } else {
            open.remove(key);
        }
        sqlx::query(
            "INSERT INTO tracker_meta (key, value) VALUES ('open_splits', ?) \
             ON CONFLICT(key) DO UPDATE SET value = excluded.value",
        )
        .bind(serde_json::to_string(&open)?)
        .execute(&mut *tx)
        .await?;
    }
    sqlx::query(
        "INSERT INTO tracker_meta (key, value) VALUES ('feed_cursor', ?) \
         ON CONFLICT(key) DO UPDATE SET value = excluded.value",
    )
    .bind(cursor.to_string())
    .execute(&mut *tx)
    .await?;
    tx.commit().await?;
    Ok(ChainDelivery {
        cursor,
        kind,
        moved,
    })
}

#[cfg(test)]
pub(crate) mod tests {
    use super::*;
    use bsv_wallet_toolbox::services::mock::{MockResponse, MockWalletServices};
    use bsv_wallet_toolbox::services::{BlockHeader, GetMerklePathResult};
    use bsv_wallet_toolbox::{
        WalletStorageWriter, BROADCAST_STATUS_ACCEPTED, BROADCAST_STATUS_MINED,
    };
    use chrono::Utc;

    /// A clock the test sets.
    pub(crate) struct FixedClock(pub Timestamp);
    impl Clock for FixedClock {
        fn now(&self) -> Timestamp {
            self.0
        }
    }

    pub(crate) fn opts() -> TickOptions {
        TickOptions {
            age_threshold: 600,
            max_asks: 20,
            spend_depth: DEFAULT_SPEND_DEPTH,
        }
    }

    /// A migrated in-memory wallet with one user.
    pub(crate) async fn wallet_storage() -> (StorageSqlx, i64) {
        let storage = StorageSqlx::in_memory().await.unwrap();
        storage
            .migrate("tracker-tests", &("02".to_string() + &"ab".repeat(32)))
            .await
            .unwrap();
        storage.make_available().await.unwrap();
        let (user, _) = storage
            .find_or_insert_user(&("02".to_string() + &"cd".repeat(32)))
            .await
            .unwrap();
        (storage, user.user_id)
    }

    pub(crate) async fn insert_tx(storage: &StorageSqlx, user_id: i64, txid: &str, status: &str) {
        sqlx::query(
            "INSERT INTO transactions (user_id, status, reference, is_outgoing, satoshis, version, lock_time, description, txid, raw_tx, created_at, updated_at) \
             VALUES (?, ?, ?, 1, 0, 1, 0, 'd', ?, X'01000000', ?, ?)",
        )
        .bind(user_id)
        .bind(status)
        .bind(&txid[..8])
        .bind(txid)
        .bind(Utc::now())
        .bind(Utc::now())
        .execute(storage.pool())
        .await
        .unwrap();
    }

    /// A coin of ours whose creator `txid` the toolbox's monitor proved
    /// before the tracker saw it: the transaction `completed` with
    /// `proven_tx_id` set to a stored proof at `height` (its root the
    /// txid), one unspent change output of `satoshis` in the default
    /// basket. A received coin (`is_outgoing` 0), as `fund` makes one.
    pub(crate) async fn proven_coin(
        storage: &StorageSqlx,
        user_id: i64,
        txid: &str,
        height: u32,
        satoshis: i64,
    ) {
        let basket = storage
            .find_or_create_default_basket(user_id)
            .await
            .unwrap()
            .basket_id;
        let proven = sqlx::query(
            "INSERT INTO proven_txs (txid, height, idx, block_hash, merkle_root, merkle_path, raw_tx, created_at, updated_at) \
             VALUES (?, ?, 0, ?, ?, ?, X'01000000', ?, ?)",
        )
        .bind(txid)
        .bind(height as i64)
        .bind("bb".repeat(32))
        .bind(txid)
        .bind(MerklePath::from_coinbase_txid(txid, height).to_binary())
        .bind(Utc::now())
        .bind(Utc::now())
        .execute(storage.pool())
        .await
        .unwrap()
        .last_insert_rowid();
        let tx = sqlx::query(
            "INSERT INTO transactions (user_id, status, reference, is_outgoing, satoshis, version, lock_time, description, txid, raw_tx, proven_tx_id, created_at, updated_at) \
             VALUES (?, 'completed', ?, 0, ?, 1, 0, 'Internalize external funding', ?, X'01000000', ?, ?, ?)",
        )
        .bind(user_id)
        .bind(&txid[..8])
        .bind(satoshis)
        .bind(txid)
        .bind(proven)
        .bind(Utc::now())
        .bind(Utc::now())
        .execute(storage.pool())
        .await
        .unwrap()
        .last_insert_rowid();
        sqlx::query(
            "INSERT INTO outputs (user_id, transaction_id, basket_id, vout, satoshis, locking_script, txid, type, spendable, change, provided_by, purpose, output_description, created_at, updated_at) \
             VALUES (?, ?, ?, 0, ?, X'76a914dbc0a7c84983c5bf199b7b2d41b3acf0408ee5aa88ac', ?, 'P2PKH', 1, 1, 'storage', 'change', 'c', ?, ?)",
        )
        .bind(user_id)
        .bind(tx)
        .bind(basket)
        .bind(satoshis)
        .bind(txid)
        .bind(Utc::now())
        .bind(Utc::now())
        .execute(storage.pool())
        .await
        .unwrap();
    }

    /// A header at `height` whose merkle root is `root`.
    pub(crate) fn header(height: u32, root: &str, nonce: u32) -> BlockHeader {
        BlockHeader {
            version: 1,
            previous_hash: "00".repeat(32),
            merkle_root: root.to_string(),
            time: 1_700_000_000,
            bits: 0x1d00_ffff,
            nonce,
            hash: String::new(),
            height,
        }
    }

    /// A path proving `txid` alone in a block at `height`: its root is the
    /// txid.
    pub(crate) fn path_of(txid: &str, height: u32) -> String {
        MerklePath::from_coinbase_txid(txid, height).to_hex()
    }

    pub(crate) fn proof_answer(path: Option<String>) -> MockResponse<GetMerklePathResult> {
        MockResponse::Success(GetMerklePathResult {
            name: Some("Services".to_string()),
            merkle_path: path,
            header: None,
            error: None,
            notes: vec![],
        })
    }

    /// E3 (the rulings of 2026-10-09). The wallet's own unproven
    /// transaction is tracked, a broadcaster's acceptance announces it, and
    /// its age asks for a proof: one request, for that transaction. The
    /// proof checks against the header service's header and the word is
    /// `mined`. The row is in the wallet's database, and the next pass
    /// reads no mined row and asks nothing.
    #[tokio::test]
    async fn a_pass_tracks_announces_and_proves_one_named_transaction() {
        let (storage, user) = wallet_storage().await;
        let txid = "a1".repeat(32);
        insert_tx(&storage, user, &txid, "unproven").await;
        storage
            .record_broadcast_status(&txid, "arcade", BROADCAST_STATUS_ACCEPTED)
            .await
            .unwrap();
        let services = MockWalletServices::builder()
            .get_merkle_path_response(proof_answer(Some(path_of(&txid, 900))))
            .build();
        services.set_header_for_height(header(900, &txid, 0));
        let now = Utc::now().timestamp() as u64;

        // Young: announced on the hint, nothing asked.
        let report = tick(&storage, &services, &FixedClock(now), &opts())
            .await
            .unwrap();
        assert_eq!((report.adopted, report.hints, report.asked), (1, 1, 0));
        assert_eq!(
            stored_word(&storage, &txid).await.unwrap(),
            Some(("announced".to_string(), None, None))
        );
        assert_eq!(services.call_count("get_merkle_path"), 0);

        // Past the age threshold: the re-ask runs, the proof checks.
        let report = tick(&storage, &services, &FixedClock(now + 601), &opts())
            .await
            .unwrap();
        assert_eq!(report.asked, 1);
        assert_eq!(report.mined, vec![txid.clone()]);
        assert_eq!(
            stored_word(&storage, &txid).await.unwrap(),
            Some(("mined".to_string(), Some(900), None))
        );
        assert_eq!(services.call_count("get_merkle_path"), 1);

        // A mined row is not read again while no header moved.
        let report = tick(&storage, &services, &FixedClock(now + 1300), &opts())
            .await
            .unwrap();
        assert_eq!((report.tracked, report.asked), (0, 0));
        assert_eq!(services.call_count("get_merkle_path"), 1);
    }

    /// The evidence rule in the host: a broadcaster's "mined" is a hint. It
    /// asks for the proof at once; with no proof the word does not move,
    /// and the same transaction is not asked again until its pause is over.
    #[tokio::test]
    async fn a_broadcasters_mined_is_a_hint_that_asks_for_the_proof() {
        let (storage, user) = wallet_storage().await;
        let txid = "b2".repeat(32);
        insert_tx(&storage, user, &txid, "unproven").await;
        storage
            .record_broadcast_status(&txid, "arcade", BROADCAST_STATUS_MINED)
            .await
            .unwrap();
        let services = MockWalletServices::builder()
            .get_merkle_path_response(proof_answer(None))
            .build();
        let now = Utc::now().timestamp() as u64;

        let report = tick(&storage, &services, &FixedClock(now), &opts())
            .await
            .unwrap();
        assert_eq!(report.asked, 1);
        assert_eq!(report.no_proof, vec![txid.clone()]);
        assert!(report.mined.is_empty());
        let (word, height, reask) = stored_word(&storage, &txid).await.unwrap().unwrap();
        assert_eq!((word.as_str(), height), ("built", None));
        assert_eq!(reask.as_deref(), Some("hint:arcade|mined"));

        // One minute later: the pause is not over, nothing is asked.
        let report = tick(&storage, &services, &FixedClock(now + 60), &opts())
            .await
            .unwrap();
        assert_eq!(report.asked, 0);
        // After the pause it is asked again, and the pause doubles.
        let report = tick(&storage, &services, &FixedClock(now + 600), &opts())
            .await
            .unwrap();
        assert_eq!(report.asked, 1);
        let report = tick(&storage, &services, &FixedClock(now + 1200), &opts())
            .await
            .unwrap();
        assert_eq!(report.asked, 0);
        assert_eq!(services.call_count("get_merkle_path"), 2);
    }

    /// A header we stood on moved: the mined rows at or above it are read
    /// again and asked again, and a mined row below it is not touched.
    #[tokio::test]
    async fn a_moved_header_reasks_the_rows_at_or_above_it_and_no_other() {
        let (storage, user) = wallet_storage().await;
        let low = "c3".repeat(32);
        let high = "d4".repeat(32);
        let services = MockWalletServices::builder()
            .get_merkle_path_response(MockResponse::Sequence(vec![
                proof_answer(Some(path_of(&low, 100))),
                proof_answer(Some(path_of(&high, 200))),
                proof_answer(Some(path_of(&high, 201))),
            ]))
            .build();
        services.set_header_for_height(header(100, &low, 0));
        services.set_header_for_height(header(200, &high, 0));
        services.set_tip_header(header_with_hash(200, &high, 0));
        let now = Utc::now().timestamp() as u64;
        for txid in [&low, &high] {
            insert_tx(&storage, user, txid, "unproven").await;
            storage
                .record_broadcast_status(txid, "arcade", BROADCAST_STATUS_MINED)
                .await
                .unwrap();
        }
        let report = tick(&storage, &services, &FixedClock(now), &opts())
            .await
            .unwrap();
        assert_eq!(report.mined.len(), 2, "{report:?}");

        // Height 200 is replaced by a block without `high`; it is mined
        // again at 201.
        services.set_header_for_height(header(200, &"ee".repeat(32), 7));
        services.set_header_for_height(header(201, &high, 0));
        services.set_tip_header(header_with_hash(201, &high, 0));
        let report = tick(&storage, &services, &FixedClock(now + 60), &opts())
            .await
            .unwrap();
        assert_eq!(report.moved_from, Some(200));
        assert_eq!(
            report.tracked, 1,
            "only the row at or above the moved header"
        );
        assert_eq!(report.moved, vec![high.clone()]);
        assert_eq!(report.mined, vec![high.clone()]);
        assert_eq!(
            stored_word(&storage, &high).await.unwrap().unwrap().1,
            Some(201)
        );
        assert_eq!(
            stored_word(&storage, &low).await.unwrap().unwrap().1,
            Some(100)
        );
    }

    /// E2 (the rulings of 2026-10-09). A transaction a broadcaster called
    /// not final is held with its BEEF: nothing is asked or announced
    /// before its lock time; at the lock time it is announced again; "not
    /// final" again (the node's clock is behind ours) waits one pause; an
    /// acceptance announces it, and from there its age asks for the proof.
    #[tokio::test]
    async fn a_not_final_transaction_is_announced_again_at_its_lock_time() {
        use bsv_wallet_toolbox::services::mock::{
            error_post_beef_result, success_post_beef_result,
        };
        let (storage, _user) = wallet_storage().await;
        let txid = "c7".repeat(32);
        let lock = 2_000_000_000u64;
        let services = MockWalletServices::builder()
            .post_beef_response(MockResponse::Sequence(vec![
                MockResponse::Success(vec![error_post_beef_result("Arcade", "476 non-final")]),
                MockResponse::Success(vec![success_post_beef_result("Arcade", &[&txid])]),
            ]))
            .get_merkle_path_response(proof_answer(None))
            .build();

        let hint = Hint::new(
            "Arcade|rejected",
            HintStatus::Rejected {
                reason: "476 non-final".to_string(),
            },
            lock - 3600,
        );
        let again = Reannounce {
            beef: vec![1, 2, 3],
            at: lock,
            on_accept: None,
        };
        let word = heard(
            &storage,
            &FixedClock(lock - 3600),
            &opts(),
            &txid,
            hint,
            Some(lock),
            Some(again),
        )
        .await
        .unwrap();
        assert_eq!(
            word,
            (
                "built".to_string(),
                Some("hint:Arcade|rejected".to_string())
            )
        );

        // Before the lock time: nothing announced, no proof asked.
        let report = tick(&storage, &services, &FixedClock(lock - 1), &opts())
            .await
            .unwrap();
        assert_eq!((report.tracked, report.asked), (1, 0), "{report:?}");
        assert_eq!(services.call_count("post_beef"), 0);

        // At the lock time: announced again; the node still says not final.
        let report = tick(&storage, &services, &FixedClock(lock), &opts())
            .await
            .unwrap();
        assert_eq!(report.not_final, vec![txid.clone()]);
        assert_eq!(services.call_count("post_beef"), 1);
        let report = tick(&storage, &services, &FixedClock(lock + 60), &opts())
            .await
            .unwrap();
        assert!(report.not_final.is_empty() && report.reannounced.is_empty());
        assert_eq!(services.call_count("post_beef"), 1, "one pause between two");

        // One pause on: accepted. The word is the tracker's `announced`.
        let report = tick(&storage, &services, &FixedClock(lock + 600), &opts())
            .await
            .unwrap();
        assert_eq!(report.reannounced, vec![txid.clone()]);
        assert_eq!(services.call_count("post_beef"), 2);
        assert_eq!(
            stored_word(&storage, &txid).await.unwrap().unwrap().0,
            "announced"
        );
        assert_eq!(services.call_count("get_merkle_path"), 0);

        // It is not announced a third time; its age asks for the proof.
        let report = tick(&storage, &services, &FixedClock(lock + 1300), &opts())
            .await
            .unwrap();
        assert_eq!(report.asked, 1);
        assert_eq!(services.call_count("post_beef"), 2);
        assert_eq!(services.call_count("get_merkle_path"), 1);
    }

    /// A courier's mempool presence (tracker 0.5.0) survives the stored row,
    /// its courier with it; a presence row without one reads as `unknown`.
    #[test]
    fn a_mempool_presence_hint_round_trips_through_the_stored_row() {
        let hint = Hint::new(
            "arcade".to_string(),
            HintStatus::MempoolPresent {
                source: "courier-a".to_string(),
            },
            1_700_000_000,
        );
        let stored = StoredHint::of(&hint);
        let json = serde_json::to_string(&stored).unwrap();
        let back: StoredHint = serde_json::from_str(&json).unwrap();
        assert_eq!(back.hint(), hint);

        let bare = StoredHint {
            courier: None,
            ..stored
        };
        assert_eq!(bare.hint().status, HintStatus::Unknown);
    }

    /// The words that mean "not final", and one that does not.
    #[test]
    fn not_final_is_read_from_the_refusal() {
        for said in [
            "476 non-final",
            "bad-txns-nonfinal",
            "tx is not final",
            "code 476",
        ] {
            assert!(says_not_final(said), "{said}");
        }
        for said in [
            "fee too low",
            "DOUBLE_SPEND_ATTEMPTED",
            "txid 4476aa",
            "HTTP 500",
        ] {
            assert!(!says_not_final(said), "{said}");
        }
    }

    /// A tip header carrying the hash its bytes have.
    pub(crate) fn header_with_hash(height: u32, root: &str, nonce: u32) -> BlockHeader {
        let mut h = header(height, root, nonce);
        h.hash = header_of(&h.to_binary()).unwrap().hash;
        h
    }

    /// R3 (b): the spend depth is the host's, named in its config, 1 the
    /// floor. A coin whose block is the tip (one block deep) is offered at
    /// depth 1 and held at depth 3; two blocks later it is offered at 3; a
    /// depth of 0 reads as 1.
    #[tokio::test]
    async fn the_spend_depth_holds_a_coin_until_its_block_is_deep_enough() {
        let (storage, user) = wallet_storage().await;
        let txid = "e7".repeat(32);
        insert_tx(&storage, user, &txid, "unproven").await;
        storage
            .record_broadcast_status(&txid, "arcade", BROADCAST_STATUS_MINED)
            .await
            .unwrap();
        let services = MockWalletServices::builder()
            .get_merkle_path_response(proof_answer(Some(path_of(&txid, 900))))
            .build();
        services.set_header_for_height(header(900, &txid, 0));
        services.set_tip_header(header_with_hash(900, &txid, 0));
        let clock = FixedClock(Utc::now().timestamp() as u64);
        tick(&storage, &services, &clock, &opts()).await.unwrap();
        let at = |spend_depth| TickOptions {
            spend_depth,
            ..opts()
        };

        let one = spend_guard(&storage, &services, &clock, &at(1), &txid)
            .await
            .unwrap();
        let three = spend_guard(&storage, &services, &clock, &at(3), &txid)
            .await
            .unwrap();
        let zero = spend_guard(&storage, &services, &clock, &at(0), &txid)
            .await
            .unwrap();
        println!("one block deep: depth 1 {one:?}, depth 3 {three:?}, depth 0 {zero:?}");
        assert_eq!(one, Ok(true));
        assert_eq!(three, Ok(false));
        assert_eq!(zero, Ok(true), "0 reads as the floor, 1");

        services.set_tip_header(header_with_hash(902, &"0c".repeat(32), 0));
        let three = spend_guard(&storage, &services, &clock, &at(3), &txid)
            .await
            .unwrap();
        println!("three blocks deep: depth 3 {three:?}");
        assert_eq!(three, Ok(true));
        assert_eq!(
            TickOptions {
                spend_depth: 0,
                ..opts()
            }
            .floored()
            .spend_depth,
            1
        );
    }

    /// The README's wallet spend guard, run by the host over a stored row
    /// (bsv-tracker `README.md`, "Wallet: spend guard"). Three runs: the
    /// header is the one the proof was checked against; the header service
    /// cannot answer; the header moved.
    #[tokio::test]
    async fn the_readme_spend_guard_runs_as_the_host() {
        let (storage, user) = wallet_storage().await;
        let txid = "e5".repeat(32);
        insert_tx(&storage, user, &txid, "unproven").await;
        storage
            .record_broadcast_status(&txid, "arcade", BROADCAST_STATUS_MINED)
            .await
            .unwrap();
        let services = MockWalletServices::builder()
            .get_merkle_path_response(proof_answer(Some(path_of(&txid, 900))))
            .build();
        services.set_header_for_height(header(900, &txid, 0));
        let clock = FixedClock(Utc::now().timestamp() as u64);
        tick(&storage, &services, &clock, &opts()).await.unwrap();

        // 1. Mined, checked, not suspect: the coin may be spent.
        let guard = spend_guard(&storage, &services, &clock, &opts(), &txid)
            .await
            .unwrap();
        println!(
            "spend guard, header in place:   {guard:?}, stored {:?}",
            stored_word(&storage, &txid).await.unwrap().unwrap()
        );
        assert_eq!(guard, Ok(true));

        // 2. No tip from the header service: an error, and no spend.
        services.set_tip_unavailable();
        let guard = spend_guard(&storage, &services, &clock, &opts(), &txid)
            .await
            .unwrap();
        println!(
            "spend guard, no header service: {guard:?}, stored {:?}",
            stored_word(&storage, &txid).await.unwrap().unwrap()
        );
        assert!(matches!(guard, Err(CheckError::Headers(_))), "{guard:?}");
        assert_eq!(
            stored_word(&storage, &txid).await.unwrap().unwrap().0,
            "mined",
            "could not look changes no word"
        );

        // 3. The header at the height moved: not mined, a re-ask pending.
        services.set_tip_header(header_with_hash(901, &"00".repeat(32), 1));
        services.set_header_for_height(header(900, &"ee".repeat(32), 7));
        let guard = spend_guard(&storage, &services, &clock, &opts(), &txid)
            .await
            .unwrap();
        println!(
            "spend guard, header moved:      {guard:?}, stored {:?}",
            stored_word(&storage, &txid).await.unwrap().unwrap()
        );
        assert_eq!(guard, Ok(false));

        // A transaction the host does not track is not mined.
        let guard = spend_guard(&storage, &services, &clock, &opts(), &"f6".repeat(32))
            .await
            .unwrap();
        assert_eq!(guard, Ok(false));

        // The guard's own error path, on a state held in memory: mined,
        // then the header service loses the height.
        let params = opts().params();
        services.set_tip_header(header_with_hash(900, &txid, 0));
        services.set_header_for_height(header(900, &txid, 0));
        let held = HeaderSnapshot::take(&services, [900]).await;
        let row = load_row(&storage, &txid).await.unwrap().unwrap();
        let (mut state, fault) = row.replay(&params, &held);
        assert_eq!(fault, None);
        let lost = HeaderSnapshot::take(&services, []).await;
        let guard = readme_spend_guard(&mut state, &params, &lost);
        println!("spend guard, height unanswered: {guard:?}");
        assert_eq!(guard, Err(CheckError::Unavailable(900)));
    }

    /// F1 of the stack's coin read (bsv-stack-lean #164): the daemon's
    /// served pass tracks the wallet's creators. An `unproven` creator is
    /// adopted as before; a `completed` creator the toolbox's monitor proved
    /// first is adopted by replaying its stored path against the headers of
    /// the moment: `mined` with its evidence when the header at its height
    /// carries the root, the re-ask `proofFailed` and no word when it does
    /// not. A creator tracked while unproven keeps its row when the toolbox
    /// moves it to `completed`. The toolbox's rows are not written, and the
    /// next pass adopts nothing.
    #[tokio::test]
    async fn the_served_pass_adopts_an_unproven_creator_and_a_proven_completed_one() {
        let (storage, user) = wallet_storage().await;
        let unproven = "d1".repeat(32);
        let earlier = "d2".repeat(32);
        insert_tx(&storage, user, &unproven, "unproven").await;
        insert_tx(&storage, user, &earlier, "unproven").await;
        let services = MockWalletServices::builder().build();
        services.set_tip_header(header_with_hash(960, &"0a".repeat(32), 0));
        let now = Utc::now().timestamp() as u64;
        let pass = crate::broadcast_reconcile::run_served_pass(
            &storage,
            &services,
            &FixedClock(now),
            &opts(),
            None,
        )
        .await
        .unwrap();
        println!("{}", pass.summary());
        assert_eq!(pass.tracker.adopted, 2);
        assert_eq!(
            stored_word(&storage, &earlier).await.unwrap(),
            Some(("built".to_string(), None, None))
        );

        // The toolbox's monitor proves `earlier` (completed), and two coins
        // arrive already proven: one whose root the header service's
        // header at 900 carries, one whose header at 950 carries another.
        sqlx::query("UPDATE transactions SET status = 'completed' WHERE txid = ?")
            .bind(&earlier)
            .execute(storage.pool())
            .await
            .unwrap();
        let checks = "d3".repeat(32);
        let moved = "d4".repeat(32);
        proven_coin(&storage, user, &checks, 900, 3_000).await;
        proven_coin(&storage, user, &moved, 950, 2_000).await;
        services.set_header_for_height(header(900, &checks, 0));
        services.set_header_for_height(header(950, &"ee".repeat(32), 0));
        let toolbox_before: Vec<(String, String, Option<i64>)> =
            sqlx::query_as("SELECT txid, status, proven_tx_id FROM transactions ORDER BY txid")
                .fetch_all(storage.pool())
                .await
                .unwrap();

        let pass = crate::broadcast_reconcile::run_served_pass(
            &storage,
            &services,
            &FixedClock(now + 60),
            &opts(),
            None,
        )
        .await
        .unwrap();
        println!("{}", pass.summary());
        assert_eq!(pass.tracker.adopted_mined, vec![checks.clone()]);
        assert_eq!(pass.tracker.adopted_proof_failed, vec![moved.clone()]);
        assert_eq!(
            stored_word(&storage, &checks).await.unwrap(),
            Some(("mined".to_string(), Some(900), None))
        );
        let row = load_row(&storage, &checks).await.unwrap().unwrap();
        assert_eq!(
            row.bump,
            Some(path_of(&checks, 900)),
            "the evidence is the stored path"
        );
        assert!(!row.built, "a received coin's creator was not built here");
        assert_eq!(
            stored_word(&storage, &moved).await.unwrap(),
            Some((
                "unknown".to_string(),
                None,
                Some("proof_failed@950".to_string())
            ))
        );
        assert_eq!(
            stored_word(&storage, &earlier).await.unwrap(),
            Some(("built".to_string(), None, None)),
            "a creator tracked while unproven keeps its row"
        );
        let toolbox_after: Vec<(String, String, Option<i64>)> =
            sqlx::query_as("SELECT txid, status, proven_tx_id FROM transactions ORDER BY txid")
                .fetch_all(storage.pool())
                .await
                .unwrap();
        assert_eq!(
            toolbox_after, toolbox_before,
            "no write to the toolbox's rows"
        );
        assert_eq!(services.call_count("get_merkle_path"), 0);

        let pass = crate::broadcast_reconcile::run_served_pass(
            &storage,
            &services,
            &FixedClock(now + 120),
            &opts(),
            None,
        )
        .await
        .unwrap();
        println!("{}", pass.summary());
        assert!(pass.tracker.adopted_mined.is_empty());
        assert!(pass.tracker.adopted_proof_failed.is_empty());
    }
}