fynd_core/feed/

protocol_registry.rs

use std::{collections::HashMap, env, fmt, time::Duration};

use tokio_stream::Stream;
use tracing::{info, warn};
use tycho_simulation::{
    evm::{
        engine_db::tycho_db::PreCachedDB,
        protocol::{
            aerodrome_slipstreams::state::AerodromeSlipstreamsState,
            aerodrome_v1::state::AerodromeV1State,
            curve::CurveState,
            ekubo::state::EkuboState,
            ekubo_v3::state::EkuboV3State,
            erc4626::state::ERC4626State,
            filters::{
                balancer_v2_pool_filter, curve_filter, ekubo_v3_extension_filter,
                ekubo_v3_extension_filter_with_signed_exclusive_swap, erc4626_filter,
                fluid_v1_paused_pools_filter,
            },
            fluid::FluidV1,
            lunarbase::state::LunarBaseState,
            pancakeswap_v2::state::PancakeswapV2State,
            uniswap_v2::state::UniswapV2State,
            uniswap_v3::state::UniswapV3State,
            uniswap_v4::state::UniswapV4State,
            vm::state::EVMPoolState,
        },
        stream::ProtocolStreamBuilder,
        tycho_models::Chain,
    },
    price_level_stream::{config::default_served_pamms, stream::PriceLevelStreamBuilder},
    protocol::models::Update,
    rfq::{
        protocols::{
            bebop::{client_builder::BebopClientBuilder, state::BebopState},
            hashflow::{client_builder::HashflowClientBuilder, state::HashflowState},
        },
        stream::RFQStreamBuilder,
    },
    tycho_client::feed::component_tracker::ComponentFilter,
    tycho_common::models::token::Token,
    tycho_core::Bytes,
};

use super::DataFeedError;

/// Opts a protocol into streaming its exclusive pools, e.g. `exclusive:ekubo_v3`.
///
/// Fynd-side only: stripped before registration, so Tycho sees the bare system name.
const EXCLUSIVE_PREFIX: &str = "exclusive:";

/// Protocol systems that offer an exclusive-liquidity stream variant, i.e. the ones that may be
/// requested with the `exclusive:` prefix.
const EXCLUSIVE_CAPABLE_PROTOCOLS: &[&str] = &["ekubo_v3"];

/// Marks a `--protocols` entry served from the Titan pAMM price level stream rather than from
/// Tycho, e.g. `pricelevelstream:fermiswap`.
const PRICE_LEVEL_STREAM_PREFIX: &str = "pricelevelstream:";

/// Marks a component whose swaps execute through Titan's PropAMMRouter rather than against the
/// venue directly, e.g. `propammfallback:fermiswap`.
///
/// tycho-simulation gives a venue on the router's on-chain whitelist this family instead of
/// [`PRICE_LEVEL_STREAM_PREFIX`], so one `pricelevelstream:{venue}` entry can bring in components
/// under either prefix depending on the whitelist. Fynd never requests this family: it names the
/// venue, and the stream decides which of the two labels its components carry.
const PROPAMM_FALLBACK_PREFIX: &str = "propammfallback:";

/// Marks a `--protocols` entry served from an RFQ client rather than from Tycho, e.g.
/// `rfq:bebop`.
const RFQ_PREFIX: &str = "rfq:";

/// Marks a `--protocols` entry that drops a protocol system from the list rather than adding one,
/// e.g. `exclude:vm:fermiswap`.
pub const EXCLUDE_PREFIX: &str = "exclude:";

/// The only chain the Titan pAMM price level stream serves.
///
/// Tracks tycho-simulation's `default_served_pamms`, whose venue addresses are all Ethereum
/// mainnet deployments; it carries no chain of its own, so this has to move when it gains a venue
/// elsewhere.
const PRICE_LEVEL_STREAM_CHAIN: Chain = Chain::Ethereum;

/// Whether any requested protocol is streamed from Tycho.
///
/// The RFQ clients and the pAMM price level stream each connect to their own endpoint, so a list
/// naming only those needs no Tycho protocol stream at all.
pub(crate) fn has_tycho_protocols(protocols: &[String]) -> bool {
    protocols.iter().any(|protocol| {
        !protocol.starts_with(RFQ_PREFIX) && !protocol.starts_with(PRICE_LEVEL_STREAM_PREFIX)
    })
}

