satd-events-client 0.5.2

//! The streaming client: connect, `Subscribe` firehose, and the `Watch` handle
//! with its typed watch helpers. This module establishes the connection, auth,
//! and cursor-capturing event stream that the reconnect/replay resilience layer
//! (the `resilience` module) builds on.

use std::time::Duration;

use satd_events_proto::v1 as pb;
use satd_events_proto::v1::node_event_stream_client::NodeEventStreamClient;
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
use tonic::metadata::{Ascii, MetadataValue};
use tonic::transport::{Channel, Endpoint};

use crate::error::StreamError;
use crate::event::{Cursor, Event};

/// Bounded control-channel depth for a `Watch` stream.
const CONTROL_BUFFER: usize = 16;

/// Category bitfield helpers for [`SubscribeOptions::categories`].
///
/// `0` (the default, [`Categories::ALL`]) means "all categories". Combine the
/// others with `|`.
pub struct Categories;

impl Categories {
    /// All categories (the server default).
    pub const ALL: u32 = 0;
    /// Mempool events.
    pub const MEMPOOL: u32 = 1;
    /// Chain (block connect/disconnect/reorg) events.
    pub const CHAIN: u32 = 2;
    /// Heartbeats.
    pub const HEARTBEAT: u32 = 4;
    /// BIP 352 silent-payment per-block tweaks (Tier 1 client-side scan). Unlike
    /// the other bits this is **not** part of the [`ALL`](Self::ALL) (`0`)
    /// default: a subscription must request it explicitly, so an existing
    /// `0`-subscriber never starts receiving tweak volume after a node upgrade.
    /// Requires the node's tweak index (`silentpaymentindex=1`); a `tweaks`
    /// subscription against a node with it disabled is refused in-band.
    pub const TWEAKS: u32 = 8;
    /// Node-health conditions the daemon detects about itself
    /// ([`Event::Status`](crate::Event::Status)). Like
    /// [`TWEAKS`](Self::TWEAKS) this is **not** part of the [`ALL`](Self::ALL)
    /// (`0`) default — a client written against an older node never starts
    /// receiving a body it has no parser for. Unlike `TWEAKS` there is no index
    /// prerequisite: any node serves it.
    pub const STATUS: u32 = 16;
}

/// Options for a [`StreamClient::subscribe`] firehose.
#[derive(Debug, Clone, Default)]
pub struct SubscribeOptions {
    /// Category bitfield; `0` = all. See [`Categories`].
    pub categories: u32,
    /// Durable replay anchor. When set, the server replays confirmed history
    /// forward from this cursor, then joins live with no gap or duplicate.
    pub from_cursor: Option<Cursor>,
    /// Forward-only dedup filter: drop events with `seq <= since_seq`. Use after
    /// a brief reconnect within the broadcast window — not for durable replay.
    pub since_seq: Option<u64>,
    /// Silent-payment tweak dust floor (satoshis): drop [`TweakEntry`](crate::TweakEntry)
    /// whose `max_value` is below this, and flag the block `filtered`. Only
    /// meaningful with the [`TWEAKS`](Categories::TWEAKS) category. `None` = keep
    /// every tweak.
    pub tweak_dust_limit: Option<u64>,
    /// Compact tweak form: when `true`, tweak entries carry only the 33-byte
    /// tweak (no `txid`/`max_value`), minimizing bytes for a pure client-side
    /// scan. Only meaningful with the [`TWEAKS`](Categories::TWEAKS) category.
    pub tweaks_only: bool,
    /// Mempool-time tweaks ("Tier 1.5"): when `true`, the server additionally
    /// streams an [`Event::MempoolTweak`](crate::Event::MempoolTweak) at each
    /// SP-eligible transaction's admission, for mempool-latency detection without
    /// uploading a scan key. A modifier on the [`TWEAKS`](Categories::TWEAKS)
    /// category — the server rejects it if that bit is not set. `false` (default)
    /// = block tweaks only. Ephemeral and best-effort: mempool tweaks are not
    /// replayable, so a missed admission is caught at confirmation via
    /// [`Event::BlockTweaks`](crate::Event::BlockTweaks).
    pub mempool_tweaks: bool,
    /// Include the transaction's taproot outputs on each
    /// [`Event::BlockTweaks`](crate::Event::BlockTweaks) entry
    /// ([`TweakEntry::taproot_outputs`](crate::TweakEntry::taproot_outputs)), so
    /// a match is confirmed against the on-chain output without fetching the
    /// block. `false` (default) keeps the confirmed firehose lean. A modifier on
    /// the [`TWEAKS`](Categories::TWEAKS) category — the server rejects it if that
    /// bit is not set, or if the node has no block source. A
    /// [`MempoolTweak`](crate::Event::MempoolTweak) carries its outputs
    /// regardless of this flag.
    pub tweak_outputs: bool,
    /// Cut-through: drop [`Event::BlockTweaks`](crate::Event::BlockTweaks)
    /// entries whose taproot outputs are **all** already spent on the confirmed
    /// chain. A balance scan then never runs ECDH against a coin that is already
    /// gone. A surviving entry carries its **full** taproot output set —
    /// spentness decides whether an entry survives, never which outputs it
    /// carries, so BIP 352 `k` enumeration is never truncated.
    ///
    /// A **restore** that needs transaction history must leave this `false`: a
    /// payment received and later spent is omitted entirely, so the balance is
    /// right and the history is not. Spentness is evaluated at serve time, so the
    /// same historical block can yield fewer entries on a later scan. Dropped
    /// entries set the block's `filtered` flag. A modifier on the
    /// [`TWEAKS`](Categories::TWEAKS) category — the server rejects it without
    /// that bit, or on a node with no block source. Never applies to
    /// [`MempoolTweak`](crate::Event::MempoolTweak).
    pub tweak_unspent_only: bool,
}

impl SubscribeOptions {
    // `..Default::default()` is redundant against the current workspace proto but
    // retained so this published crate still compiles if the schema grows fields
    // ahead of this literal (the same resilience the released-proto pin needs).
    #[allow(clippy::needless_update)]
    fn into_request(self) -> pb::SubscribeRequest {
        pb::SubscribeRequest {
            categories: self.categories,
            since_seq: self.since_seq,
            from_cursor: self.from_cursor,
            tweak_dust_limit: self.tweak_dust_limit,
            // Send the flag only when set, so a default subscription is
            // byte-identical to one built before these knobs existed.
            tweaks_only: self.tweaks_only.then_some(true),
            mempool_tweaks: self.mempool_tweaks.then_some(true),
            tweak_outputs: self.tweak_outputs.then_some(true),
            tweak_unspent_only: self.tweak_unspent_only.then_some(true),
            ..Default::default()
        }
    }
}

/// Server-enforced cap on the number of *distinct* labels one scan-key target
/// may carry (mirrors the node's `MAX_SP_LABELS_PER_TARGET`). The server rejects
/// an over-label target; enforcing the same bound client-side turns that into a
/// deterministic error at the call site instead of a silent skip.
pub const MAX_SP_LABELS_PER_TARGET: usize = 16;

/// Server-enforced cap on the number of scan-key targets per connection (mirrors
/// the node's `MAX_SP_TARGETS_PER_CONNECTION`). The server silently sheds an
/// over-cap add; a stateful [`ResilientWatch`](crate::ResilientWatch) enforces
/// this bound before recording/sending so it never mirrors and replays a target
/// the server dropped.
pub const MAX_SP_TARGETS_PER_CONNECTION: usize = 16;

/// A BIP 352 scan-key watch target (Tier 2): a `(scan_secret, spend_pubkey)`
/// pair plus optional labels, registered via
/// [`WatchHandle::add_silent_payments`]. The node runs the ECDH match and pushes
/// [`Event::SilentPaymentMatched`](crate::Event::SilentPaymentMatched).
///
/// `scan_secret` is a **watch credential**, not a spending key: disclosing it
/// lets the node (and anyone who compromises it) learn *which* outputs are
/// yours, but never spend them — spend authority stays with `B_spend`'s private
/// half, which is never sent. `Debug` redacts the scan secret.
#[derive(Clone)]
pub struct SilentPaymentTarget {
    /// 32-byte scan secret `b_scan`.
    pub scan_secret: [u8; 32],
    /// 33-byte compressed spend public key `B_spend` (public half only).
    pub spend_pubkey: [u8; 33],
    /// Receiver label integers to also match. Include `0` to catch change.
    pub labels: Vec<u32>,
}

