satd-events-client 0.6.3

//! Typed event model — an ergonomic mirror of the `satd.events.v1` wire
//! `NodeEvent.body` tagged union, so consumers `match` on a flat enum instead
//! of unwrapping nested protobuf `Option`s.
//!
//! Hashes and txids are carried as raw `Vec<u8>` in **internal (consensus)
//! byte order** — the order the wire and key derivation use, *not* the reversed
//! order block explorers and Bitcoin Core JSON-RPC display. Convert at the edge:
//! [`parse_txid`] returns a typed [`bitcoin::Txid`] (whose `Display` is explorer
//! order, and which drops into `bitcoin`-ecosystem APIs) with the `bitcoin`
//! feature, and [`display_hex`] returns a display-order hex string with no extra
//! dependency. Only hashes/txids are byte-reversed for display — do not apply
//! either helper to a public key or tweak (`output_pubkey`, `tweak`).

use satd_events_proto::v1 as pb;

/// Render a raw wire hash/txid (32 bytes, **internal / consensus byte order**,
/// as every hash and txid on this API is carried) as a lowercase hex string in
/// the reversed **display order** used by block explorers and Bitcoin Core
/// JSON-RPC.
///
/// Use it on a hash or txid field (`txid`, `block_hash`, …). Do **not** apply
/// it to a public key or tweak: those are raw bytes but are *not* reversed for
/// display. Needs no optional dependency; for a typed value use [`parse_txid`].
///
/// ```
/// # use satd_events_client::display_hex;
/// let internal: Vec<u8> = (0u8..32).collect(); // 00,01,…,1f
/// assert!(display_hex(&internal).starts_with("1f1e1d")); // reversed
/// ```
pub fn display_hex(internal_hash: &[u8]) -> String {
    internal_hash
        .iter()
        .rev()
        .map(|b| format!("{b:02x}"))
        .collect()
}

/// Parse a raw wire txid (32 bytes, internal/consensus byte order) into a typed
/// [`bitcoin::Txid`]. The returned value's `Display` is the usual explorer /
/// JSON-RPC order, and it hands straight to `bitcoin`-ecosystem APIs
/// (`bitcoincore-rpc`, comparison against `getrawtransaction`, …).
///
/// A thin wrapper over [`bitcoin::Txid::from_slice`]; the block-hash equivalent
/// is `bitcoin::BlockHash::from_slice(bytes)`.
///
/// ```
/// # use satd_events_client::{parse_txid, display_hex};
/// let raw: Vec<u8> = (0u8..32).collect();
/// let txid = parse_txid(&raw).unwrap();
/// assert_eq!(txid.to_string(), display_hex(&raw)); // both reverse to display order
/// ```
#[cfg(feature = "bitcoin")]
pub fn parse_txid(raw: &[u8]) -> Result<bitcoin::Txid, bitcoin::hashes::FromSliceError> {
    use bitcoin::hashes::Hash as _;
    bitcoin::Txid::from_slice(raw)
}

/// Durable resume position, re-exported from the wire schema. Persist the
/// value returned by [`EventStream::cursor`](crate::EventStream::cursor) and
/// present it again as `from_cursor` to resume.
pub use satd_events_proto::v1::Cursor;

/// An outpoint (`txid:vout`), raw bytes.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Outpoint {
    /// Transaction id, 32 raw bytes (internal byte order — [`parse_txid`] or
    /// [`display_hex`] for a typed value / explorer-order string).
    pub txid: Vec<u8>,
    /// Output index.
    pub vout: u32,
}

/// Descriptor attribution for a [`ScriptMatched`](Event::ScriptMatched): which
/// descriptor watch a matched scripthash belongs to, and the exact coordinate the
/// server derived it at.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DescriptorMatch {
    /// The descriptor string the watch was registered with.
    pub descriptor: String,
    /// The 0-based BIP-389 multipath branch the matched script came from (`<0;1>`
    /// → external = 0, change = 1; always 0 for a single-path descriptor).
    pub branch: u32,
    /// The absolute derivation index of the matched script — ready to use, no
    /// `gap_limit` arithmetic. `(branch, derivation_index)` is exactly what the
    /// server derived (correct for fixed and multipath descriptors alike); the
    /// server still tracks no derivation progress — advancing your gap limit
    /// remains your concern.
    pub derivation_index: u32,
}

/// Why a transaction left the mempool by policy.
///
/// Like the status enums, this distinguishes an **unset** field
/// ([`Unspecified`](Self::Unspecified), proto3's zero value) from a value this
/// build does not recognize ([`Unknown`](Self::Unknown)) — two different facts
/// that must not collapse into one variant.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum EvictReason {
    /// The producer did not set a reason (proto3's zero value).
    Unspecified,
    /// Evicted because the pool hit its byte budget.
    FullPool,
    /// Evicted on mempool-expiry.
    Expiry,
    /// Evicted because a connected block conflicts with it.
    BlockConflict,
    /// Evicted from the quarantine class on a fee-rate byte-budget overflow.
    Policy,
    /// Evicted because a chain reorganization made the transaction invalid
    /// (locktime no longer satisfied, coinbase became immature, or BIP 68
    /// sequence lock unsatisfied).
    Reorg,
    /// An eviction reason this client build does not recognize.
    Unknown(i32),
}

/// Which node-health condition an [`Event::Status`] describes.
///
/// Open by design: a newer node may report a kind this client build predates,
/// which surfaces as [`Unknown`](StatusKind::Unknown) rather than an error. The
/// event's `severity` and `message` stay meaningful in that case, so a generic
/// "log it and page on critical" handler keeps working across upgrades.
///
/// `#[non_exhaustive]` for the reason the `Unknown` arm exists: this crate is
/// published, the node's taxonomy grows independently of it, and a downstream
/// `match` that listed every variant *including* `Unknown(_)` would be
/// exhaustive — so adding a kind here would be a breaking release. Forcing a
/// wildcard arm keeps a node-side addition additive for every consumer.
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StatusKind {
    /// The producer did not set a kind (proto3's zero value). Distinct from
    /// [`Unknown`](Self::Unknown), which is a kind that *was* set and this
    /// build does not recognize.
    Unspecified,
    /// Initial block download finished (one-shot).
    IbdComplete,
    /// No block connected inside the configured window.
    ///
    /// Not suppressed during initial block download: the node's IBD predicate
    /// is the tip header's age, not a sync flag, so a node that was caught up
    /// and then wedged re-enters it exactly when you want to hear about it. A
    /// genuinely syncing node connects blocks and never crosses the threshold.
    TipStall,
    /// Free space on the watched directory fell below the configured floor.
    DiskLow,
    /// Mempool occupancy crossed the configured share of its byte cap.
    MempoolCongested,
    /// Connected peers stayed below the configured floor.
    PeerFloor,
    /// A reorg at least the configured depth landed (one-shot).
    DeepReorg,
    /// A block template the node built failed its own validity check.
    TemplateInvalid,
    /// A condition this client build does not recognize.
    Unknown(i32),
}

impl StatusKind {
    fn from_proto(v: i32) -> Self {
        match pb::StatusKind::try_from(v) {
            Ok(pb::StatusKind::Unspecified) => StatusKind::Unspecified,
            Ok(pb::StatusKind::IbdComplete) => StatusKind::IbdComplete,
            Ok(pb::StatusKind::TipStall) => StatusKind::TipStall,
            Ok(pb::StatusKind::DiskLow) => StatusKind::DiskLow,
            Ok(pb::StatusKind::MempoolCongested) => StatusKind::MempoolCongested,
            Ok(pb::StatusKind::PeerFloor) => StatusKind::PeerFloor,
            Ok(pb::StatusKind::DeepReorg) => StatusKind::DeepReorg,
            Ok(pb::StatusKind::TemplateInvalid) => StatusKind::TemplateInvalid,
            _ => StatusKind::Unknown(v),
        }
    }
}

