rustango 0.60.7

//! Per-request tenant resolvers — turn an incoming HTTP request into
//! an [`Org`] from the registry.
//!
//! The headline routing mode (locked in v0.5 design) is **subdomain**:
//! `acme.app.com` → org with `host_pattern = "acme.app.com"`. Cookie
//! isolation by subdomain is the security win — tenant A's session
//! cookie is unreadable from tenant B's domain by browser policy.
//!
//! # Built-in resolvers
//!
//! | Resolver               | Matches against                 | Notes                                          |
//! |------------------------|---------------------------------|------------------------------------------------|
//! | [`SubdomainResolver`]  | `Org.host_pattern` ↔ `Host` hdr | Default — composed first in `ChainResolver::standard()`. |
//! | [`PathPrefixResolver`] | `Org.path_prefix` ↔ URL path    | Opt-in — caller adds explicitly.               |
//! | [`HeaderResolver`]     | `Org.slug` ↔ user header value  | API-only deployments.                          |
//! | [`PortResolver`]       | `Org.port` ↔ listener port      | Niche — hard-isolated tenant ports.            |
//! | [`ChainResolver`]      | tries each in order             | Operator builds the chain in `main.rs`.        |
//!
//! # `ChainResolver::standard()`
//!
//! `[Subdomain, RegisteredHost]`. Header and path-prefix are **not** in
//! it — the operator opts in explicitly when they need those modes.
//!
//! # Apex (no subdomain) handling
//!
//! Bare `app.com` does **not** resolve to any tenant. The default
//! routing decision is that the apex hosts only operator UI
//! (`/operator/*` in Slice 6); other apex paths return 404. The
//! resolver returning `Ok(None)` is the signal — the caller (the
//! tenant-aware admin in Slice 4) translates that into a 404 for
//! tenant routes and bypasses the resolver entirely for `/operator`.

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};
use crate::sql::PoolId;
use std::sync::{Arc, PoisonError};

/// Resolve an HTTP request to an [`Org`] from the registry.
///
/// Implementations live in `rustango-tenancy::resolver` (built-ins)
/// or in user crates (custom resolvers — JWT claims, API key
/// prefixes, geo routing, etc.).
///
/// `Ok(None)` means "no tenant matched"; the caller decides whether
/// that's a 404 or a fallthrough (e.g. the apex hits the operator
/// UI without ever calling the resolver).
#[async_trait]
pub trait OrgResolver: Send + Sync + 'static {
    /// Resolve `parts` against `registry` to find the matching tenant.
    ///
    /// # Errors
    /// Returns [`TenancyError::Driver`] for SQL failures during the
    /// lookup, [`TenancyError::Validation`] for malformed registry
    /// rows, or [`TenancyError::Resolution`] for explicit "no
    /// tenant" with diagnostic context (rare — most no-match cases
    /// return `Ok(None)`).
    async fn resolve(&self, parts: &Parts, registry: &Pool) -> Result<Option<Org>, TenancyError>;
}

// ---------------- SubdomainResolver ----------------

/// Match the request's `Host` header against `Org.host_pattern`.
///
/// A request to `acme.app.com` finds the org whose `host_pattern` is
/// exactly `"acme.app.com"`. The apex (`app.com`) never matches —
/// only orgs with an explicit non-null `host_pattern` are eligible.
///
/// `apex_domain` is informational — used in error messages and as
/// part of the boot pre-flight check that warns when `apex_domain`
/// looks malformed. It does NOT auto-derive slugs from subdomains;
/// `cargo run -- create-tenant <slug>` writes the full `host_pattern`
/// to the Org row, and the resolver matches it verbatim.
pub struct SubdomainResolver {
    pub apex_domain: String,
}

