reddb_wire/

conn_string.rs

//! Connection-string parser shared across `reddb`, `reddb-client`,
//! `red_client`, and every language driver.
//!
//! Pure function over a string; no I/O, no allocation beyond what the
//! returned [`ConnectionTarget`] needs. The grammar is defined by
//! `docs/clients/connection-strings.md`; this module is the canonical
//! parser and is the single source of truth consumed by the rest of
//! the workspace.
//!
//! The parser ports the logic that previously lived in
//! `drivers/rust/src/connect.rs` (which keeps a thin re-export layer
//! for backwards compatibility while drivers migrate over). Cluster
//! URIs (`grpc://primary,replica:port`), default ports per scheme,
//! and the `?route=primary` override behave identically to the
//! original.

use std::path::PathBuf;

use url::Url;

/// Stable error code for parser failures.
///
/// Mirrors the `ErrorCode` shape used by the language drivers so that
/// downstream wrappers can map 1:1 without information loss.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ParseErrorKind {
    /// The input was empty.
    Empty,
    /// `url::Url` rejected the string, or a transport-specific
    /// invariant (missing host, empty cluster entry, bad port…) was
    /// violated.
    InvalidUri,
    /// The scheme is not in the documented vocabulary.
    UnsupportedScheme,
    /// A DoS guardrail in [`ConnStringLimits`] was tripped.
    /// `message` carries the limit name + the offending value so
    /// downstream wrappers can surface the structured detail.
    LimitExceeded,
}

impl ParseErrorKind {
    pub fn as_str(self) -> &'static str {
        match self {
            ParseErrorKind::Empty => "EMPTY",
            ParseErrorKind::InvalidUri => "INVALID_URI",
            ParseErrorKind::UnsupportedScheme => "UNSUPPORTED_SCHEME",
            ParseErrorKind::LimitExceeded => "LIMIT_EXCEEDED",
        }
    }
}

/// Error returned by [`parse`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ParseError {
    pub kind: ParseErrorKind,
    pub message: String,
}

impl ParseError {
    pub fn new(kind: ParseErrorKind, message: impl Into<String>) -> Self {
        Self {
            kind,
            message: message.into(),
        }
    }
}

impl std::fmt::Display for ParseError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}: {}", self.kind.as_str(), self.message)
    }
}

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

/// Default port per documented scheme. Centralised so other crates
/// (the connector, server-side dispatch) can stay consistent.
pub const DEFAULT_PORT_RED: u16 = 5050;
pub const DEFAULT_PORT_GRPC: u16 = 55055;
pub const DEFAULT_PORT_GRPCS: u16 = 55555;
/// Default ports for `ws://` / `red+ws://` and `wss://` / `red+wss://` — align with the
/// standard WS / WSS browser defaults (80 and 443) so a hosted endpoint
/// like `*.db.reddb.io` works without an explicit port.
pub const DEFAULT_PORT_WS: u16 = 80;
pub const DEFAULT_PORT_WSS: u16 = 443;

/// URI schemes accepted by the connection-string parser.
///
/// This enum is the connection-layer source for generated agent knowledge:
/// [`crate::knowledge`] iterates [`SUPPORTED_SCHEMES`] instead of carrying a
/// separate hand-maintained scheme list.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConnectionScheme {
    Memory,
    File,
    Red,
    Reds,
    RedWs,
    RedWss,
    Ws,
    Wss,
    Grpc,
    Grpcs,
    Http,
    Https,
}

/// Stable parser-owned list of supported URI schemes.
pub const SUPPORTED_SCHEMES: &[ConnectionScheme] = &[
    ConnectionScheme::Red,
    ConnectionScheme::Reds,
    ConnectionScheme::Grpc,
    ConnectionScheme::Grpcs,
    ConnectionScheme::Http,
    ConnectionScheme::Https,
    ConnectionScheme::Memory,
    ConnectionScheme::File,
    ConnectionScheme::RedWs,
    ConnectionScheme::RedWss,
    ConnectionScheme::Ws,
    ConnectionScheme::Wss,
];