/// Level-triggered lifecycle of a health condition.
///
/// `#[non_exhaustive]` — see [`StatusKind`].
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StatusState {
    /// The producer did not set a state (proto3's zero value).
    ///
    /// A consumer tracking standing conditions cannot infer a lifecycle from
    /// this — it is neither a raise nor a clear — so treat it as "unusable" and
    /// say so, rather than silently dropping it. Dropping it is how a condition
    /// whose *clear* arrived unset stays standing forever.
    Unspecified,
    /// The condition was entered; it stands until a matching `Cleared`.
    Raised,
    /// The condition recovered.
    Cleared,
    /// A one-shot observation; no `Cleared` will follow.
    Edge,
    /// A state this client build does not recognize.
    Unknown(i32),
}

impl StatusState {
    fn from_proto(v: i32) -> Self {
        match pb::StatusState::try_from(v) {
            Ok(pb::StatusState::Unspecified) => StatusState::Unspecified,
            Ok(pb::StatusState::Raised) => StatusState::Raised,
            Ok(pb::StatusState::Cleared) => StatusState::Cleared,
            Ok(pb::StatusState::Edge) => StatusState::Edge,
            _ => StatusState::Unknown(v),
        }
    }
}

/// How loud a health condition is.
///
/// Ordering is by *severity rank*, not by proto value, so a client filters with
/// a comparison (`severity >= StatusSeverity::Warning`). The rank is:
///
/// ```text
/// Unspecified  <  Info  <  Warning  <  Critical  <  Unknown(_)
/// ```
///
/// `Unknown(_)` sorting above `Critical` is deliberate: a severity this build
/// cannot name is not one to quietly filter out, so an additive taxonomy change
/// fails loud rather than silent.
///
/// [`Unspecified`](Self::Unspecified) is the opposite case and is why the two
/// are separate variants. `STATUS_SEVERITY_UNSPECIFIED` is proto3's zero value,
/// which is what an absent field decodes to — a producer that never set the
/// field at all. That is an absence of information, not a loud condition, so it
/// ranks lowest. Folding it into `Unknown` (as this type once did) made an
/// unset severity outrank `Critical` and page on the documented filter.
///
/// **satd itself never emits a zero severity** — the node's own enum has three
/// variants and the encoder writes one of them explicitly — so this is defence
/// against a third-party producer, an in-path relay that re-encodes the event,
/// or a future schema, not a bug you can have hit against a stock node.
///
/// If you do see `Unspecified` and need a level, `kind` is the recovery: in v1
/// severity is a fixed function of the condition, so the kind's documented
/// severity is the one the originating node would have sent.
///
/// `#[non_exhaustive]` — see [`StatusKind`].
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StatusSeverity {
    /// The producer did not set a severity (proto3's zero value).
    ///
    /// Ranks below `Info`: the event is still delivered and still carries a
    /// `kind` and `message`, but a severity floor cannot promote a level that
    /// was never stated.
    Unspecified,
    /// Worth knowing, not worth waking anyone (IBD completing).
    Info,
    /// Degraded but functioning (peer starvation, a congested mempool).
    Warning,
    /// The node is not doing its job, or is about to stop (a stalled tip, a
    /// filling disk, a deep reorg).
    Critical,
    /// A severity this client build does not recognize. Ranks above `Critical`.
    Unknown(i32),
}

impl StatusSeverity {
    fn from_proto(v: i32) -> Self {
        match pb::StatusSeverity::try_from(v) {
            Ok(pb::StatusSeverity::Unspecified) => StatusSeverity::Unspecified,
            Ok(pb::StatusSeverity::Info) => StatusSeverity::Info,
            Ok(pb::StatusSeverity::Warning) => StatusSeverity::Warning,
            Ok(pb::StatusSeverity::Critical) => StatusSeverity::Critical,
            _ => StatusSeverity::Unknown(v),
        }
    }

    /// Sort key: the severity rank, then the raw value.
    ///
    /// The second element only ever distinguishes two `Unknown`s, and it is
    /// required rather than cosmetic: `Ord` must agree with `Eq`, and the
    /// derived `Eq` says `Unknown(5) != Unknown(9)`. Ranking both as simply
    /// "above Critical" and stopping there would report them equal, which
    /// breaks the total order and, with it, `BTreeMap`/`sort` on this type.
    fn rank(self) -> (u8, i32) {
        match self {
            StatusSeverity::Unspecified => (0, 0),
            StatusSeverity::Info => (1, 0),
            StatusSeverity::Warning => (2, 0),
            StatusSeverity::Critical => (3, 0),
            StatusSeverity::Unknown(v) => (4, v),
        }
    }
}

impl Ord for StatusSeverity {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        self.rank().cmp(&other.rank())
    }
}

impl PartialOrd for StatusSeverity {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

impl From<pb::EvictReason> for EvictReason {
    fn from(r: pb::EvictReason) -> Self {
        match r {
            pb::EvictReason::Unspecified => EvictReason::Unspecified,
            pb::EvictReason::FullPool => EvictReason::FullPool,
            pb::EvictReason::Expiry => EvictReason::Expiry,
            pb::EvictReason::BlockConflict => EvictReason::BlockConflict,
            pb::EvictReason::Policy => EvictReason::Policy,
            pb::EvictReason::Reorg => EvictReason::Reorg,
        }
    }
}

impl EvictReason {
    /// Decode from the raw wire value.
    ///
    /// This exists because prost's generated `reason()` accessor is
    /// `try_from(..).unwrap_or_default()`, which turns *any* unrecognized value
    /// into the zero variant. Going through it made `EvictReason::Unknown`
    /// unconstructible and reported a reason added by a newer node as
    /// "unspecified" — indistinguishable from a producer that omitted the
    /// field. Decode from the `i32` instead, as every other open enum here
    /// does.
    fn from_proto(v: i32) -> Self {
        match pb::EvictReason::try_from(v) {
            Ok(r) => r.into(),
            Err(_) => EvictReason::Unknown(v),
        }
    }
}

/// A k-bit prefix of `sha256(scriptPubKey)`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ScriptPrefix {
    /// The top `ceil(bits/8)` bytes, big-endian.
    pub prefix: Vec<u8>,
    /// Prefix length in bits.
    pub bits: u32,
}

/// One of a transaction's taproot outputs — a scan candidate. Carried in
/// [`TweakEntry::taproot_outputs`] so a client can confirm a derived output key
/// against the actual on-chain output without fetching the transaction.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TaprootOutput {
    /// Output index within the transaction.
    pub vout: u32,
    /// 32-byte x-only taproot output key, internal (consensus) byte order — the
    /// raw `scriptPubKey` push, unreversed. Compare it directly against a
    /// derived `P_k.x_only_public_key().serialize()` with no byte flip.
    pub output_pubkey: Vec<u8>,
    /// Output value in satoshis.
    pub value: u64,
}

/// One transaction's public silent-payment tweak, carried in a
/// [`Event::BlockTweaks`] for Tier 1 client-side scanning.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TweakEntry {
    /// 33-byte compressed public tweak `T = input_hash · A` for the transaction.
    /// Always present — this is what a client feeds its own `b_scan` into to
    /// scan the transaction's outputs locally.
    pub tweak: Vec<u8>,
    /// The transaction's txid (internal/consensus byte order — [`parse_txid`] /
    /// [`display_hex`]). Empty when the subscription set `tweaks_only` (the
    /// compact, tweak-alone form).
    pub txid: Vec<u8>,
    /// The largest taproot output value in the transaction, in satoshis — a cap
    /// on what a payment here could be worth, for client-side dust triage. `0`
    /// under `tweaks_only`.
    pub max_value: u64,
    /// The transaction's taproot outputs (the scan candidates). Always populated
    /// on a [`Event::MempoolTweak`] — the whole point of Tier 1.5, since there
    /// is no block to fetch to recover them. On a [`Event::BlockTweaks`] entry it
    /// is present only when the subscription set `tweak_outputs`; empty
    /// otherwise (the confirmed block is the fallback).
    pub taproot_outputs: Vec<TaprootOutput>,
}

