use std::path::PathBuf;
use url::Url;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ParseErrorKind {
Empty,
InvalidUri,
UnsupportedScheme,
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",
}
}
}
#[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 {}
pub const DEFAULT_PORT_RED: u16 = 5050;
pub const DEFAULT_PORT_GRPC: u16 = 55055;
pub const DEFAULT_PORT_GRPCS: u16 = 55555;
pub const DEFAULT_PORT_WS: u16 = 80;
pub const DEFAULT_PORT_WSS: u16 = 443;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConnectionScheme {
Memory,
File,
Red,
Reds,
RedWs,
RedWss,
Ws,
Wss,
Grpc,
Grpcs,
Http,
Https,
}
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.",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ConnStringLimits {
pub max_uri_bytes: usize,
pub max_query_params: usize,
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,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ConnectionTarget {
Memory,
File { path: PathBuf },
Grpc { endpoint: String },
GrpcCluster {
primary: String,
replicas: Vec<String>,
force_primary: bool,
},
Http { base_url: String },
RedWire { host: String, port: u16, tls: bool },
WsNative { host: String, port: u16, tls: bool },
}
#[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(_))
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConnectionSpec {
pub target: ConnectionTarget,
pub auth: ConnectionAuth,
pub redacted_uri: String,
}
pub fn parse(uri: &str) -> Result<ConnectionTarget, ParseError> {
parse_with_limits(uri, ConnStringLimits::default())
}
pub fn parse_with_auth(uri: &str) -> Result<ConnectionSpec, ParseError> {
let normalised = normalise_scheme(uri);
let redacted_uri = redact_uri_userinfo(&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
}
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:///")
}
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(),
),
));
}
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: PathBuf::from(rest),
});
}
if let Some(cluster) = try_parse_grpc_cluster(uri, &limits)? {
return Ok(cluster);
}
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")
})?;
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
}
});
Ok(ConnectionTarget::Grpc {
endpoint: format!("http://{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 normalise_scheme(uri: &str) -> String {
match uri.find(':') {
Some(i) => {
let scheme = &uri[..i];
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);
}
let parsed = Url::parse(uri)
.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_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 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(())
}
fn try_parse_grpc_cluster(
uri: &str,
limits: &ConnStringLimits,
) -> Result<Option<ConnectionTarget>, ParseError> {
let (rest, default_port) = if let Some(r) = uri.strip_prefix("grpc://") {
(r, DEFAULT_PORT_GRPC)
} else if let Some(r) = uri.strip_prefix("grpcs://") {
(r, DEFAULT_PORT_GRPCS)
} else if let Some(r) = uri
.strip_prefix("red://")
.or_else(|| uri.strip_prefix("reds://"))
{
(r, DEFAULT_PORT_RED)
} else {
return Ok(None);
};
let (host_part, query_part) = match rest.find('?') {
Some(i) => (&rest[..i], Some(&rest[i + 1..])),
None => (rest, None),
};
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",
));
}
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!("http://{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,
}))
}