impl ConnectionScheme {
    pub fn from_uri_scheme(scheme: &str) -> Option<Self> {
        match scheme {
            "memory" => Some(Self::Memory),
            "file" => Some(Self::File),
            "red" => Some(Self::Red),
            "reds" => Some(Self::Reds),
            "red+ws" => Some(Self::RedWs),
            "red+wss" => Some(Self::RedWss),
            "ws" => Some(Self::Ws),
            "wss" => Some(Self::Wss),
            "grpc" => Some(Self::Grpc),
            "grpcs" => Some(Self::Grpcs),
            "http" => Some(Self::Http),
            "https" => Some(Self::Https),
            _ => None,
        }
    }

    pub fn uri_prefix(self) -> &'static str {
        match self {
            Self::Memory => "memory://",
            Self::File => "file://",
            Self::Red => "red://",
            Self::Reds => "reds://",
            Self::RedWs => "red+ws://",
            Self::RedWss => "red+wss://",
            Self::Ws => "ws://",
            Self::Wss => "wss://",
            Self::Grpc => "grpc://",
            Self::Grpcs => "grpcs://",
            Self::Http => "http://",
            Self::Https => "https://",
        }
    }

    pub fn transport(self) -> &'static str {
        match self {
            Self::Memory => "embedded in-memory engine",
            Self::File => "embedded file-backed engine",
            Self::Red | Self::Reds => "RedWire TCP",
            Self::RedWs | Self::RedWss | Self::Ws | Self::Wss => "RedWire WebSocket",
            Self::Grpc | Self::Grpcs => "gRPC",
            Self::Http | Self::Https => "HTTP REST",
        }
    }

    pub fn mode(self) -> &'static str {
        match self {
            Self::Memory | Self::File => "embedded",
            Self::Red
            | Self::Reds
            | Self::RedWs
            | Self::RedWss
            | Self::Ws
            | Self::Wss
            | Self::Grpc
            | Self::Grpcs
            | Self::Http
            | Self::Https => "remote",
        }
    }

    pub fn example(self) -> &'static str {
        match self {
            Self::Memory => "memory://",
            Self::File => "file:///var/lib/reddb/app.db",
            Self::Red => "red://db.example.com:5050",
            Self::Reds => "reds://db.example.com:5050",
            Self::RedWs => "red+ws://db.example.com",
            Self::RedWss => "red+wss://db.example.com",
            Self::Ws => "ws://db.example.com",
            Self::Wss => "wss://db.example.com",
            Self::Grpc => "grpc://db.example.com:55055",
            Self::Grpcs => "grpcs://db.example.com:55555",
            Self::Http => "http://db.example.com:80",
            Self::Https => "https://db.example.com:443",
        }
    }

    pub fn notes(self) -> &'static str {
        match self {
            Self::Memory => "Zero-config ephemeral engine, commonly used by local MCP hosts.",
            Self::File => "Embedded durable engine rooted at the URI path.",
            Self::Red => "Principal RedWire transport without TLS.",
            Self::Reds => "Principal RedWire transport with TLS.",
            Self::RedWs => "Browser-native RedWire over WebSocket without TLS.",
            Self::RedWss => "Browser-native RedWire over WebSocket with TLS.",
            Self::Ws => "Browser-friendly alias for RedWire over WebSocket without TLS.",
            Self::Wss => "Browser-friendly alias for RedWire over WebSocket with TLS.",
            Self::Grpc => "Compatibility transport for existing gRPC clients.",
            Self::Grpcs => "TLS variant of the gRPC compatibility transport.",
            Self::Http => "REST/admin transport without TLS.",
            Self::Https => "REST/admin transport with TLS.",
        }
    }
}