impl SubdomainResolver {
    /// Construct from the apex (e.g. `"app.example.com"`). The apex
    /// is not validated here; the boot pre-flight in Slice 4 will
    /// fail-fast on missing or malformed apex.
    #[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);
        };
        // Apex only — no tenant resolution.
        if host.eq_ignore_ascii_case(&self.apex_domain) {
            return Ok(None);
        }
        // Pick up a tenant created or suspended by another pod before
        // trusting anything cached here. Throttled to one aggregate per
        // interval — see `sync_org_generation`.
        let st = state(registry);
        sync_org_generation(registry, &st).await;
        match st.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, &st, Org::host_pattern.eq(host.clone())).await?;
        // Cache the miss too: this resolver runs first for *every*
        // request, so an unregistered host would otherwise be a free
        // registry query per request. See `RegistryState::org_cache`.
        st.org_cache.put(&host, found.clone());
        Ok(found)
    }
}

// ---------------- RegisteredHostResolver ----------------

/// Match the `Host` header against an extra hostname registered in
/// [`rustango_org_hosts`](super::OrgHost), so one tenant can serve several
/// domains beyond its base `Org.host_pattern`.
///
/// Composed AFTER [`SubdomainResolver`] in the standard chain: the base
/// host keeps resolving exactly as it always did, and this only ever runs
/// on a host that would otherwise have found nothing. The feature is
/// purely additive — it cannot change where an existing request lands.
///
/// ## Missing table is not an error
///
/// The `rustango_org_hosts` table arrives with the generated system
/// migration chain, so a deployment that upgrades the binary and has not
/// yet run `migrate` does not have it. Propagating that error would turn
/// "you haven't migrated yet" into a 500 on **every** request, including
/// for tenants that never use extra hosts. So a failed lookup is logged
/// once and treated as "no match", which is the pre-upgrade behaviour to
/// the byte.
pub struct RegisteredHostResolver;

/// Resolver caches, polls and breakers for one registry pool. Kept per
/// registry so two registries in one process never share them (#2077).
struct RegistryState {
    /// Base `Host` → [`Org`] (`None` = known miss).
    ///
    /// [`SubdomainResolver`] runs first in the standard chain, so uncached
    /// it cost one registry `SELECT` per request — the registry's hot-path
    /// SPOF. The whole `Org` is stored, not its id, so a hit costs zero
    /// queries. Bounded and negatively cached: the keys are attacker-supplied.
    org_cache: HostTtlCache<Org>,
    /// Extra hostname → [`Org`] (`None` = known miss). Same reasons as
    /// `org_cache`; negative entries matter most here, as only hosts the
    /// base resolver missed reach this one.
    host_cache: HostTtlCache<Org>,
    /// Last `rustango_orgs` fingerprint seen, so this pod notices another
    /// pod's writes without a shared cache. See [`sync_org_generation`].
    org_gen: GenerationPoll<super::org::OrgGeneration>,
    /// Same as `org_gen`, for `rustango_org_hosts`.
    host_gen: GenerationPoll<super::org_host::Generation>,
    /// Fails fast while the registry is unreachable. Tenant resolution runs
    /// before everything else, so an outage is otherwise paid per request.
    /// The caller still gets `Err`, just in microseconds.
    registry_down: Breaker,
    /// Backs off while the host-table lookup fails. An error is not a
    /// miss, so it can't be cached as one; without this the error path is
    /// uncached and unthrottled.
    host_table_down: Breaker,
}

impl RegistryState {
    const fn new() -> Self {
        Self {
            org_cache: HostTtlCache::new(CACHE_TTL, CACHE_MAX),
            host_cache: HostTtlCache::new(CACHE_TTL, CACHE_MAX),
            org_gen: GenerationPoll::new(),
            host_gen: GenerationPoll::new(),
            registry_down: Breaker::new(),
            host_table_down: Breaker::new(),
        }
    }

    /// Drop both caches: each holds whole `Org` rows. See [`invalidate_org_cache`].
    fn invalidate_orgs(&self) {
        self.org_cache.clear();
        self.host_cache.clear();
    }
}