/// Whether the components labelled `protocol_system` are the ones a `--protocols` entry asked for.
///
/// Most entries name their own label. A `pricelevelstream:{venue}` entry names the venue to
/// stream, and its components arrive labelled `propammfallback:{venue}` when that venue is on the
/// PropAMMRouter whitelist, so both prefixes answer for the same entry.
pub fn matches_streamed_system(entry: &str, protocol_system: &str) -> bool {
    if entry == protocol_system {
        return true;
    }
    match (
        entry.strip_prefix(PRICE_LEVEL_STREAM_PREFIX),
        protocol_system.strip_prefix(PROPAMM_FALLBACK_PREFIX),
    ) {
        (Some(requested_venue), Some(streamed_venue)) => requested_venue == streamed_venue,
        _ => false,
    }
}

/// Whether any requested protocol is served by an RFQ client.
pub(crate) fn has_rfq_protocols(protocols: &[String]) -> bool {
    protocols
        .iter()
        .any(|protocol| protocol.starts_with(RFQ_PREFIX))
}

/// The `exclusive:` prefix was applied to a protocol system that has no exclusive variant.
#[derive(Debug, thiserror::Error)]
#[error(
    "protocol '{requested}' has no exclusive-liquidity variant; '{EXCLUSIVE_PREFIX}' is only \
     supported for: {supported}",
    supported = EXCLUSIVE_CAPABLE_PROTOCOLS.join(", ")
)]
pub struct UnsupportedExclusiveProtocol {
    /// The protocol system the prefix was applied to.
    requested: String,
}

/// A requested protocol system together with the liquidity variant to stream for it.
///
/// `parse` and the `Display` impl round-trip: displaying one yields a `--protocols` entry that
/// parses back to the same value.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProtocolSpec {
    /// Tycho protocol system name. Never carries the `exclusive:` prefix.
    pub system: String,
    /// Whether to register the filter that also admits exclusive pools.
    pub exclusive: bool,
}

impl ProtocolSpec {
    /// A protocol system streaming public liquidity only.
    pub fn public(system: impl Into<String>) -> Self {
        Self { system: system.into(), exclusive: false }
    }

    /// Parses a single `--protocols` entry.
    ///
    /// # Errors
    ///
    /// Returns `UnsupportedExclusiveProtocol` when the `exclusive:` prefix is applied to a protocol
    /// system that has no exclusive variant. Unrecognised protocol systems without the prefix are
    /// accepted here and skipped with a warning during registration.
    pub fn parse(entry: &str) -> Result<Self, UnsupportedExclusiveProtocol> {
        let Some(system) = entry.strip_prefix(EXCLUSIVE_PREFIX) else {
            return Ok(Self::public(entry));
        };
        if !EXCLUSIVE_CAPABLE_PROTOCOLS.contains(&system) {
            return Err(UnsupportedExclusiveProtocol { requested: system.to_string() });
        }
        Ok(Self { system: system.to_string(), exclusive: true })
    }
}

/// Parses a `--protocols` entry that drops a protocol system, e.g. `exclude:vm:fermiswap`.
///
/// Returns `None` for an entry that names a protocol to stream instead. The part after the prefix
/// goes through [`ProtocolSpec::parse`], so `exclude:ekubo_v3` and `exclude:exclusive:ekubo_v3`
/// both name the system `ekubo_v3` and a malformed exclusion fails the same way a malformed
/// request does. An entry naming nothing (`exclude:`) yields an empty system for the caller to
/// reject.
///
/// # Errors
///
/// Returns `UnsupportedExclusiveProtocol` when the excluded entry carries the `exclusive:` prefix
/// for a protocol system that has no exclusive variant.
pub fn parse_exclusion(entry: &str) -> Option<Result<String, UnsupportedExclusiveProtocol>> {
    let excluded = entry.strip_prefix(EXCLUDE_PREFIX)?;
    Some(ProtocolSpec::parse(excluded).map(|protocol| protocol.system))
}

impl fmt::Display for ProtocolSpec {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        if self.exclusive {
            write!(f, "{EXCLUSIVE_PREFIX}{}", self.system)
        } else {
            f.write_str(&self.system)
        }
    }
}

