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    /// `red://`/`reds://`/`grpc://` 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    let parsed = Url::parse(uri)
443        .map_err(|e| ParseError::new(ParseErrorKind::InvalidUri, format!("{e}: {uri}")))?;
444
445    enforce_query_param_limit(&parsed, &limits)?;
446
447    match ConnectionScheme::from_uri_scheme(parsed.scheme()) {
448        Some(ConnectionScheme::Red | ConnectionScheme::Reds) => {
449            let host = parsed.host_str().ok_or_else(|| {
450                ParseError::new(ParseErrorKind::InvalidUri, "red:// URI is missing a host")
451            })?;
452            let port = parsed.port().unwrap_or(DEFAULT_PORT_RED);
453            Ok(ConnectionTarget::RedWire {
454                host: host.to_string(),
455                port,
456                tls: parsed.scheme() == "reds",
457            })
458        }
459        Some(
460            ConnectionScheme::RedWs
461            | ConnectionScheme::RedWss
462            | ConnectionScheme::Ws
463            | ConnectionScheme::Wss,
464        ) => {
465            let host = parsed.host_str().ok_or_else(|| {
466                ParseError::new(
467                    ParseErrorKind::InvalidUri,
468                    "RedWire WebSocket URI is missing a host",
469                )
470            })?;
471            let tls = parsed.scheme() == "red+wss" || parsed.scheme() == "wss";
472            let port = parsed.port().unwrap_or(if tls {
473                DEFAULT_PORT_WSS
474            } else {
475                DEFAULT_PORT_WS
476            });
477            Ok(ConnectionTarget::WsNative {
478                host: host.to_string(),
479                port,
480                tls,
481            })
482        }
483        Some(ConnectionScheme::Grpc | ConnectionScheme::Grpcs) => {
484            let host = parsed.host_str().ok_or_else(|| {
485                ParseError::new(ParseErrorKind::InvalidUri, "grpc:// URI is missing a host")
486            })?;
487            let port = parsed.port().unwrap_or_else(|| {
488                if parsed.scheme() == "grpcs" {
489                    DEFAULT_PORT_GRPCS
490                } else {
491                    DEFAULT_PORT_GRPC
492                }
493            });
494            Ok(ConnectionTarget::Grpc {
495                endpoint: format!("http://{host}:{port}"),
496            })
497        }
498        Some(ConnectionScheme::Http | ConnectionScheme::Https) => {
499            let host = parsed.host_str().ok_or_else(|| {
500                ParseError::new(
501                    ParseErrorKind::InvalidUri,
502                    "http(s):// URI is missing a host",
503                )
504            })?;
505            let scheme = parsed.scheme();
506            let port = parsed
507                .port()
508                .unwrap_or(if scheme == "https" { 443 } else { 80 });
509            Ok(ConnectionTarget::Http {
510                base_url: format!("{scheme}://{host}:{port}"),
511            })
512        }
513        Some(ConnectionScheme::Memory | ConnectionScheme::File) | None => Err(ParseError::new(
514            ParseErrorKind::UnsupportedScheme,
515            format!("unsupported scheme: {}", parsed.scheme()),
516        )),
517    }
518}
519
520/// Lowercase only the scheme portion (everything before the first
521/// `:`), leaving host/path/query untouched. Returns the original
522/// string when no scheme separator is present so the downstream
523/// `Url::parse` path produces the canonical "missing scheme" error
524/// instead of being masked here.
525fn normalise_scheme(uri: &str) -> String {
526    match uri.find(':') {
527        Some(i) => {
528            let scheme = &uri[..i];
529            // Only ASCII alphanumerics + `+ . -` are valid scheme
530            // bytes per RFC 3986. If the prefix violates that we
531            // leave it alone so `Url::parse` can produce the
532            // structured error.
533            if scheme.is_empty()
534                || !scheme
535                    .bytes()
536                    .all(|b| b.is_ascii_alphanumeric() || b == b'+' || b == b'.' || b == b'-')
537            {
538                return uri.to_string();
539            }
540            let mut out = String::with_capacity(uri.len());
541            out.push_str(&scheme.to_ascii_lowercase());
542            out.push_str(&uri[i..]);
543            out
544        }
545        None => uri.to_string(),
546    }
547}
548
549fn auth_from_uri_userinfo(uri: &str) -> Result<ConnectionAuth, ParseError> {
550    if uri == "memory://" || uri == "memory:" || uri.starts_with("file://") {
551        return Ok(ConnectionAuth::Anonymous);
552    }
553    let parsed = Url::parse(uri)
554        .map_err(|e| ParseError::new(ParseErrorKind::InvalidUri, format!("{e}: {uri}")))?;
555    let username = parsed.username();
556    if username.is_empty() {
557        return Ok(ConnectionAuth::Anonymous);
558    }
559    match parsed.password() {
560        Some(pass) => Ok(ConnectionAuth::Basic {
561            user: username.to_string(),
562            pass: pass.to_string(),
563        }),
564        None => Ok(ConnectionAuth::ApiKey(username.to_string())),
565    }
566}
567
568fn redact_uri_userinfo(uri: &str) -> String {
569    let Some(scheme_end) = uri.find("://") else {
570        return uri.to_string();
571    };
572    let authority_start = scheme_end + 3;
573    let authority_end = uri[authority_start..]
574        .find(['/', '?', '#'])
575        .map(|i| authority_start + i)
576        .unwrap_or(uri.len());
577    let authority = &uri[authority_start..authority_end];
578    let Some(at) = authority.rfind('@') else {
579        return uri.to_string();
580    };
581    let userinfo = &authority[..at];
582    let replacement = if userinfo.contains(':') {
583        "<redacted>:<redacted>"
584    } else {
585        "<redacted>"
586    };
587    format!(
588        "{}{}{}",
589        &uri[..authority_start],
590        replacement,
591        &uri[authority_start + at..]
592    )
593}
594
595fn enforce_query_param_limit(url: &Url, limits: &ConnStringLimits) -> Result<(), ParseError> {
596    let Some(q) = url.query() else {
597        return Ok(());
598    };
599    if q.is_empty() {
600        return Ok(());
601    }
602    let count = q.split('&').count();
603    if count > limits.max_query_params {
604        return Err(ParseError::new(
605            ParseErrorKind::LimitExceeded,
606            format!(
607                "max_query_params exceeded: limit={} actual={}",
608                limits.max_query_params, count,
609            ),
610        ));
611    }
612    Ok(())
613}
614
615/// Try to parse a multi-host gRPC URI. `Ok(None)` means "this is a
616/// single-host URI — fall through to the standard parser".
617fn try_parse_grpc_cluster(
618    uri: &str,
619    limits: &ConnStringLimits,
620) -> Result<Option<ConnectionTarget>, ParseError> {
621    let (rest, default_port) = if let Some(r) = uri.strip_prefix("grpc://") {
622        (r, DEFAULT_PORT_GRPC)
623    } else if let Some(r) = uri.strip_prefix("grpcs://") {
624        (r, DEFAULT_PORT_GRPCS)
625    } else if let Some(r) = uri
626        .strip_prefix("red://")
627        .or_else(|| uri.strip_prefix("reds://"))
628    {
629        (r, DEFAULT_PORT_RED)
630    } else {
631        return Ok(None);
632    };
633
634    let (host_part, query_part) = match rest.find('?') {
635        Some(i) => (&rest[..i], Some(&rest[i + 1..])),
636        None => (rest, None),
637    };
638
639    if !host_part.contains(',') {
640        return Ok(None);
641    }
642
643    let raw_count = host_part.split(',').count();
644    if raw_count > limits.max_cluster_hosts {
645        return Err(ParseError::new(
646            ParseErrorKind::LimitExceeded,
647            format!(
648                "max_cluster_hosts exceeded: limit={} actual={}",
649                limits.max_cluster_hosts, raw_count,
650            ),
651        ));
652    }
653
654    let mut endpoints: Vec<String> = Vec::with_capacity(raw_count);
655    for raw in host_part.split(',') {
656        let raw = raw.trim();
657        if raw.is_empty() {
658            return Err(ParseError::new(
659                ParseErrorKind::InvalidUri,
660                "grpc cluster URI has an empty host entry",
661            ));
662        }
663        // Bracketed IPv6 literal: `[::1]:5050` or `[::1]`.
664        let (host, port) = if let Some(after_bracket) = raw.strip_prefix('[') {
665            let end = after_bracket.find(']').ok_or_else(|| {
666                ParseError::new(
667                    ParseErrorKind::InvalidUri,
668                    format!("unterminated IPv6 bracket in cluster URI: {raw}"),
669                )
670            })?;
671            let host = &after_bracket[..end];
672            let tail = &after_bracket[end + 1..];
673            let port = if tail.is_empty() {
674                default_port
675            } else if let Some(p) = tail.strip_prefix(':') {
676                p.parse::<u16>().map_err(|_| {
677                    ParseError::new(
678                        ParseErrorKind::InvalidUri,
679                        format!("invalid port in cluster URI: {raw}"),
680                    )
681                })?
682            } else {
683                return Err(ParseError::new(
684                    ParseErrorKind::InvalidUri,
685                    format!("trailing junk after IPv6 bracket in cluster URI: {raw}"),
686                ));
687            };
688            (format!("[{host}]"), port)
689        } else {
690            match raw.rsplit_once(':') {
691                Some((h, p)) => {
692                    let port: u16 = p.parse().map_err(|_| {
693                        ParseError::new(
694                            ParseErrorKind::InvalidUri,
695                            format!("invalid port in cluster URI: {raw}"),
696                        )
697                    })?;
698                    (h.to_string(), port)
699                }
700                None => (raw.to_string(), default_port),
701            }
702        };
703        if host.is_empty() || host == "[]" {
704            return Err(ParseError::new(
705                ParseErrorKind::InvalidUri,
706                "grpc cluster URI has an empty host entry",
707            ));
708        }
709        endpoints.push(format!("http://{host}:{port}"));
710    }
711
712    if let Some(q) = query_part {
713        let qcount = if q.is_empty() {
714            0
715        } else {
716            q.split('&').count()
717        };
718        if qcount > limits.max_query_params {
719            return Err(ParseError::new(
720                ParseErrorKind::LimitExceeded,
721                format!(
722                    "max_query_params exceeded: limit={} actual={}",
723                    limits.max_query_params, qcount,
724                ),
725            ));
726        }
727    }
728
729    let force_primary = query_part
730        .map(|q| {
731            q.split('&').any(|kv| {
732                let mut parts = kv.splitn(2, '=');
733                let k = parts.next().unwrap_or("");
734                let v = parts.next().unwrap_or("");
735                k.eq_ignore_ascii_case("route") && v.eq_ignore_ascii_case("primary")
736            })
737        })
738        .unwrap_or(false);
739
740    let mut iter = endpoints.into_iter();
741    let primary = iter.next().expect("split on ',' yields at least one entry");
742    let replicas: Vec<String> = iter.collect();
743
744    Ok(Some(ConnectionTarget::GrpcCluster {
745        primary,
746        replicas,
747        force_primary,
748    }))
749}
reddb_wire/conn_string.rs

reddb_wire/
conn_string.rs