/// Registries tracked at once. Each entry holds a clone of its pool, so a
/// dropped registry's idle connections stay open until evicted or
/// `close()`d; accepted, as the cap bounds it. Closed pools go first, then
/// the oldest, which only restarts from empty caches.
const REGISTRY_MAX: usize = 8;

/// One [`RegistryState`] per registry pool, oldest first.
static STATES: std::sync::RwLock<Vec<(PoolId, Arc<RegistryState>)>> =
    std::sync::RwLock::new(Vec::new());

/// The state for `registry`, created on first use.
fn state(registry: &Pool) -> Arc<RegistryState> {
    let find = |v: &[(PoolId, Arc<RegistryState>)]| {
        v.iter()
            .find(|(id, _)| id.is(registry))
            .map(|(_, s)| s.clone())
    };
    if let Some(s) = find(&STATES.read().unwrap_or_else(PoisonError::into_inner)) {
        return s;
    }
    let mut states = STATES.write().unwrap_or_else(PoisonError::into_inner);
    if let Some(s) = find(&states) {
        return s;
    }
    if states.len() >= REGISTRY_MAX {
        let victim = states
            .iter()
            .position(|(id, _)| id.is_closed())
            .unwrap_or(0);
        states.remove(victim);
    }
    let s = Arc::new(RegistryState::new());
    states.push((PoolId::of(registry), s.clone()));
    s
}

/// Run `f` on every registry's state.
fn each_state(f: impl Fn(&RegistryState)) {
    for (_, s) in STATES.read().unwrap_or_else(PoisonError::into_inner).iter() {
        f(s);
    }
}

/// How long a failed host-table lookup suppresses further attempts.
/// Shares the fingerprint poll's interval, so one env var tunes both.
fn host_table_retry_after() -> std::time::Duration {
    gen_check_every().unwrap_or(GEN_CHECK_EVERY_DEFAULT)
}

/// One TTL and one cap for both caches, so they cannot drift apart the
/// way the two hand-written copies did.
pub(crate) const CACHE_TTL: std::time::Duration = std::time::Duration::from_secs(30);
const CACHE_MAX: usize = 1024;

/// Drop every cached resolution that carries `Org` data — **both**
/// caches.
///
/// Call after any write to `rustango_orgs`, so the pod that made the
/// change sees it immediately rather than waiting out the TTL. Other
/// pods converge via the registry fingerprint — see
/// [`sync_org_generation`].
///
/// Clearing the extra-host cache as well is not incidental. It holds whole
/// `Org` rows now, and it previously re-ran the active-filtered lookup on every
/// hit — so suspending a tenant used to stop its extra hostnames
/// *immediately*, per request. Losing that re-check without clearing
/// here would leave a suspended tenant's extra hosts serving while its
/// base host correctly 404s, which is worse than either behaviour alone.
/// One function so a caller cannot get half of it right.
///
/// Clears the caches of every registry in the process.
pub fn invalidate_org_cache() {
    each_state(RegistryState::invalidate_orgs);
}

/// Drop the base-host cache if another process changed `rustango_orgs`.
///
/// Cheap to call on every resolve: it does nothing at all until the poll
/// interval has passed. Shares [`gen_check_every`] with the host-table
/// poll, so one env var tunes both and a deployment that disables
/// polling disables both.
async fn sync_org_generation(registry: &Pool, st: &RegistryState) {
    st.org_gen
        .sync(
            gen_check_every(),
            || async {
                // Through the breaker, not around it. A direct
                // `org::generation` call would wait out the pool's
                // acquire timeout on an unreachable registry — and
                // base-host traffic reaches this poll on every request,
                // so that would put the stall #1311 removed straight
                // back on the hot path. Declining is not a failure: the
                // breaker is already throttling, and the claim has
                // already backed this poll off for a full interval.
                if registry_is_down(st) {
                    return Ok(None);
                }
                super::org::generation(registry).await.map(Some)
            },
            |e| {
                st.registry_down.open();
                st.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"
                    );
                });
            },
            // Clears the extra-host cache as well, and must: it holds
            // whole `Org` rows too, so an extra hostname pointing at a
            // tenant that has since been suspended has to stop resolving
            // on the same bound as its base host does.
            || st.invalidate_orgs(),
        )
        .await;
}