/// DoS guardrails applied by [`parse`] before any URI work happens.
///
/// The connection-string parser is the only entry point an attacker
/// can reach BEFORE auth, so every limit here is enforced eagerly
/// and surfaces as a structured [`ParseErrorKind::LimitExceeded`]
/// error rather than a panic, hang, or unbounded allocation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ConnStringLimits {
    /// Maximum length of the input URI in bytes. Default `8 KiB`.
    pub max_uri_bytes: usize,
    /// Maximum number of `key=value` query parameters. Default `32`.
    pub max_query_params: usize,
    /// Maximum number of comma-separated cluster hosts allowed in a
    /// `grpc://`/`grpcs://` cluster URI. Default `64`.
    pub max_cluster_hosts: usize,
}

impl Default for ConnStringLimits {
    fn default() -> Self {
        Self {
            max_uri_bytes: 8 * 1024,
            max_query_params: 32,
            max_cluster_hosts: 64,
        }
    }
}

/// Normalised target produced by [`parse`].
///
/// Variants intentionally mirror the public Rust client target shape
/// so callers can keep a thin compatibility layer without duplicating
/// parser behavior.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ConnectionTarget {
    /// `memory://` — ephemeral, in-memory backend.
    Memory,
    /// `file:///abs/path` — embedded engine on disk.
    File { path: PathBuf },
    /// Single remote endpoint over `grpc://` or `grpcs://`. Stored as a
    /// normalised `http://` or `https://` endpoint for tonic.
    Grpc { endpoint: String },
    /// Multi-host gRPC URI: primary + read replicas. Writes hit the
    /// primary; reads round-robin across replicas unless
    /// `force_primary` is set.
    GrpcCluster {
        primary: String,
        replicas: Vec<String>,
        force_primary: bool,
    },
    /// `http://host:port` / `https://host:port` — REST endpoint.
    Http { base_url: String },
    /// `red://host:port` (plain TCP) or `reds://host:port` (TLS).
    /// RedWire binary frame protocol per ADR 0001. The connector
    /// speaks framed binary directly; it does NOT route through
    /// tonic.
    RedWire { host: String, port: u16, tls: bool },
    /// `red+ws://host:port` / `ws://host:port` (plain WS) or
    /// `red+wss://host:port` / `wss://host:port` (WSS).
    /// Browser-native WebSocket transport (ADR 0047 direct-when-reachable).
    /// The UI connects directly — no local RedWire-over-TCP bridge needed.
    WsNative { host: String, port: u16, tls: bool },
}

/// Authentication material derived from a connection string or adjacent CLI
/// fallback. `Debug` deliberately redacts every caller-supplied credential.
#[derive(Clone, PartialEq, Eq)]
pub enum ConnectionAuth {
    Anonymous,
    Bearer(String),
    Basic { user: String, pass: String },
    ApiKey(String),
}

impl std::fmt::Debug for ConnectionAuth {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Anonymous => f.write_str("Anonymous"),
            Self::Bearer(_) => f.debug_tuple("Bearer").field(&"<redacted>").finish(),
            Self::Basic { .. } => f
                .debug_struct("Basic")
                .field("user", &"<redacted>")
                .field("pass", &"<redacted>")
                .finish(),
            Self::ApiKey(_) => f.debug_tuple("ApiKey").field(&"<redacted>").finish(),
        }
    }
}

impl ConnectionAuth {
    pub fn bearer(token: impl Into<String>) -> Self {
        Self::Bearer(token.into())
    }

    pub fn is_bearer(&self) -> bool {
        matches!(self, Self::Bearer(_))
    }
}

/// Connection target plus URL-derived auth metadata.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConnectionSpec {
    pub target: ConnectionTarget,
    pub auth: ConnectionAuth,
    pub redacted_uri: String,
}

