use crate::core::Column as _;
use crate::sql::FetcherPool;
use crate::sql::Pool;
use async_trait::async_trait;
use http::request::Parts;
use http::HeaderName;
use super::error::TenancyError;
use super::org::Org;
use super::resolver_cache::{Breaker, Cached, GenerationPoll, HostTtlCache};
#[async_trait]
pub trait OrgResolver: Send + Sync + 'static {
async fn resolve(&self, parts: &Parts, registry: &Pool) -> Result<Option<Org>, TenancyError>;
}
pub struct SubdomainResolver {
pub apex_domain: String,
}
impl SubdomainResolver {
#[must_use]
pub fn new(apex_domain: impl Into<String>) -> Self {
Self {
apex_domain: apex_domain.into(),
}
}
}
#[async_trait]
impl OrgResolver for SubdomainResolver {
async fn resolve(&self, parts: &Parts, registry: &Pool) -> Result<Option<Org>, TenancyError> {
let Some(host) = host_from_parts(parts) else {
return Ok(None);
};
if host == self.apex_domain {
return Ok(None);
}
sync_org_generation(registry).await;
match ORG_CACHE.get(&host) {
Cached::Miss => return Ok(None),
Cached::Hit(org) => return Ok(Some(org)),
Cached::Absent => {}
}
let found = find_active_org_by(registry, Org::host_pattern.eq(host.clone())).await?;
ORG_CACHE.put(&host, found.clone());
Ok(found)
}
}
pub struct RegisteredHostResolver;
static HOST_TABLE_DOWN: Breaker = Breaker::new();
fn host_table_retry_after() -> std::time::Duration {
gen_check_every().unwrap_or(GEN_CHECK_EVERY_DEFAULT)
}
static HOST_CACHE: HostTtlCache<Org> = HostTtlCache::new(CACHE_TTL, CACHE_MAX);
const CACHE_TTL: std::time::Duration = std::time::Duration::from_secs(30);
const CACHE_MAX: usize = 1024;
static ORG_CACHE: HostTtlCache<Org> = HostTtlCache::new(CACHE_TTL, CACHE_MAX);
pub fn invalidate_org_cache() {
ORG_CACHE.clear();
invalidate_host_cache();
}
static ORG_GEN: GenerationPoll<super::org::OrgGeneration> = GenerationPoll::new();
async fn sync_org_generation(registry: &Pool) {
ORG_GEN
.sync(
gen_check_every(),
|| async {
if registry_is_down() {
return Ok(None);
}
super::org::generation(registry).await.map(Some)
},
|e| {
REGISTRY_DOWN.open();
ORG_GEN.warn_once(|| {
tracing::warn!(
target: "rustango::tenancy::resolver",
error = %e,
"could not read the rustango_orgs fingerprint; this pod will \
not notice tenants created or suspended by other processes \
until its own cache entries expire"
);
});
},
invalidate_org_cache,
)
.await;
}
#[cfg(any(test, feature = "testkit"))]
pub(crate) fn reset_org_cache() {
ORG_GEN.reset();
invalidate_org_cache();
}
static HOST_GEN: GenerationPoll<super::org_host::Generation> = GenerationPoll::new();
const GEN_CHECK_EVERY_DEFAULT: std::time::Duration = std::time::Duration::from_secs(5);
const GEN_CHECK_ENV: &str = "RUSTANGO_HOST_GEN_CHECK_SECS";
fn gen_check_every() -> Option<std::time::Duration> {
static CACHED: std::sync::OnceLock<Option<std::time::Duration>> = std::sync::OnceLock::new();
*CACHED.get_or_init(|| match std::env::var(GEN_CHECK_ENV) {
Ok(raw) => match raw.trim().parse::<u64>() {
Ok(0) => None,
Ok(secs) => Some(std::time::Duration::from_secs(secs)),
Err(_) => {
tracing::warn!(
target: "rustango::tenancy::resolver",
value = %raw,
"{GEN_CHECK_ENV} is not a whole number of seconds; \
falling back to the default interval"
);
Some(GEN_CHECK_EVERY_DEFAULT)
}
},
Err(_) => Some(GEN_CHECK_EVERY_DEFAULT),
})
}
async fn sync_generation(registry: &Pool) {
HOST_GEN
.sync(
gen_check_every(),
|| async { super::org_host::generation(registry).await.map(Some) },
|e| {
HOST_GEN.warn_once(|| {
tracing::warn!(
target: "rustango::tenancy::resolver",
error = %e,
"could not read the rustango_org_hosts fingerprint; this pod \
will not notice host changes made by other processes until \
its own cache entries expire"
);
});
},
invalidate_host_cache,
)
.await;
}
#[cfg(any(test, feature = "testkit"))]
pub(crate) fn reset_generation() {
HOST_GEN.reset();
HOST_TABLE_DOWN.close();
}
#[cfg(any(test, feature = "testkit"))]
pub(crate) fn expire_generation() {
let interval = gen_check_every().unwrap_or(GEN_CHECK_EVERY_DEFAULT);
HOST_GEN.expire(interval * 2, *PROCESS_START);
}
#[cfg(any(test, feature = "testkit"))]
static PROCESS_START: std::sync::LazyLock<std::time::Instant> =
std::sync::LazyLock::new(std::time::Instant::now);
pub fn invalidate_host_cache() {
HOST_CACHE.clear();
}
#[async_trait]
impl OrgResolver for RegisteredHostResolver {
async fn resolve(&self, parts: &Parts, registry: &Pool) -> Result<Option<Org>, TenancyError> {
let Some(host) = host_from_parts(parts) else {
return Ok(None);
};
sync_generation(registry).await;
match HOST_CACHE.get(&host) {
Cached::Miss => return Ok(None),
Cached::Hit(org) => return Ok(Some(org)),
Cached::Absent => {}
}
if HOST_TABLE_DOWN.is_open(host_table_retry_after()) {
return Ok(None);
}
let rows = match super::OrgHost::objects()
.where_(super::OrgHost::hostname.eq(host.clone()))
.where_(super::OrgHost::enabled.eq(true))
.fetch(registry)
.await
{
Ok(rows) => {
HOST_TABLE_DOWN.close();
rows
}
Err(e) => {
HOST_TABLE_DOWN.open();
HOST_TABLE_DOWN.warn_once(|| {
tracing::warn!(
target: "rustango::tenancy::resolver",
error = %e,
"could not read rustango_org_hosts; extra tenant hostnames \
are inactive until `migrate` creates the table"
);
});
return Ok(None);
}
};
let Some(row) = rows.into_iter().next() else {
HOST_CACHE.put(&host, None);
return Ok(None);
};
let found = find_active_org_by(registry, Org::id.eq(row.org_id)).await?;
HOST_CACHE.put(&host, found.clone());
Ok(found)
}
}
pub struct PathPrefixResolver;
#[async_trait]
impl OrgResolver for PathPrefixResolver {
async fn resolve(&self, parts: &Parts, registry: &Pool) -> Result<Option<Org>, TenancyError> {
let path = parts.uri.path();
let Some(first) = path
.trim_start_matches('/')
.split('/')
.next()
.filter(|s| !s.is_empty())
else {
return Ok(None);
};
let candidate = format!("/{first}");
find_active_org_by(registry, Org::path_prefix.eq(candidate)).await
}
}
pub struct HeaderResolver {
pub header_name: HeaderName,
allowed_slugs: Option<std::collections::HashSet<String>>,
}
impl HeaderResolver {
#[must_use]
pub fn new(header_name: HeaderName) -> Self {
Self {
header_name,
allowed_slugs: None,
}
}
#[must_use]
pub fn allow_only<I, S>(mut self, slugs: I) -> Self
where
I: IntoIterator<Item = S>,
S: Into<String>,
{
self.allowed_slugs = Some(slugs.into_iter().map(Into::into).collect());
self
}
}
impl Default for HeaderResolver {
fn default() -> Self {
Self {
header_name: HeaderName::from_static("x-org"),
allowed_slugs: None,
}
}
}
#[async_trait]
impl OrgResolver for HeaderResolver {
async fn resolve(&self, parts: &Parts, registry: &Pool) -> Result<Option<Org>, TenancyError> {
let Some(value) = parts.headers.get(&self.header_name) else {
return Ok(None);
};
let slug = match value.to_str() {
Ok(s) => s.trim(),
Err(_) => return Ok(None),
};
if slug.is_empty() {
return Ok(None);
}
if let Some(allowed) = &self.allowed_slugs {
if !allowed.contains(slug) {
return Ok(None);
}
}
find_active_org_by(registry, Org::slug.eq(slug.to_owned())).await
}
}
pub struct PortResolver;
#[async_trait]
impl OrgResolver for PortResolver {
async fn resolve(&self, parts: &Parts, registry: &Pool) -> Result<Option<Org>, TenancyError> {
let Some(port) = parts.uri.port_u16() else {
return Ok(None);
};
find_active_org_by(registry, Org::port.eq(i32::from(port))).await
}
}
pub struct ChainResolver {
resolvers: Vec<Box<dyn OrgResolver>>,
}
impl ChainResolver {
#[must_use]
pub fn new() -> Self {
Self {
resolvers: Vec::new(),
}
}
#[must_use]
pub fn push<R: OrgResolver>(mut self, resolver: R) -> Self {
self.resolvers.push(Box::new(resolver));
self
}
#[must_use]
pub fn standard(apex_domain: impl Into<String>) -> Self {
Self::new()
.push(SubdomainResolver::new(apex_domain))
.push(RegisteredHostResolver)
.push(HeaderResolver::default())
}
}
impl Default for ChainResolver {
fn default() -> Self {
Self::new()
}
}
#[async_trait]
impl OrgResolver for ChainResolver {
async fn resolve(&self, parts: &Parts, registry: &Pool) -> Result<Option<Org>, TenancyError> {
for resolver in &self.resolvers {
match resolver.resolve(parts, registry).await? {
Some(org) => return Ok(Some(org)),
None => continue,
}
}
Ok(None)
}
}
fn host_from_parts(parts: &Parts) -> Option<String> {
if let Some(value) = parts.headers.get(http::header::HOST) {
if let Ok(s) = value.to_str() {
return Some(s.split(':').next().unwrap_or(s).to_ascii_lowercase());
}
}
parts.uri.host().map(str::to_ascii_lowercase)
}
static REGISTRY_DOWN: Breaker = Breaker::new();
const REGISTRY_RETRY_AFTER: std::time::Duration = std::time::Duration::from_secs(1);
fn registry_is_down() -> bool {
REGISTRY_DOWN.is_open(REGISTRY_RETRY_AFTER)
}
fn registry_unavailable() -> TenancyError {
TenancyError::Driver(sqlx::Error::PoolTimedOut)
}
async fn find_active_org_by<F>(registry: &Pool, filter: F) -> Result<Option<Org>, TenancyError>
where
F: Into<rustango::core::TypedFilter<Org>>,
{
if registry_is_down() {
return Err(registry_unavailable());
}
let typed: rustango::core::TypedFilter<Org> = filter.into();
let result: Result<Vec<Org>, _> = Org::objects()
.where_(typed)
.where_(Org::active.eq(true))
.fetch(registry)
.await;
match result {
Ok(rows) => {
REGISTRY_DOWN.close();
Ok(rows.into_iter().next())
}
Err(e) => {
REGISTRY_DOWN.open();
REGISTRY_DOWN.warn_once(|| {
tracing::warn!(
target: "rustango::tenancy::resolver",
error = %e,
"registry lookup failed; tenant resolution is failing fast for \
up to {}s at a time until it recovers",
REGISTRY_RETRY_AFTER.as_secs(),
);
});
Err(TenancyError::Driver(driver_from_exec(e)))
}
}
}
#[cfg(any(test, feature = "testkit"))]
pub(crate) fn reset_registry_breaker() {
REGISTRY_DOWN.close();
}
fn driver_from_exec(e: rustango::sql::ExecError) -> rustango::sql::sqlx::Error {
use crate::sql::ExecError;
match e {
ExecError::Driver(err) => err,
other => rustango::sql::sqlx::Error::Protocol(format!("resolver query: {other}")),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn default_has_no_allowlist() {
let r = HeaderResolver::default();
assert!(
r.allowed_slugs.is_none(),
"default resolver should not have an allowlist"
);
}
#[test]
fn allow_only_stores_slugs() {
let r = HeaderResolver::default().allow_only(["acme", "globex"]);
let set = r.allowed_slugs.expect("allow_only sets the slug set");
assert_eq!(set.len(), 2);
assert!(set.contains("acme"));
assert!(set.contains("globex"));
assert!(!set.contains("attacker-tenant"));
}
#[test]
fn allow_only_accepts_owned_strings() {
let slugs: Vec<String> = vec!["a".into(), "b".into()];
let r = HeaderResolver::default().allow_only(slugs);
assert_eq!(
r.allowed_slugs.expect("allow_only sets the slug set").len(),
2
);
}
}