/// Forget the org fingerprint and the base-host cache of every registry
/// (test hook — see [`crate::testkit`]).
#[cfg(any(test, feature = "testkit"))]
pub(crate) fn reset_org_cache() {
    each_state(|s| {
        s.org_gen.reset();
        s.invalidate_orgs();
    });
}

/// Default interval between fingerprint reads — the bound on how long
/// another pod can serve a stale answer.
const GEN_CHECK_EVERY_DEFAULT: std::time::Duration = std::time::Duration::from_secs(5);

/// Env override for [`GEN_CHECK_EVERY_DEFAULT`], in seconds. `0` disables
/// cross-process polling entirely.
///
/// A hard-coded constant is the wrong shape for this: a single-process
/// deployment gains nothing from the poll, a cross-region registry wants a
/// longer interval, and an operator who needs tighter convergence wants a
/// shorter one. None of them should have to fork the crate, and this runs
/// on the path that must not acquire new ways to fail.
const GEN_CHECK_ENV: &str = "RUSTANGO_HOST_GEN_CHECK_SECS";

/// Resolved once — reading and parsing an env var on every resolve would
/// cost more than the check it is gating.
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),
    })
}

/// Drop the local cache if another process changed the host table.
///
/// Cheap to call on every resolve: it does nothing at all until
/// [`GEN_CHECK_EVERY`] has passed.
async fn sync_generation(registry: &Pool, st: &RegistryState) {
    st.host_gen
        .sync(
            gen_check_every(),
            || async { super::org_host::generation(registry).await.map(Some) },
            |e| {
                // Never fail a request over a cache refresh — but do not
                // fail silently either. Cross-pod invalidation being dead
                // is invisible from the outside until a host 404s on some
                // pods and not others, which is near-impossible to
                // correlate after the fact.
                st.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"
                    );
                });
            },
            // Clears negative entries too, and must: a hostname cached as
            // a known-miss is exactly what an add on another pod
            // invalidates. The cost is that a registry with a steady
            // write stream keeps every pod's negative cache short-lived,
            // which is the deliberate trade — a stale 404 on a real
            // customer domain is worse than a repeated lookup on a
            // sprayed one, and `CACHE_MAX` still bounds the latter.
            || st.host_cache.clear(),
        )
        .await;
}

/// Forget the generation state entirely.
///
/// `pub(crate)` and surfaced through [`crate::testkit`] rather than being
/// `pub` here: as public API of `rustango::tenancy` these would be
/// permanent semver surface that any downstream crate could call in
/// production, silently defeating cross-process invalidation for the life
/// of the process. `#[doc(hidden)]` hides a name from rustdoc, not from
/// the compiler.
#[cfg(any(test, feature = "testkit"))]
pub(crate) fn reset_generation() {
    each_state(|s| {
        s.host_gen.reset();
        // A test that exercised an absent table would otherwise suppress
        // the lookup for its next resolve.
        s.host_table_down.close();
    });
}

/// Make the next resolve re-read the fingerprint immediately instead of
/// waiting out the poll interval, so a cross-pod test proves the bound
/// without sleeping through it.
#[cfg(any(test, feature = "testkit"))]
pub(crate) fn expire_generation() {
    // Back-date past the *effective* interval, not the default one —
    // with a longer interval configured, subtracting the default would
    // leave the entry un-due and make this hook a no-op again.
    let interval = gen_check_every().unwrap_or(GEN_CHECK_EVERY_DEFAULT);
    each_state(|s| s.host_gen.expire(interval * 2, *PROCESS_START));
}