/// Parse a connection URI into a [`ConnectionTarget`] under the
/// default DoS limits.
///
/// Pure function, no side effects. Behaviour matches
/// `drivers/rust/src/connect.rs::parse` 1:1 with two additions:
///   - Mixed-case schemes (e.g. `Red://`, `REDS://`) are normalised
///     to lowercase before dispatch.
///   - Inputs exceeding [`ConnStringLimits`] return a structured
///     [`ParseErrorKind::LimitExceeded`] error instead of being
///     processed.
pub fn parse(uri: &str) -> Result<ConnectionTarget, ParseError> {
    parse_with_limits(uri, ConnStringLimits::default())
}

/// Parse a connection URI and derive auth from RFC 3986 userinfo.
///
/// The existing [`parse`] function remains target-only for compatibility.
/// New client-mode entry points should use this richer shape so credentials
/// are identified and redacted before transport connectors are built.
pub fn parse_with_auth(uri: &str) -> Result<ConnectionSpec, ParseError> {
    let normalised = normalise_scheme(uri);
    let redacted_uri = redact_connection_uri(&normalised);
    let target =
        parse(uri).map_err(|err| redact_parse_error(err, uri, &normalised, &redacted_uri))?;
    let auth = auth_from_uri_userinfo(&normalised)
        .map_err(|err| redact_parse_error(err, uri, &normalised, &redacted_uri))?;
    Ok(ConnectionSpec {
        target,
        auth,
        redacted_uri,
    })
}

fn redact_parse_error(
    mut err: ParseError,
    raw_uri: &str,
    normalised_uri: &str,
    redacted_uri: &str,
) -> ParseError {
    err.message = err
        .message
        .replace(raw_uri, redacted_uri)
        .replace(normalised_uri, redacted_uri);
    err
}

/// Return true for documented embedded aliases that must not resolve to
/// a remote transport target.
///
/// This is intentionally separate from [`parse`]: legacy clients may need
/// to reject embedded targets before mapping `red://host` onto a remote
/// compatibility transport.
pub fn is_embedded_connection_uri(uri: &str) -> bool {
    let trimmed = uri.trim();
    matches!(
        trimmed,
        "red://" | "red:" | "red:///" | "red://:memory" | "red://:memory:"
    ) || trimmed.starts_with("red:///")
}