impl core::fmt::Debug for SilentPaymentTarget {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.debug_struct("SilentPaymentTarget")
            .field("scan_secret", &"<redacted>")
            .field("spend_pubkey", &hex_lower(&self.spend_pubkey))
            .field("labels", &self.labels)
            .finish()
    }
}

impl Drop for SilentPaymentTarget {
    /// Scrub the retained scan secret on drop. `b_scan` is a watch credential —
    /// the long-lived copies held by a `ResilientWatch` mirror (and any clones)
    /// must not linger in freed memory once a target is removed or the mirror is
    /// torn down. (The transient `Vec<u8>` a `to_proto` copy places on the wire
    /// is owned by the outbound message and cleared with it; this covers every
    /// in-process copy the SDK retains.)
    fn drop(&mut self) {
        use zeroize::Zeroize;
        self.scan_secret.zeroize();
    }
}

impl SilentPaymentTarget {
    pub(crate) fn to_proto(&self) -> pb::SilentPaymentTarget {
        pb::SilentPaymentTarget {
            scan_secret: self.scan_secret.to_vec(),
            spend_pubkey: self.spend_pubkey.to_vec(),
            labels: self.labels.clone(),
        }
    }

    /// The target's identity `b_scan·G` (33-byte compressed) — the key used to
    /// remove it ([`WatchHandle::remove_silent_payments`]) and how a
    /// [`ResilientWatch`](crate::ResilientWatch) tracks it. Requires the
    /// `bitcoin` feature (enabled by default) for the elliptic-curve derivation.
    #[cfg(feature = "bitcoin")]
    pub fn scan_pubkey(&self) -> Result<[u8; 33], StreamError> {
        use bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey};
        let sk = SecretKey::from_slice(&self.scan_secret)
            .map_err(|_| StreamError::InvalidArgument("scan_secret is not a valid scalar".into()))?;
        let secp = Secp256k1::signing_only();
        Ok(PublicKey::from_secret_key(&secp, &sk).serialize())
    }

    /// Validate **both** curve values — the `scan_secret` scalar and the
    /// `spend_pubkey` point — and return the identity `b_scan·G`. Client-side
    /// validation turns a malformed target into a deterministic
    /// [`InvalidArgument`](StreamError::InvalidArgument) at the call site instead
    /// of a silent server-side skip that would return `Ok(())` while installing
    /// no watch (and repeat the invalid target on every resilient reconnect).
    /// Requires the `bitcoin` feature (enabled by default).
    #[cfg(feature = "bitcoin")]
    pub fn validate(&self) -> Result<[u8; 33], StreamError> {
        let id = self.scan_pubkey()?; // validates the b_scan scalar
        bitcoin::secp256k1::PublicKey::from_slice(&self.spend_pubkey)
            .map_err(|_| StreamError::InvalidArgument("spend_pubkey is not a valid point".into()))?;
        // Mirror the server's per-target label cap: an over-label target is
        // otherwise silently skipped by the node while the client believes it
        // was installed (and a resilient reconnect replays it forever).
        let mut labels = self.labels.clone();
        labels.sort_unstable();
        labels.dedup();
        if labels.len() > MAX_SP_LABELS_PER_TARGET {
            return Err(StreamError::InvalidArgument(format!(
                "too many silent-payment labels: {} distinct (max {MAX_SP_LABELS_PER_TARGET})",
                labels.len()
            )));
        }
        Ok(id)
    }
}

/// Lowercase hex, for redacting `Debug` impls (no external dep).
fn hex_lower(bytes: &[u8]) -> String {
    let mut s = String::with_capacity(bytes.len() * 2);
    for b in bytes {
        s.push_str(&format!("{b:02x}"));
    }
    s
}

/// A live stream of typed [`Event`]s, shared by `Subscribe` and `Watch`.
///
/// As confirmed events flow, the stream captures their durable [`Cursor`];
/// [`cursor`](Self::cursor) returns the latest, which a consumer persists and
/// presents as [`SubscribeOptions::from_cursor`] to resume.
pub struct EventStream {
    inner: tonic::Streaming<pb::NodeEvent>,
    last_cursor: Option<Cursor>,
}

impl EventStream {
    /// Await the next event. Returns `Ok(None)` when the server closes the
    /// stream.
    pub async fn message(&mut self) -> Result<Option<Event>, StreamError> {
        match self.inner.message().await {
            Ok(Some(ev)) => {
                if let Some(c) = ev.cursor {
                    self.last_cursor = Some(c);
                }
                Ok(Some(Event::from(ev)))
            }
            Ok(None) => Ok(None),
            Err(status) => Err(StreamError::from_status(status)),
        }
    }

    /// The most recent durable cursor seen on this stream, if any. Persist it
    /// to resume after a disconnect.
    pub fn cursor(&self) -> Option<&Cursor> {
        self.last_cursor.as_ref()
    }
}

/// Auto-close policy for a transaction lifecycle watch: self-evict and emit
/// `TxidFinalized` once the tx is buried this many confirmations deep.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum AutoClose {
    /// Never auto-close; the lifecycle watch persists until removed.
    #[default]
    Never,
    /// Auto-close once the tx reaches this confirmation depth (>= 1).
    AtDepth(u32),
}

impl AutoClose {
    fn depth(self) -> u32 {
        match self {
            AutoClose::Never => 0,
            AutoClose::AtDepth(d) => d,
        }
    }
}

/// A handle to send control messages on a bidirectional `Watch` stream.
///
/// Typed helpers build and send the right [`SubscribeControl`](pb::SubscribeControl)
/// for each watch kind; [`send_control`](Self::send_control) remains for raw
/// access. Empty inputs are no-ops (nothing is sent). Dropping the handle closes
/// the control channel, which the server uses to tear the stream down.
pub struct WatchHandle {
    tx: mpsc::Sender<pb::SubscribeControl>,
}

impl WatchHandle {
    /// Send a raw [`SubscribeControl`](pb::SubscribeControl) message.
    pub async fn send_control(&self, ctrl: pb::SubscribeControl) -> Result<(), StreamError> {
        self.tx.send(ctrl).await.map_err(|_| StreamError::ControlClosed)
    }

    async fn send_msg(&self, msg: pb::subscribe_control::Msg) -> Result<(), StreamError> {
        self.send_control(pb::SubscribeControl { msg: Some(msg) }).await
    }

    /// Add scripthashes (each `sha256(scriptPubKey)`), with an optional per-script
    /// `min_value` floor in satoshis. Matches below a script's floor are
    /// suppressed (a `None` floor delivers everything). Re-asserting a held
    /// scripthash updates its floor. Charges one watch-quota unit per scripthash.
    pub async fn add_scripts(
        &self,
        items: impl IntoIterator<Item = ([u8; 32], Option<u64>)>,
    ) -> Result<(), StreamError> {
        let items: Vec<_> = items.into_iter().collect();
        if items.is_empty() {
            return Ok(());
        }
        let scripthashes = items.iter().map(|(h, _)| h.to_vec()).collect();
        // `min_values` is empty when no floor is set on any script; otherwise it
        // must be parallel to `scripthashes`, with 0 (deliver-all) for the
        // unfloored entries.
        let min_values = if items.iter().any(|(_, f)| f.is_some()) {
            items.iter().map(|(_, f)| f.unwrap_or(0)).collect()
        } else {
            Vec::new()
        };
        self.send_msg(pb::subscribe_control::Msg::AddScripts(pb::AddScripts {
            scripthashes,
            min_values,
        }))
        .await
    }

    /// Remove scripthashes from the watch-set, releasing their quota.
    pub async fn remove_scripts(
        &self,
        scripthashes: impl IntoIterator<Item = [u8; 32]>,
    ) -> Result<(), StreamError> {
        let scripthashes: Vec<Vec<u8>> = scripthashes.into_iter().map(|h| h.to_vec()).collect();
        if scripthashes.is_empty() {
            return Ok(());
        }
        self.send_msg(pb::subscribe_control::Msg::RemoveScripts(pb::RemoveScripts {
            scripthashes,
        }))
        .await
    }