/// Earliest `Instant` this process can name. Used as a saturating floor
/// by [`expire_generation`].
#[cfg(any(test, feature = "testkit"))]
static PROCESS_START: std::sync::LazyLock<std::time::Instant> =
    std::sync::LazyLock::new(std::time::Instant::now);

/// Drop cached resolutions. Called after a host is bound, unbound or
/// toggled so the admin's next request reflects the change instead of
/// waiting out the TTL.
///
/// Clears everything rather than one key: a rename is a remove plus an add,
/// and the map is small and cheap to refill. Clears every registry's cache.
pub fn invalidate_host_cache() {
    each_state(|s| s.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);
        };
        // Pick up a host added or toggled by another pod before trusting
        // anything cached here. Throttled — see `GEN_CHECK_EVERY`.
        let st = state(registry);
        sync_generation(registry, &st).await;
        // Cached, including the miss — see `RegistryState::host_cache`.
        match st.host_cache.get(&host) {
            Cached::Miss => return Ok(None),
            Cached::Hit(org) => return Ok(Some(org)),
            Cached::Absent => {}
        }
        // The table was failing very recently — don't re-ask on every
        // request. See `RegistryState::host_table_down`. Pre-upgrade
        // behaviour is "no match", which is what we return anyway, so
        // backing off costs nothing and stops a per-request storm against
        // a registry that is already in trouble.
        if st.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) => {
                st.host_table_down.close();
                rows
            }
            Err(e) => {
                st.host_table_down.open();
                st.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 {
            st.host_cache.put(&host, None);
            return Ok(None);
        };
        // Resolve the row to its Org once, then cache the Org itself.
        // Caching `row.org_id` instead would make every later hit pay
        // this same fetch — the 2.00-queries-per-request a cache hit
        // used to cost.
        let found = find_active_org_by(registry, &st, Org::id.eq(row.org_id)).await?;
        st.host_cache.put(&host, found.clone());
        Ok(found)
    }
}

// ---------------- PathPrefixResolver ----------------

/// Match the request URL's first path segment against
/// `Org.path_prefix`. Opt-in — not in `ChainResolver::standard()`.
///
/// A request to `app.com/acme/dashboard` matches an org with
/// `path_prefix = "/acme"`. The leading slash is required in the
/// stored value; the resolver builds the candidate as `"/<segment>"`
/// before lookup. Empty path or apex (`"/"`) returns `Ok(None)`.
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, &state(registry), Org::path_prefix.eq(candidate)).await
    }
}

// ---------------- HeaderResolver ----------------

/// Match a user-chosen HTTP header value against `Org.slug`.
///
/// Useful for API-only deployments where every request carries an
/// explicit tenant identifier (`X-Org: acme`). The default header
/// is `X-Org`; configure via [`HeaderResolver::new`].
///
/// # Security
///
/// The header value comes straight from the client. Without an
/// [`Self::allow_only`] allowlist, an attacker who reaches the
/// resolver chain can request *any* tenant by name; defense-in-depth
/// against IDOR therefore depends on every downstream handler
/// checking `Tenant`-scoped authorization. For API-key or
/// JWT-authenticated deployments where the credential is itself
/// tenant-scoped this is fine; for ambient-cookie deployments,
/// always pair this resolver with `allow_only`.
///
/// Not in [`ChainResolver::standard`]: push it yourself (#1856).
#[derive(Clone)]
pub struct HeaderResolver {
    pub header_name: HeaderName,
    allowed_slugs: Option<std::collections::HashSet<String>>,
}

impl HeaderResolver {
    /// Construct with a custom header name (case-insensitive).
    #[must_use]
    pub fn new(header_name: HeaderName) -> Self {
        Self {
            header_name,
            allowed_slugs: None,
        }
    }