/// A spent prevout that matched a prefix bucket (the spend side of a
/// [`Event::PrefixMatched`]).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SpentPrevout {
    /// The outpoint consumed.
    pub outpoint: Outpoint,
    /// The `scriptPubKey` it paid. Empty when the server did not retain it
    /// (mempool spend below the `full` retention tier) — resolve the outpoint
    /// yourself in that case.
    pub script_pubkey: Vec<u8>,
    /// The prevout value in satoshis. `None` when not retained (distinct from a
    /// genuine 0-value prevout, which is `Some(0)`).
    pub amount: Option<u64>,
}

impl From<pb::SpentPrevout> for SpentPrevout {
    fn from(s: pb::SpentPrevout) -> Self {
        SpentPrevout {
            outpoint: Outpoint { txid: s.outpoint_txid, vout: s.outpoint_vout },
            script_pubkey: s.script_pubkey,
            amount: s.has_amount.then_some(s.amount),
        }
    }
}

/// A transaction that fell inside a watched script-prefix bucket. Carries the
/// full serialized tx so the client filters the bucket against its real scripts
/// locally (the privacy contract — the server only learns the bucket).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PrefixMatch {
    /// The registered bucket that fired.
    pub prefix: ScriptPrefix,
    /// Consensus-serialized matching transaction.
    pub raw_tx: Vec<u8>,
    /// `false` = mempool, `true` = connected block.
    pub confirmed: bool,
    /// Block height when confirmed; 0 in the mempool.
    pub height: u32,
    /// Spend-side matched prevouts; empty for a pure funding (output) match.
    pub matched_prevouts: Vec<SpentPrevout>,
}