/// Register DEX protocol decoders for test tooling (record-market).
///
/// Wrapper over [`register_exchanges`] so the recorder builds the same protocol stream as
/// production without exposing the crate-private `DataFeedError`.
#[cfg(feature = "test-utils")]
pub fn register_exchanges_for_recording(
    builder: ProtocolStreamBuilder,
    tvl_filter: ComponentFilter,
    entries: &[String],
) -> Result<ProtocolStreamBuilder, String> {
    register_exchanges(builder, tvl_filter, entries).map_err(|e| e.to_string())
}

/// Parses every `--protocols` entry, rejecting a list with no unambiguous reading.
///
/// Registration is keyed by protocol system, so naming one system both with and without the
/// `exclusive:` prefix would silently keep whichever entry came last. Callers that expand a
/// protocol list (`fynd_rpc::protocols::resolve_protocols`) merge the variants before getting here;
/// a hand-assembled list gets an error instead of an order-dependent stream.
fn parse_protocols(entries: &[String]) -> Result<Vec<ProtocolSpec>, DataFeedError> {
    let mut protocols = Vec::with_capacity(entries.len());
    for entry in entries {
        protocols
            .push(ProtocolSpec::parse(entry).map_err(|e| DataFeedError::Config(e.to_string()))?);
    }

    let mut variants: HashMap<&str, bool> = HashMap::new();
    for protocol in &protocols {
        if variants
            .insert(protocol.system.as_str(), protocol.exclusive)
            .is_some_and(|previous| previous != protocol.exclusive)
        {
            return Err(DataFeedError::Config(format!(
                "protocol '{}' requested both with and without the '{EXCLUSIVE_PREFIX}' prefix",
                protocol.system
            )));
        }
    }
    Ok(protocols)
}