    /// Restrict accepted header values to a fixed allowlist of slugs.
    /// Requests whose header value is not in the set resolve to
    /// `None` (the chain falls through to the next resolver, then to
    /// the operator-console fallback if nothing matches). v0.43.
    ///
    /// ```ignore
    /// HeaderResolver::default().allow_only(["acme", "globex"])
    /// ```
    #[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 {
    /// `X-Org` is the recommended default header.
    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);
        }
        // v0.43 — allowlist short-circuit, before the DB lookup.
        if let Some(allowed) = &self.allowed_slugs {
            if !allowed.contains(slug) {
                return Ok(None);
            }
        }
        find_active_org_by(registry, &state(registry), Org::slug.eq(slug.to_owned())).await
    }
}

// ---------------- PortResolver ----------------

/// Match the port the request was accepted on against `Org.port`.
/// Niche — used for hard-isolated tenant ports in compliance /
/// pen-test scenarios. Most deployments don't need this and
/// shouldn't put it in their resolver chain.
///
/// Reads the [`ListenerPort`] request extension, never the URI: an
/// absolute-form request line lets the client pick that port (#1856).
pub struct PortResolver;

/// The local port of the listener that accepted the request.
///
/// `server::Builder::serve` inserts it; when you serve yourself, add
/// `axum::Extension(ListenerPort(port))` per listener.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ListenerPort(pub u16);

#[async_trait]
impl OrgResolver for PortResolver {
    async fn resolve(&self, parts: &Parts, registry: &Pool) -> Result<Option<Org>, TenancyError> {
        let Some(ListenerPort(port)) = parts.extensions.get::<ListenerPort>().copied() else {
            return Ok(None);
        };
        find_active_org_by(registry, &state(registry), Org::port.eq(i32::from(port))).await
    }
}

// ---------------- ChainResolver ----------------

/// Try each resolver in order; first `Ok(Some(_))` wins. `Ok(None)`
/// from one resolver falls through to the next; an `Err` short-
/// circuits (the caller usually surfaces it as 500).
///
/// [`ChainResolver::standard`] is `[SubdomainResolver, RegisteredHostResolver]`.
/// `HeaderResolver` and `PathPrefixResolver` are **not** in it —
/// operators add them explicitly when they need them.
pub struct ChainResolver {
    resolvers: Vec<Box<dyn OrgResolver>>,
}

impl ChainResolver {
    /// Empty chain. Use [`ChainResolver::push`] to add resolvers.
    #[must_use]
    pub fn new() -> Self {
        Self {
            resolvers: Vec::new(),
        }
    }

    /// Append a resolver to the chain. Returns `self` for builder
    /// ergonomics.
    #[must_use]
    pub fn push<R: OrgResolver>(mut self, resolver: R) -> Self {
        self.resolvers.push(Box::new(resolver));
        self
    }

    /// Standard chain: the tenant's host, then its extra registered
    /// hosts. The `apex_domain` is the bare app domain (e.g.
    /// `"app.example.com"`) — `app.com` itself never resolves to a
    /// tenant; only its subdomains do.
    ///
    /// No `X-Org` fallback: a client header must not pick the tenant on
    /// a host nobody registered (#1856). Push a [`HeaderResolver`],
    /// ideally with [`HeaderResolver::allow_only`], to opt in.
    #[must_use]
    pub fn standard(apex_domain: impl Into<String>) -> Self {
        Self::new()
            .push(SubdomainResolver::new(apex_domain))
            // After the base host: an extra hostname must never outrank a
            // tenant's own `host_pattern`, and until an operator registers
            // one the table is empty, so no existing deployment changes
            // behaviour.
            .push(RegisteredHostResolver)
    }
}