/// A typed streaming event — the flat mirror of `NodeEvent.body`.
///
/// `Eq` is not derived: the [`Lagged`](Event::Lagged) resume cursor is a
/// prost-generated type that is only `PartialEq`.
#[derive(Debug, Clone, PartialEq)]
#[non_exhaustive]
pub enum Event {
    /// A transaction entered the mempool.
    MempoolEnter {
        /// Transaction id.
        txid: Vec<u8>,
        /// Fee in satoshis.
        fee: u64,
        /// Virtual size in vbytes.
        vsize: u64,
        /// Fee rate in sat/kvB.
        fee_rate_sat_per_kvb: u64,
        /// Admission time, seconds since the Unix epoch.
        time: u64,
    },
    /// A mempool transaction was confirmed in a block.
    MempoolLeaveConfirmed {
        /// Transaction id.
        txid: Vec<u8>,
        /// Confirming block hash.
        block_hash: Vec<u8>,
        /// Confirming block height.
        height: u32,
    },
    /// A mempool transaction was evicted by policy.
    MempoolLeaveEvicted {
        /// Transaction id.
        txid: Vec<u8>,
        /// Eviction reason.
        reason: EvictReason,
    },
    /// A mempool transaction was replaced (RBF).
    MempoolLeaveReplaced {
        /// Transaction id.
        txid: Vec<u8>,
        /// The replacing transaction id.
        replacing_txid: Vec<u8>,
    },
    /// A block was connected to the active chain.
    BlockConnected {
        /// Block hash.
        hash: Vec<u8>,
        /// Block height.
        height: u32,
    },
    /// A block was disconnected (reorg).
    BlockDisconnected {
        /// Block hash.
        hash: Vec<u8>,
        /// Block height.
        height: u32,
    },
    /// A reorg marker, emitted once before the disconnect/connect sequence.
    Reorg {
        /// Height of the abandoned tip.
        from_height: u32,
        /// Hash of the abandoned tip.
        old_tip: Vec<u8>,
        /// Height of the new active tip.
        to_height: u32,
        /// Hash of the new active tip.
        new_tip: Vec<u8>,
    },
    /// A periodic heartbeat.
    Heartbeat {
        /// Publisher uptime in nanoseconds.
        uptime_ns: u64,
    },
    /// A node-health condition the daemon detected about **itself** — a stalled
    /// tip, a filling disk, peer starvation. Arrives only for a subscription
    /// that set the [`STATUS`](crate::Categories::STATUS) category bit, which
    /// is deliberately not part of the `0` ("all") default.
    ///
    /// Standing conditions are level-triggered: exactly one
    /// [`StatusState::Raised`] on entry and one [`StatusState::Cleared`] on
    /// recovery, with hysteresis so a value hovering at the threshold does not
    /// flap. Observations with no recovered state (IBD finishing, a deep reorg
    /// landing) are [`StatusState::Edge`].
    ///
    /// **Not replayable, and the gap is not observable.** A status event carries
    /// no cursor, so a `from_cursor` resume never yields one; the node re-raises
    /// standing conditions after a restart instead. A client that connects
    /// *after* a condition was raised will not see it until the condition
    /// changes.
    ///
    /// This is not something the reconnect layer can paper over.
    /// [`ResilientSubscription`](crate::ResilientSubscription) hides reconnects
    /// by design, and its one synthetic notice
    /// ([`Event::ReplayGap`]) is cursor-anchored — which status has no part in.
    /// So a drop during which one condition raised and another cleared leaves a
    /// client's picture silently wrong, with nothing to key recovery off.
    ///
    /// Treat this stream as the low-latency edge signal and **`getwarnings` over
    /// JSON-RPC as the authority** on what is wrong right now: poll it on a slow
    /// timer as well as consuming this, and a missed transition self-corrects at
    /// the next poll instead of persisting for the life of the process. Seeding
    /// once on connect is not enough, because reconnects after that are
    /// invisible to you.
    ///
    /// Match on [`kind`](StatusKind) and read [`details`](Event::Status) for the
    /// numbers; treat [`StatusKind::Unknown`] as informational rather than an
    /// error, since new conditions are added additively. See
    /// [`examples/health_watch.rs`].
    Status {
        /// Which condition this describes.
        kind: StatusKind,
        /// Whether it was entered, recovered, or is a one-shot observation.
        state: StatusState,
        /// How loud it is. Ordered, so a severity floor is a comparison.
        severity: StatusSeverity,
        /// Human-readable one-liner. Log it or page on it; do not parse it.
        message: String,
        /// Kind-specific structured fields, as strings.
        ///
        /// Mostly decimal numbers (free bytes and the threshold, seconds since
        /// the last block, a reorg's depth and fork height, …) but **not
        /// uniformly** — a `cleared` event can carry a `reason` token such as
        /// `detector_disabled` or `mempool_cap_zero`. Parse per key, not with
        /// one integer parse over the map.
        ///
        /// The watched *path* is deliberately absent: it reaches every status
        /// subscriber and every webhook receiver, and an absolute datadir path
        /// usually names the account the node runs under.
        ///
        /// Tolerate unknown keys and absent optional ones.
        details: std::collections::BTreeMap<String, String>,
    },
    /// A watched outpoint was spent.
    OutpointSpent {
        /// The spent outpoint.
        outpoint: Outpoint,
        /// The spending transaction id.
        spending_txid: Vec<u8>,
        /// The spending input index.
        spending_vin: u32,
        /// `false` = seen in mempool, `true` = confirmed.
        confirmed: bool,
    },
    /// A watched script was matched on the funding or spending side.
    ScriptMatched {
        /// The matched scripthash.
        scripthash: Vec<u8>,
        /// The matching transaction id.
        txid: Vec<u8>,
        /// `true` = funding (output), `false` = spending (input).
        is_output: bool,
        /// vout if `is_output`, else vin.
        index: u32,
        /// `false` = mempool, `true` = confirmed.
        confirmed: bool,
        /// Matched value in satoshis: the funded output value (`is_output =
        /// true`) or the spent-prevout value (`is_output = false`). `Some` on
        /// the funding side and for confirmed spends; `Some` for mempool spends
        /// when the node retained the prevout value (`streamprevoutmeta >=
        /// amount`), else `None` (hash tier). `None` lets the consumer skip the
        /// enrichment `getrawtransaction` for the common single-coin case.
        amount: Option<u64>,
        /// Full consensus-serialized matching transaction, present only when
        /// this stream opted in via
        /// [`set_watch_options`](crate::WatchControls::set_watch_options) with
        /// `include_raw_tx = true`; `None` otherwise.
        raw_tx: Option<Vec<u8>>,
        /// Descriptor attribution: the descriptor watch(es) this scripthash
        /// belongs to, if it was registered via `add_descriptor`. Empty for a
        /// directly-watched script. See [`DescriptorMatch`].
        descriptors: Vec<DescriptorMatch>,
    },
    /// A watched txid appeared in the mempool or a connected block.
    TxidMatched {
        /// Transaction id.
        txid: Vec<u8>,
        /// `false` = mempool, `true` = confirmed.
        confirmed: bool,
        /// Block height when confirmed; 0 in the mempool.
        height: u32,
    },
    /// A watched tx was replaced in the mempool (RBF).
    TxidReplaced {
        /// Transaction id.
        txid: Vec<u8>,
        /// The replacing transaction id.
        replacing_txid: Vec<u8>,
    },
    /// A watched tx left the mempool by policy.
    TxidEvicted {
        /// Transaction id.
        txid: Vec<u8>,
        /// Free-text reason (`"full_pool"` | `"expiry"` | `"block_conflict"` |
        /// `"policy"`).
        reason: String,
    },
    /// A watched tx's confirming block was rolled back by a reorg.
    TxidUnconfirmed {
        /// Transaction id.
        txid: Vec<u8>,
        /// Height it had been confirmed at, now disconnected.
        prev_height: u32,
    },
    /// A depth alarm fired (single-shot).
    TxidDepthReached {
        /// Transaction id.
        txid: Vec<u8>,
        /// Confirmations reached (>= the requested depth).
        depth: u32,
        /// Active-chain height where the tx is confirmed.
        height: u32,
    },
    /// A lifecycle watch's `auto_close_depth` was reached (terminal).
    TxidFinalized {
        /// Transaction id.
        txid: Vec<u8>,
        /// Confirmations reached (>= `auto_close_depth`).
        depth: u32,
        /// Active-chain height where the tx is confirmed.
        height: u32,
    },
    /// A transaction fell inside a watched script-prefix bucket.
    PrefixMatched(PrefixMatch),
    /// In-band slow-consumer lag notice. Not an error: reconnect (Subscribe) or
    /// re-anchor (Watch) with `resume_cursor` to recover the dropped events.
    Lagged {
        /// Number of events dropped before this notice.
        dropped_count: u64,
        /// The anchor to resume from to recover the gap.
        resume_cursor: Option<Cursor>,
    },
    /// **SDK-synthesized — not a wire event.** Emitted by
    /// [`ResilientSubscription`](crate::ResilientSubscription) when a durable
    /// replay was clamped by the server to the most recent `MAX_REPLAY_BLOCKS`
    /// (10,000) blocks, so the confirmed history in `[resume_height,
    /// first_height - 1]` was skipped. The live stream continues correctly from
    /// `first_height`; the gap is unrecoverable via this stream, so full-resync
    /// the skipped range from another source (e.g. RPC `getblock`). Emitted once
    /// per resume, immediately before the first replayed block.
    ///
    /// **Detection limit — absence of this event is not proof of a complete
    /// scan.** It is synthesized by comparing the resume cursor against the
    /// first *`BlockConnected`* delivered. A subscription that filtered chain
    /// events out — a tweaks-only silent-payment scan, say — receives no
    /// `BlockConnected`, so a clamp it suffered passes unremarked here. Such a
    /// subscription is normally protected server-side instead (a tweaks-only
    /// replay against a complete index is exempt from the clamp, and a node
    /// whose index is incomplete refuses the subscribe outright rather than
    /// clamping it), but that protection lives on the node, not in this event.
    /// Do not treat this variant as a wallet's only guard against a silent hole.
    ReplayGap {
        /// The height the resume cursor expected next (`cursor.height + 1`).
        resume_height: u32,
        /// The first height the server actually delivered (`> resume_height`).
        first_height: u32,
    },
    /// A mid-stream re-anchor ([`WatchHandle::set_cursor`](crate::WatchHandle::set_cursor))
    /// was **admitted**. Confirmed-history replay follows this event (in height
    /// order) before the live tail resumes. When `clamped` is true the requested
    /// cursor predated the server's replay window: replay still runs, but only
    /// from `earliest_replayed`, so full-resync history below it from another
    /// source. This is the deterministic "accepted, replaying from X" signal.
    CursorAccepted {
        /// The cursor the server anchored to.
        from: Option<Cursor>,
        /// `true` → the replay window truncated the lower end of the gap.
        clamped: bool,
        /// First height the server will replay.
        earliest_replayed: u32,
    },
    /// A mid-stream re-anchor ([`WatchHandle::set_cursor`](crate::WatchHandle::set_cursor))
    /// was **not** admitted. The live stream is unchanged (still emitting from
    /// its prior position). Decide whether to retry, back off, or escalate to a
    /// full resnapshot based on `reason`; `current_head` is where the server is
    /// now.
    CursorRejected {
        /// Why the re-anchor was declined.
        reason: CursorRejectReason,
        /// The server's current resume position.
        current_head: Option<Cursor>,
    },
    /// An atomic watch-set replace ([`ResilientWatch::reload`](crate::ResilientWatch::reload),
    /// carried as `SetWatchSet`) was applied. The live watch-set now equals the
    /// reloaded truth; the counts are the server's authoritative diff by
    /// **effective coverage** (a scripthash covered by both the old and new set —
    /// even via a different mechanism — counts as `unchanged`, never gapped).
    WatchSetReplaced {
        /// Items newly watched.
        added: u32,
        /// Items released.
        removed: u32,
        /// Items in both the old and new set (kept without re-registration).
        unchanged: u32,
    },
    /// An atomic watch-set replace was **rejected**; the live watch-set is
    /// UNCHANGED (the prior set is still in effect). `reason` says which ceiling
    /// refused it, and `required`/`quota` are in the matching unit:
    /// [`QuotaExceeded`](WatchSetRejectReason::QuotaExceeded) — `required` units
    /// vs the `quota` ceiling; [`CapExceeded`](WatchSetRejectReason::CapExceeded)
    /// — `required` entries vs the per-connection entry cap (`quota`);
    /// [`Malformed`](WatchSetRejectReason::Malformed) — a client bug (bad element),
    /// `required`/`quota` are 0 and retrying the same set will not help. In every
    /// case the client's mirror still reflects the (unapplied) reloaded set, so a
    /// consumer that ignores this keeps an over-claiming mirror; react by
    /// reloading a set the server accepts.
    WatchSetRejected {
        /// Why the replace was refused.
        reason: WatchSetRejectReason,
        /// What the rejected target needs: units (`QuotaExceeded`) or entries
        /// (`CapExceeded`); 0 for `Malformed`.
        required: u64,
        /// The ceiling that refused it: unit quota (`QuotaExceeded`) or entry cap
        /// (`CapExceeded`); 0 for `Malformed`.
        quota: u64,
    },
    /// A bounded historical rescan ([`ResilientWatch::rescan`](crate::ResilientWatch::rescan))
    /// was **admitted**. Confirmed watch-matches for the scanned range follow
    /// this event (in height order), terminated by a [`RescanComplete`](Event::RescanComplete).
    /// `from_height`/`to_height` are the range the server will ACTUALLY scan:
    /// `clamped` is true when the requested upper bound exceeded the tip and was
    /// narrowed to it. A rescan is a side query — it does not advance the durable
    /// cursor, and its match events carry no resume cursor.
    RescanAccepted {
        /// First height that will be scanned (post-clamp).
        from_height: u32,
        /// Last height that will be scanned (post-clamp).
        to_height: u32,
        /// `true` → the requested range was narrowed to what the node holds.
        clamped: bool,
    },
    /// A bounded historical rescan was **not** admitted; no matches follow and
    /// the live stream is unchanged. `tip_height` is the server's current tip so
    /// a client can re-scope the range and retry.
    RescanRejected {
        /// Why the rescan was declined.
        reason: RescanRejectReason,
        /// The server's current active-chain tip height.
        tip_height: u32,
    },
    /// Terminal marker for a bounded historical rescan: the range has been fully
    /// scanned and every match delivered. `matches` counts the match events this
    /// rescan emitted (0 when the range held none). After this the stream resumes
    /// its prior live position.
    RescanComplete {
        /// The scanned range (post-clamp), echoing [`RescanAccepted`](Event::RescanAccepted).
        from_height: u32,
        /// The scanned range upper bound (post-clamp).
        to_height: u32,
        /// Number of match events emitted for this rescan.
        matches: u64,
    },
    /// One connected block's silent-payment tweaks — the Tier 1 (client-side
    /// scan, zero-custody) firehose payload. Arrives only for a subscription that
    /// set the [`TWEAKS`](crate::Categories::TWEAKS) category bit. For each entry,
    /// a client computes `T · b_scan` with its own scan secret and derives the
    /// candidate output key(s), so the scan key never leaves the device. See
    /// [`examples/sp_light_scan.rs`].
    BlockTweaks {
        /// The block these tweaks describe (internal/consensus byte order).
        block_hash: Vec<u8>,
        /// The block's height.
        height: u32,
        /// One entry per silent-payment-eligible transaction in the block.
        entries: Vec<TweakEntry>,
        /// `true` when a subscription filter dropped entries from this block —
        /// the `tweak_dust_limit` floor, or `tweak_unspent_only` cut-through —
        /// so an empty `entries` may mean "filtered out", not "none present".
        /// A surviving entry always carries its full taproot output set; no
        /// filter trims one.
        filtered: bool,
    },
    /// One accepted-but-unconfirmed transaction's silent-payment tweak — the
    /// Tier 1.5 (mempool-time, zero-custody) firehose payload. Arrives only for a
    /// subscription that set the [`TWEAKS`](crate::Categories::TWEAKS) category
    /// bit **and** [`SubscribeOptions::mempool_tweaks`](crate::SubscribeOptions::mempool_tweaks).
    /// Scan it exactly like a [`BlockTweaks`](Event::BlockTweaks) entry — compute
    /// `T · b_scan` and derive the candidate output key(s) — for mempool-latency
    /// detection. Ephemeral and best-effort: not replayable, no retraction on
    /// RBF/eviction; dedup by `entry.txid` (always present) against the confirmed
    /// `BlockTweaks`. See [`examples/sp_light_scan.rs`].
    MempoolTweak {
        /// The admitted transaction's tweak. Always full (`txid` present) —
        /// `tweaks_only` does not apply to mempool tweaks.
        entry: TweakEntry,
    },
    /// A BIP 352 silent payment paid a registered scan key — the Tier 2
    /// (scan-key watch, convenience) match, delivered on the `Watch` stream to a
    /// connection that registered the target via
    /// [`WatchHandle::add_silent_payments`](crate::WatchHandle::add_silent_payments).
    /// The node ran the ECDH; `tweak` + `k` let the wallet re-derive the output
    /// key — and, with its `b_spend`, the spending key — **offline**. See
    /// [`examples/sp_wallet.rs`].
    SilentPaymentMatched {
        /// The registered target's identity `b_scan·G` (33 bytes) this output
        /// paid — echoes which of your scan keys matched, never the secret.
        scan_pubkey: Vec<u8>,
        /// The paying transaction's txid (internal byte order — [`parse_txid`] /
        /// [`display_hex`]).
        txid: Vec<u8>,
        /// The matched output's index.
        vout: u32,
        /// The matched taproot output key (32-byte x-only).
        output_pubkey: Vec<u8>,
        /// The output value in satoshis.
        amount: u64,
        /// The transaction's 33-byte public tweak `T`. With `k`, re-derive the
        /// full output key offline: `P_k = B_spend + hash(b_scan·T || k)·G`.
        tweak: Vec<u8>,
        /// The BIP 352 output counter for this match — the `k` in the derivation.
        k: u32,
        /// The matched label integer, when the output paid a registered label
        /// (e.g. change is commonly label `0`); `None` for an unlabeled match.
        label: Option<u32>,
        /// `true` once seen in a connected block; `false` while only in the
        /// mempool (re-emitted `true` on confirmation).
        confirmed: bool,
        /// The confirming block height; `None` while unconfirmed.
        height: Option<u32>,
        /// The full serialized transaction, only when this connection opted in
        /// via `SetWatchOptions.include_raw_tx`; `None` otherwise.
        raw_tx: Option<Vec<u8>>,
    },
    /// An incremental watch add (`add_scripts`, `add_outpoints`, …) that the
    /// server did **not** register: over the quota, over the per-add rate
    /// limit, over a per-connection cap, without `stream:watch`, or malformed.
    /// None of the items it names is watched. Items the add re-asserted
    /// (already watched) are not named and stay watched. The server sends
    /// nothing for an add that registered.
    ///
    /// [`ResilientWatch`](crate::ResilientWatch) re-sends a
    /// [`RateLimited`](WatchAddRejectReason::RateLimited) add itself after the
    /// server's retry hint, within its backoff budget, and hands the event on
    /// only once that budget runs out. For any other reason, or then, it drops
    /// the named items from its mirror before handing this on, so a reconnect
    /// does not re-register them; re-add them yourself if you want another try.
    WatchAddRejected(WatchAddRejected),
    /// A body this client build does not recognize (a newer server arm), or an
    /// event with no body set. Ignored by well-behaved consumers.
    Unknown,
}