/// Register DEX protocol decoders on a [`ProtocolStreamBuilder`].
///
/// Entries may carry the `exclusive:` prefix to select the protocol's exclusive-liquidity stream
/// variant; doing so for a protocol without one is a configuration error, as is naming one protocol
/// both with and without the prefix.
pub(crate) fn register_exchanges(
    mut builder: ProtocolStreamBuilder,
    tvl_filter: ComponentFilter,
    entries: &[String],
) -> Result<ProtocolStreamBuilder, DataFeedError> {
    for protocol in parse_protocols(entries)? {
        match protocol.system.as_str() {
            "uniswap_v2" => {
                builder =
                    builder.exchange::<UniswapV2State>("uniswap_v2", tvl_filter.clone(), None);
            }
            "sushiswap_v2" => {
                builder =
                    builder.exchange::<UniswapV2State>("sushiswap_v2", tvl_filter.clone(), None);
            }
            "pancakeswap_v2" => {
                builder = builder.exchange::<PancakeswapV2State>(
                    "pancakeswap_v2",
                    tvl_filter.clone(),
                    None,
                );
            }
            "uniswap_v3" => {
                builder =
                    builder.exchange::<UniswapV3State>("uniswap_v3", tvl_filter.clone(), None);
            }
            "pancakeswap_v3" => {
                builder =
                    builder.exchange::<UniswapV3State>("pancakeswap_v3", tvl_filter.clone(), None);
            }
            "vm:balancer_v2" => {
                builder = builder.exchange::<EVMPoolState<PreCachedDB>>(
                    "vm:balancer_v2",
                    tvl_filter.clone(),
                    Some(balancer_v2_pool_filter),
                );
            }
            "uniswap_v4" => {
                builder =
                    builder.exchange::<UniswapV4State>("uniswap_v4", tvl_filter.clone(), None);
            }
            "ekubo_v2" => {
                builder = builder.exchange::<EkuboState>("ekubo_v2", tvl_filter.clone(), None);
            }
            "vm:curve" => {
                // The hybrid CurveState with tycho-simulation's own curve_filter, which drops
                // the components CurveState cannot quote correctly (oracle/rate-bearing/rebasing
                // coins) — the source of the overestimation that forced the temporary
                // full-EVM fallback (see #318); fixed upstream in tycho-simulation 0.338.0.
                builder = builder.exchange::<CurveState>(
                    "vm:curve",
                    tvl_filter.clone(),
                    Some(curve_filter),
                );
            }
            "uniswap_v4_hooks" => {
                builder = builder.exchange::<UniswapV4State>(
                    "uniswap_v4_hooks",
                    tvl_filter.clone(),
                    None,
                );
            }
            "vm:maverick_v2" => {
                builder = builder.exchange::<EVMPoolState<PreCachedDB>>(
                    "vm:maverick_v2",
                    tvl_filter.clone(),
                    None,
                );
            }
            "vm:bopamm" => {
                builder = builder.exchange::<EVMPoolState<PreCachedDB>>(
                    "vm:bopamm",
                    tvl_filter.clone(),
                    None,
                );
            }
            "vm:fermiswap" => {
                builder = builder.exchange::<EVMPoolState<PreCachedDB>>(
                    "vm:fermiswap",
                    tvl_filter.clone(),
                    None,
                );
            }
            "fluid_v1" => {
                builder = builder.exchange::<FluidV1>(
                    "fluid_v1",
                    tvl_filter.clone(),
                    Some(fluid_v1_paused_pools_filter),
                );
            }
            "aerodrome_v1" => {
                builder =
                    builder.exchange::<AerodromeV1State>("aerodrome_v1", tvl_filter.clone(), None);
            }
            "aerodrome_slipstreams" => {
                builder = builder.exchange::<AerodromeSlipstreamsState>(
                    "aerodrome_slipstreams",
                    tvl_filter.clone(),
                    None,
                );
            }
            "erc4626" => {
                builder = builder.exchange::<ERC4626State>(
                    "erc4626",
                    tvl_filter.clone(),
                    Some(erc4626_filter),
                );
            }
            "velodrome_slipstreams" => {
                builder = builder.exchange::<AerodromeSlipstreamsState>(
                    "velodrome_slipstreams",
                    tvl_filter.clone(),
                    None,
                );
            }
            "ekubo_v3" => {
                // SignedExclusiveSwap pools need a controller signature per swap, so they are
                // only streamed when the deployment explicitly opts in.
                let filter = if protocol.exclusive {
                    info!("Including exclusive liquidity for ekubo_v3");
                    ekubo_v3_extension_filter_with_signed_exclusive_swap
                } else {
                    ekubo_v3_extension_filter
                };
                builder =
                    builder.exchange::<EkuboV3State>("ekubo_v3", tvl_filter.clone(), Some(filter));
            }
            "quickswap_v2" => {
                builder =
                    builder.exchange::<UniswapV2State>("quickswap_v2", tvl_filter.clone(), None);
            }
            "lunarbase" => {
                builder = builder.exchange::<LunarBaseState>("lunarbase", tvl_filter.clone(), None);
            }
            p if p.starts_with(RFQ_PREFIX) || p.starts_with(PRICE_LEVEL_STREAM_PREFIX) => {
                // Handled by register_rfq and open_price_level_stream, which stream from their
                // own endpoints rather than from Tycho.
                continue;
            }
            _ => {
                warn!("Skipping unknown protocol: {}", protocol);
            }
        }
    }
    Ok(builder)
}

pub(crate) fn register_rfq(
    mut rfq_stream_builder: RFQStreamBuilder,
    chain: Chain,
    min_tvl: f64,
    protocols: &[String],
    rfq_tokens: std::collections::HashSet<Bytes>,
) -> Result<RFQStreamBuilder, DataFeedError> {
    for protocol in protocols {
        match protocol.as_str() {
            "rfq:bebop" => {
                let key = get_env("BEBOP_KEY")?;
                info!("Adding {protocol} RFQ client...");
                let bebop_client = BebopClientBuilder::new(chain, key)
                    .tokens(rfq_tokens.clone())
                    .tvl_threshold(min_tvl)
                    .build()
                    .map_err(|e| DataFeedError::StreamError(e.to_string()))?;
                rfq_stream_builder =
                    rfq_stream_builder.add_client::<BebopState>("bebop", Box::new(bebop_client));
            }
            "rfq:hashflow" => {
                let user = get_env("HASHFLOW_USER")?;
                let key = get_env("HASHFLOW_KEY")?;
                info!("Adding {protocol} RFQ client...");
                let hashflow_client = HashflowClientBuilder::new(chain, user, key)
                    .tokens(rfq_tokens.clone())
                    .tvl_threshold(min_tvl)
                    .poll_time(Duration::from_secs(30))
                    .build()
                    .map_err(|e| DataFeedError::StreamError(e.to_string()))?;
                rfq_stream_builder = rfq_stream_builder
                    .add_client::<HashflowState>("hashflow", Box::new(hashflow_client));
            }
            p if p.starts_with(RFQ_PREFIX) => {
                warn!("Skipping unknown RFQ protocol: {}", p);
            }
            _ => {}
        }
    }
    Ok(rfq_stream_builder)
}