/// An **empty** chain, same as [`ChainResolver::new`]: it resolves no
/// tenant. Use [`ChainResolver::standard`] for a working chain.
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) => {
                    // The one funnel every request path goes through —
                    // extractors, tenant admin, operator console. Publish
                    // the identity here and every log line downstream can
                    // name it. See `crate::tenant_log`.
                    crate::tenant_log::record(&org.slug, org.id.get().copied());
                    return Ok(Some(org));
                }
                None => continue,
            }
        }
        Ok(None)
    }
}

// ---------------- helpers ----------------

/// Whether the request's host is the apex, ignoring case like every
/// other host comparison here (#1856).
pub(crate) fn host_is_apex(headers: &http::HeaderMap, uri: &http::Uri, apex: &str) -> bool {
    host_of(headers, uri).is_some_and(|h| h.eq_ignore_ascii_case(apex))
}

/// Pull the host name from the request, lowercased and without a port.
/// Tries the `Host` header, then `uri.host()` (HTTP/2 `:authority`,
/// absolute-form URIs).
///
/// Case-folding is not cosmetic: hostnames are case-insensitive per RFC
/// 4343, but the caches below are keyed on this string, and the string
/// comes straight from a client-supplied `Host` header. Without folding,
/// `ACME.app.test` and `AcMe.app.test` are distinct keys for one tenant —
/// so a few thousand case variants of a real host fill a bounded map with
/// duplicates, evicting genuine entries and taking a write lock on the
/// shared cache each time.
fn host_from_parts(parts: &Parts) -> Option<String> {
    host_of(&parts.headers, &parts.uri)
}

fn host_of(headers: &http::HeaderMap, uri: &http::Uri) -> Option<String> {
    if let Some(value) = headers.get(http::header::HOST) {
        if let Ok(s) = value.to_str() {
            // `Host` header may include `:port` — strip it.
            return crate::urls::split_host_port(s).map(|(h, _)| h.to_ascii_lowercase());
        }
    }
    uri.host().map(str::to_ascii_lowercase)
}

/// How long a recorded failure suppresses further attempts.
///
/// Short on purpose. It caps the damage of an outage without meaningfully
/// delaying recovery: at most one probe per second reaches a registry
/// that is still down, and a registry that has come back is picked up
/// within a second of the next request.
const REGISTRY_RETRY_AFTER: std::time::Duration = std::time::Duration::from_secs(1);

/// True when a recent registry lookup failed and the retry window has
/// not elapsed.
fn registry_is_down(st: &RegistryState) -> bool {
    st.registry_down.is_open(REGISTRY_RETRY_AFTER)
}

/// The error returned while the breaker is open. Mirrors what the pool
/// itself produces so callers cannot tell the two apart.
fn registry_unavailable() -> TenancyError {
    TenancyError::Driver(sqlx::Error::PoolTimedOut)
}

/// Run `Org::objects().where_(filter).where_(active=true)` and
/// return the first match. Helper extracted because every resolver
/// shape ends in this same query.
async fn find_active_org_by<F>(
    registry: &Pool,
    st: &RegistryState,
    filter: F,
) -> Result<Option<Org>, TenancyError>
where
    F: Into<rustango::core::TypedFilter<Org>>,
{
    // A very recent failure means the registry is unreachable; don't
    // queue behind it. See `RegistryState::registry_down`.
    if registry_is_down(st) {
        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))
        // Deterministic if two rows ever match: the oldest org wins (#1931).
        .order_by(&[("id", false)])
        .fetch(registry)
        .await;

    match result {
        Ok(rows) => {
            st.registry_down.close();
            Ok(rows.into_iter().next())
        }
        Err(e) => {
            st.registry_down.open();
            st.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)))
        }
    }
}

/// Close every registry's breaker (test hook — see [`crate::testkit`]).
#[cfg(any(test, feature = "testkit"))]
pub(crate) fn reset_registry_breaker() {
    each_state(|s| s.registry_down.close());
}