/// The payload of [`Event::WatchAddRejected`]: which add was refused, why, and
/// the items it named. Only the item field for [`kind`](Self::kind) is set.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct WatchAddRejected {
    /// Which kind of add was refused.
    pub kind: WatchAddKind,
    /// Why it was refused.
    pub reason: WatchAddRejectReason,
    /// [`QuotaExceeded`](WatchAddRejectReason::QuotaExceeded): the units the
    /// refused items cost. [`CapExceeded`](WatchAddRejectReason::CapExceeded):
    /// the count the add would have reached. 0 otherwise.
    pub required: u64,
    /// `QuotaExceeded`: the units the token already holds. 0 otherwise.
    pub held: u64,
    /// `QuotaExceeded`: the token's unit quota. `CapExceeded`: the cap. 0
    /// otherwise.
    pub quota: u64,
    /// [`RateLimited`](WatchAddRejectReason::RateLimited): seconds until the
    /// rate limit admits another add. 0 otherwise.
    pub retry_after_secs: u32,
    /// Refused scripthashes (32 bytes each).
    pub scripthashes: Vec<Vec<u8>>,
    /// Refused outpoints, as `(txid, vout)` (txid in internal byte order).
    pub outpoints: Vec<(Vec<u8>, u32)>,
    /// Refused lifecycle watches, by txid.
    pub txids: Vec<Vec<u8>>,
    /// Refused depth alarms, as `(txid, depth)`.
    pub depth_alarms: Vec<(Vec<u8>, u32)>,
    /// The refused descriptor and the window it asked for.
    pub descriptor: Option<RejectedDescriptor>,
    /// Refused script prefixes, as `(prefix, bits)` with the prefix masked to
    /// `bits`.
    pub prefixes: Vec<(Vec<u8>, u32)>,
    /// Refused silent-payment targets, by scan pubkey `b_scan·G` (33 bytes).
    pub scan_pubkeys: Vec<Vec<u8>>,
}

/// A descriptor named by a [`WatchAddRejected`].
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct RejectedDescriptor {
    /// The descriptor string the add carried.
    pub descriptor: String,
    /// The window size the add asked for.
    pub gap_limit: u32,
    /// The window start the add asked for.
    pub start: u32,
    /// `true` when an earlier window of this descriptor stays watched (a refused
    /// slide); `false` when the descriptor is not watched at all.
    pub kept: bool,
}

/// Which kind of add a [`WatchAddRejected`] refers to.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum WatchAddKind {
    /// `add_scripts`.
    Scripts,
    /// `add_outpoints`.
    Outpoints,
    /// `add_tx_lifecycle`.
    Transactions,
    /// `add_depth_alarms`.
    DepthAlarms,
    /// `add_descriptor`.
    Descriptor,
    /// `add_script_prefixes`.
    ScriptPrefixes,
    /// `add_silent_payments`.
    SilentPayments,
    /// A kind code this client build does not recognize (a newer server).
    Unknown,
}