    /// Add outpoints (`txid:vout`) to the watch-set. Charges one unit each.
    pub async fn add_outpoints(
        &self,
        outpoints: impl IntoIterator<Item = ([u8; 32], u32)>,
    ) -> Result<(), StreamError> {
        let outpoints: Vec<pb::Outpoint> = outpoints
            .into_iter()
            .map(|(txid, vout)| pb::Outpoint { txid: txid.to_vec(), vout })
            .collect();
        if outpoints.is_empty() {
            return Ok(());
        }
        self.send_msg(pb::subscribe_control::Msg::AddOutpoints(pb::AddOutpoints { outpoints }))
            .await
    }

    /// Remove outpoints from the watch-set, releasing their quota.
    pub async fn remove_outpoints(
        &self,
        outpoints: impl IntoIterator<Item = ([u8; 32], u32)>,
    ) -> Result<(), StreamError> {
        let outpoints: Vec<pb::Outpoint> = outpoints
            .into_iter()
            .map(|(txid, vout)| pb::Outpoint { txid: txid.to_vec(), vout })
            .collect();
        if outpoints.is_empty() {
            return Ok(());
        }
        self.send_msg(pb::subscribe_control::Msg::RemoveOutpoints(pb::RemoveOutpoints {
            outpoints,
        }))
        .await
    }

    /// Add persistent lifecycle watches on txids (seen → confirmed → replaced /
    /// evicted / unconfirmed). `auto_close` optionally self-evicts the watch once
    /// the tx is buried. Charges one unit per txid.
    pub async fn add_tx_lifecycle(
        &self,
        txids: impl IntoIterator<Item = [u8; 32]>,
        auto_close: AutoClose,
    ) -> Result<(), StreamError> {
        let txids: Vec<Vec<u8>> = txids.into_iter().map(|t| t.to_vec()).collect();
        if txids.is_empty() {
            return Ok(());
        }
        self.send_msg(pb::subscribe_control::Msg::AddTransactions(pb::AddTransactions {
            txids,
            min_depths: Vec::new(),
            auto_close_depth: auto_close.depth(),
        }))
        .await
    }

    /// Remove lifecycle watches on txids.
    pub async fn remove_tx_lifecycle(
        &self,
        txids: impl IntoIterator<Item = [u8; 32]>,
    ) -> Result<(), StreamError> {
        let txids: Vec<Vec<u8>> = txids.into_iter().map(|t| t.to_vec()).collect();
        if txids.is_empty() {
            return Ok(());
        }
        self.send_msg(pb::subscribe_control::Msg::RemoveTransactions(pb::RemoveTransactions {
            txids,
            min_depths: Vec::new(),
        }))
        .await
    }

    /// Arm single-shot depth alarms over the **cross product** of `txids` and
    /// `depths`: each txid fires `TxidDepthReached` once it is `depth`
    /// confirmations deep, then self-evicts. Charges one unit per (txid, depth).
    ///
    /// Depths must be `>= 1`; depths `< 1` are dropped client-side. If that
    /// leaves no valid depths (or no txids), this is a no-op — importantly, it
    /// does **not** send an empty `min_depths`, which the server would otherwise
    /// reinterpret as a *lifecycle* add rather than a depth alarm.
    pub async fn add_depth_alarms(
        &self,
        txids: impl IntoIterator<Item = [u8; 32]>,
        depths: impl IntoIterator<Item = u32>,
    ) -> Result<(), StreamError> {
        let txids: Vec<Vec<u8>> = txids.into_iter().map(|t| t.to_vec()).collect();
        let min_depths: Vec<u32> = depths.into_iter().filter(|d| *d >= 1).collect();
        if txids.is_empty() || min_depths.is_empty() {
            return Ok(());
        }
        self.send_msg(pb::subscribe_control::Msg::AddTransactions(pb::AddTransactions {
            txids,
            min_depths,
            auto_close_depth: 0,
        }))
        .await
    }

    /// Remove depth alarms over the cross product of `txids` and `depths`.
    /// Depths `< 1` are dropped client-side; an all-invalid (or empty) call is a
    /// no-op and never sends an empty `min_depths` (which would target lifecycle
    /// watches instead).
    pub async fn remove_depth_alarms(
        &self,
        txids: impl IntoIterator<Item = [u8; 32]>,
        depths: impl IntoIterator<Item = u32>,
    ) -> Result<(), StreamError> {
        let txids: Vec<Vec<u8>> = txids.into_iter().map(|t| t.to_vec()).collect();
        let min_depths: Vec<u32> = depths.into_iter().filter(|d| *d >= 1).collect();
        if txids.is_empty() || min_depths.is_empty() {
            return Ok(());
        }
        self.send_msg(pb::subscribe_control::Msg::RemoveTransactions(pb::RemoveTransactions {
            txids,
            min_depths,
        }))
        .await
    }

    /// Expand a public output descriptor into a watch-set over the window
    /// `[start, start + gap_limit)`. The server retains the descriptor →
    /// scripthashes membership, so the client owns gap-limit advancement by
    /// re-sending with an advanced `start` (the server reconciles the slid
    /// window) and can drop the whole window with
    /// [`remove_descriptor`](Self::remove_descriptor).
    pub async fn add_descriptor(
        &self,
        descriptor: impl Into<String>,
        gap_limit: u32,
        start: u32,
    ) -> Result<(), StreamError> {
        self.send_msg(pb::subscribe_control::Msg::AddDescriptor(pb::AddDescriptor {
            descriptor: descriptor.into(),
            gap_limit,
            start,
        }))
        .await
    }

    /// Remove a descriptor previously added with [`add_descriptor`](Self::add_descriptor),
    /// releasing every scripthash its window contributed whose last owner this
    /// drops. A scripthash the descriptor shares with a direct `add_scripts` or
    /// another descriptor stays watched. `descriptor` must byte-match the string
    /// it was added with; removing an unknown descriptor is a no-op.
    pub async fn remove_descriptor(
        &self,
        descriptor: impl Into<String>,
    ) -> Result<(), StreamError> {
        self.send_msg(pb::subscribe_control::Msg::RemoveDescriptor(pb::RemoveDescriptor {
            descriptor: descriptor.into(),
        }))
        .await
    }

    /// Add privacy-preserving script-prefix buckets: each is a `bits`-bit prefix
    /// of `sha256(scriptPubKey)`, carried as the top `ceil(bits/8)` bytes. The
    /// server delivers every tx in the `2^-bits` bucket and the client filters
    /// locally, so the server learns only the bucket. Charged by coarseness
    /// (smaller `bits` costs more). `bits` must be in `1..=32` (the server
    /// buckets on the top 32 bits of the scripthash; wider is meaningless and
    /// silently dropped) and `prefix` exactly `ceil(bits/8)` bytes; the server
    /// additionally enforces its configured
    /// `[streamprefixminbits, streamprefixmaxbits]` range.
    pub async fn add_script_prefixes(
        &self,
        prefixes: impl IntoIterator<Item = (Vec<u8>, u32)>,
    ) -> Result<(), StreamError> {
        let prefixes: Vec<pb::ScriptPrefix> = prefixes
            .into_iter()
            .map(|(prefix, bits)| validate_prefix(prefix, bits))
            .collect::<Result<_, _>>()?;
        if prefixes.is_empty() {
            return Ok(());
        }
        self.send_msg(pb::subscribe_control::Msg::AddScriptPrefixes(pb::AddScriptPrefixes {
            prefixes,
        }))
        .await
    }

    /// Remove script-prefix buckets, releasing their quota.
    pub async fn remove_script_prefixes(
        &self,
        prefixes: impl IntoIterator<Item = (Vec<u8>, u32)>,
    ) -> Result<(), StreamError> {
        let prefixes: Vec<pb::ScriptPrefix> = prefixes
            .into_iter()
            .map(|(prefix, bits)| validate_prefix(prefix, bits))
            .collect::<Result<_, _>>()?;
        if prefixes.is_empty() {
            return Ok(());
        }
        self.send_msg(pb::subscribe_control::Msg::RemoveScriptPrefixes(
            pb::RemoveScriptPrefixes { prefixes },
        ))
        .await
    }

