use std::fmt;
use std::sync::Arc;
use std::task::{Context, Poll};
use futures_util::future::{self, Either, Ready};
use futures_util::TryFutureExt;
use http::header::{HeaderName, HeaderValue};
use http::{HeaderMap, Uri};
use serde::Serialize;
use tower::{Layer, Service};
use url::Url;
use crate::request_options::Applied;
use crate::{BoxError, Error};
type MapPath = Arc<dyn Fn(&str) -> String + Send + Sync>;
#[derive(Clone)]
pub struct ProviderLayer {
base: Url,
rewrite_from: Option<Url>,
headers: HeaderMap,
query: Vec<(String, String)>,
map_path: Option<MapPath>,
}
impl ProviderLayer {
pub fn new(base_url: &str) -> Result<Self, Error> {
Ok(Self {
base: parse_base(base_url)?,
rewrite_from: None,
headers: HeaderMap::new(),
query: Vec::new(),
map_path: None,
})
}
pub fn rewrite_from(mut self, base_url: &str) -> Result<Self, Error> {
self.rewrite_from = Some(parse_base(base_url)?);
Ok(self)
}
pub fn header<K, V>(mut self, name: K, value: V) -> Result<Self, Error>
where
HeaderName: TryFrom<K>,
<HeaderName as TryFrom<K>>::Error: Into<http::Error>,
HeaderValue: TryFrom<V>,
<HeaderValue as TryFrom<V>>::Error: Into<http::Error>,
{
let (name, value) = header(name, value)?;
self.headers.insert(name, value);
Ok(self)
}
pub fn sensitive_header<K, V>(mut self, name: K, value: V) -> Result<Self, Error>
where
HeaderName: TryFrom<K>,
<HeaderName as TryFrom<K>>::Error: Into<http::Error>,
HeaderValue: TryFrom<V>,
<HeaderValue as TryFrom<V>>::Error: Into<http::Error>,
{
let (name, mut value) = header(name, value)?;
value.set_sensitive(true);
self.headers.insert(name, value);
Ok(self)
}
pub fn headers(mut self, headers: HeaderMap) -> Self {
self.headers.extend(headers);
self
}
pub fn query<T: Serialize + ?Sized>(mut self, query: &T) -> Result<Self, Error> {
let encoded = serde_urlencoded::to_string(query).map_err(Error::builder)?;
let pairs: Vec<(String, String)> = url::form_urlencoded::parse(encoded.as_bytes())
.into_owned()
.collect();
self.query
.retain(|(name, _)| !pairs.iter().any(|(new, _)| new == name));
self.query.extend(pairs);
Ok(self)
}
pub fn map_path<F>(mut self, f: F) -> Self
where
F: Fn(&str) -> String + Send + Sync + 'static,
{
self.map_path = Some(Arc::new(f));
self
}
pub fn base_url(&self) -> &str {
self.base.as_str()
}
pub fn header_map(&self) -> &HeaderMap {
&self.headers
}
pub fn query_pairs(&self) -> &[(String, String)] {
&self.query
}
pub fn rewrite_uri(&self, uri: &Uri) -> Result<Uri, Error> {
Ok(self.rewrite(uri, None)?.unwrap_or_else(|| uri.clone()))
}
fn rewrite(&self, uri: &Uri, applied: Option<&Applied>) -> Result<Option<Uri>, Error> {
let Ok(mut url) = Url::parse(&uri.to_string()) else {
return Ok(None);
};
let rewrite_from = self
.rewrite_from
.as_ref()
.filter(|_| !applied.is_some_and(|applied| applied.base));
let rest = rewrite_from
.into_iter()
.chain([&self.base])
.find_map(|base| rest_of_path(&url, base));
let Some(rest) = rest else {
return Ok(None);
};
let rest = match &self.map_path {
Some(map_path) => map_path(&rest),
None => rest,
};
let mut path = trimmed_path(&self.base).to_owned();
path.push_str(&rest);
let _ = url.set_scheme(self.base.scheme());
url.set_host(self.base.host_str()).map_err(Error::builder)?;
let _ = url.set_port(self.base.port());
url.set_path(if path.is_empty() { "/" } else { &path });
self.set_query(&mut url, applied);
url.as_str()
.parse::<Uri>()
.map(Some)
.map_err(Error::builder)
}
fn set_query(&self, url: &mut Url, applied: Option<&Applied>) {
let query: Vec<&(String, String)> = self
.query
.iter()
.filter(|(name, _)| !applied.is_some_and(|applied| applied.has_query(name)))
.collect();
if query.is_empty() {
return;
}
let replaced = url
.query_pairs()
.any(|(name, _)| query.iter().any(|(new, _)| *new == name));
if replaced {
let kept: Vec<(String, String)> = url
.query_pairs()
.filter(|(name, _)| !query.iter().any(|(new, _)| new == name))
.map(|(name, value)| (name.into_owned(), value.into_owned()))
.collect();
url.set_query(None);
url.query_pairs_mut().extend_pairs(kept);
}
url.query_pairs_mut()
.extend_pairs(query.iter().map(|(name, value)| (name, value)));
}
}
impl fmt::Debug for ProviderLayer {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("ProviderLayer")
.field("base", &self.base.as_str())
.field("rewrite_from", &self.rewrite_from.as_ref().map(Url::as_str))
.field("headers", &self.headers)
.field("query", &self.query)
.field("map_path", &self.map_path.is_some())
.finish()
}
}
impl<S> Layer<S> for ProviderLayer {
type Service = Provider<S>;
fn layer(&self, inner: S) -> Self::Service {
Provider {
inner,
provider: Arc::new(self.clone()),
}
}
}
#[derive(Clone)]
pub struct Provider<S> {
inner: S,
provider: Arc<ProviderLayer>,
}
impl<S: fmt::Debug> fmt::Debug for Provider<S> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Provider")
.field("inner", &self.inner)
.field("provider", &self.provider)
.finish()
}
}
impl<S, B> Service<http::Request<B>> for Provider<S>
where
S: Service<http::Request<B>>,
S::Error: Into<BoxError>,
{
type Response = S::Response;
type Error = BoxError;
#[allow(clippy::type_complexity)]
type Future = Either<
Ready<Result<S::Response, BoxError>>,
futures_util::future::MapErr<S::Future, fn(S::Error) -> BoxError>,
>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.inner.poll_ready(cx).map_err(Into::into)
}
fn call(&mut self, mut request: http::Request<B>) -> Self::Future {
let applied = request.extensions().get::<Applied>().cloned();
match self.provider.rewrite(request.uri(), applied.as_ref()) {
Err(error) => Either::Left(future::ready(Err(error.into()))),
Ok(None) => Either::Right(
self.inner
.call(request)
.map_err(into_box_error::<S::Error> as fn(_) -> _),
),
Ok(Some(uri)) => {
*request.uri_mut() = uri;
match &applied {
None => request.headers_mut().extend(self.provider.headers.clone()),
Some(applied) => {
let mut kept = HeaderMap::new();
for (name, value) in &self.provider.headers {
if !applied.has_header(name) {
kept.append(name.clone(), value.clone());
}
}
request.headers_mut().extend(kept);
}
}
Either::Right(
self.inner
.call(request)
.map_err(into_box_error::<S::Error> as fn(_) -> _),
)
}
}
}
}
fn into_box_error<E: Into<BoxError>>(error: E) -> BoxError {
error.into()
}
fn header<K, V>(name: K, value: V) -> Result<(HeaderName, HeaderValue), Error>
where
HeaderName: TryFrom<K>,
<HeaderName as TryFrom<K>>::Error: Into<http::Error>,
HeaderValue: TryFrom<V>,
<HeaderValue as TryFrom<V>>::Error: Into<http::Error>,
{
let name = HeaderName::try_from(name).map_err(|e| Error::builder(e.into()))?;
let value = HeaderValue::try_from(value).map_err(|e| Error::builder(e.into()))?;
Ok((name, value))
}
pub(crate) fn parse_base(base_url: &str) -> Result<Url, Error> {
let url = Url::parse(base_url).map_err(Error::builder)?;
if url.cannot_be_a_base() || url.host_str().is_none() {
return Err(Error::builder(format!(
"not a base URL with a host: {base_url}"
)));
}
if url.query().is_some() || url.fragment().is_some() {
return Err(Error::builder(format!(
"a base URL has no query or fragment (set query parameters with `query`): {base_url}"
)));
}
Ok(url)
}
pub(crate) fn trimmed_path(base: &Url) -> &str {
base.path().trim_end_matches('/')
}
pub(crate) fn rest_of_path(url: &Url, base: &Url) -> Option<String> {
if url.scheme() != base.scheme()
|| url.host_str() != base.host_str()
|| url.port_or_known_default() != base.port_or_known_default()
{
return None;
}
let rest = url.path().strip_prefix(trimmed_path(base))?;
(rest.is_empty() || rest.starts_with('/')).then(|| rest.to_owned())
}
#[cfg(test)]
mod tests {
use super::*;
fn rewrite(layer: &ProviderLayer, uri: &str) -> String {
layer
.rewrite_uri(&uri.parse().unwrap())
.unwrap()
.to_string()
}
#[test]
fn moves_requests_under_the_declared_base() {
let layer = ProviderLayer::new("https://openrouter.ai/api/v1/")
.unwrap()
.rewrite_from("https://api.openai.com/v1")
.unwrap();
assert_eq!(
rewrite(&layer, "https://api.openai.com/v1/chat/completions"),
"https://openrouter.ai/api/v1/chat/completions"
);
assert_eq!(
rewrite(
&layer,
"https://api.openai.com:443/v1/models?limit=2&after=a%20b"
),
"https://openrouter.ai/api/v1/models?limit=2&after=a%20b"
);
assert_eq!(
rewrite(&layer, "https://api.openai.com/v1"),
"https://openrouter.ai/api/v1"
);
for uri in [
"https://api.openai.com/v10/models",
"https://api.openai.com/models",
"http://api.openai.com/v1/models",
"https://api.openai.com:8443/v1/models",
"https://files.example.com/v1/download",
] {
assert_eq!(rewrite(&layer, uri), uri);
}
}
#[test]
fn a_request_already_at_the_provider_gets_its_path_mapped_once() {
let layer = ProviderLayer::new("http://localhost:11434/v1")
.unwrap()
.map_path(|path| format!("/x{path}"));
assert_eq!(
rewrite(&layer, "http://localhost:11434/v1/models"),
"http://localhost:11434/v1/x/models"
);
assert_eq!(
rewrite(&layer, "http://localhost:11435/v1/models"),
"http://localhost:11435/v1/models"
);
}
#[test]
fn root_bases() {
let layer = ProviderLayer::new("https://proxy.example")
.unwrap()
.rewrite_from("https://api.example.com/")
.unwrap();
assert_eq!(
rewrite(&layer, "https://api.example.com/items?a=1"),
"https://proxy.example/items?a=1"
);
assert_eq!(
rewrite(&layer, "https://api.example.com"),
"https://proxy.example/"
);
let layer = ProviderLayer::new("https://proxy.example/prefix")
.unwrap()
.rewrite_from("https://api.example.com")
.unwrap();
assert_eq!(
rewrite(&layer, "https://api.example.com/items"),
"https://proxy.example/prefix/items"
);
}
#[test]
fn query_parameters_replace_those_of_their_name() {
let layer = ProviderLayer::new("https://p.example/openai")
.unwrap()
.query(&[("api-version", "old")])
.unwrap()
.query(&[("api-version", "2024-10-21"), ("tag", "a"), ("tag", "b")])
.unwrap();
assert_eq!(
layer.query_pairs(),
[
("api-version".to_owned(), "2024-10-21".to_owned()),
("tag".to_owned(), "a".to_owned()),
("tag".to_owned(), "b".to_owned()),
]
);
assert_eq!(
rewrite(&layer, "https://p.example/openai/models"),
"https://p.example/openai/models?api-version=2024-10-21&tag=a&tag=b"
);
assert_eq!(
rewrite(&layer, "https://p.example/openai/models?limit=2&q=a%20b"),
"https://p.example/openai/models?limit=2&q=a%20b&api-version=2024-10-21&tag=a&tag=b"
);
assert_eq!(
rewrite(
&layer,
"https://p.example/openai/models?api-version=x&limit=2&tag=z"
),
"https://p.example/openai/models?limit=2&api-version=2024-10-21&tag=a&tag=b"
);
}
#[test]
fn invalid_input_is_a_builder_error() {
for base in [
"not a url",
"mailto:someone@example.com",
"https://p.example/v1?a=1",
"https://p.example/v1#top",
] {
assert!(ProviderLayer::new(base).unwrap_err().is_builder(), "{base}");
}
let layer = ProviderLayer::new("https://p.example").unwrap();
assert!(layer.clone().header("x bad", "v").unwrap_err().is_builder());
assert!(layer
.clone()
.sensitive_header("x-key", "line\nbreak")
.unwrap_err()
.is_builder());
assert!(layer.query(&[("a", [1, 2])]).unwrap_err().is_builder());
}
#[test]
fn debug_hides_sensitive_headers() {
let layer = ProviderLayer::new("https://p.example")
.unwrap()
.sensitive_header("api-key", "secret-123")
.unwrap()
.header("x-title", "app")
.unwrap();
let debug = format!("{layer:?}");
assert!(!debug.contains("secret-123"), "{debug}");
assert!(debug.contains("app"), "{debug}");
assert!(layer.header_map()["api-key"].is_sensitive());
assert_eq!(layer.base_url(), "https://p.example/");
}
}