/// Why the server refused an incremental add (see [`Event::WatchAddRejected`]).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum WatchAddRejectReason {
    /// The refused items' unit cost does not fit the token's watch quota. Remove
    /// watches, or ask the operator for a larger quota.
    QuotaExceeded,
    /// The token's per-add rate limit is spent. The same add can succeed after
    /// `retry_after_secs`.
    RateLimited,
    /// A per-connection cap: 16 silent-payment targets, 256 descriptors, or the
    /// WebSocket entry cap. Remove watches on this connection first.
    CapExceeded,
    /// The token lacks the `stream:watch` capability.
    PermissionDenied,
    /// The add could not be applied as a whole (a `min_values` list not parallel
    /// to its scripthashes, an invalid descriptor, too many txid × depth
    /// pairs). A client bug: the same add will fail again.
    Malformed,
    /// A reason code this client build does not recognize (a newer server).
    Unknown,
}

impl WatchAddRejected {
    pub(crate) fn from_proto(r: pb::WatchAddRejected) -> Self {
        use pb::watch_add_rejected::{Kind, Reason};
        let kind = match Kind::try_from(r.kind) {
            Ok(Kind::Scripts) => WatchAddKind::Scripts,
            Ok(Kind::Outpoints) => WatchAddKind::Outpoints,
            Ok(Kind::Transactions) => WatchAddKind::Transactions,
            Ok(Kind::DepthAlarms) => WatchAddKind::DepthAlarms,
            Ok(Kind::Descriptor) => WatchAddKind::Descriptor,
            Ok(Kind::ScriptPrefixes) => WatchAddKind::ScriptPrefixes,
            Ok(Kind::SilentPayments) => WatchAddKind::SilentPayments,
            Ok(Kind::Unspecified) | Err(_) => WatchAddKind::Unknown,
        };
        let reason = match Reason::try_from(r.reason) {
            Ok(Reason::QuotaExceeded) => WatchAddRejectReason::QuotaExceeded,
            Ok(Reason::RateLimited) => WatchAddRejectReason::RateLimited,
            Ok(Reason::CapExceeded) => WatchAddRejectReason::CapExceeded,
            Ok(Reason::PermissionDenied) => WatchAddRejectReason::PermissionDenied,
            Ok(Reason::Malformed) => WatchAddRejectReason::Malformed,
            Ok(Reason::Unspecified) | Err(_) => WatchAddRejectReason::Unknown,
        };
        let descriptor = (kind == WatchAddKind::Descriptor).then_some(RejectedDescriptor {
            descriptor: r.descriptor,
            gap_limit: r.gap_limit,
            start: r.start,
            kept: r.descriptor_kept,
        });
        WatchAddRejected {
            kind,
            reason,
            required: r.required,
            held: r.held,
            quota: r.quota,
            retry_after_secs: r.retry_after_secs,
            scripthashes: r.scripthashes,
            outpoints: r.outpoints.into_iter().map(|o| (o.txid, o.vout)).collect(),
            txids: r.txids,
            depth_alarms: r.depth_alarms.into_iter().map(|d| (d.txid, d.depth)).collect(),
            descriptor,
            prefixes: r.prefixes.into_iter().map(|p| (p.prefix, p.bits)).collect(),
            scan_pubkeys: r.scan_pubkeys,
        }
    }
}

/// Why a mid-stream re-anchor was declined by the server (see
/// [`Event::CursorRejected`]).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum CursorRejectReason {
    /// Per-principal re-anchor rate limit exceeded — retry after a backoff.
    RateLimited,
    /// Another re-anchor is already draining (only one runs at a time) — retry
    /// once it completes.
    ConcurrentReanchor,
    /// The request carried no cursor (client bug).
    EmptyCursor,
    /// The server has no block source to replay from.
    NoSource,
    /// A reason code this client build does not recognize (a newer server).
    Unknown,
}

/// Why an atomic watch-set replace was declined by the server (see
/// [`Event::WatchSetRejected`]).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum WatchSetRejectReason {
    /// The target set's total unit cost exceeds the principal's quota (`required`
    /// units vs the `quota` ceiling) — shed items and retry. Transient: a smaller
    /// set fits.
    QuotaExceeded,
    /// The target's watch-set **entry** count (`required`) exceeds the
    /// per-connection cap (`quota`, `streamwsmaxsubscriptions`). Distinct from
    /// `QuotaExceeded`: this bound applies even to a no-auth connection with no
    /// quota, and counts entries (a prefix is one) not units — shed entries and
    /// retry.
    CapExceeded,
    /// The server could not parse or expand an element of the snapshot (a bad
    /// scripthash, outpoint, txid, descriptor, or prefix). A full replace is
    /// all-or-nothing, so the whole snapshot was refused. This is a client bug —
    /// retrying the same set will not help.
    Malformed,
    /// A reason code this client build does not recognize (a newer server).
    Unknown,
}

/// Map the proto `WatchSetRejected.Reason` enum (an `i32` on the wire) to the
/// client enum, treating an unrecognized code as [`WatchSetRejectReason::Unknown`].
fn watch_set_reject_reason(reason: i32) -> WatchSetRejectReason {
    match pb::watch_set_rejected::Reason::try_from(reason) {
        Ok(pb::watch_set_rejected::Reason::QuotaExceeded) => WatchSetRejectReason::QuotaExceeded,
        Ok(pb::watch_set_rejected::Reason::CapExceeded) => WatchSetRejectReason::CapExceeded,
        Ok(pb::watch_set_rejected::Reason::Malformed) => WatchSetRejectReason::Malformed,
        Ok(pb::watch_set_rejected::Reason::Unspecified) | Err(_) => WatchSetRejectReason::Unknown,
    }
}

/// Why a bounded historical rescan was declined by the server (see
/// [`Event::RescanRejected`]).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum RescanRejectReason {
    /// Per-principal rescan rate limit exceeded — retry after a backoff.
    RateLimited,
    /// Another rescan is already draining on this connection (only one runs at a
    /// time) — retry once it completes.
    ConcurrentRescan,
    /// `to_height < from_height`, or the range lies entirely above the tip.
    InvalidRange,
    /// The (clamped) span exceeds the server cap — page the range into smaller
    /// rescans.
    RangeTooLarge,
    /// The server has no block-scan source (no local block bodies/undo).
    NoSource,
    /// The connection watches nothing — a rescan could match nothing. Register a
    /// watch-set first.
    EmptyWatchSet,
    /// A reason code this client build does not recognize (a newer server).
    Unknown,
}

/// Map the proto `RescanRejected.Reason` enum (an `i32` on the wire) to the
/// client enum, treating an unrecognized code as [`RescanRejectReason::Unknown`].
fn rescan_reject_reason(reason: i32) -> RescanRejectReason {
    match pb::rescan_rejected::Reason::try_from(reason) {
        Ok(pb::rescan_rejected::Reason::RateLimited) => RescanRejectReason::RateLimited,
        Ok(pb::rescan_rejected::Reason::ConcurrentRescan) => RescanRejectReason::ConcurrentRescan,
        Ok(pb::rescan_rejected::Reason::InvalidRange) => RescanRejectReason::InvalidRange,
        Ok(pb::rescan_rejected::Reason::RangeTooLarge) => RescanRejectReason::RangeTooLarge,
        Ok(pb::rescan_rejected::Reason::NoSource) => RescanRejectReason::NoSource,
        Ok(pb::rescan_rejected::Reason::EmptyWatchSet) => RescanRejectReason::EmptyWatchSet,
        Ok(pb::rescan_rejected::Reason::Unspecified) | Err(_) => RescanRejectReason::Unknown,
    }
}

