1use std::path::PathBuf;
18
19use url::Url;
20
21#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub enum ParseErrorKind {
27 Empty,
29 InvalidUri,
33 UnsupportedScheme,
35 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#[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
76pub const DEFAULT_PORT_RED: u16 = 5050;
79pub const DEFAULT_PORT_GRPC: u16 = 55055;
80pub const DEFAULT_PORT_GRPCS: u16 = 55555;
81pub const DEFAULT_PORT_WS: u16 = 80;
85pub const DEFAULT_PORT_WSS: u16 = 443;
86
87#[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
108pub 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#[derive(Debug, Clone, Copy, PartialEq, Eq)]
229pub struct ConnStringLimits {
230 pub max_uri_bytes: usize,
232 pub max_query_params: usize,
234 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#[derive(Debug, Clone, PartialEq, Eq)]
255pub enum ConnectionTarget {
256 Memory,
258 File { path: PathBuf },
260 Grpc { endpoint: String },
263 GrpcCluster {
267 primary: String,
268 replicas: Vec<String>,
269 force_primary: bool,
270 },
271 Http { base_url: String },
273 RedWire { host: String, port: u16, tls: bool },
278 WsNative { host: String, port: u16, tls: bool },
283}
284
285#[derive(Clone, PartialEq, Eq)]
288pub enum ConnectionAuth {
289 Anonymous,
290 Bearer(String),
291 Basic { user: String, pass: String },
292 ApiKey(String),
293}
294
295impl std::fmt::Debug for ConnectionAuth {
296 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
297 match self {
298 Self::Anonymous => f.write_str("Anonymous"),
299 Self::Bearer(_) => f.debug_tuple("Bearer").field(&"<redacted>").finish(),
300 Self::Basic { .. } => f
301 .debug_struct("Basic")
302 .field("user", &"<redacted>")
303 .field("pass", &"<redacted>")
304 .finish(),
305 Self::ApiKey(_) => f.debug_tuple("ApiKey").field(&"<redacted>").finish(),
306 }
307 }
308}
309
310impl ConnectionAuth {
311 pub fn bearer(token: impl Into<String>) -> Self {
312 Self::Bearer(token.into())
313 }
314
315 pub fn is_bearer(&self) -> bool {
316 matches!(self, Self::Bearer(_))
317 }
318}
319
320#[derive(Debug, Clone, PartialEq, Eq)]
322pub struct ConnectionSpec {
323 pub target: ConnectionTarget,
324 pub auth: ConnectionAuth,
325 pub redacted_uri: String,
326}
327
328pub fn parse(uri: &str) -> Result<ConnectionTarget, ParseError> {
339 parse_with_limits(uri, ConnStringLimits::default())
340}
341
342pub fn parse_with_auth(uri: &str) -> Result<ConnectionSpec, ParseError> {
348 let normalised = normalise_scheme(uri);
349 let redacted_uri = redact_connection_uri(&normalised);
350 let target =
351 parse(uri).map_err(|err| redact_parse_error(err, uri, &normalised, &redacted_uri))?;
352 let auth = auth_from_uri_userinfo(&normalised)
353 .map_err(|err| redact_parse_error(err, uri, &normalised, &redacted_uri))?;
354 Ok(ConnectionSpec {
355 target,
356 auth,
357 redacted_uri,
358 })
359}
360
361fn redact_parse_error(
362 mut err: ParseError,
363 raw_uri: &str,
364 normalised_uri: &str,
365 redacted_uri: &str,
366) -> ParseError {
367 err.message = err
368 .message
369 .replace(raw_uri, redacted_uri)
370 .replace(normalised_uri, redacted_uri);
371 err
372}
373
374pub fn is_embedded_connection_uri(uri: &str) -> bool {
381 let trimmed = uri.trim();
382 matches!(
383 trimmed,
384 "red://" | "red:" | "red:///" | "red://:memory" | "red://:memory:"
385 ) || trimmed.starts_with("red:///")
386}
387
388pub fn parse_with_limits(
393 uri: &str,
394 limits: ConnStringLimits,
395) -> Result<ConnectionTarget, ParseError> {
396 if uri.is_empty() {
397 return Err(ParseError::new(
398 ParseErrorKind::Empty,
399 "empty connection string",
400 ));
401 }
402
403 if uri.len() > limits.max_uri_bytes {
404 return Err(ParseError::new(
405 ParseErrorKind::LimitExceeded,
406 format!(
407 "max_uri_bytes exceeded: limit={} actual={}",
408 limits.max_uri_bytes,
409 uri.len(),
410 ),
411 ));
412 }
413
414 let normalised = normalise_scheme(uri);
419 let uri = normalised.as_str();
420
421 if uri == "memory://" || uri == "memory:" {
422 return Ok(ConnectionTarget::Memory);
423 }
424
425 if let Some(rest) = uri.strip_prefix("file://") {
426 if rest.is_empty() {
427 return Err(ParseError::new(
428 ParseErrorKind::InvalidUri,
429 "file:// URI is missing a path",
430 ));
431 }
432 return Ok(ConnectionTarget::File {
433 path: file_uri_path(rest),
434 });
435 }
436
437 if let Some(cluster) = try_parse_grpc_cluster(uri, &limits)? {
438 return Ok(cluster);
439 }
440
441 if let Some(rest) = uri
447 .strip_prefix("red://")
448 .or_else(|| uri.strip_prefix("reds://"))
449 {
450 let authority = rest.split(['/', '?', '#']).next().unwrap_or(rest);
451 if authority.contains(',') {
452 return Err(ParseError::new(
453 ParseErrorKind::InvalidUri,
454 "RedWire does not support comma-separated cluster hosts; \
455 use grpc:// or grpcs:// for a cluster URI",
456 ));
457 }
458 }
459
460 let parsed = Url::parse(uri)
461 .map_err(|e| ParseError::new(ParseErrorKind::InvalidUri, format!("{e}: {uri}")))?;
462
463 enforce_query_param_limit(&parsed, &limits)?;
464
465 match ConnectionScheme::from_uri_scheme(parsed.scheme()) {
466 Some(ConnectionScheme::Red | ConnectionScheme::Reds) => {
467 let host = parsed.host_str().ok_or_else(|| {
468 ParseError::new(ParseErrorKind::InvalidUri, "red:// URI is missing a host")
469 })?;
470 if host.contains(',') {
471 return Err(ParseError::new(
472 ParseErrorKind::InvalidUri,
473 "RedWire does not support comma-separated cluster hosts",
474 ));
475 }
476 let port = parsed.port().unwrap_or(DEFAULT_PORT_RED);
477 Ok(ConnectionTarget::RedWire {
478 host: host.to_string(),
479 port,
480 tls: parsed.scheme() == "reds",
481 })
482 }
483 Some(
484 ConnectionScheme::RedWs
485 | ConnectionScheme::RedWss
486 | ConnectionScheme::Ws
487 | ConnectionScheme::Wss,
488 ) => {
489 let host = parsed.host_str().ok_or_else(|| {
490 ParseError::new(
491 ParseErrorKind::InvalidUri,
492 "RedWire WebSocket URI is missing a host",
493 )
494 })?;
495 let tls = parsed.scheme() == "red+wss" || parsed.scheme() == "wss";
496 let port = parsed.port().unwrap_or(if tls {
497 DEFAULT_PORT_WSS
498 } else {
499 DEFAULT_PORT_WS
500 });
501 Ok(ConnectionTarget::WsNative {
502 host: host.to_string(),
503 port,
504 tls,
505 })
506 }
507 Some(ConnectionScheme::Grpc | ConnectionScheme::Grpcs) => {
508 let host = parsed.host_str().ok_or_else(|| {
509 ParseError::new(ParseErrorKind::InvalidUri, "grpc:// URI is missing a host")
510 })?;
511 let port = parsed.port().unwrap_or_else(|| {
512 if parsed.scheme() == "grpcs" {
513 DEFAULT_PORT_GRPCS
514 } else {
515 DEFAULT_PORT_GRPC
516 }
517 });
518 let endpoint_scheme = if parsed.scheme() == "grpcs" {
519 "https"
520 } else {
521 "http"
522 };
523 Ok(ConnectionTarget::Grpc {
524 endpoint: format!("{endpoint_scheme}://{host}:{port}"),
525 })
526 }
527 Some(ConnectionScheme::Http | ConnectionScheme::Https) => {
528 let host = parsed.host_str().ok_or_else(|| {
529 ParseError::new(
530 ParseErrorKind::InvalidUri,
531 "http(s):// URI is missing a host",
532 )
533 })?;
534 let scheme = parsed.scheme();
535 let port = parsed
536 .port()
537 .unwrap_or(if scheme == "https" { 443 } else { 80 });
538 Ok(ConnectionTarget::Http {
539 base_url: format!("{scheme}://{host}:{port}"),
540 })
541 }
542 Some(ConnectionScheme::Memory | ConnectionScheme::File) | None => Err(ParseError::new(
543 ParseErrorKind::UnsupportedScheme,
544 format!("unsupported scheme: {}", parsed.scheme()),
545 )),
546 }
547}
548
549fn file_uri_path(path: &str) -> PathBuf {
550 #[cfg(windows)]
551 {
552 let bytes = path.as_bytes();
553 if bytes.len() >= 4
557 && bytes[0] == b'/'
558 && bytes[1].is_ascii_alphabetic()
559 && bytes[2] == b':'
560 && matches!(bytes[3], b'/' | b'\\')
561 {
562 return PathBuf::from(&path[1..]);
563 }
564 }
565 PathBuf::from(path)
566}
567
568fn normalise_scheme(uri: &str) -> String {
574 match uri.find(':') {
575 Some(i) => {
576 let scheme = &uri[..i];
577 if scheme.is_empty()
582 || !scheme
583 .bytes()
584 .all(|b| b.is_ascii_alphanumeric() || b == b'+' || b == b'.' || b == b'-')
585 {
586 return uri.to_string();
587 }
588 let mut out = String::with_capacity(uri.len());
589 out.push_str(&scheme.to_ascii_lowercase());
590 out.push_str(&uri[i..]);
591 out
592 }
593 None => uri.to_string(),
594 }
595}
596
597fn auth_from_uri_userinfo(uri: &str) -> Result<ConnectionAuth, ParseError> {
598 if uri == "memory://" || uri == "memory:" || uri.starts_with("file://") {
599 return Ok(ConnectionAuth::Anonymous);
600 }
601 let parsed = Url::parse(uri)
602 .map_err(|e| ParseError::new(ParseErrorKind::InvalidUri, format!("{e}: {uri}")))?;
603 let username = parsed.username();
604 if username.is_empty() {
605 return Ok(ConnectionAuth::Anonymous);
606 }
607 match parsed.password() {
608 Some(pass) => Ok(ConnectionAuth::Basic {
609 user: username.to_string(),
610 pass: pass.to_string(),
611 }),
612 None => Ok(ConnectionAuth::ApiKey(username.to_string())),
613 }
614}
615
616fn redact_connection_uri(uri: &str) -> String {
617 let uri = redact_uri_userinfo(uri);
618 redact_uri_query_secrets(&uri)
619}
620
621fn redact_uri_userinfo(uri: &str) -> String {
622 let Some(scheme_end) = uri.find("://") else {
623 return uri.to_string();
624 };
625 let authority_start = scheme_end + 3;
626 let authority_end = uri[authority_start..]
627 .find(['/', '?', '#'])
628 .map(|i| authority_start + i)
629 .unwrap_or(uri.len());
630 let authority = &uri[authority_start..authority_end];
631 let Some(at) = authority.rfind('@') else {
632 return uri.to_string();
633 };
634 let userinfo = &authority[..at];
635 let replacement = if userinfo.contains(':') {
636 "<redacted>:<redacted>"
637 } else {
638 "<redacted>"
639 };
640 format!(
641 "{}{}{}",
642 &uri[..authority_start],
643 replacement,
644 &uri[authority_start + at..]
645 )
646}
647
648fn redact_uri_query_secrets(uri: &str) -> String {
649 let Some(query_start) = uri.find('?') else {
650 return uri.to_string();
651 };
652 let query_end = uri[query_start + 1..]
653 .find('#')
654 .map(|offset| query_start + 1 + offset)
655 .unwrap_or(uri.len());
656 let query = &uri[query_start + 1..query_end];
657 let mut redacted = String::with_capacity(uri.len());
658 redacted.push_str(&uri[..query_start + 1]);
659 for (index, parameter) in query.split('&').enumerate() {
660 if index > 0 {
661 redacted.push('&');
662 }
663 let (name, value) = parameter.split_once('=').unwrap_or((parameter, ""));
664 redacted.push_str(name);
665 if parameter.contains('=') {
666 redacted.push('=');
667 if sensitive_query_name(name) {
668 redacted.push_str("<redacted>");
669 } else {
670 redacted.push_str(value);
671 }
672 }
673 }
674 redacted.push_str(&uri[query_end..]);
675 redacted
676}
677
678fn sensitive_query_name(raw: &str) -> bool {
679 let mut decoded = Vec::with_capacity(raw.len());
680 let bytes = raw.as_bytes();
681 let mut index = 0;
682 while index < bytes.len() {
683 match bytes[index] {
684 b'%' if index + 2 < bytes.len() => {
685 let high = (bytes[index + 1] as char).to_digit(16);
686 let low = (bytes[index + 2] as char).to_digit(16);
687 if let (Some(high), Some(low)) = (high, low) {
688 decoded.push(((high << 4) | low) as u8);
689 index += 3;
690 continue;
691 }
692 decoded.push(bytes[index]);
693 }
694 b'+' => decoded.push(b' '),
695 byte => decoded.push(byte),
696 }
697 index += 1;
698 }
699 let normalized: String = String::from_utf8_lossy(&decoded)
700 .chars()
701 .filter(|ch| *ch != '_' && *ch != '-')
702 .flat_map(char::to_lowercase)
703 .collect();
704 matches!(
705 normalized.as_str(),
706 "token" | "password" | "pass" | "passwd" | "secret" | "cert" | "key" | "ca" | "apikey"
707 ) || normalized.ends_with("token")
708 || normalized.ends_with("password")
709 || normalized.ends_with("passwd")
710 || normalized.ends_with("secret")
711 || normalized.ends_with("apikey")
712 || normalized.ends_with("privatekey")
713 || normalized.ends_with("keyfile")
714 || normalized.ends_with("certfile")
715 || normalized.ends_with("cafile")
716 || ((normalized.starts_with("ssl") || normalized.starts_with("tls"))
717 && (normalized.ends_with("key")
718 || normalized.ends_with("cert")
719 || normalized.ends_with("ca")))
720}
721
722fn enforce_query_param_limit(url: &Url, limits: &ConnStringLimits) -> Result<(), ParseError> {
723 let Some(q) = url.query() else {
724 return Ok(());
725 };
726 if q.is_empty() {
727 return Ok(());
728 }
729 let count = q.split('&').count();
730 if count > limits.max_query_params {
731 return Err(ParseError::new(
732 ParseErrorKind::LimitExceeded,
733 format!(
734 "max_query_params exceeded: limit={} actual={}",
735 limits.max_query_params, count,
736 ),
737 ));
738 }
739 Ok(())
740}
741
742fn try_parse_grpc_cluster(
745 uri: &str,
746 limits: &ConnStringLimits,
747) -> Result<Option<ConnectionTarget>, ParseError> {
748 let (rest, default_port, endpoint_scheme) = if let Some(r) = uri.strip_prefix("grpc://") {
749 (r, DEFAULT_PORT_GRPC, "http")
750 } else if let Some(r) = uri.strip_prefix("grpcs://") {
751 (r, DEFAULT_PORT_GRPCS, "https")
752 } else {
753 return Ok(None);
754 };
755
756 let (host_part, query_part) = match rest.find('?') {
757 Some(i) => (&rest[..i], Some(&rest[i + 1..])),
758 None => (rest, None),
759 };
760
761 if !host_part.contains(',') {
762 return Ok(None);
763 }
764
765 let raw_count = host_part.split(',').count();
766 if raw_count > limits.max_cluster_hosts {
767 return Err(ParseError::new(
768 ParseErrorKind::LimitExceeded,
769 format!(
770 "max_cluster_hosts exceeded: limit={} actual={}",
771 limits.max_cluster_hosts, raw_count,
772 ),
773 ));
774 }
775
776 let mut endpoints: Vec<String> = Vec::with_capacity(raw_count);
777 for raw in host_part.split(',') {
778 let raw = raw.trim();
779 if raw.is_empty() {
780 return Err(ParseError::new(
781 ParseErrorKind::InvalidUri,
782 "grpc cluster URI has an empty host entry",
783 ));
784 }
785 let (host, port) = if let Some(after_bracket) = raw.strip_prefix('[') {
787 let end = after_bracket.find(']').ok_or_else(|| {
788 ParseError::new(
789 ParseErrorKind::InvalidUri,
790 format!("unterminated IPv6 bracket in cluster URI: {raw}"),
791 )
792 })?;
793 let host = &after_bracket[..end];
794 let tail = &after_bracket[end + 1..];
795 let port = if tail.is_empty() {
796 default_port
797 } else if let Some(p) = tail.strip_prefix(':') {
798 p.parse::<u16>().map_err(|_| {
799 ParseError::new(
800 ParseErrorKind::InvalidUri,
801 format!("invalid port in cluster URI: {raw}"),
802 )
803 })?
804 } else {
805 return Err(ParseError::new(
806 ParseErrorKind::InvalidUri,
807 format!("trailing junk after IPv6 bracket in cluster URI: {raw}"),
808 ));
809 };
810 (format!("[{host}]"), port)
811 } else {
812 match raw.rsplit_once(':') {
813 Some((h, p)) => {
814 let port: u16 = p.parse().map_err(|_| {
815 ParseError::new(
816 ParseErrorKind::InvalidUri,
817 format!("invalid port in cluster URI: {raw}"),
818 )
819 })?;
820 (h.to_string(), port)
821 }
822 None => (raw.to_string(), default_port),
823 }
824 };
825 if host.is_empty() || host == "[]" {
826 return Err(ParseError::new(
827 ParseErrorKind::InvalidUri,
828 "grpc cluster URI has an empty host entry",
829 ));
830 }
831 endpoints.push(format!("{endpoint_scheme}://{host}:{port}"));
832 }
833
834 if let Some(q) = query_part {
835 let qcount = if q.is_empty() {
836 0
837 } else {
838 q.split('&').count()
839 };
840 if qcount > limits.max_query_params {
841 return Err(ParseError::new(
842 ParseErrorKind::LimitExceeded,
843 format!(
844 "max_query_params exceeded: limit={} actual={}",
845 limits.max_query_params, qcount,
846 ),
847 ));
848 }
849 }
850
851 let force_primary = query_part
852 .map(|q| {
853 q.split('&').any(|kv| {
854 let mut parts = kv.splitn(2, '=');
855 let k = parts.next().unwrap_or("");
856 let v = parts.next().unwrap_or("");
857 k.eq_ignore_ascii_case("route") && v.eq_ignore_ascii_case("primary")
858 })
859 })
860 .unwrap_or(false);
861
862 let mut iter = endpoints.into_iter();
863 let primary = iter.next().expect("split on ',' yields at least one entry");
864 let replicas: Vec<String> = iter.collect();
865
866 Ok(Some(ConnectionTarget::GrpcCluster {
867 primary,
868 replicas,
869 force_primary,
870 }))
871}