reddb_wire/
conn_string.rs

1//! Connection-string parser shared across `reddb`, `reddb-client`,
2//! `red_client`, and every language driver.
3//!
4//! Pure function over a string; no I/O, no allocation beyond what the
5//! returned [`ConnectionTarget`] needs. The grammar is defined by
6//! `docs/clients/connection-strings.md`; this module is the canonical
7//! parser and is the single source of truth consumed by the rest of
8//! the workspace.
9//!
10//! The parser ports the logic that previously lived in
11//! `drivers/rust/src/connect.rs` (which keeps a thin re-export layer
12//! for backwards compatibility while drivers migrate over). Cluster
13//! URIs (`grpc://primary,replica:port`), default ports per scheme,
14//! and the `?route=primary` override behave identically to the
15//! original.
16
17use std::path::PathBuf;
18
19use url::Url;
20
21/// Stable error code for parser failures.
22///
23/// Mirrors the `ErrorCode` shape used by the language drivers so that
24/// downstream wrappers can map 1:1 without information loss.
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub enum ParseErrorKind {
27    /// The input was empty.
28    Empty,
29    /// `url::Url` rejected the string, or a transport-specific
30    /// invariant (missing host, empty cluster entry, bad port…) was
31    /// violated.
32    InvalidUri,
33    /// The scheme is not in the documented vocabulary.
34    UnsupportedScheme,
35    /// A DoS guardrail in [`ConnStringLimits`] was tripped.
36    /// `message` carries the limit name + the offending value so
37    /// downstream wrappers can surface the structured detail.
38    LimitExceeded,
39}
40
41impl ParseErrorKind {
42    pub fn as_str(self) -> &'static str {
43        match self {
44            ParseErrorKind::Empty => "EMPTY",
45            ParseErrorKind::InvalidUri => "INVALID_URI",
46            ParseErrorKind::UnsupportedScheme => "UNSUPPORTED_SCHEME",
47            ParseErrorKind::LimitExceeded => "LIMIT_EXCEEDED",
48        }
49    }
50}
51
52/// Error returned by [`parse`].
53#[derive(Debug, Clone, PartialEq, Eq)]
54pub struct ParseError {
55    pub kind: ParseErrorKind,
56    pub message: String,
57}
58
59impl ParseError {
60    pub fn new(kind: ParseErrorKind, message: impl Into<String>) -> Self {
61        Self {
62            kind,
63            message: message.into(),
64        }
65    }
66}
67
68impl std::fmt::Display for ParseError {
69    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
70        write!(f, "{}: {}", self.kind.as_str(), self.message)
71    }
72}
73
74impl std::error::Error for ParseError {}
75
76/// Default port per documented scheme. Centralised so other crates
77/// (the connector, server-side dispatch) can stay consistent.
78pub const DEFAULT_PORT_RED: u16 = 5050;
79pub const DEFAULT_PORT_GRPC: u16 = 55055;
80pub const DEFAULT_PORT_GRPCS: u16 = 55555;
81/// Default ports for `ws://` / `red+ws://` and `wss://` / `red+wss://` — align with the
82/// standard WS / WSS browser defaults (80 and 443) so a hosted endpoint
83/// like `*.db.reddb.io` works without an explicit port.
84pub const DEFAULT_PORT_WS: u16 = 80;
85pub const DEFAULT_PORT_WSS: u16 = 443;
86
87/// URI schemes accepted by the connection-string parser.
88///
89/// This enum is the connection-layer source for generated agent knowledge:
90/// [`crate::knowledge`] iterates [`SUPPORTED_SCHEMES`] instead of carrying a
91/// separate hand-maintained scheme list.
92#[derive(Debug, Clone, Copy, PartialEq, Eq)]
93pub enum ConnectionScheme {
94    Memory,
95    File,
96    Red,
97    Reds,
98    RedWs,
99    RedWss,
100    Ws,
101    Wss,
102    Grpc,
103    Grpcs,
104    Http,
105    Https,
106}
107
108/// Stable parser-owned list of supported URI schemes.
109pub const SUPPORTED_SCHEMES: &[ConnectionScheme] = &[
110    ConnectionScheme::Red,
111    ConnectionScheme::Reds,
112    ConnectionScheme::Grpc,
113    ConnectionScheme::Grpcs,
114    ConnectionScheme::Http,
115    ConnectionScheme::Https,
116    ConnectionScheme::Memory,
117    ConnectionScheme::File,
118    ConnectionScheme::RedWs,
119    ConnectionScheme::RedWss,
120    ConnectionScheme::Ws,
121    ConnectionScheme::Wss,
122];
123
124impl ConnectionScheme {
125    pub fn from_uri_scheme(scheme: &str) -> Option<Self> {
126        match scheme {
127            "memory" => Some(Self::Memory),
128            "file" => Some(Self::File),
129            "red" => Some(Self::Red),
130            "reds" => Some(Self::Reds),
131            "red+ws" => Some(Self::RedWs),
132            "red+wss" => Some(Self::RedWss),
133            "ws" => Some(Self::Ws),
134            "wss" => Some(Self::Wss),
135            "grpc" => Some(Self::Grpc),
136            "grpcs" => Some(Self::Grpcs),
137            "http" => Some(Self::Http),
138            "https" => Some(Self::Https),
139            _ => None,
140        }
141    }
142
143    pub fn uri_prefix(self) -> &'static str {
144        match self {
145            Self::Memory => "memory://",
146            Self::File => "file://",
147            Self::Red => "red://",
148            Self::Reds => "reds://",
149            Self::RedWs => "red+ws://",
150            Self::RedWss => "red+wss://",
151            Self::Ws => "ws://",
152            Self::Wss => "wss://",
153            Self::Grpc => "grpc://",
154            Self::Grpcs => "grpcs://",
155            Self::Http => "http://",
156            Self::Https => "https://",
157        }
158    }
159
160    pub fn transport(self) -> &'static str {
161        match self {
162            Self::Memory => "embedded in-memory engine",
163            Self::File => "embedded file-backed engine",
164            Self::Red | Self::Reds => "RedWire TCP",
165            Self::RedWs | Self::RedWss | Self::Ws | Self::Wss => "RedWire WebSocket",
166            Self::Grpc | Self::Grpcs => "gRPC",
167            Self::Http | Self::Https => "HTTP REST",
168        }
169    }
170
171    pub fn mode(self) -> &'static str {
172        match self {
173            Self::Memory | Self::File => "embedded",
174            Self::Red
175            | Self::Reds
176            | Self::RedWs
177            | Self::RedWss
178            | Self::Ws
179            | Self::Wss
180            | Self::Grpc
181            | Self::Grpcs
182            | Self::Http
183            | Self::Https => "remote",
184        }
185    }
186
187    pub fn example(self) -> &'static str {
188        match self {
189            Self::Memory => "memory://",
190            Self::File => "file:///var/lib/reddb/app.db",
191            Self::Red => "red://db.example.com:5050",
192            Self::Reds => "reds://db.example.com:5050",
193            Self::RedWs => "red+ws://db.example.com",
194            Self::RedWss => "red+wss://db.example.com",
195            Self::Ws => "ws://db.example.com",
196            Self::Wss => "wss://db.example.com",
197            Self::Grpc => "grpc://db.example.com:55055",
198            Self::Grpcs => "grpcs://db.example.com:55555",
199            Self::Http => "http://db.example.com:80",
200            Self::Https => "https://db.example.com:443",
201        }
202    }
203
204    pub fn notes(self) -> &'static str {
205        match self {
206            Self::Memory => "Zero-config ephemeral engine, commonly used by local MCP hosts.",
207            Self::File => "Embedded durable engine rooted at the URI path.",
208            Self::Red => "Principal RedWire transport without TLS.",
209            Self::Reds => "Principal RedWire transport with TLS.",
210            Self::RedWs => "Browser-native RedWire over WebSocket without TLS.",
211            Self::RedWss => "Browser-native RedWire over WebSocket with TLS.",
212            Self::Ws => "Browser-friendly alias for RedWire over WebSocket without TLS.",
213            Self::Wss => "Browser-friendly alias for RedWire over WebSocket with TLS.",
214            Self::Grpc => "Compatibility transport for existing gRPC clients.",
215            Self::Grpcs => "TLS variant of the gRPC compatibility transport.",
216            Self::Http => "REST/admin transport without TLS.",
217            Self::Https => "REST/admin transport with TLS.",
218        }
219    }
220}
221
222/// DoS guardrails applied by [`parse`] before any URI work happens.
223///
224/// The connection-string parser is the only entry point an attacker
225/// can reach BEFORE auth, so every limit here is enforced eagerly
226/// and surfaces as a structured [`ParseErrorKind::LimitExceeded`]
227/// error rather than a panic, hang, or unbounded allocation.
228#[derive(Debug, Clone, Copy, PartialEq, Eq)]
229pub struct ConnStringLimits {
230    /// Maximum length of the input URI in bytes. Default `8 KiB`.
231    pub max_uri_bytes: usize,
232    /// Maximum number of `key=value` query parameters. Default `32`.
233    pub max_query_params: usize,
234    /// Maximum number of comma-separated cluster hosts allowed in a
235    /// `grpc://`/`grpcs://` cluster URI. Default `64`.
236    pub max_cluster_hosts: usize,
237}
238
239impl Default for ConnStringLimits {
240    fn default() -> Self {
241        Self {
242            max_uri_bytes: 8 * 1024,
243            max_query_params: 32,
244            max_cluster_hosts: 64,
245        }
246    }
247}
248
249/// Normalised target produced by [`parse`].
250///
251/// Variants intentionally mirror the public Rust client target shape
252/// so callers can keep a thin compatibility layer without duplicating
253/// parser behavior.
254#[derive(Debug, Clone, PartialEq, Eq)]
255pub enum ConnectionTarget {
256    /// `memory://` — ephemeral, in-memory backend.
257    Memory,
258    /// `file:///abs/path` — embedded engine on disk.
259    File { path: PathBuf },
260    /// Single remote endpoint over `grpc://` or `grpcs://`. Stored
261    /// as a normalised `http://host:port` string because tonic's
262    /// `Endpoint` consumes that form.
263    Grpc { endpoint: String },
264    /// Multi-host gRPC URI: primary + read replicas. Writes hit the
265    /// primary; reads round-robin across replicas unless
266    /// `force_primary` is set.
267    GrpcCluster {
268        primary: String,
269        replicas: Vec<String>,
270        force_primary: bool,
271    },
272    /// `http://host:port` / `https://host:port` — REST endpoint.
273    Http { base_url: String },
274    /// `red://host:port` (plain TCP) or `reds://host:port` (TLS).
275    /// RedWire binary frame protocol per ADR 0001. The connector
276    /// speaks framed binary directly; it does NOT route through
277    /// tonic.
278    RedWire { host: String, port: u16, tls: bool },
279    /// `red+ws://host:port` / `ws://host:port` (plain WS) or
280    /// `red+wss://host:port` / `wss://host:port` (WSS).
281    /// Browser-native WebSocket transport (ADR 0047 direct-when-reachable).
282    /// The UI connects directly — no local RedWire-over-TCP bridge needed.
283    WsNative { host: String, port: u16, tls: bool },
284}
285
286/// Authentication material derived from a connection string or adjacent CLI
287/// fallback. `Debug` deliberately redacts every caller-supplied credential.
288#[derive(Clone, PartialEq, Eq)]
289pub enum ConnectionAuth {
290    Anonymous,
291    Bearer(String),
292    Basic { user: String, pass: String },
293    ApiKey(String),
294}
295
296impl std::fmt::Debug for ConnectionAuth {
297    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
298        match self {
299            Self::Anonymous => f.write_str("Anonymous"),
300            Self::Bearer(_) => f.debug_tuple("Bearer").field(&"<redacted>").finish(),
301            Self::Basic { .. } => f
302                .debug_struct("Basic")
303                .field("user", &"<redacted>")
304                .field("pass", &"<redacted>")
305                .finish(),
306            Self::ApiKey(_) => f.debug_tuple("ApiKey").field(&"<redacted>").finish(),
307        }
308    }
309}
310
311impl ConnectionAuth {
312    pub fn bearer(token: impl Into<String>) -> Self {
313        Self::Bearer(token.into())
314    }
315
316    pub fn is_bearer(&self) -> bool {
317        matches!(self, Self::Bearer(_))
318    }
319}
320
321/// Connection target plus URL-derived auth metadata.
322#[derive(Debug, Clone, PartialEq, Eq)]
323pub struct ConnectionSpec {
324    pub target: ConnectionTarget,
325    pub auth: ConnectionAuth,
326    pub redacted_uri: String,
327}
328
329/// Parse a connection URI into a [`ConnectionTarget`] under the
330/// default DoS limits.
331///
332/// Pure function, no side effects. Behaviour matches
333/// `drivers/rust/src/connect.rs::parse` 1:1 with two additions:
334///   - Mixed-case schemes (e.g. `Red://`, `REDS://`) are normalised
335///     to lowercase before dispatch.
336///   - Inputs exceeding [`ConnStringLimits`] return a structured
337///     [`ParseErrorKind::LimitExceeded`] error instead of being
338///     processed.
339pub fn parse(uri: &str) -> Result<ConnectionTarget, ParseError> {
340    parse_with_limits(uri, ConnStringLimits::default())
341}
342
343/// Parse a connection URI and derive auth from RFC 3986 userinfo.
344///
345/// The existing [`parse`] function remains target-only for compatibility.
346/// New client-mode entry points should use this richer shape so credentials
347/// are identified and redacted before transport connectors are built.
348pub fn parse_with_auth(uri: &str) -> Result<ConnectionSpec, ParseError> {
349    let normalised = normalise_scheme(uri);
350    let redacted_uri = redact_uri_userinfo(&normalised);
351    let target =
352        parse(uri).map_err(|err| redact_parse_error(err, uri, &normalised, &redacted_uri))?;
353    let auth = auth_from_uri_userinfo(&normalised)
354        .map_err(|err| redact_parse_error(err, uri, &normalised, &redacted_uri))?;
355    Ok(ConnectionSpec {
356        target,
357        auth,
358        redacted_uri,
359    })
360}
361
362fn redact_parse_error(
363    mut err: ParseError,
364    raw_uri: &str,
365    normalised_uri: &str,
366    redacted_uri: &str,
367) -> ParseError {
368    err.message = err
369        .message
370        .replace(raw_uri, redacted_uri)
371        .replace(normalised_uri, redacted_uri);
372    err
373}
374
375/// Return true for documented embedded aliases that must not resolve to
376/// a remote transport target.
377///
378/// This is intentionally separate from [`parse`]: legacy clients may need
379/// to reject embedded targets before mapping `red://host` onto a remote
380/// compatibility transport.
381pub fn is_embedded_connection_uri(uri: &str) -> bool {
382    let trimmed = uri.trim();
383    matches!(
384        trimmed,
385        "red://" | "red:" | "red:///" | "red://:memory" | "red://:memory:"
386    ) || trimmed.starts_with("red:///")
387}
388
389/// Same as [`parse`] but with caller-supplied DoS guardrails.
390/// Useful for tests that need tighter limits or for callers (a
391/// future admin tool, an offline validator) that need to relax the
392/// defaults.
393pub fn parse_with_limits(
394    uri: &str,
395    limits: ConnStringLimits,
396) -> Result<ConnectionTarget, ParseError> {
397    if uri.is_empty() {
398        return Err(ParseError::new(
399            ParseErrorKind::Empty,
400            "empty connection string",
401        ));
402    }
403
404    if uri.len() > limits.max_uri_bytes {
405        return Err(ParseError::new(
406            ParseErrorKind::LimitExceeded,
407            format!(
408                "max_uri_bytes exceeded: limit={} actual={}",
409                limits.max_uri_bytes,
410                uri.len(),
411            ),
412        ));
413    }
414
415    // Lowercase the scheme so `Red://Host`, `REDS://Host`, etc.
416    // dispatch identically to the canonical lowercase forms. The
417    // host and path retain original casing — host is downcased by
418    // `url::Url` for IDN per spec, path stays verbatim.
419    let normalised = normalise_scheme(uri);
420    let uri = normalised.as_str();
421
422    if uri == "memory://" || uri == "memory:" {
423        return Ok(ConnectionTarget::Memory);
424    }
425
426    if let Some(rest) = uri.strip_prefix("file://") {
427        if rest.is_empty() {
428            return Err(ParseError::new(
429                ParseErrorKind::InvalidUri,
430                "file:// URI is missing a path",
431            ));
432        }
433        return Ok(ConnectionTarget::File {
434            path: PathBuf::from(rest),
435        });
436    }
437
438    if let Some(cluster) = try_parse_grpc_cluster(uri, &limits)? {
439        return Ok(cluster);
440    }
441
442    // Before `Url::parse`, because a multi-host authority like
443    // `red://a:1,b:2` fails URL parsing with "invalid port number" — a
444    // message that hides the real problem. `red://` no longer folds to
445    // gRPC (#2159), and RedWire has no multi-host form, so a cluster URI
446    // has to name the transport that supports one.
447    if let Some(rest) = uri
448        .strip_prefix("red://")
449        .or_else(|| uri.strip_prefix("reds://"))
450    {
451        let authority = rest.split(['/', '?', '#']).next().unwrap_or(rest);
452        if authority.contains(',') {
453            return Err(ParseError::new(
454                ParseErrorKind::InvalidUri,
455                "RedWire does not support comma-separated cluster hosts; \
456                 use grpc:// or grpcs:// for a cluster URI",
457            ));
458        }
459    }
460
461    let parsed = Url::parse(uri)
462        .map_err(|e| ParseError::new(ParseErrorKind::InvalidUri, format!("{e}: {uri}")))?;
463
464    enforce_query_param_limit(&parsed, &limits)?;
465
466    match ConnectionScheme::from_uri_scheme(parsed.scheme()) {
467        Some(ConnectionScheme::Red | ConnectionScheme::Reds) => {
468            let host = parsed.host_str().ok_or_else(|| {
469                ParseError::new(ParseErrorKind::InvalidUri, "red:// URI is missing a host")
470            })?;
471            if host.contains(',') {
472                return Err(ParseError::new(
473                    ParseErrorKind::InvalidUri,
474                    "RedWire does not support comma-separated cluster hosts",
475                ));
476            }
477            let port = parsed.port().unwrap_or(DEFAULT_PORT_RED);
478            Ok(ConnectionTarget::RedWire {
479                host: host.to_string(),
480                port,
481                tls: parsed.scheme() == "reds",
482            })
483        }
484        Some(
485            ConnectionScheme::RedWs
486            | ConnectionScheme::RedWss
487            | ConnectionScheme::Ws
488            | ConnectionScheme::Wss,
489        ) => {
490            let host = parsed.host_str().ok_or_else(|| {
491                ParseError::new(
492                    ParseErrorKind::InvalidUri,
493                    "RedWire WebSocket URI is missing a host",
494                )
495            })?;
496            let tls = parsed.scheme() == "red+wss" || parsed.scheme() == "wss";
497            let port = parsed.port().unwrap_or(if tls {
498                DEFAULT_PORT_WSS
499            } else {
500                DEFAULT_PORT_WS
501            });
502            Ok(ConnectionTarget::WsNative {
503                host: host.to_string(),
504                port,
505                tls,
506            })
507        }
508        Some(ConnectionScheme::Grpc | ConnectionScheme::Grpcs) => {
509            let host = parsed.host_str().ok_or_else(|| {
510                ParseError::new(ParseErrorKind::InvalidUri, "grpc:// URI is missing a host")
511            })?;
512            let port = parsed.port().unwrap_or_else(|| {
513                if parsed.scheme() == "grpcs" {
514                    DEFAULT_PORT_GRPCS
515                } else {
516                    DEFAULT_PORT_GRPC
517                }
518            });
519            Ok(ConnectionTarget::Grpc {
520                endpoint: format!("http://{host}:{port}"),
521            })
522        }
523        Some(ConnectionScheme::Http | ConnectionScheme::Https) => {
524            let host = parsed.host_str().ok_or_else(|| {
525                ParseError::new(
526                    ParseErrorKind::InvalidUri,
527                    "http(s):// URI is missing a host",
528                )
529            })?;
530            let scheme = parsed.scheme();
531            let port = parsed
532                .port()
533                .unwrap_or(if scheme == "https" { 443 } else { 80 });
534            Ok(ConnectionTarget::Http {
535                base_url: format!("{scheme}://{host}:{port}"),
536            })
537        }
538        Some(ConnectionScheme::Memory | ConnectionScheme::File) | None => Err(ParseError::new(
539            ParseErrorKind::UnsupportedScheme,
540            format!("unsupported scheme: {}", parsed.scheme()),
541        )),
542    }
543}
544
545/// Lowercase only the scheme portion (everything before the first
546/// `:`), leaving host/path/query untouched. Returns the original
547/// string when no scheme separator is present so the downstream
548/// `Url::parse` path produces the canonical "missing scheme" error
549/// instead of being masked here.
550fn normalise_scheme(uri: &str) -> String {
551    match uri.find(':') {
552        Some(i) => {
553            let scheme = &uri[..i];
554            // Only ASCII alphanumerics + `+ . -` are valid scheme
555            // bytes per RFC 3986. If the prefix violates that we
556            // leave it alone so `Url::parse` can produce the
557            // structured error.
558            if scheme.is_empty()
559                || !scheme
560                    .bytes()
561                    .all(|b| b.is_ascii_alphanumeric() || b == b'+' || b == b'.' || b == b'-')
562            {
563                return uri.to_string();
564            }
565            let mut out = String::with_capacity(uri.len());
566            out.push_str(&scheme.to_ascii_lowercase());
567            out.push_str(&uri[i..]);
568            out
569        }
570        None => uri.to_string(),
571    }
572}
573
574fn auth_from_uri_userinfo(uri: &str) -> Result<ConnectionAuth, ParseError> {
575    if uri == "memory://" || uri == "memory:" || uri.starts_with("file://") {
576        return Ok(ConnectionAuth::Anonymous);
577    }
578    let parsed = Url::parse(uri)
579        .map_err(|e| ParseError::new(ParseErrorKind::InvalidUri, format!("{e}: {uri}")))?;
580    let username = parsed.username();
581    if username.is_empty() {
582        return Ok(ConnectionAuth::Anonymous);
583    }
584    match parsed.password() {
585        Some(pass) => Ok(ConnectionAuth::Basic {
586            user: username.to_string(),
587            pass: pass.to_string(),
588        }),
589        None => Ok(ConnectionAuth::ApiKey(username.to_string())),
590    }
591}
592
593fn redact_uri_userinfo(uri: &str) -> String {
594    let Some(scheme_end) = uri.find("://") else {
595        return uri.to_string();
596    };
597    let authority_start = scheme_end + 3;
598    let authority_end = uri[authority_start..]
599        .find(['/', '?', '#'])
600        .map(|i| authority_start + i)
601        .unwrap_or(uri.len());
602    let authority = &uri[authority_start..authority_end];
603    let Some(at) = authority.rfind('@') else {
604        return uri.to_string();
605    };
606    let userinfo = &authority[..at];
607    let replacement = if userinfo.contains(':') {
608        "<redacted>:<redacted>"
609    } else {
610        "<redacted>"
611    };
612    format!(
613        "{}{}{}",
614        &uri[..authority_start],
615        replacement,
616        &uri[authority_start + at..]
617    )
618}
619
620fn enforce_query_param_limit(url: &Url, limits: &ConnStringLimits) -> Result<(), ParseError> {
621    let Some(q) = url.query() else {
622        return Ok(());
623    };
624    if q.is_empty() {
625        return Ok(());
626    }
627    let count = q.split('&').count();
628    if count > limits.max_query_params {
629        return Err(ParseError::new(
630            ParseErrorKind::LimitExceeded,
631            format!(
632                "max_query_params exceeded: limit={} actual={}",
633                limits.max_query_params, count,
634            ),
635        ));
636    }
637    Ok(())
638}
639
640/// Try to parse a multi-host gRPC URI. `Ok(None)` means "this is a
641/// single-host URI — fall through to the standard parser".
642fn try_parse_grpc_cluster(
643    uri: &str,
644    limits: &ConnStringLimits,
645) -> Result<Option<ConnectionTarget>, ParseError> {
646    let (rest, default_port) = if let Some(r) = uri.strip_prefix("grpc://") {
647        (r, DEFAULT_PORT_GRPC)
648    } else if let Some(r) = uri.strip_prefix("grpcs://") {
649        (r, DEFAULT_PORT_GRPCS)
650    } else {
651        return Ok(None);
652    };
653
654    let (host_part, query_part) = match rest.find('?') {
655        Some(i) => (&rest[..i], Some(&rest[i + 1..])),
656        None => (rest, None),
657    };
658
659    if !host_part.contains(',') {
660        return Ok(None);
661    }
662
663    let raw_count = host_part.split(',').count();
664    if raw_count > limits.max_cluster_hosts {
665        return Err(ParseError::new(
666            ParseErrorKind::LimitExceeded,
667            format!(
668                "max_cluster_hosts exceeded: limit={} actual={}",
669                limits.max_cluster_hosts, raw_count,
670            ),
671        ));
672    }
673
674    let mut endpoints: Vec<String> = Vec::with_capacity(raw_count);
675    for raw in host_part.split(',') {
676        let raw = raw.trim();
677        if raw.is_empty() {
678            return Err(ParseError::new(
679                ParseErrorKind::InvalidUri,
680                "grpc cluster URI has an empty host entry",
681            ));
682        }
683        // Bracketed IPv6 literal: `[::1]:5050` or `[::1]`.
684        let (host, port) = if let Some(after_bracket) = raw.strip_prefix('[') {
685            let end = after_bracket.find(']').ok_or_else(|| {
686                ParseError::new(
687                    ParseErrorKind::InvalidUri,
688                    format!("unterminated IPv6 bracket in cluster URI: {raw}"),
689                )
690            })?;
691            let host = &after_bracket[..end];
692            let tail = &after_bracket[end + 1..];
693            let port = if tail.is_empty() {
694                default_port
695            } else if let Some(p) = tail.strip_prefix(':') {
696                p.parse::<u16>().map_err(|_| {
697                    ParseError::new(
698                        ParseErrorKind::InvalidUri,
699                        format!("invalid port in cluster URI: {raw}"),
700                    )
701                })?
702            } else {
703                return Err(ParseError::new(
704                    ParseErrorKind::InvalidUri,
705                    format!("trailing junk after IPv6 bracket in cluster URI: {raw}"),
706                ));
707            };
708            (format!("[{host}]"), port)
709        } else {
710            match raw.rsplit_once(':') {
711                Some((h, p)) => {
712                    let port: u16 = p.parse().map_err(|_| {
713                        ParseError::new(
714                            ParseErrorKind::InvalidUri,
715                            format!("invalid port in cluster URI: {raw}"),
716                        )
717                    })?;
718                    (h.to_string(), port)
719                }
720                None => (raw.to_string(), default_port),
721            }
722        };
723        if host.is_empty() || host == "[]" {
724            return Err(ParseError::new(
725                ParseErrorKind::InvalidUri,
726                "grpc cluster URI has an empty host entry",
727            ));
728        }
729        endpoints.push(format!("http://{host}:{port}"));
730    }
731
732    if let Some(q) = query_part {
733        let qcount = if q.is_empty() {
734            0
735        } else {
736            q.split('&').count()
737        };
738        if qcount > limits.max_query_params {
739            return Err(ParseError::new(
740                ParseErrorKind::LimitExceeded,
741                format!(
742                    "max_query_params exceeded: limit={} actual={}",
743                    limits.max_query_params, qcount,
744                ),
745            ));
746        }
747    }
748
749    let force_primary = query_part
750        .map(|q| {
751            q.split('&').any(|kv| {
752                let mut parts = kv.splitn(2, '=');
753                let k = parts.next().unwrap_or("");
754                let v = parts.next().unwrap_or("");
755                k.eq_ignore_ascii_case("route") && v.eq_ignore_ascii_case("primary")
756            })
757        })
758        .unwrap_or(false);
759
760    let mut iter = endpoints.into_iter();
761    let primary = iter.next().expect("split on ',' yields at least one entry");
762    let replicas: Vec<String> = iter.collect();
763
764    Ok(Some(ConnectionTarget::GrpcCluster {
765        primary,
766        replicas,
767        force_primary,
768    }))
769}
reddb_wire/conn_string.rs

reddb_wire/
conn_string.rs