impl From<pb::NodeEvent> for Event {
    fn from(ev: pb::NodeEvent) -> Self {
        use pb::node_event::Body;
        let Some(body) = ev.body else { return Event::Unknown };
        match body {
            Body::Mempool(m) => mempool_event(m),
            Body::Chain(c) => chain_event(c),
            Body::Heartbeat(h) => Event::Heartbeat { uptime_ns: h.uptime_ns },
            Body::Status(s) => Event::Status {
                kind: StatusKind::from_proto(s.kind),
                state: StatusState::from_proto(s.state),
                severity: StatusSeverity::from_proto(s.severity),
                message: s.message,
                details: s.details.into_iter().collect(),
            },
            Body::OutpointSpent(o) => Event::OutpointSpent {
                outpoint: Outpoint { txid: o.outpoint_txid, vout: o.outpoint_vout },
                spending_txid: o.spending_txid,
                spending_vin: o.spending_vin,
                confirmed: o.confirmed,
            },
            Body::ScriptMatched(s) => Event::ScriptMatched {
                scripthash: s.scripthash,
                txid: s.txid,
                is_output: s.is_output,
                index: s.index,
                confirmed: s.confirmed,
                amount: s.has_amount.then_some(s.amount),
                raw_tx: (!s.raw_tx.is_empty()).then_some(s.raw_tx),
                descriptors: s
                    .descriptor_matches
                    .into_iter()
                    .map(|d| DescriptorMatch {
                        descriptor: d.descriptor,
                        branch: d.branch,
                        derivation_index: d.derivation_index,
                    })
                    .collect(),
            },
            Body::TxidMatched(t) => Event::TxidMatched {
                txid: t.txid,
                confirmed: t.confirmed,
                height: t.height,
            },
            Body::TxidReplaced(t) => Event::TxidReplaced {
                txid: t.txid,
                replacing_txid: t.replacing_txid,
            },
            Body::TxidEvicted(t) => Event::TxidEvicted { txid: t.txid, reason: t.reason },
            Body::TxidUnconfirmed(t) => Event::TxidUnconfirmed {
                txid: t.txid,
                prev_height: t.prev_height,
            },
            Body::TxidDepthReached(t) => Event::TxidDepthReached {
                txid: t.txid,
                depth: t.depth,
                height: t.height,
            },
            Body::TxidFinalized(t) => Event::TxidFinalized {
                txid: t.txid,
                depth: t.depth,
                height: t.height,
            },
            // A PrefixMatched without its bucket is a degenerate message the
            // local re-filter cannot use (bits:0 matches nothing meaningfully);
            // surface it as Unknown rather than a structurally-valid-looking 0.
            Body::PrefixMatched(p) => match p.prefix {
                Some(sp) => Event::PrefixMatched(PrefixMatch {
                    prefix: ScriptPrefix { prefix: sp.prefix, bits: sp.bits },
                    raw_tx: p.raw_tx,
                    confirmed: p.confirmed,
                    height: p.height,
                    matched_prevouts: p.matched_prevouts.into_iter().map(Into::into).collect(),
                }),
                None => Event::Unknown,
            },
            Body::Lagged(l) => Event::Lagged {
                dropped_count: l.dropped_count,
                resume_cursor: l.resume_cursor,
            },
            Body::SetCursorResult(r) => match r.outcome {
                Some(pb::set_cursor_result::Outcome::Accepted(a)) => Event::CursorAccepted {
                    from: a.from,
                    clamped: a.clamped,
                    earliest_replayed: a.earliest_replayed,
                },
                Some(pb::set_cursor_result::Outcome::Rejected(rj)) => Event::CursorRejected {
                    reason: cursor_reject_reason(rj.reason),
                    current_head: rj.current_head,
                },
                // A result frame with no outcome set is a degenerate message.
                None => Event::Unknown,
            },
            Body::SetWatchSetResult(r) => match r.outcome {
                Some(pb::watch_set_result::Outcome::Accepted(a)) => Event::WatchSetReplaced {
                    added: a.added,
                    removed: a.removed,
                    unchanged: a.unchanged,
                },
                Some(pb::watch_set_result::Outcome::Rejected(rj)) => Event::WatchSetRejected {
                    reason: watch_set_reject_reason(rj.reason),
                    required: rj.required,
                    quota: rj.quota,
                },
                None => Event::Unknown,
            },
            Body::WatchAddRejected(r) => Event::WatchAddRejected(WatchAddRejected::from_proto(r)),
            Body::RescanResult(r) => match r.outcome {
                Some(pb::rescan_result::Outcome::Accepted(a)) => Event::RescanAccepted {
                    from_height: a.from_height,
                    to_height: a.to_height,
                    clamped: a.clamped,
                },
                Some(pb::rescan_result::Outcome::Rejected(rj)) => Event::RescanRejected {
                    reason: rescan_reject_reason(rj.reason),
                    tip_height: rj.tip_height,
                },
                None => Event::Unknown,
            },
            Body::RescanComplete(c) => Event::RescanComplete {
                from_height: c.from_height,
                to_height: c.to_height,
                matches: c.matches,
            },
            // BIP 352 silent payments (§7.7). Tier 1 firehose tweaks and the
            // Tier 2 scan-key match.
            Body::BlockTweaks(b) => Event::BlockTweaks {
                block_hash: b.block_hash,
                height: b.height,
                entries: b
                    .entries
                    .into_iter()
                    .map(|e| TweakEntry {
                        tweak: e.tweak,
                        txid: e.txid,
                        max_value: e.max_value,
                        taproot_outputs: e
                            .taproot_outputs
                            .into_iter()
                            .map(taproot_output_from_proto)
                            .collect(),
                    })
                    .collect(),
                filtered: b.filtered,
            },
            Body::MempoolTweak(m) => Event::MempoolTweak {
                entry: m
                    .entry
                    .map(|e| TweakEntry {
                        tweak: e.tweak,
                        txid: e.txid,
                        max_value: e.max_value,
                        taproot_outputs: e
                            .taproot_outputs
                            .into_iter()
                            .map(taproot_output_from_proto)
                            .collect(),
                    })
                    // A well-formed MempoolTweak always carries its entry; an
                    // absent one (malformed wire) degrades to an empty entry
                    // rather than dropping the event.
                    .unwrap_or(TweakEntry {
                        tweak: Vec::new(),
                        txid: Vec::new(),
                        max_value: 0,
                        taproot_outputs: Vec::new(),
                    }),
            },
            Body::SilentPaymentMatched(s) => Event::SilentPaymentMatched {
                scan_pubkey: s.scan_pubkey,
                txid: s.txid,
                vout: s.vout,
                output_pubkey: s.output_pubkey,
                amount: s.amount,
                tweak: s.tweak,
                k: s.k,
                label: s.has_label.then_some(s.label),
                confirmed: s.confirmed,
                // `height` is 0 on the wire while unconfirmed; surface `None`.
                height: s.confirmed.then_some(s.height),
                raw_tx: (!s.raw_tx.is_empty()).then_some(s.raw_tx),
            },
            // Forward-compatible catch-all: any body a still-newer proto adds maps
            // to `Unknown`. It is unreachable against the *current* workspace proto
            // (every body above is now typed), but retained so this crate keeps
            // compiling if the schema grows ahead of these arms — the same
            // resilience the SP bodies themselves relied on before they were typed.
            #[allow(unreachable_patterns)]
            _ => Event::Unknown,
        }
    }
}

/// Map a proto [`pb::TaprootOutput`] to the client [`TaprootOutput`]. Bytes are
/// carried through unchanged (x-only key, internal byte order).
fn taproot_output_from_proto(o: pb::TaprootOutput) -> TaprootOutput {
    TaprootOutput {
        vout: o.vout,
        output_pubkey: o.output_pubkey,
        value: o.value,
    }
}

/// Map the proto `CursorRejected.Reason` enum (an `i32` on the wire) to the
/// client enum, treating an unrecognized code as [`CursorRejectReason::Unknown`].
fn cursor_reject_reason(reason: i32) -> CursorRejectReason {
    match pb::cursor_rejected::Reason::try_from(reason) {
        Ok(pb::cursor_rejected::Reason::RateLimited) => CursorRejectReason::RateLimited,
        Ok(pb::cursor_rejected::Reason::ConcurrentReanchor) => {
            CursorRejectReason::ConcurrentReanchor
        }
        Ok(pb::cursor_rejected::Reason::EmptyCursor) => CursorRejectReason::EmptyCursor,
        Ok(pb::cursor_rejected::Reason::NoSource) => CursorRejectReason::NoSource,
        Ok(pb::cursor_rejected::Reason::Unspecified) | Err(_) => CursorRejectReason::Unknown,
    }
}