    /// Register BIP 352 scan-key watch targets (Tier 2). The node runs the ECDH
    /// match server-side and pushes an
    /// [`Event::SilentPaymentMatched`](crate::Event::SilentPaymentMatched) for
    /// every output paying one of them. Up to 16 targets per connection; each
    /// charges one watch-quota unit. Re-registering an existing target (same
    /// `b_scan·G`) refreshes its labels.
    ///
    /// The `scan_secret` is a watch credential disclosed to the node — see
    /// [`SilentPaymentTarget`]. With the `bitcoin` feature (the default) each
    /// target's `scan_secret` and `spend_pubkey` are validated as curve values
    /// up front (see [`SilentPaymentTarget::validate`]), so a malformed target is
    /// a deterministic [`InvalidArgument`](StreamError::InvalidArgument) here
    /// rather than a silent server-side skip that returns `Ok(())` while
    /// installing no watch. Without the feature this helper moves bytes only —
    /// validate the field widths yourself or let the server reject a malformed
    /// target.
    pub async fn add_silent_payments(
        &self,
        targets: impl IntoIterator<Item = SilentPaymentTarget>,
    ) -> Result<(), StreamError> {
        let targets: Vec<SilentPaymentTarget> = targets.into_iter().collect();
        #[cfg(feature = "bitcoin")]
        for t in &targets {
            t.validate()?;
        }
        let targets: Vec<pb::SilentPaymentTarget> = targets.iter().map(|t| t.to_proto()).collect();
        if targets.is_empty() {
            return Ok(());
        }
        self.send_msg(pb::subscribe_control::Msg::AddSilentPayments(
            pb::AddSilentPayments { targets },
        ))
        .await
    }

    /// Remove scan-key watch targets by their identity `b_scan·G` (33-byte
    /// compressed, from [`SilentPaymentTarget::scan_pubkey`]), releasing quota.
    pub async fn remove_silent_payments(
        &self,
        scan_pubkeys: impl IntoIterator<Item = [u8; 33]>,
    ) -> Result<(), StreamError> {
        let scan_pubkeys: Vec<Vec<u8>> = scan_pubkeys.into_iter().map(|k| k.to_vec()).collect();
        if scan_pubkeys.is_empty() {
            return Ok(());
        }
        self.send_msg(pb::subscribe_control::Msg::RemoveSilentPayments(
            pb::RemoveSilentPayments { scan_pubkeys },
        ))
        .await
    }

    /// Adjust the live firehose category bitfield (see [`Categories`]). Applies
    /// immediately; does not affect the watch-set.
    pub async fn set_categories(&self, categories: u32) -> Result<(), StreamError> {
        self.send_msg(pb::subscribe_control::Msg::SetCategories(pb::SetCategories {
            categories,
        }))
        .await
    }

    /// Set per-stream delivery options. With `include_raw_tx = true`, subsequent
    /// [`Event::ScriptMatched`](crate::Event::ScriptMatched) on this stream carry
    /// the full serialized matching transaction in
    /// [`raw_tx`](crate::Event::ScriptMatched); `false` restores the default
    /// (empty). Applies immediately; does not affect the watch-set. Bandwidth-
    /// heavy — the value fields (`amount`) already cover the common case, so
    /// enable this only when you need the whole transaction.
    ///
    /// When driving a [`ResilientWatch`](crate::ResilientWatch), prefer its
    /// [`set_watch_options`](crate::ResilientWatch::set_watch_options) so the
    /// opt-in is re-applied across reconnects.
    pub async fn set_watch_options(&self, include_raw_tx: bool) -> Result<(), StreamError> {
        self.send_msg(pb::subscribe_control::Msg::SetWatchOptions(
            pb::SetWatchOptions { include_raw_tx },
        ))
        .await
    }

    /// Mid-stream re-anchor: replay confirmed history forward from `cursor`, then
    /// resume live, without tearing down the watch-set. Rate-limited per
    /// principal; only one re-anchor drains at a time.
    ///
    /// `Ok(())` means the request was written to the control stream — **not**
    /// that the re-anchor ran. The outcome arrives **in-band on the event
    /// stream** as exactly one of (#439):
    ///
    /// - [`Event::CursorAccepted`](crate::Event::CursorAccepted) — admitted;
    ///   confirmed-history replay follows it (watch for `clamped` to learn the
    ///   server truncated the lower end and you must full-resync below
    ///   `earliest_replayed`).
    /// - [`Event::CursorRejected`](crate::Event::CursorRejected) — declined
    ///   (rate-limited, concurrent re-anchor, empty cursor, or no block source);
    ///   the live stream is unchanged. Retry, back off, or escalate to a full
    ///   resnapshot per the [`reason`](crate::CursorRejectReason).
    ///
    /// A consumer that needs at-least-once delivery should drive its catch-up
    /// state machine off these events rather than treating `Ok(())` as success.
    pub async fn set_cursor(&self, cursor: Cursor) -> Result<(), StreamError> {
        self.send_msg(pb::subscribe_control::Msg::SetCursor(pb::SetCursor {
            cursor: Some(cursor),
        }))
        .await
    }

    /// Request a bounded historical rescan of the current watch-set over the
    /// inclusive height range `[from_height, to_height]`. The outcome arrives
    /// in-band on the event stream as [`Event`](crate::Event)`::RescanAccepted` /
    /// `RescanRejected`; on accept, confirmed watch-matches for the range follow
    /// in height order, terminated by `RescanComplete`.
    ///
    /// A side query: it does not move the durable cursor and runs independently
    /// of the live tail or any in-flight re-anchor. The server span-caps the
    /// range and admits at most one rescan at a time (a second is rejected
    /// `ConcurrentRescan`). `Ok(())` means the request was sent, not that it was
    /// admitted — drive off the in-band result.
    pub async fn rescan(&self, from_height: u32, to_height: u32) -> Result<(), StreamError> {
        self.send_msg(pb::subscribe_control::Msg::RescanBlocks(pb::RescanBlocks {
            from_height,
            to_height,
        }))
        .await
    }
}

/// The maximum meaningful prefix width. The server buckets on the top 32 bits of
/// `sha256(scriptPubKey)` (its mask saturates at 32) and its default
/// `streamprefixmaxbits` is 32, so a `bits > 32` registration can never be more
/// selective and is silently dropped server-side (the control path has no ack).
/// We reject it client-side rather than let a caller believe a watch was
/// installed. A server may further *lower* the max via `streamprefixmaxbits`;
/// that bound is not advertised over the wire, so an over-precise (but ≤ 32)
/// prefix can still be silently dropped — there is no client-side signal for it.
pub(crate) const MAX_PREFIX_BITS: u32 = 32;

/// Validate a prefix/bits pair before it reaches the wire.
pub(crate) fn validate_prefix(
    prefix: Vec<u8>,
    bits: u32,
) -> Result<pb::ScriptPrefix, StreamError> {
    if !(1..=MAX_PREFIX_BITS).contains(&bits) {
        return Err(StreamError::InvalidArgument(format!(
            "prefix bits {bits} out of range 1..={MAX_PREFIX_BITS}"
        )));
    }
    let want = bits.div_ceil(8) as usize;
    if prefix.len() != want {
        return Err(StreamError::InvalidArgument(format!(
            "prefix for {bits} bits must be {want} bytes, got {}",
            prefix.len()
        )));
    }
    Ok(pb::ScriptPrefix { prefix, bits })
}

/// Whether `endpoint` carries the `https://` scheme.
///
/// tonic selects TLS from the URI scheme alone, so the scheme — not the
/// builder's `tls` field — is what decides whether bytes are encrypted. A
/// scheme-less endpoint (`host:port`, the common form) is plaintext.
///
/// Only the TLS-capable build consults this: without the `tls` feature the
/// scheme carries no security meaning at all, which is the whole point of
/// [`endpoint_is_encrypted`]. Kept compiled in both so the unit test that
/// pins its behaviour runs in both.
#[cfg_attr(not(feature = "tls"), allow(dead_code))]
fn endpoint_is_https(endpoint: &str) -> bool {
    endpoint
        .split_once("://")
        .is_some_and(|(scheme, _)| scheme.eq_ignore_ascii_case("https"))
}