/// Serializes lib tests that resolve through this module's globals and
/// starts each from a closed breaker and empty caches (#2067).
#[cfg(all(test, feature = "sqlite"))]
pub(crate) async fn isolated() -> tokio::sync::MutexGuard<'static, ()> {
    static LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
    let guard = LOCK.lock().await;
    reset_registry_breaker();
    reset_org_cache();
    reset_generation();
    invalidate_host_cache();
    guard
}

/// Convert an `ExecError` into the `sqlx::Error` shape `TenancyError`
/// stores. We don't want `TenancyError::Driver` to wrap the full
/// `ExecError` because it carries `QueryError`/`SqlError` shapes
/// that don't apply to resolver lookups; we surface only the
/// driver-level cause.
fn driver_from_exec(e: rustango::sql::ExecError) -> rustango::sql::sqlx::Error {
    use crate::sql::ExecError;
    match e {
        ExecError::Driver(err) => err,
        // Resolver queries are simple `where_(... eq ...)` lookups
        // — Query/Sql shape errors here would be a rustango bug,
        // not a user error. Wrap as a synthetic driver error so
        // callers don't need to match every ExecError variant.
        other => rustango::sql::sqlx::Error::Protocol(format!("resolver query: {other}")),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    // v0.43 — lock in the HeaderResolver allowlist behaviour at the
    // builder level. The full resolve() path requires a live Pool +
    // Org row; that's exercised in the tenant_extractor_*_live tests.
    // These unit tests guarantee the builder semantics so a refactor
    // can't silently drop the allowlist set.

    #[test]
    fn default_has_no_allowlist() {
        let r = HeaderResolver::default();
        assert!(
            r.allowed_slugs.is_none(),
            "default resolver should not have an allowlist"
        );
    }

    /// #1856 — `APP.test` is the apex, not a tenant host to fall through on.
    #[test]
    fn apex_match_ignores_case_and_port() {
        let mut h = http::HeaderMap::new();
        h.insert(http::header::HOST, "APP.Test:8080".parse().unwrap());
        let path: http::Uri = "/".parse().unwrap();
        assert!(host_is_apex(&h, &path, "app.test"));
        assert!(!host_is_apex(&h, &path, "acme.app.test"));
        let none = http::HeaderMap::new();
        assert!(!host_is_apex(&none, &path, "app.test"));
        // HTTP/2 carries the host in `:authority`, not `Host`.
        let h2: http::Uri = "https://app.test/x".parse().unwrap();
        assert!(host_is_apex(&none, &h2, "app.test"));
    }

    /// An IPv6 `Host` keeps its address, not just `[` (#2043).
    #[test]
    fn ipv6_host_keeps_the_address() {
        let mut h = http::HeaderMap::new();
        h.insert(http::header::HOST, "[::1]:8080".parse().unwrap());
        let path: http::Uri = "/".parse().unwrap();
        assert_eq!(host_of(&h, &path).as_deref(), Some("[::1]"));
        assert!(host_is_apex(&h, &path, "[::1]"));
    }

    /// At the cap a closed pool is evicted before the oldest live one.
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn a_full_list_evicts_a_closed_pool_first() {
        let _iso = isolated().await;
        STATES.write().unwrap().clear();
        let mut pools = Vec::new();
        for _ in 0..REGISTRY_MAX {
            let p = Pool::connect("sqlite::memory:").await.unwrap();
            state(&p);
            pools.push(p);
        }
        pools[3].close().await;
        let extra = Pool::connect("sqlite::memory:").await.unwrap();
        state(&extra);
        let kept = |p: &Pool| STATES.read().unwrap().iter().any(|(id, _)| id.is(p));
        assert!(kept(&pools[0]), "the oldest live pool was evicted");
        assert!(!kept(&pools[3]), "the closed pool was kept");
        STATES.write().unwrap().clear();
    }

    #[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() {
        // `Into<String>` bound — verify Vec<String> also works.
        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
        );
    }
}