/// Same as [`parse`] but with caller-supplied DoS guardrails.
/// Useful for tests that need tighter limits or for callers (a
/// future admin tool, an offline validator) that need to relax the
/// defaults.
pub fn parse_with_limits(
    uri: &str,
    limits: ConnStringLimits,
) -> Result<ConnectionTarget, ParseError> {
    if uri.is_empty() {
        return Err(ParseError::new(
            ParseErrorKind::Empty,
            "empty connection string",
        ));
    }

    if uri.len() > limits.max_uri_bytes {
        return Err(ParseError::new(
            ParseErrorKind::LimitExceeded,
            format!(
                "max_uri_bytes exceeded: limit={} actual={}",
                limits.max_uri_bytes,
                uri.len(),
            ),
        ));
    }

    // Lowercase the scheme so `Red://Host`, `REDS://Host`, etc.
    // dispatch identically to the canonical lowercase forms. The
    // host and path retain original casing — host is downcased by
    // `url::Url` for IDN per spec, path stays verbatim.
    let normalised = normalise_scheme(uri);
    let uri = normalised.as_str();

    if uri == "memory://" || uri == "memory:" {
        return Ok(ConnectionTarget::Memory);
    }

    if let Some(rest) = uri.strip_prefix("file://") {
        if rest.is_empty() {
            return Err(ParseError::new(
                ParseErrorKind::InvalidUri,
                "file:// URI is missing a path",
            ));
        }
        return Ok(ConnectionTarget::File {
            path: file_uri_path(rest),
        });
    }

    if let Some(cluster) = try_parse_grpc_cluster(uri, &limits)? {
        return Ok(cluster);
    }

    // Before `Url::parse`, because a multi-host authority like
    // `red://a:1,b:2` fails URL parsing with "invalid port number" — a
    // message that hides the real problem. `red://` no longer folds to
    // gRPC (#2159), and RedWire has no multi-host form, so a cluster URI
    // has to name the transport that supports one.
    if let Some(rest) = uri
        .strip_prefix("red://")
        .or_else(|| uri.strip_prefix("reds://"))
    {
        let authority = rest.split(['/', '?', '#']).next().unwrap_or(rest);
        if authority.contains(',') {
            return Err(ParseError::new(
                ParseErrorKind::InvalidUri,
                "RedWire does not support comma-separated cluster hosts; \
                 use grpc:// or grpcs:// for a cluster URI",
            ));
        }
    }

    let parsed = Url::parse(uri)
        .map_err(|e| ParseError::new(ParseErrorKind::InvalidUri, format!("{e}: {uri}")))?;

    enforce_query_param_limit(&parsed, &limits)?;

    match ConnectionScheme::from_uri_scheme(parsed.scheme()) {
        Some(ConnectionScheme::Red | ConnectionScheme::Reds) => {
            let host = parsed.host_str().ok_or_else(|| {
                ParseError::new(ParseErrorKind::InvalidUri, "red:// URI is missing a host")
            })?;
            if host.contains(',') {
                return Err(ParseError::new(
                    ParseErrorKind::InvalidUri,
                    "RedWire does not support comma-separated cluster hosts",
                ));
            }
            let port = parsed.port().unwrap_or(DEFAULT_PORT_RED);
            Ok(ConnectionTarget::RedWire {
                host: host.to_string(),
                port,
                tls: parsed.scheme() == "reds",
            })
        }
        Some(
            ConnectionScheme::RedWs
            | ConnectionScheme::RedWss
            | ConnectionScheme::Ws
            | ConnectionScheme::Wss,
        ) => {
            let host = parsed.host_str().ok_or_else(|| {
                ParseError::new(
                    ParseErrorKind::InvalidUri,
                    "RedWire WebSocket URI is missing a host",
                )
            })?;
            let tls = parsed.scheme() == "red+wss" || parsed.scheme() == "wss";
            let port = parsed.port().unwrap_or(if tls {
                DEFAULT_PORT_WSS
            } else {
                DEFAULT_PORT_WS
            });
            Ok(ConnectionTarget::WsNative {
                host: host.to_string(),
                port,
                tls,
            })
        }
        Some(ConnectionScheme::Grpc | ConnectionScheme::Grpcs) => {
            let host = parsed.host_str().ok_or_else(|| {
                ParseError::new(ParseErrorKind::InvalidUri, "grpc:// URI is missing a host")
            })?;
            let port = parsed.port().unwrap_or_else(|| {
                if parsed.scheme() == "grpcs" {
                    DEFAULT_PORT_GRPCS
                } else {
                    DEFAULT_PORT_GRPC
                }
            });
            let endpoint_scheme = if parsed.scheme() == "grpcs" {
                "https"
            } else {
                "http"
            };
            Ok(ConnectionTarget::Grpc {
                endpoint: format!("{endpoint_scheme}://{host}:{port}"),
            })
        }
        Some(ConnectionScheme::Http | ConnectionScheme::Https) => {
            let host = parsed.host_str().ok_or_else(|| {
                ParseError::new(
                    ParseErrorKind::InvalidUri,
                    "http(s):// URI is missing a host",
                )
            })?;
            let scheme = parsed.scheme();
            let port = parsed
                .port()
                .unwrap_or(if scheme == "https" { 443 } else { 80 });
            Ok(ConnectionTarget::Http {
                base_url: format!("{scheme}://{host}:{port}"),
            })
        }
        Some(ConnectionScheme::Memory | ConnectionScheme::File) | None => Err(ParseError::new(
            ParseErrorKind::UnsupportedScheme,
            format!("unsupported scheme: {}", parsed.scheme()),
        )),
    }
}