/// Whether connecting to `endpoint` will actually encrypt the bytes.
///
/// This is *not* the same question as [`endpoint_is_https`], and conflating the
/// two is how a credential leaks. tonic only honours the `https://` scheme when
/// it is itself compiled with its `tls` feature: the scheme test and the whole
/// TLS branch of its connector are `#[cfg(feature = "tls")]`, and it disables
/// `enforce_http`. In a build without that feature an `https://` URI therefore
/// opens an ordinary TCP connection and speaks cleartext h2c — no error, no
/// downgrade warning. So the scheme is transport-truthful only in a TLS-capable
/// build; in a build without one, nothing is, and this returns `false` for every
/// endpoint.
#[cfg(feature = "tls")]
fn endpoint_is_encrypted(endpoint: &str) -> bool {
    endpoint_is_https(endpoint)
}

#[cfg(not(feature = "tls"))]
fn endpoint_is_encrypted(_endpoint: &str) -> bool {
    false
}

/// Strip `user:password@` userinfo out of an endpoint before it is put into an
/// error string. Refusals are normally logged, and a password in a URL is a
/// credential too — an error whose whole purpose is to stop a credential
/// reaching the wire must not spill a different one into the log.
fn redact_endpoint_userinfo(endpoint: &str) -> String {
    let Some((scheme, rest)) = endpoint.split_once("://") else {
        // No scheme, so the whole string is an authority (`host:port`).
        return match endpoint.rsplit_once('@') {
            Some((_, host)) => format!("***@{host}"),
            None => endpoint.to_string(),
        };
    };
    // Userinfo lives in the authority only: everything before the first `/`.
    // Splitting there keeps a later `@` in a path from being mistaken for it.
    let (authority, path) = match rest.find('/') {
        Some(i) => rest.split_at(i),
        None => (rest, ""),
    };
    match authority.rsplit_once('@') {
        Some((_, host)) => format!("{scheme}://***@{host}{path}"),
        None => endpoint.to_string(),
    }
}

/// The endpoint as it should appear in an [`StreamError::InsecureCredential`],
/// naming the reason when the reason is not visible in the endpoint itself.
#[cfg(feature = "tls")]
fn insecure_credential_reason(endpoint: &str) -> String {
    redact_endpoint_userinfo(endpoint)
}

#[cfg(not(feature = "tls"))]
fn insecure_credential_reason(endpoint: &str) -> String {
    format!(
        "{} — this build of satd-events-client has the `tls` feature disabled, \
         so no endpoint is encrypted here, https:// included",
        redact_endpoint_userinfo(endpoint)
    )
}

/// TLS settings assembled by the `StreamClientBuilder::tls*` methods and applied
/// in [`connect`](StreamClientBuilder::connect). `Some(..)` on the builder means
/// TLS is enabled; the fields refine trust and identity.
#[cfg(feature = "tls")]
#[derive(Clone, Default)]
struct TlsSettings {
    /// PEM CA (or self-signed leaf) to verify the server cert. `None` → trust the
    /// bundled Mozilla webpki roots (public-CA servers).
    ca_pem: Option<Vec<u8>>,
    /// mTLS client identity: `(cert_pem, key_pem)`.
    identity: Option<(Vec<u8>, Vec<u8>)>,
    /// SNI / certificate domain-name override (e.g. when connecting by IP).
    domain: Option<String>,
}

/// Builder for a [`StreamClient`].
///
/// `Debug` is hand-written to redact the bearer token (and never leak TLS key
/// material) — never derive it here.
#[derive(Clone)]
pub struct StreamClientBuilder {
    endpoint: String,
    token: Option<String>,
    /// Set only by [`StreamClientBuilder::insecure_bearer_token`]: the caller
    /// has explicitly accepted sending the token over an unencrypted endpoint.
    allow_insecure_token: bool,
    keepalive: Option<(Duration, Duration)>,
    #[cfg(feature = "tls")]
    tls: Option<TlsSettings>,
}

impl std::fmt::Debug for StreamClientBuilder {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut d = f.debug_struct("StreamClientBuilder");
        d.field("endpoint", &self.endpoint)
            .field("token", &self.token.as_ref().map(|_| "<redacted>"))
            // Not the token, but worth seeing in a bug report: it says whether
            // the plaintext waiver is in force.
            .field("allow_insecure_token", &self.allow_insecure_token)
            .field("keepalive", &self.keepalive);
        #[cfg(feature = "tls")]
        d.field("tls", &self.tls.is_some());
        d.finish()
    }
}

impl StreamClientBuilder {
    /// Attach a bearer token, sent as `authorization: Bearer <token>` metadata
    /// on every RPC.
    ///
    /// The token is only honored when the server enforces auth
    /// (`-eventsgrpcauth`); a no-auth (loopback-trust) server ignores it.
    ///
    /// **The endpoint must be encrypted.** [`connect`](Self::connect) returns
    /// [`StreamError::InsecureCredential`] for a token combined with a
    /// non-`https://` endpoint rather than putting it on the wire in the clear.
    /// Pair this with [`tls`](Self::tls) or [`tls_ca_pem`](Self::tls_ca_pem),
    /// or use [`insecure_bearer_token`](Self::insecure_bearer_token) to accept
    /// the risk explicitly.
    pub fn bearer_token(mut self, token: impl Into<String>) -> Self {
        self.token = Some(token.into());
        self.allow_insecure_token = false;
        self
    }

    /// Like [`bearer_token`](Self::bearer_token), but permits sending the token
    /// over an unencrypted endpoint.
    ///
    /// Only for loopback and test harnesses. Over anything else the token is
    /// readable by every host on the path, and so is everything the stream
    /// carries — including the BIP 352 scan keys a Tier 2 watch registers,
    /// which disclose which outputs belong to the receiver.
    ///
    /// This is a separate method rather than a flag so that the unsafe choice
    /// has to be named at the call site, and so that a later switch back to
    /// [`bearer_token`](Self::bearer_token) cannot silently leave the waiver
    /// behind.
    pub fn insecure_bearer_token(mut self, token: impl Into<String>) -> Self {
        self.token = Some(token.into());
        self.allow_insecure_token = true;
        self
    }

    /// Configure client-side HTTP/2 keepalive. Defaults match the server
    /// (30s interval / 20s timeout) when [`keepalive_default`](Self::keepalive_default)
    /// is used; this overrides with explicit seconds.
    pub fn keepalive(mut self, interval_secs: u64, timeout_secs: u64) -> Self {
        self.keepalive =
            Some((Duration::from_secs(interval_secs), Duration::from_secs(timeout_secs)));
        self
    }

    /// Use the server-matching keepalive defaults (30s interval / 20s timeout).
    pub fn keepalive_default(self) -> Self {
        self.keepalive(30, 20)
    }

    /// Enable TLS for the connection, trusting the bundled Mozilla root CAs
    /// (for a server with a publicly-trusted certificate). For a private or
    /// self-signed CA — the usual case for a satd node — use
    /// [`tls_ca_pem`](Self::tls_ca_pem) instead. Requires the `tls` feature
    /// (on by default).
    #[cfg(feature = "tls")]
    #[cfg_attr(docsrs, doc(cfg(feature = "tls")))]
    pub fn tls(mut self) -> Self {
        self.tls.get_or_insert_with(TlsSettings::default);
        self
    }

    /// Enable TLS and verify the server certificate against the PEM CA (or
    /// self-signed leaf) in `pem` — the usual choice for a satd node serving its
    /// own certificate. Replaces the bundled roots for this connection.
    #[cfg(feature = "tls")]
    #[cfg_attr(docsrs, doc(cfg(feature = "tls")))]
    pub fn tls_ca_pem(mut self, pem: impl Into<Vec<u8>>) -> Self {
        self.tls.get_or_insert_with(TlsSettings::default).ca_pem = Some(pem.into());
        self
    }

    /// Enable mutual TLS: present this PEM client certificate + private key to a
    /// server configured for mTLS (`-eventsgrpcmtls`). Combine with
    /// [`tls_ca_pem`](Self::tls_ca_pem) to pin the server's CA — without it the
    /// *server* certificate is verified against the bundled public Mozilla roots,
    /// so a self-signed satd node (the usual mTLS case) will fail the handshake
    /// unless you also call `tls_ca_pem`.
    #[cfg(feature = "tls")]
    #[cfg_attr(docsrs, doc(cfg(feature = "tls")))]
    pub fn tls_client_identity(
        mut self,
        cert_pem: impl Into<Vec<u8>>,
        key_pem: impl Into<Vec<u8>>,
    ) -> Self {
        self.tls.get_or_insert_with(TlsSettings::default).identity =
            Some((cert_pem.into(), key_pem.into()));
        self
    }