/// Opens the Titan pAMM price level stream for the requested `pricelevelstream:` venues.
///
/// Returns `None` when no entry names the stream. Every named venue must be one of the venues
/// tycho-simulation knows how to execute against ([`default_served_pamms`]); a name outside that
/// set is a configuration error rather than a warning, because these entries are always written
/// by hand and a typo would otherwise silently stream nothing.
///
/// The stream reconnects on its own for as long as it is polled, so — unlike the RFQ clients —
/// it needs no supervising task.
///
/// A venue served here may also be integrated as a Tycho protocol system (FermiSwap is also
/// `vm:fermiswap`), in which case both price the same maker inventory. Streaming both
/// double-counts that liquidity, so drop the Tycho one from `--protocols` instead.
///
/// # Errors
///
/// Returns [`DataFeedError::Config`] if the chain is not [`PRICE_LEVEL_STREAM_CHAIN`], or if an
/// entry names a venue that is not served.
pub(crate) fn open_price_level_stream(
    chain: Chain,
    protocols: &[String],
    tokens: &HashMap<Bytes, Token>,
) -> Result<Option<impl Stream<Item = Update> + Send>, DataFeedError> {
    let venues: Vec<&str> = protocols
        .iter()
        .filter_map(|protocol| protocol.strip_prefix(PRICE_LEVEL_STREAM_PREFIX))
        .collect();
    if venues.is_empty() {
        return Ok(None);
    }
    if chain != PRICE_LEVEL_STREAM_CHAIN {
        return Err(DataFeedError::Config(format!(
            "the pAMM price level stream serves {PRICE_LEVEL_STREAM_CHAIN} only, but this feed \
             runs on {chain}"
        )));
    }

    let served = default_served_pamms();
    let mut builder = PriceLevelStreamBuilder::new().with_tokens(tokens.clone());
    for venue in venues {
        let Some(config) = served
            .iter()
            .find(|config| config.protocol == venue)
        else {
            return Err(DataFeedError::Config(format!(
                "unknown pAMM '{venue}' for the price level stream; served venues are: {}",
                served
                    .iter()
                    .map(|config| config.protocol.as_str())
                    .collect::<Vec<_>>()
                    .join(", ")
            )));
        };
        info!("Adding {PRICE_LEVEL_STREAM_PREFIX}{venue} price level venue...");
        builder = builder.add_pamm(config.clone());
    }
    Ok(Some(builder.build()))
}

/// Opens the pAMM price level stream for test tooling (the benchmark's live capture).
///
/// Wrapper over [`open_price_level_stream`] so the capture serves the same venues as
/// production without exposing the crate-private `DataFeedError`.
#[cfg(feature = "test-utils")]
pub fn open_price_level_stream_for_recording(
    chain: Chain,
    protocols: &[String],
    tokens: &HashMap<Bytes, Token>,
) -> Result<Option<impl Stream<Item = Update> + Send>, String> {
    open_price_level_stream(chain, protocols, tokens).map_err(|e| e.to_string())
}

fn get_env(var: &str) -> Result<String, DataFeedError> {
    env::var(var).map_err(|_| DataFeedError::Config(format!("{} env var not set", var)))
}

#[cfg(test)]
mod tests {
    use tycho_simulation::price_level_stream::config::PRICE_LEVEL_STREAM_FAMILY;

    use super::*;

    fn register(entries: &[&str]) -> Result<ProtocolStreamBuilder, DataFeedError> {
        register_exchanges(
            ProtocolStreamBuilder::new("localhost:0", Chain::Ethereum),
            ComponentFilter::with_tvl_range(1.0, 10.0),
            &entries
                .iter()
                .map(|entry| (*entry).to_string())
                .collect::<Vec<_>>(),
        )
    }