fn file_uri_path(path: &str) -> PathBuf {
    #[cfg(windows)]
    {
        let bytes = path.as_bytes();
        // RFC 8089 spells a Windows drive path as file:///C:/path. After
        // removing `file://`, discard only the URI separator before a drive
        // letter; retaining it produces the invalid Windows path `\C:\...`.
        if bytes.len() >= 4
            && bytes[0] == b'/'
            && bytes[1].is_ascii_alphabetic()
            && bytes[2] == b':'
            && matches!(bytes[3], b'/' | b'\\')
        {
            return PathBuf::from(&path[1..]);
        }
    }
    PathBuf::from(path)
}

/// Lowercase only the scheme portion (everything before the first
/// `:`), leaving host/path/query untouched. Returns the original
/// string when no scheme separator is present so the downstream
/// `Url::parse` path produces the canonical "missing scheme" error
/// instead of being masked here.
fn normalise_scheme(uri: &str) -> String {
    match uri.find(':') {
        Some(i) => {
            let scheme = &uri[..i];
            // Only ASCII alphanumerics + `+ . -` are valid scheme
            // bytes per RFC 3986. If the prefix violates that we
            // leave it alone so `Url::parse` can produce the
            // structured error.
            if scheme.is_empty()
                || !scheme
                    .bytes()
                    .all(|b| b.is_ascii_alphanumeric() || b == b'+' || b == b'.' || b == b'-')
            {
                return uri.to_string();
            }
            let mut out = String::with_capacity(uri.len());
            out.push_str(&scheme.to_ascii_lowercase());
            out.push_str(&uri[i..]);
            out
        }
        None => uri.to_string(),
    }
}

fn auth_from_uri_userinfo(uri: &str) -> Result<ConnectionAuth, ParseError> {
    if uri == "memory://" || uri == "memory:" || uri.starts_with("file://") {
        return Ok(ConnectionAuth::Anonymous);
    }
    // URL parsers cannot parse multiple independently ported authorities.
    // Authentication belongs to the entire seed list, not its first host.
    let mut auth_seed = uri.to_string();
    if uri.starts_with("grpc://") || uri.starts_with("grpcs://") {
        if let Some((scheme, rest)) = uri.split_once("://") {
            let end = rest.find(['/', '?', '#']).unwrap_or(rest.len());
            let authority = &rest[..end];
            let (userinfo, hosts) = authority
                .rsplit_once('@')
                .map(|(userinfo, hosts)| (format!("{userinfo}@"), hosts))
                .unwrap_or_else(|| (String::new(), authority));
            if hosts.contains(',') {
                auth_seed = format!(
                    "{scheme}://{userinfo}{}{}",
                    hosts.split(',').next().unwrap_or_default(),
                    &rest[end..]
                );
            }
        }
    }
    let parsed = Url::parse(&auth_seed)
        .map_err(|e| ParseError::new(ParseErrorKind::InvalidUri, format!("{e}: {uri}")))?;
    let username = parsed.username();
    if username.is_empty() {
        return Ok(ConnectionAuth::Anonymous);
    }
    match parsed.password() {
        Some(pass) => Ok(ConnectionAuth::Basic {
            user: username.to_string(),
            pass: pass.to_string(),
        }),
        None => Ok(ConnectionAuth::ApiKey(username.to_string())),
    }
}

fn redact_connection_uri(uri: &str) -> String {
    let uri = redact_uri_userinfo(uri);
    redact_uri_query_secrets(&uri)
}

fn redact_uri_userinfo(uri: &str) -> String {
    let Some(scheme_end) = uri.find("://") else {
        return uri.to_string();
    };
    let authority_start = scheme_end + 3;
    let authority_end = uri[authority_start..]
        .find(['/', '?', '#'])
        .map(|i| authority_start + i)
        .unwrap_or(uri.len());
    let authority = &uri[authority_start..authority_end];
    let Some(at) = authority.rfind('@') else {
        return uri.to_string();
    };
    let userinfo = &authority[..at];
    let replacement = if userinfo.contains(':') {
        "<redacted>:<redacted>"
    } else {
        "<redacted>"
    };
    format!(
        "{}{}{}",
        &uri[..authority_start],
        replacement,
        &uri[authority_start + at..]
    )
}