    /// Override the certificate domain name verified during the TLS handshake
    /// (SNI). Use when the endpoint host differs from the certificate subject —
    /// e.g. connecting by IP, or through a proxy. Enables TLS if not already.
    #[cfg(feature = "tls")]
    #[cfg_attr(docsrs, doc(cfg(feature = "tls")))]
    pub fn tls_domain(mut self, domain: impl Into<String>) -> Self {
        self.tls.get_or_insert_with(TlsSettings::default).domain = Some(domain.into());
        self
    }

    /// Connect the transport and return a ready [`StreamClient`].
    pub async fn connect(self) -> Result<StreamClient, StreamError> {
        // Captured before the match below moves `self.token`; the credential
        // check that reads it is ordered later on purpose (see there).
        let has_token = self.token.is_some();
        let auth = match self.token {
            Some(t) => Some(
                format!("Bearer {t}")
                    .parse::<MetadataValue<Ascii>>()
                    .map_err(|_| StreamError::InvalidToken)?,
            ),
            None => None,
        };

        // TLS is selected by tonic purely from the URI scheme, *not* from
        // whether a `ClientTlsConfig` is attached: an `http://` (or scheme-less)
        // endpoint silently connects in cleartext even with TLS configured,
        // leaking the bearer token and the whole event stream while the caller
        // believes the link is encrypted. Fail closed rather than downgrade.
        #[cfg(feature = "tls")]
        if self.tls.is_some() && !endpoint_is_https(&self.endpoint) {
            return Err(StreamError::InvalidEndpoint(
                "TLS was requested (tls / tls_ca_pem / tls_client_identity / tls_domain) but \
                 the endpoint scheme is not https:// — refusing to connect in cleartext; use \
                 an https:// endpoint"
                    .to_string(),
            ));
        }

        // A bearer token on an unencrypted endpoint crosses the network in the
        // clear, and so does everything the stream carries — including the BIP
        // 352 scan keys a Tier 2 watch registers. Nothing downstream catches
        // this: the token is attached as ordinary metadata, so tonic has no
        // credential to apply a transport-security requirement to, and a
        // remote-bound node with `eventsgrpcauth=1` and no TLS is a supported
        // server configuration. Both ends behave as configured while the
        // credential is on the wire (#521).
        //
        // The predicate is `endpoint_is_encrypted`, not the builder's `tls`
        // field and not the scheme on its own: tonic selects TLS from the URI
        // scheme, but only in a build that has its `tls` feature. Without that
        // feature an `https://` endpoint connects in cleartext rather than
        // failing, so this crate's own `tls` feature is part of the question.
        // See `endpoint_is_encrypted`.
        //
        // Ordered after the TLS-scheme check above so the more specific
        // diagnosis wins when a caller asked for TLS *and* gave an http:// URL,
        // matching the Go SDK, where the scheme conflict is likewise reported
        // first.
        if has_token && !self.allow_insecure_token && !endpoint_is_encrypted(&self.endpoint) {
            return Err(StreamError::InsecureCredential(insecure_credential_reason(
                &self.endpoint,
            )));
        }

        let mut endpoint = Endpoint::from_shared(self.endpoint)
            .map_err(|e| StreamError::InvalidEndpoint(e.to_string()))?;
        if let Some((interval, timeout)) = self.keepalive {
            endpoint = endpoint
                .http2_keep_alive_interval(interval)
                .keep_alive_timeout(timeout)
                .keep_alive_while_idle(true);
        }
        #[cfg(feature = "tls")]
        if let Some(tls) = self.tls {
            ensure_crypto_provider();
            let mut cfg = tonic::transport::ClientTlsConfig::new();
            // A pinned CA verifies exactly that authority (the satd self-signed
            // case); otherwise fall back to the bundled public roots.
            cfg = match tls.ca_pem {
                Some(ca) => cfg.ca_certificate(tonic::transport::Certificate::from_pem(ca)),
                None => cfg.with_webpki_roots(),
            };
            if let Some((cert, key)) = tls.identity {
                cfg = cfg.identity(tonic::transport::Identity::from_pem(cert, key));
            }
            if let Some(domain) = tls.domain {
                cfg = cfg.domain_name(domain);
            }
            endpoint = endpoint.tls_config(cfg)?;
        }

        let channel = endpoint.connect().await?;
        Ok(StreamClient { inner: NodeEventStreamClient::new(channel), auth })
    }
}

/// Install `ring` as the process-default rustls `CryptoProvider`, once. In this
/// workspace `tonic/tls` resolves to ring only, so rustls already auto-selects
/// it; this is belt-and-suspenders for a build that *also* compiled `aws-lc-rs`
/// (a downstream consumer that doesn't pin the provider), where the default
/// would otherwise be ambiguous and panic. Best-effort — if the application
/// already installed a provider, that one stays.
#[cfg(feature = "tls")]
fn ensure_crypto_provider() {
    use std::sync::Once;
    static ONCE: Once = Once::new();
    ONCE.call_once(|| {
        let _ = tokio_rustls::rustls::crypto::ring::default_provider().install_default();
    });
}

/// An async client for the satd `satd.events.v1` streaming API.
///
/// `Debug` is hand-written to redact the auth credential — never derive it here.
#[derive(Clone)]
pub struct StreamClient {
    inner: NodeEventStreamClient<Channel>,
    auth: Option<MetadataValue<Ascii>>,
}

impl std::fmt::Debug for StreamClient {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("StreamClient")
            .field("auth", &self.auth.as_ref().map(|_| "<redacted>"))
            .finish_non_exhaustive()
    }
}

impl StreamClient {
    /// Start building a client for `endpoint` (e.g. `http://node:50051`, or
    /// `https://node:50051` with [`tls`](StreamClientBuilder::tls)).
    pub fn builder(endpoint: impl Into<String>) -> StreamClientBuilder {
        StreamClientBuilder {
            endpoint: endpoint.into(),
            token: None,
            allow_insecure_token: false,
            keepalive: None,
            #[cfg(feature = "tls")]
            tls: None,
        }
    }

    /// Wrap a message in a request carrying the configured auth metadata.
    fn authed<T>(&self, msg: T) -> tonic::Request<T> {
        let mut req = tonic::Request::new(msg);
        if let Some(value) = &self.auth {
            req.metadata_mut().insert("authorization", value.clone());
        }
        req
    }

    /// Open a server-streaming firehose. Requires the `stream:subscribe`
    /// capability when the server enforces auth.
    pub async fn subscribe(
        &mut self,
        opts: SubscribeOptions,
    ) -> Result<EventStream, StreamError> {
        let req = self.authed(opts.into_request());
        let inner = self.inner.subscribe(req).await?.into_inner();
        Ok(EventStream { inner, last_cursor: None })
    }

    /// Open a reconnect-and-replay-aware firehose. Unlike [`subscribe`](Self::subscribe),
    /// the returned [`ResilientSubscription`](crate::ResilientSubscription)
    /// reconnects with backoff, persists and replays the resume cursor, recovers
    /// from `Lagged` per the [`LagPolicy`](crate::LagPolicy), and surfaces
    /// replay-truncation gaps. Connects lazily on the first
    /// [`next`](crate::ResilientSubscription::next).
    pub fn resilient_subscribe(
        &self,
        opts: SubscribeOptions,
        config: crate::ResilientConfig,
    ) -> crate::ResilientSubscription {
        crate::ResilientSubscription::new(self.clone(), opts, config)
    }

    /// Open a reconnect-and-replay-aware bidirectional watch. Unlike
    /// [`watch`](Self::watch), the returned [`ResilientWatch`](crate::ResilientWatch)
    /// mirrors the watch-set and re-registers it on reconnect, re-anchors off the
    /// deterministic `set_cursor` results, and persists the resume cursor.
    /// Connects lazily on the first [`next`](crate::ResilientWatch::next).
    pub fn resilient_watch(&self, config: crate::ResilientWatchConfig) -> crate::ResilientWatch {
        crate::ResilientWatch::new(self.clone(), config)
    }