    fn price_level_stream(
        chain: Chain,
        entries: &[&str],
    ) -> Result<Option<impl Stream<Item = Update> + Send>, DataFeedError> {
        open_price_level_stream(
            chain,
            &entries
                .iter()
                .map(|entry| (*entry).to_string())
                .collect::<Vec<_>>(),
            &HashMap::new(),
        )
    }

    #[test]
    fn test_parse_plain_protocol() {
        let protocol = ProtocolSpec::parse("uniswap_v3").unwrap();
        assert_eq!(protocol, ProtocolSpec { system: "uniswap_v3".to_string(), exclusive: false });
    }

    #[test]
    fn test_parse_exclusive_protocol() {
        let protocol = ProtocolSpec::parse("exclusive:ekubo_v3").unwrap();
        assert_eq!(protocol, ProtocolSpec { system: "ekubo_v3".to_string(), exclusive: true });
    }

    #[test]
    fn test_parse_exclusive_unsupported_protocol() {
        let err = ProtocolSpec::parse("exclusive:uniswap_v3").unwrap_err();
        assert!(err
            .to_string()
            .contains("has no exclusive-liquidity variant"));
    }

    #[test]
    fn test_parse_exclusive_without_protocol() {
        assert!(ProtocolSpec::parse("exclusive:").is_err());
    }

    #[test]
    fn test_parse_leaves_other_prefixes_intact() {
        assert_eq!(
            ProtocolSpec::parse("rfq:bebop").unwrap(),
            ProtocolSpec { system: "rfq:bebop".to_string(), exclusive: false }
        );
        assert_eq!(
            ProtocolSpec::parse("vm:curve").unwrap(),
            ProtocolSpec { system: "vm:curve".to_string(), exclusive: false }
        );
    }

    #[test]
    fn test_parse_exclusion() {
        assert_eq!(
            parse_exclusion("exclude:vm:fermiswap")
                .unwrap()
                .unwrap(),
            "vm:fermiswap"
        );
    }

    #[test]
    fn test_parse_exclusion_strips_the_exclusive_prefix() {
        assert_eq!(
            parse_exclusion("exclude:exclusive:ekubo_v3")
                .unwrap()
                .unwrap(),
            "ekubo_v3"
        );
    }

    #[test]
    fn test_parse_exclusion_rejects_unsupported_exclusive() {
        assert!(parse_exclusion("exclude:exclusive:uniswap_v3")
            .unwrap()
            .is_err());
    }

    #[test]
    fn test_parse_exclusion_without_protocol() {
        assert_eq!(
            parse_exclusion("exclude:")
                .unwrap()
                .unwrap(),
            ""
        );
    }

    #[test]
    fn test_parse_exclusion_of_a_plain_entry() {
        assert!(parse_exclusion("uniswap_v3").is_none());
    }

    #[test]
    fn test_display_round_trips() {
        for entry in ["uniswap_v3", "exclusive:ekubo_v3", "rfq:bebop", "vm:curve"] {
            let protocol = ProtocolSpec::parse(entry).unwrap();
            assert_eq!(protocol.to_string(), entry);
            assert_eq!(ProtocolSpec::parse(&protocol.to_string()).unwrap(), protocol);
        }
    }

    #[test]
    fn test_register_exchanges_accepts_exclusive_ekubo_v3() {
        assert!(register(&["uniswap_v3", "exclusive:ekubo_v3"]).is_ok());
    }

    #[test]
    fn test_register_exchanges_rejects_unsupported_exclusive() {
        let Err(err) = register(&["exclusive:uniswap_v3"]) else {
            panic!("expected `exclusive:uniswap_v3` to be rejected");
        };
        assert!(matches!(err, DataFeedError::Config(_)), "expected a config error, got {err:?}");
    }

    #[test]
    fn test_register_exchanges_skips_unknown_protocol() {
        assert!(register(&["not_a_protocol"]).is_ok());
    }