fn redact_uri_query_secrets(uri: &str) -> String {
    let Some(query_start) = uri.find('?') else {
        return uri.to_string();
    };
    let query_end = uri[query_start + 1..]
        .find('#')
        .map(|offset| query_start + 1 + offset)
        .unwrap_or(uri.len());
    let query = &uri[query_start + 1..query_end];
    let mut redacted = String::with_capacity(uri.len());
    redacted.push_str(&uri[..query_start + 1]);
    for (index, parameter) in query.split('&').enumerate() {
        if index > 0 {
            redacted.push('&');
        }
        let (name, value) = parameter.split_once('=').unwrap_or((parameter, ""));
        redacted.push_str(name);
        if parameter.contains('=') {
            redacted.push('=');
            if sensitive_query_name(name) {
                redacted.push_str("<redacted>");
            } else {
                redacted.push_str(value);
            }
        }
    }
    redacted.push_str(&uri[query_end..]);
    redacted
}

fn sensitive_query_name(raw: &str) -> bool {
    let mut decoded = Vec::with_capacity(raw.len());
    let bytes = raw.as_bytes();
    let mut index = 0;
    while index < bytes.len() {
        match bytes[index] {
            b'%' if index + 2 < bytes.len() => {
                let high = (bytes[index + 1] as char).to_digit(16);
                let low = (bytes[index + 2] as char).to_digit(16);
                if let (Some(high), Some(low)) = (high, low) {
                    decoded.push(((high << 4) | low) as u8);
                    index += 3;
                    continue;
                }
                decoded.push(bytes[index]);
            }
            b'+' => decoded.push(b' '),
            byte => decoded.push(byte),
        }
        index += 1;
    }
    let normalized: String = String::from_utf8_lossy(&decoded)
        .chars()
        .filter(|ch| *ch != '_' && *ch != '-')
        .flat_map(char::to_lowercase)
        .collect();
    matches!(
        normalized.as_str(),
        "token" | "password" | "pass" | "passwd" | "secret" | "cert" | "key" | "ca" | "apikey"
    ) || normalized.ends_with("token")
        || normalized.ends_with("password")
        || normalized.ends_with("passwd")
        || normalized.ends_with("secret")
        || normalized.ends_with("apikey")
        || normalized.ends_with("privatekey")
        || normalized.ends_with("keyfile")
        || normalized.ends_with("certfile")
        || normalized.ends_with("cafile")
        || ((normalized.starts_with("ssl") || normalized.starts_with("tls"))
            && (normalized.ends_with("key")
                || normalized.ends_with("cert")
                || normalized.ends_with("ca")))
}

fn enforce_query_param_limit(url: &Url, limits: &ConnStringLimits) -> Result<(), ParseError> {
    let Some(q) = url.query() else {
        return Ok(());
    };
    if q.is_empty() {
        return Ok(());
    }
    let count = q.split('&').count();
    if count > limits.max_query_params {
        return Err(ParseError::new(
            ParseErrorKind::LimitExceeded,
            format!(
                "max_query_params exceeded: limit={} actual={}",
                limits.max_query_params, count,
            ),
        ));
    }
    Ok(())
}