    /// Open a bidirectional watch stream, returning a [`WatchHandle`] to send
    /// control messages and an [`EventStream`] of matches + firehose.
    pub async fn watch(&mut self) -> Result<(WatchHandle, EventStream), StreamError> {
        let (tx, rx) = mpsc::channel::<pb::SubscribeControl>(CONTROL_BUFFER);
        let req = self.authed(ReceiverStream::new(rx));
        let inner = self.inner.watch(req).await?.into_inner();
        Ok((WatchHandle { tx }, EventStream { inner, last_cursor: None }))
    }
}

#[cfg(test)]
impl WatchHandle {
    /// A handle wired to an inspectable receiver — for unit tests in sibling
    /// modules (e.g. `ResilientWatch::reload`) that assert the control messages a
    /// layer sends. No server needed.
    pub(crate) fn for_test() -> (Self, mpsc::Receiver<pb::SubscribeControl>) {
        let (tx, rx) = mpsc::channel(64);
        (WatchHandle { tx }, rx)
    }
}

#[cfg(test)]
impl StreamClient {
    /// Build a client over a lazily-connected channel to a dummy endpoint, for
    /// unit tests of layers (e.g. [`ResilientSubscription`](crate::ResilientSubscription))
    /// that need a `StreamClient` value but drive their logic via injected events
    /// and never actually poll the transport. `connect_lazy` performs no I/O, so
    /// no server is required; it must be called from within a Tokio runtime.
    pub(crate) fn for_test() -> Self {
        let channel = Endpoint::from_static("http://127.0.0.1:1").connect_lazy();
        StreamClient { inner: NodeEventStreamClient::new(channel), auth: None }
    }
}

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

    /// A handle wired to a receiver we can inspect — no server needed.
    fn handle() -> (WatchHandle, mpsc::Receiver<pb::SubscribeControl>) {
        let (tx, rx) = mpsc::channel(16);
        (WatchHandle { tx }, rx)
    }

    fn next(rx: &mut mpsc::Receiver<pb::SubscribeControl>) -> Msg {
        rx.try_recv().expect("a control message was sent").msg.expect("msg set")
    }

    #[test]
    fn only_an_https_scheme_counts_as_encrypted() {
        assert!(endpoint_is_https("https://node.example:50051"));
        assert!(endpoint_is_https("HTTPS://node.example:50051"));
        assert!(!endpoint_is_https("http://node.example:50051"));
        // The common scheme-less form. tonic treats it as plaintext, so this
        // must not be mistaken for encrypted.
        assert!(!endpoint_is_https("node.example:50051"));
        assert!(!endpoint_is_https("127.0.0.1:50051"));
    }

    /// The scheme decides only in a build that can actually do TLS. tonic gates
    /// its whole https branch on its own `tls` feature and turns `enforce_http`
    /// off, so without that feature an `https://` URI connects in cleartext h2c
    /// instead of failing — which would have let a token past the gate below in
    /// exactly the configuration the gate advertises as safe.
    #[test]
    fn encryption_requires_the_tls_feature_and_not_just_the_scheme() {
        assert_eq!(endpoint_is_encrypted("https://node.example:50051"), cfg!(feature = "tls"));
        assert!(!endpoint_is_encrypted("http://node.example:50051"));
        assert!(!endpoint_is_encrypted("node.example:50051"));
    }

    /// The refusal is normally logged, so it must not spill a password that was
    /// embedded in the endpoint. A later `@` in the path is not userinfo.
    #[test]
    fn the_refusal_does_not_echo_url_userinfo() {
        assert_eq!(
            redact_endpoint_userinfo("http://user:hunter2@node.example:50051"),
            "http://***@node.example:50051"
        );
        assert_eq!(
            redact_endpoint_userinfo("http://user:hunter2@node.example:50051/a@b"),
            "http://***@node.example:50051/a@b"
        );
        assert_eq!(redact_endpoint_userinfo("user:hunter2@node.example:50051"), "***@node.example:50051");
        // Untouched when there is nothing to redact.
        assert_eq!(
            redact_endpoint_userinfo("https://node.example:50051/path"),
            "https://node.example:50051/path"
        );
        assert_eq!(redact_endpoint_userinfo("node.example:50051"), "node.example:50051");
    }

    #[tokio::test]
    async fn a_password_in_the_endpoint_is_not_quoted_back() {
        let err = StreamClient::builder("http://user:hunter2@node.example:50051")
            .bearer_token("SECRET")
            .connect()
            .await
            .expect_err("a bearer token over plaintext must be refused");
        assert!(!err.to_string().contains("hunter2"), "leaked the URL password: {err}");
        assert!(!err.to_string().contains("SECRET"), "leaked the token: {err}");
    }

    /// The #521 guard. The token is attached as ordinary metadata, so there is
    /// no `PerRPCCredentials` for tonic to apply a transport-security
    /// requirement to, and a remote-bound node with `eventsgrpcauth=1` and no
    /// TLS is a supported server configuration — nothing else stops the
    /// credential reaching the wire in the clear.
    #[tokio::test]
    async fn a_bearer_token_over_plaintext_is_refused() {
        for endpoint in ["http://127.0.0.1:1", "http://node.example:50051", "node.example:50051"] {
            let err = StreamClient::builder(endpoint)
                .bearer_token("SECRET")
                .connect()
                .await
                .expect_err("a bearer token over plaintext must be refused");
            assert!(
                matches!(err, StreamError::InsecureCredential(_)),
                "{endpoint}: got {err:?}"
            );
            assert!(
                !err.to_string().contains("SECRET"),
                "{endpoint}: the error must not quote the token: {err}"
            );
        }
    }

    /// Reaching the transport is what proves the gate did not fire. Asserting
    /// merely "not `InsecureCredential`" would also hold if `connect()` started
    /// failing for some unrelated wrong reason, so name the errors a connect to
    /// a dead port is allowed to produce.
    fn assert_reached_the_transport(err: &StreamError, ctx: &str) {
        assert!(
            matches!(err, StreamError::Connect(_) | StreamError::Transport(_)),
            "{ctx}: expected a transport/connect error, got {err:?}"
        );
    }

    /// A plaintext connection with no token is the default local setup and is
    /// unaffected: it gets as far as the transport (and fails there, since
    /// nothing is listening on port 1).
    #[tokio::test]
    async fn plaintext_without_a_token_is_unaffected() {
        let err = StreamClient::builder("http://127.0.0.1:1")
            .connect()
            .await
            .expect_err("nothing is listening on port 1");
        assert_reached_the_transport(&err, "no token");
    }

    /// The named waiver gets past the gate — it too fails at the transport,
    /// which is the proof that the credential check no longer short-circuits.
    #[tokio::test]
    async fn the_named_waiver_permits_a_token_over_plaintext() {
        let err = StreamClient::builder("http://127.0.0.1:1")
            .insecure_bearer_token("SECRET")
            .connect()
            .await
            .expect_err("nothing is listening on port 1");
        assert_reached_the_transport(&err, "waiver");
    }

    /// A later `bearer_token` must clear an earlier waiver rather than inherit
    /// it — the reason the waiver is a token-carrying method and not a flag.
    #[tokio::test]
    async fn the_waiver_does_not_leak_into_a_subsequent_bearer_token() {
        let err = StreamClient::builder("http://127.0.0.1:1")
            .insecure_bearer_token("SECRET")
            .bearer_token("SECRET")
            .connect()
            .await
            .expect_err("the waiver must not survive a later bearer_token");
        assert!(
            matches!(err, StreamError::InsecureCredential(_)),
            "got {err:?}"
        );
    }

    /// An `https://` endpoint is accepted with a token whether or not a `tls*`
    /// method was called — tonic selects TLS from the scheme, so the scheme is
    /// what decides.
    #[cfg(feature = "tls")]
    #[tokio::test]
    async fn https_permits_a_token() {
        let err = StreamClient::builder("https://127.0.0.1:1")
            .tls()
            .bearer_token("SECRET")
            .connect()
            .await
            .expect_err("nothing is listening on port 1");
        assert!(
            !matches!(err, StreamError::InsecureCredential(_)),
            "got {err:?}"
        );
    }

    #[test]
    fn subscribe_options_map_tweak_knobs_only_when_set() {
        // A default subscription sends no tweak knobs (byte-identical to one
        // built before they existed); each flag maps to Some(true) when set.
        let default = SubscribeOptions { categories: Categories::TWEAKS, ..Default::default() }
            .into_request();
        assert_eq!(default.mempool_tweaks, None);
        assert_eq!(default.tweak_outputs, None);
        assert_eq!(default.tweak_unspent_only, None);

        let full = SubscribeOptions {
            categories: Categories::TWEAKS,
            mempool_tweaks: true,
            tweak_outputs: true,
            tweaks_only: true,
            tweak_unspent_only: true,
            ..Default::default()
        }
        .into_request();
        assert_eq!(full.mempool_tweaks, Some(true));
        assert_eq!(full.tweak_outputs, Some(true));
        assert_eq!(full.tweaks_only, Some(true));
        assert_eq!(full.tweak_unspent_only, Some(true));
    }

    #[tokio::test]
    async fn add_scripts_mixed_floors_emits_parallel_min_values() {
        let (h, mut rx) = handle();
        h.add_scripts([([1u8; 32], Some(5_000)), ([2u8; 32], None)]).await.unwrap();
        match next(&mut rx) {
            Msg::AddScripts(a) => {
                assert_eq!(a.scripthashes.len(), 2);
                // Unfloored entry becomes a 0 floor to keep the vec parallel.
                assert_eq!(a.min_values, vec![5_000, 0]);
            }
            other => panic!("expected AddScripts, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn add_scripts_no_floors_emits_empty_min_values() {
        let (h, mut rx) = handle();
        h.add_scripts([([1u8; 32], None), ([2u8; 32], None)]).await.unwrap();
        match next(&mut rx) {
            Msg::AddScripts(a) => assert!(a.min_values.is_empty()),
            other => panic!("expected AddScripts, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn empty_inputs_send_nothing() {
        let (h, mut rx) = handle();
        h.add_scripts([]).await.unwrap();
        h.add_outpoints([]).await.unwrap();
        h.add_depth_alarms([[1u8; 32]], []).await.unwrap(); // empty depths
        h.add_depth_alarms([], [3]).await.unwrap(); // empty txids
        // depths < 1 are filtered out; an all-invalid call must NOT send an
        // empty min_depths (the server would treat that as a lifecycle add).
        h.add_depth_alarms([[1u8; 32]], [0]).await.unwrap();
        assert!(rx.try_recv().is_err(), "no message should have been sent");
    }

    #[tokio::test]
    async fn lifecycle_vs_depth_alarms_dispatch_on_min_depths() {
        let (h, mut rx) = handle();
        h.add_tx_lifecycle([[7u8; 32]], AutoClose::AtDepth(6)).await.unwrap();
        match next(&mut rx) {
            Msg::AddTransactions(a) => {
                assert!(a.min_depths.is_empty(), "lifecycle => empty min_depths");
                assert_eq!(a.auto_close_depth, 6);
            }
            other => panic!("expected AddTransactions, got {other:?}"),
        }

        h.add_depth_alarms([[7u8; 32], [8u8; 32]], [3, 6]).await.unwrap();
        match next(&mut rx) {
            Msg::AddTransactions(a) => {
                assert_eq!(a.txids.len(), 2);
                assert_eq!(a.min_depths, vec![3, 6]); // cross product, server-side
                assert_eq!(a.auto_close_depth, 0);
            }
            other => panic!("expected AddTransactions, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn prefix_validation_rejects_bad_length_and_bits() {
        let (h, mut rx) = handle();
        // 16 bits needs 2 bytes; give 1.
        let err = h.add_script_prefixes([(vec![0xab], 16)]).await.unwrap_err();
        assert!(matches!(err, StreamError::InvalidArgument(_)));
        // bits out of range (too small).
        let err = h.add_script_prefixes([(vec![], 0)]).await.unwrap_err();
        assert!(matches!(err, StreamError::InvalidArgument(_)));
        // bits above the server's 32-bit ceiling: rejected client-side rather
        // than sent and silently dropped by the server (5 bytes = ceil(33/8)).
        let err = h.add_script_prefixes([(vec![0u8; 5], 33)]).await.unwrap_err();
        assert!(matches!(err, StreamError::InvalidArgument(_)));
        // Nothing reached the wire.
        assert!(rx.try_recv().is_err());

        // Valid boundary: exactly 32 bits, 4 bytes.
        h.add_script_prefixes([(vec![0xde, 0xad, 0xbe, 0xef], 32)]).await.unwrap();
        match next(&mut rx) {
            Msg::AddScriptPrefixes(a) => assert_eq!(a.prefixes[0].bits, 32),
            other => panic!("expected AddScriptPrefixes, got {other:?}"),
        }

        // Valid: 16 bits, 2 bytes.
        h.add_script_prefixes([(vec![0xab, 0xcd], 16)]).await.unwrap();
        match next(&mut rx) {
            Msg::AddScriptPrefixes(a) => {
                assert_eq!(a.prefixes.len(), 1);
                assert_eq!(a.prefixes[0].bits, 16);
                assert_eq!(a.prefixes[0].prefix, vec![0xab, 0xcd]);
            }
            other => panic!("expected AddScriptPrefixes, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn set_cursor_wraps_cursor() {
        let (h, mut rx) = handle();
        let cursor = Cursor { height: 9, tx_index: 0, mempool_seq: 1, instance_id: 2 };
        h.set_cursor(cursor).await.unwrap();
        match next(&mut rx) {
            Msg::SetCursor(s) => assert_eq!(s.cursor, Some(cursor)),
            other => panic!("expected SetCursor, got {other:?}"),
        }
    }

    /// The builder's `Debug` must show only that TLS is enabled — never the CA
    /// bytes, the client key, or the bearer token.
    #[cfg(feature = "tls")]
    #[test]
    fn tls_builder_debug_redacts_secrets() {
        let b = StreamClient::builder("https://node:50051")
            .bearer_token("SUPER_SECRET_TOKEN")
            .tls_ca_pem(b"-----BEGIN CERTIFICATE-----SECRET_CA_BYTES".to_vec())
            .tls_client_identity(b"CLIENT_CERT".to_vec(), b"CLIENT_PRIVATE_KEY".to_vec());
        let dbg = format!("{b:?}");
        assert!(dbg.contains("tls: true"), "expected tls: true, got {dbg}");
        assert!(dbg.contains("<redacted>"), "token should be redacted: {dbg}");
        assert!(!dbg.contains("SUPER_SECRET_TOKEN"));
        assert!(!dbg.contains("SECRET_CA_BYTES"));
        assert!(!dbg.contains("CLIENT_PRIVATE_KEY"));
    }

    /// The TLS connect path assembles a rustls config and installs the `ring`
    /// provider without panicking; against a dead port it fails cleanly (a
    /// transport error, not `InvalidEndpoint`/`InvalidToken`). Exercises the
    /// whole `tls()` → `ClientTlsConfig::with_webpki_roots` → `tls_config`
    /// assembly end to end, no server required.
    #[cfg(feature = "tls")]
    #[tokio::test]
    async fn tls_connect_assembles_and_fails_cleanly() {
        let err = StreamClient::builder("https://127.0.0.1:1")
            .tls()
            .tls_domain("node.example")
            .connect()
            .await
            .expect_err("connect to a dead port must fail");
        assert!(
            !matches!(err, StreamError::InvalidEndpoint(_) | StreamError::InvalidToken),
            "expected a transport/connect error, got {err:?}",
        );
    }

    /// TLS requested over a plaintext `http://` endpoint must fail closed at
    /// `connect()` rather than silently connecting in cleartext — tonic selects
    /// TLS from the URI scheme alone, so without this guard the bearer token and
    /// event stream would leak on the wire. Covers `http://` and a scheme-less
    /// endpoint; the `https://` happy path is exercised above.
    #[cfg(feature = "tls")]
    #[tokio::test]
    async fn tls_over_non_https_endpoint_is_refused() {
        for ep in ["http://127.0.0.1:1", "127.0.0.1:1"] {
            let err = StreamClient::builder(ep)
                .tls_ca_pem(b"-----BEGIN CERTIFICATE-----".to_vec())
                .bearer_token("SECRET")
                .connect()
                .await
                .expect_err("TLS over non-https must be refused");
            match err {
                StreamError::InvalidEndpoint(msg) => assert!(
                    msg.contains("https"),
                    "expected an https-scheme error for {ep}, got {msg}"
                ),
                other => panic!("expected InvalidEndpoint for {ep}, got {other:?}"),
            }
        }
    }
}