    #[test]
    fn test_register_exchanges_rejects_conflicting_variants() {
        for protocols in [["ekubo_v3", "exclusive:ekubo_v3"], ["exclusive:ekubo_v3", "ekubo_v3"]] {
            let Err(err) = register(&protocols) else {
                panic!("expected {protocols:?} to be rejected");
            };
            assert!(
                err.to_string()
                    .contains("both with and without"),
                "unexpected error for {protocols:?}: {err}"
            );
        }
    }

    #[test]
    fn test_register_exchanges_allows_repeated_protocol() {
        assert!(register(&["uniswap_v3", "uniswap_v3"]).is_ok());
    }
    #[test]
    fn test_price_level_stream_prefix_matches_family() {
        assert_eq!(PRICE_LEVEL_STREAM_PREFIX, format!("{PRICE_LEVEL_STREAM_FAMILY}:"));
    }

    #[test]
    fn test_matches_streamed_system() {
        assert!(matches_streamed_system("uniswap_v3", "uniswap_v3"));
        assert!(matches_streamed_system(
            "pricelevelstream:fermiswap",
            "pricelevelstream:fermiswap"
        ));
        // The whitelisted venue arrives under the router's family for the same entry.
        assert!(matches_streamed_system("pricelevelstream:fermiswap", "propammfallback:fermiswap"));
    }

    #[test]
    fn test_matches_streamed_system_rejects_another_venue() {
        assert!(!matches_streamed_system("pricelevelstream:fermiswap", "propammfallback:kipseli"));
        assert!(!matches_streamed_system("pricelevelstream:fermiswap", "vm:fermiswap"));
        assert!(!matches_streamed_system("uniswap_v3", "propammfallback:fermiswap"));
        assert!(!matches_streamed_system("vm:fermiswap", "propammfallback:fermiswap"));
    }

    #[test]
    fn test_has_tycho_protocols() {
        assert!(has_tycho_protocols(&["uniswap_v3".to_string()]));
        assert!(has_tycho_protocols(&["rfq:bebop".to_string(), "uniswap_v3".to_string()]));
        assert!(!has_tycho_protocols(&[
            "rfq:bebop".to_string(),
            "pricelevelstream:fermiswap".to_string(),
        ]));
        assert!(!has_tycho_protocols(&[]));
    }

    #[test]
    fn test_register_exchanges_skips_price_level_entries() {
        assert!(register(&["uniswap_v3", "pricelevelstream:fermiswap"]).is_ok());
    }

    #[test]
    fn test_open_price_level_stream_without_entries() {
        let Ok(None) = price_level_stream(Chain::Ethereum, &["uniswap_v3", "rfq:bebop"]) else {
            panic!("expected no price level stream without a `pricelevelstream:` entry");
        };
    }

    #[test]
    fn test_open_price_level_stream_served_venue() {
        let Ok(Some(_)) = price_level_stream(Chain::Ethereum, &["pricelevelstream:fermiswap"])
        else {
            panic!("expected fermiswap to be served");
        };
    }

    #[test]
    fn test_open_price_level_stream_several_venues() {
        let Ok(Some(_)) = price_level_stream(
            Chain::Ethereum,
            &["pricelevelstream:fermiswap", "pricelevelstream:kipseli"],
        ) else {
            panic!("expected both venues to be served");
        };
    }

    #[test]
    fn test_open_price_level_stream_unknown_venue() {
        for entries in [
            vec!["pricelevelstream:nope"],
            vec!["pricelevelstream:fermiswap", "pricelevelstream:nope"],
        ] {
            let Err(err) = price_level_stream(Chain::Ethereum, &entries) else {
                panic!("expected an unserved venue to be rejected in {entries:?}");
            };
            assert!(
                err.to_string()
                    .contains("unknown pAMM 'nope'"),
                "got {err}"
            );
        }
    }

    #[test]
    fn test_open_price_level_stream_without_entries_off_ethereum() {
        let Ok(None) = price_level_stream(Chain::Base, &["uniswap_v3"]) else {
            panic!("expected chains without a `pricelevelstream:` entry to be left alone");
        };
    }

    #[test]
    fn test_open_price_level_stream_other_chain() {
        let Err(err) = price_level_stream(Chain::Base, &["pricelevelstream:fermiswap"]) else {
            panic!("expected the price level stream to be rejected off Ethereum");
        };
        assert!(
            err.to_string()
                .contains("serves ethereum only"),
            "got {err}"
        );
    }
}