/// Try to parse a multi-host gRPC URI. `Ok(None)` means "this is a
/// single-host URI — fall through to the standard parser".
fn try_parse_grpc_cluster(
    uri: &str,
    limits: &ConnStringLimits,
) -> Result<Option<ConnectionTarget>, ParseError> {
    let (rest, default_port, endpoint_scheme) = if let Some(r) = uri.strip_prefix("grpc://") {
        (r, DEFAULT_PORT_GRPC, "http")
    } else if let Some(r) = uri.strip_prefix("grpcs://") {
        (r, DEFAULT_PORT_GRPCS, "https")
    } else {
        return Ok(None);
    };

    let (host_part, query_part) = match rest.find('?') {
        Some(i) => (&rest[..i], Some(&rest[i + 1..])),
        None => (rest, None),
    };
    let host_part = host_part
        .rsplit_once('@')
        .map(|(_, hosts)| hosts)
        .unwrap_or(host_part);

    if !host_part.contains(',') {
        return Ok(None);
    }

    let raw_count = host_part.split(',').count();
    if raw_count > limits.max_cluster_hosts {
        return Err(ParseError::new(
            ParseErrorKind::LimitExceeded,
            format!(
                "max_cluster_hosts exceeded: limit={} actual={}",
                limits.max_cluster_hosts, raw_count,
            ),
        ));
    }

    let mut endpoints: Vec<String> = Vec::with_capacity(raw_count);
    for raw in host_part.split(',') {
        let raw = raw.trim();
        if raw.is_empty() {
            return Err(ParseError::new(
                ParseErrorKind::InvalidUri,
                "grpc cluster URI has an empty host entry",
            ));
        }
        // Bracketed IPv6 literal: `[::1]:5050` or `[::1]`.
        let (host, port) = if let Some(after_bracket) = raw.strip_prefix('[') {
            let end = after_bracket.find(']').ok_or_else(|| {
                ParseError::new(
                    ParseErrorKind::InvalidUri,
                    format!("unterminated IPv6 bracket in cluster URI: {raw}"),
                )
            })?;
            let host = &after_bracket[..end];
            let tail = &after_bracket[end + 1..];
            let port = if tail.is_empty() {
                default_port
            } else if let Some(p) = tail.strip_prefix(':') {
                p.parse::<u16>().map_err(|_| {
                    ParseError::new(
                        ParseErrorKind::InvalidUri,
                        format!("invalid port in cluster URI: {raw}"),
                    )
                })?
            } else {
                return Err(ParseError::new(
                    ParseErrorKind::InvalidUri,
                    format!("trailing junk after IPv6 bracket in cluster URI: {raw}"),
                ));
            };
            (format!("[{host}]"), port)
        } else {
            match raw.rsplit_once(':') {
                Some((h, p)) => {
                    let port: u16 = p.parse().map_err(|_| {
                        ParseError::new(
                            ParseErrorKind::InvalidUri,
                            format!("invalid port in cluster URI: {raw}"),
                        )
                    })?;
                    (h.to_string(), port)
                }
                None => (raw.to_string(), default_port),
            }
        };
        if host.is_empty() || host == "[]" {
            return Err(ParseError::new(
                ParseErrorKind::InvalidUri,
                "grpc cluster URI has an empty host entry",
            ));
        }
        endpoints.push(format!("{endpoint_scheme}://{host}:{port}"));
    }

    if let Some(q) = query_part {
        let qcount = if q.is_empty() {
            0
        } else {
            q.split('&').count()
        };
        if qcount > limits.max_query_params {
            return Err(ParseError::new(
                ParseErrorKind::LimitExceeded,
                format!(
                    "max_query_params exceeded: limit={} actual={}",
                    limits.max_query_params, qcount,
                ),
            ));
        }
    }

    let force_primary = query_part
        .map(|q| {
            q.split('&').any(|kv| {
                let mut parts = kv.splitn(2, '=');
                let k = parts.next().unwrap_or("");
                let v = parts.next().unwrap_or("");
                k.eq_ignore_ascii_case("route") && v.eq_ignore_ascii_case("primary")
            })
        })
        .unwrap_or(false);

    let mut iter = endpoints.into_iter();
    let primary = iter.next().expect("split on ',' yields at least one entry");
    let replicas: Vec<String> = iter.collect();

    Ok(Some(ConnectionTarget::GrpcCluster {
        primary,
        replicas,
        force_primary,
    }))
}