fn mempool_event(m: pb::MempoolEvent) -> Event {
    use pb::mempool_event::Body;
    match m.body {
        Some(Body::Enter(e)) => Event::MempoolEnter {
            txid: e.txid,
            fee: e.fee,
            vsize: e.vsize,
            fee_rate_sat_per_kvb: e.fee_rate_sat_per_kvb,
            time: e.time,
        },
        Some(Body::LeaveConfirmed(e)) => Event::MempoolLeaveConfirmed {
            txid: e.txid,
            block_hash: e.block_hash,
            height: e.height,
        },
        Some(Body::LeaveEvicted(e)) => {
            // NOT `e.reason()` — that accessor collapses an unrecognized value
            // into the zero variant. See `EvictReason::from_proto`.
            let reason = EvictReason::from_proto(e.reason);
            Event::MempoolLeaveEvicted { txid: e.txid, reason }
        }
        Some(Body::LeaveReplaced(e)) => Event::MempoolLeaveReplaced {
            txid: e.txid,
            replacing_txid: e.replacing_txid,
        },
        None => Event::Unknown,
    }
}

fn chain_event(c: pb::ChainEvent) -> Event {
    use pb::chain_event::Body;
    match c.body {
        Some(Body::BlockConnected(b)) => Event::BlockConnected { hash: b.hash, height: b.height },
        Some(Body::BlockDisconnected(b)) => {
            Event::BlockDisconnected { hash: b.hash, height: b.height }
        }
        Some(Body::Reorg(r)) => Event::Reorg {
            from_height: r.from_height,
            old_tip: r.old_tip,
            to_height: r.to_height,
            new_tip: r.new_tip,
        },
        None => Event::Unknown,
    }
}

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

    #[test]
    fn an_unset_severity_decodes_to_unspecified_not_unknown() {
        // proto3's zero value. Before this was its own variant it decoded to
        // `Unknown(0)`, which ranked above `Critical`.
        assert_eq!(StatusSeverity::from_proto(0), StatusSeverity::Unspecified);
    }

    #[test]
    fn an_unset_severity_ranks_below_info() {
        assert!(StatusSeverity::Unspecified < StatusSeverity::Info);
        assert!(StatusSeverity::Unspecified < StatusSeverity::Critical);
        // The regression this exists for: the documented filter must not page
        // on a severity the producer never set.
        assert!(StatusSeverity::from_proto(0) < StatusSeverity::Warning);
    }

    #[test]
    fn known_severities_order_by_loudness() {
        assert!(StatusSeverity::Info < StatusSeverity::Warning);
        assert!(StatusSeverity::Warning < StatusSeverity::Critical);
        for (v, want) in [
            (1, StatusSeverity::Info),
            (2, StatusSeverity::Warning),
            (3, StatusSeverity::Critical),
        ] {
            assert_eq!(StatusSeverity::from_proto(v), want);
        }
    }

    #[test]
    fn an_unrecognized_severity_still_outranks_critical() {
        // Deliberate and unchanged: a level this build cannot name is not one
        // to quietly filter out. A future `notice` at proto 4 would land here
        // and page — the accepted cost of failing loud on an open taxonomy.
        let future = StatusSeverity::from_proto(4);
        assert_eq!(future, StatusSeverity::Unknown(4));
        assert!(future > StatusSeverity::Critical);
        assert!(future >= StatusSeverity::Warning);
    }

    #[test]
    fn ord_agrees_with_eq_for_distinct_unknowns() {
        // `Ord` must report `Equal` exactly when `Eq` does. Ranking every
        // `Unknown` as simply "above Critical" would call these equal while
        // `Eq` calls them different, breaking the total order.
        let a = StatusSeverity::Unknown(5);
        let b = StatusSeverity::Unknown(9);
        assert_ne!(a, b);
        assert_ne!(a.cmp(&b), std::cmp::Ordering::Equal);
        assert!(a < b);
        assert_eq!(a.cmp(&a), std::cmp::Ordering::Equal);
    }

    #[test]
    fn sorting_puts_the_loudest_last_and_the_unset_first() {
        let mut all = vec![
            StatusSeverity::Critical,
            StatusSeverity::Unknown(7),
            StatusSeverity::Unspecified,
            StatusSeverity::Warning,
            StatusSeverity::Info,
        ];
        all.sort();
        assert_eq!(all, vec![
            StatusSeverity::Unspecified,
            StatusSeverity::Info,
            StatusSeverity::Warning,
            StatusSeverity::Critical,
            StatusSeverity::Unknown(7),
        ]);
    }

    #[test]
    fn the_documented_filter_idiom_selects_what_it_should() {
        let page = |s: StatusSeverity| s >= StatusSeverity::Warning;
        assert!(!page(StatusSeverity::Unspecified));
        assert!(!page(StatusSeverity::Info));
        assert!(page(StatusSeverity::Warning));
        assert!(page(StatusSeverity::Critical));
        assert!(page(StatusSeverity::Unknown(4)));
    }

    #[test]
    fn unspecified_and_unknown_zero_are_different_values() {
        // Both mean "proto 0", but only `from_proto` produces `Unspecified`.
        // `Unknown(0)` is still constructible by hand (`#[non_exhaustive]`
        // blocks exhaustive matching, not construction) and deliberately keeps
        // the loud rank — the asymmetry is documented rather than papered over,
        // because nothing in the decode path can produce it.
        assert_ne!(StatusSeverity::Unspecified, StatusSeverity::Unknown(0));
        assert!(StatusSeverity::Unknown(0) > StatusSeverity::Critical);
        assert!(StatusSeverity::Unspecified < StatusSeverity::Info);
    }

    #[test]
    fn the_order_is_total_across_mixed_sign_unknowns() {
        // `rank()`'s second element exists to keep this a total order; a
        // rank-only key would report every `Unknown` equal and break sorting.
        let mut all = vec![
            StatusSeverity::Unknown(i32::MAX),
            StatusSeverity::Critical,
            StatusSeverity::Unknown(i32::MIN),
            StatusSeverity::Unspecified,
            StatusSeverity::Unknown(0),
            StatusSeverity::Info,
        ];
        all.sort();
        assert_eq!(all, vec![
            StatusSeverity::Unspecified,
            StatusSeverity::Info,
            StatusSeverity::Critical,
            StatusSeverity::Unknown(i32::MIN),
            StatusSeverity::Unknown(0),
            StatusSeverity::Unknown(i32::MAX),
        ]);
        // Every unknown, including the extremes, still outranks Critical.
        for v in [i32::MIN, -1, 0, 1, i32::MAX] {
            assert!(StatusSeverity::Unknown(v) > StatusSeverity::Critical, "Unknown({v})");
        }
        // Transitivity spot-check across the variant boundary.
        assert!(StatusSeverity::Unspecified < StatusSeverity::Warning);
        assert!(StatusSeverity::Warning < StatusSeverity::Unknown(i32::MIN));
        assert!(StatusSeverity::Unspecified < StatusSeverity::Unknown(i32::MIN));
    }

    #[test]
    fn a_negative_severity_is_unrecognized_not_unset() {
        assert_eq!(StatusSeverity::from_proto(-1), StatusSeverity::Unknown(-1));
        assert!(StatusSeverity::from_proto(-1) > StatusSeverity::Critical);
    }

    #[test]
    fn the_sibling_enums_split_unset_from_unrecognized_too() {
        assert_eq!(StatusKind::from_proto(0), StatusKind::Unspecified);
        assert_eq!(StatusKind::from_proto(99), StatusKind::Unknown(99));
        assert_eq!(StatusState::from_proto(0), StatusState::Unspecified);
        assert_eq!(StatusState::from_proto(99), StatusState::Unknown(99));
        assert_eq!(EvictReason::from_proto(0), EvictReason::Unspecified);
        assert_eq!(EvictReason::from_proto(99), EvictReason::Unknown(99));
        assert_eq!(EvictReason::from_proto(1), EvictReason::FullPool);
    }
}