autumn-web 0.9.0

//! Tenant custom domains with per-domain certificates (issue #1635).
//!
//! A B2B tenant connects its own hostname (`app.clientco.com`) instead of a
//! subdomain of the operator's base domain. This module owns the parts that do
//! not need a CA:
//!
//! - the hostname → tenant [registry](CustomDomainRegistry) and its persistence
//!   ([`CustomDomainStore`]),
//! - the tenant-facing [DNS instructions](DnsInstructions) — CNAME for a
//!   subdomain, A/AAAA for an apex, which cannot carry a CNAME,
//! - the **verification gate** ([`grade_dns_verification`] and
//!   [`grade_ownership`]): no ACME order is created until the hostname is
//!   independently observed to point at this deployment AND its
//!   [`VERIFICATION_LABEL`] TXT record carries this registration's token,
//! - the [issuance budget](IssuanceLimiter) — per-domain and global caps plus
//!   exponential backoff — that keeps a misconfigured tenant from burning the
//!   CA's rate limits.
//!
//! Certificate selection by SNI ([`SniCertResolver`]) is behind the `tls`
//! feature; the ACME orchestration that drives this state machine is in
//! `crate::acme::tenant_domains`, behind the `acme` feature.
//!
//! # State machine
//!
//! ```text
//!   register ──▶ PendingDns ──verified──▶ Verified ──order──▶ Issuing ──▶ Active
//!                    ▲                                            │          │
//!                    └──────────── failure (reason + backoff) ◀───┘          │
//!                                                                            │
//!                              renewal failure keeps Active (still serving) ─┘
//! ```
//!
//! A failure never destroys a working certificate: a domain that is already
//! `Active` stays `Active` with a `failure_reason` set, so one tenant's failed
//! renewal cannot stop it — or anyone else — being served.

// autumn-determinism-gate: production code in this module must read time and
// mint identifiers through the framework's injected seams (ClockSource /
// Entropy), never `Instant::now()` / `Utc::now()` / `SystemTime::now()` /
// `Uuid::new_v4()` directly. See CONTRIBUTING.md "Determinism seam gate"
// (issue #1797). Justify exceptions with
// #[allow(clippy::disallowed_methods, reason = "…")] at the narrowest scope.
#![cfg_attr(not(test), deny(clippy::disallowed_methods))]

use std::collections::HashMap;
use std::future::Future;
use std::io;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
use std::path::PathBuf;
use std::pin::Pin;
use std::sync::{Arc, RwLock};

use serde::{Deserialize, Serialize};

/// Boxed future returned by the async [`CustomDomainStore`] methods.
pub type StoreFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;

/// Longest hostname the DNS wire format allows, minus the root label.
const MAX_HOSTNAME_LEN: usize = 253;
/// Longest single DNS label.
const MAX_LABEL_LEN: usize = 63;

// ── Hostname normalisation ───────────────────────────────────────────────

/// Normalise a hostname to the form the registry, SNI, and ACME all key on:
/// lowercase, no trailing dot, no port.
///
/// DNS names are case-insensitive and SNI arrives lowercase, so normalising at
/// the registration boundary is what makes a `Host:` header, a TLS SNI value,
/// and a stored record compare equal.
///
/// # Errors
///
/// Returns a message naming the problem for anything that cannot be a
/// certificate subject: an empty name, a wildcard (a tenant proves control of
/// one name, not a whole zone), an IP literal, a single-label name, or a label
/// that breaks DNS syntax.
pub fn normalize_hostname(raw: &str) -> Result<String, String> {
    let trimmed = raw.trim().trim_end_matches('.');
    // Strip a port: `Host:` carries one, a certificate subject never does.
    let without_port = trimmed.split(':').next().unwrap_or(trimmed);
    let host = without_port.trim().to_ascii_lowercase();

    if host.is_empty() {
        return Err("hostname must not be empty".to_owned());
    }
    if host.len() > MAX_HOSTNAME_LEN {
        return Err(format!(
            "hostname is {} bytes; DNS allows at most {MAX_HOSTNAME_LEN}",
            host.len()
        ));
    }
    if host.contains('*') {
        return Err(
            "wildcard hostnames are not supported for custom domains: a tenant proves control of \
             one name, not a whole zone"
                .to_owned(),
        );
    }
    if host.parse::<IpAddr>().is_ok() {
        return Err("an IP address cannot be a custom domain".to_owned());
    }
    let labels: Vec<&str> = host.split('.').collect();
    if labels.len() < 2 {
        return Err(format!(
            "{host} is a single label; a custom domain needs at least one dot"
        ));
    }
    for label in &labels {
        if label.is_empty() {
            return Err(format!("{host} has an empty DNS label"));
        }
        if label.len() > MAX_LABEL_LEN {
            return Err(format!("label {label} exceeds {MAX_LABEL_LEN} bytes"));
        }
        if label.starts_with('-') || label.ends_with('-') {
            return Err(format!("label {label} must not start or end with '-'"));
        }
        if !label
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || b == b'-')
        {
            return Err(format!(
                "label {label} may contain only letters, digits and '-' (punycode an \
                 internationalised name before registering it)"
            ));
        }
    }
    Ok(host)
}

/// Returns true if `hostname` is the top name of a DNS zone. Such a name
/// cannot have a CNAME.
///
/// Uses the Public Suffix List (ICANN and private sections), so
/// `clientco.co.uk` is an apex (#2643, ADR 0026). A public suffix itself is
/// also an apex. A TLD that is not on the list uses the two-label rule.
///
/// This function does not find a delegated sub-zone. [`grade_dns_verification`]
/// accepts A/AAAA records too, so such a tenant can still verify.
#[must_use]
pub fn is_apex(hostname: &str) -> bool {
    let host = hostname.trim_end_matches('.').to_ascii_lowercase();
    psl::domain(host.as_bytes()).is_none_or(|domain| domain.as_bytes() == host.as_bytes())
}

// ── Tenant-facing DNS instructions ───────────────────────────────────────

/// Where this deployment's ingress lives, as the tenant must point DNS at it.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ExpectedIngress {
    /// The operator's ingress hostname, the CNAME target for subdomains.
    pub hostname: Option<String>,
    /// The ingress IPv4 addresses, for apex domains.
    pub ipv4: Vec<Ipv4Addr>,
    /// The ingress IPv6 addresses, for apex domains.
    pub ipv6: Vec<Ipv6Addr>,
}

impl ExpectedIngress {
    /// Is there anything a tenant could be told to point at?
    #[must_use]
    pub const fn is_empty(&self) -> bool {
        self.hostname.is_none() && self.ipv4.is_empty() && self.ipv6.is_empty()
    }

    /// The normalised CNAME target, if one is configured.
    #[must_use]
    fn cname_target(&self) -> Option<String> {
        self.hostname
            .as_ref()
            .map(|h| h.trim().trim_end_matches('.').to_ascii_lowercase())
    }
}

/// The record that routes a custom domain to this deployment.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum RoutingRecord {
    /// A subdomain: one CNAME at the operator's ingress hostname.
    Cname {
        /// The record name the tenant creates.
        name: String,
        /// The CNAME target.
        value: String,
    },
    /// An apex: A and/or AAAA records, because RFC 1034 forbids a CNAME
    /// alongside the SOA/NS records an apex must carry.
    Address {
        /// The record name the tenant creates.
        name: String,
        /// A-record values.
        ipv4: Vec<String>,
        /// AAAA-record values.
        ipv6: Vec<String>,
    },
}

impl RoutingRecord {
    /// Build the routing record for `hostname` against this deployment's
    /// ingress.
    ///
    /// # Errors
    ///
    /// Returns a message when the hostname is invalid, or when the ingress
    /// carries nothing of the kind this hostname needs (no CNAME target for a
    /// subdomain, no addresses for an apex).
    pub fn for_hostname(hostname: &str, ingress: &ExpectedIngress) -> Result<Self, String> {
        let host = normalize_hostname(hostname)?;
        if is_apex(&host) {
            if ingress.ipv4.is_empty() && ingress.ipv6.is_empty() {
                return Err(format!(
                    "{host} is an apex domain, which cannot carry a CNAME, but no ingress \
                     addresses are configured; set [server.tls.acme.custom_domains] ingress_ipv4 \
                     / ingress_ipv6"
                ));
            }
            return Ok(Self::Address {
                name: host,
                ipv4: ingress.ipv4.iter().map(ToString::to_string).collect(),
                ipv6: ingress.ipv6.iter().map(ToString::to_string).collect(),
            });
        }
        let value = ingress.cname_target().ok_or_else(|| {
            format!(
                "{host} needs a CNAME target, but no ingress hostname is configured; set \
                 [server.tls.acme.custom_domains] ingress_hostname"
            )
        })?;
        Ok(Self::Cname { name: host, value })
    }

    /// The record as tab-separated lines a tenant-facing screen can print.
    #[must_use]
    pub fn render(&self) -> String {
        match self {
            Self::Cname { name, value } => format!("{name}\tCNAME\t{value}"),
            Self::Address { name, ipv4, ipv6 } => {
                use std::fmt::Write as _;
                let mut out = String::new();
                for (kind, addrs) in [("A", ipv4), ("AAAA", ipv6)] {
                    for addr in addrs {
                        let _ = writeln!(out, "{name}\t{kind}\t{addr}");
                    }
                }
                out.trim_end().to_owned()
            }
        }
    }
}

/// The TXT record that proves a tenant controls a custom domain (#2642).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct VerificationRecord {
    /// The record name: [`VERIFICATION_LABEL`] under the hostname.
    pub name: String,
    /// The TXT value: this registration's token.
    pub value: String,
}

/// Every DNS record a tenant must publish for one custom domain.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DnsInstructions {
    /// The record that points the hostname at this deployment.
    pub routing: RoutingRecord,
    /// The TXT record that proves the tenant controls the hostname.
    ///
    /// `None` only for a grandfathered record (see
    /// [`CustomDomain::verification_token`]), which needs no proof.
    pub verification: Option<VerificationRecord>,
}

impl DnsInstructions {
    /// Build the instructions for one registered domain.
    ///
    /// # Errors
    ///
    /// As [`RoutingRecord::for_hostname`].
    pub fn for_domain(domain: &CustomDomain, ingress: &ExpectedIngress) -> Result<Self, String> {
        Ok(Self {
            routing: RoutingRecord::for_hostname(&domain.hostname, ingress)?,
            verification: domain
                .verification_token
                .as_ref()
                .map(|token| VerificationRecord {
                    name: verification_record_name(&domain.hostname),
                    value: token.clone(),
                }),
        })
    }

    /// The instructions as tab-separated lines a tenant-facing screen can
    /// print verbatim: the routing record(s), then the TXT record.
    #[must_use]
    pub fn render(&self) -> String {
        use std::fmt::Write as _;
        let mut out = self.routing.render();
        if let Some(record) = &self.verification {
            let _ = write!(out, "\n{}\tTXT\t{}", record.name, record.value);
        }
        out
    }
}

// ── Ownership token ──────────────────────────────────────────────────────

/// The label under a custom domain that holds its ownership TXT record.
pub const VERIFICATION_LABEL: &str = "_autumn-challenge";

/// The TXT record name that proves ownership of `hostname`.
#[must_use]
pub fn verification_record_name(hostname: &str) -> String {
    format!("{VERIFICATION_LABEL}.{hostname}")
}

/// A fresh ownership token: 128 random bits as 32 hex digits.
fn mint_verification_token() -> io::Result<String> {
    use std::fmt::Write as _;
    let mut bytes = [0_u8; 16];
    getrandom::fill(&mut bytes)
        .map_err(|e| io::Error::other(format!("the OS random source failed: {e}")))?;
    let mut out = String::with_capacity(32);
    for byte in bytes {
        let _ = write!(out, "{byte:02x}");
    }
    Ok(out)
}

// ── Verification ─────────────────────────────────────────────────────────

/// What a DNS lookup actually observed for a custom domain.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ObservedTarget {
    /// The name is a CNAME pointing at this value.
    Cname(String),
    /// The name resolves to these addresses.
    Addresses(Vec<IpAddr>),
    /// The name does not resolve.
    None,
}

/// Whether a custom domain points at this deployment.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum VerificationOutcome {
    /// Confirmed: the ACME order may proceed.
    PointsHere,
    /// The name resolves, but not here. Naming what was observed is what makes
    /// a stuck domain self-service for the tenant.
    PointsElsewhere {
        /// What was seen, for the status surface.
        detail: String,
    },
    /// The name does not resolve yet — the ordinary state right after a tenant
    /// is handed the instructions.
    Unresolved,
    /// The name points here, but its TXT record does not carry this
    /// registration's token. A hostname that points here proves nothing on its
    /// own: another tenant's dangling record also points here (#2642).
    OwnershipUnproven {
        /// The TXT record name the token must be published at.
        record: String,
        /// What was seen, for the status surface.
        detail: String,
    },
}

impl VerificationOutcome {
    /// Did verification confirm the hostname points here?
    #[must_use]
    pub const fn is_verified(&self) -> bool {
        matches!(self, Self::PointsHere)
    }

    /// The reason to show a tenant whose domain is stuck.
    #[must_use]
    pub fn reason(&self) -> Option<String> {
        match self {
            Self::PointsHere => None,
            Self::PointsElsewhere { detail } => Some(format!(
                "DNS does not point at this deployment yet: {detail}"
            )),
            Self::Unresolved => Some(
                "DNS does not resolve yet. Publish the record above; propagation can take up to \
                 the record's TTL."
                    .to_owned(),
            ),
            Self::OwnershipUnproven { record, detail } => Some(format!(
                "DNS points here, but the TXT record {record} does not carry this registration's \
                 token yet: {detail}. Publish the TXT record above."
            )),
        }
    }
}

/// What a TXT lookup observed at a domain's [`verification_record_name`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ObservedTxt {
    /// The TXT values published at the name. Empty when there are none.
    Values(Vec<String>),
    /// No resolver answered, so nothing is known.
    Unanswered(String),
}

/// Grade a TXT observation against `domain`'s ownership token (pure).
///
/// Passes only when one of the observed values is exactly this registration's
/// token. Each registration mints a new token, so a record left by an earlier
/// tenant proves nothing for the next one. A record without a token fails:
/// only an `active` record is grandfathered, and that grant is applied by the
/// caller, not here.
#[must_use]
pub fn grade_ownership(domain: &CustomDomain, observed: &ObservedTxt) -> VerificationOutcome {
    let record = verification_record_name(&domain.hostname);
    let unproven = |detail: String| VerificationOutcome::OwnershipUnproven {
        record: record.clone(),
        detail,
    };
    let Some(token) = domain.verification_token.as_deref() else {
        return unproven("this registration has no token yet".to_owned());
    };
    match observed {
        ObservedTxt::Values(values) if values.iter().any(|v| v.trim() == token) => {
            VerificationOutcome::PointsHere
        }
        ObservedTxt::Values(values) if values.is_empty() => {
            unproven("no TXT record is published".to_owned())
        }
        ObservedTxt::Values(values) => unproven(format!(
            "it carries {} other value(s), possibly a token from an earlier registration",
            values.len()
        )),
        ObservedTxt::Unanswered(error) => unproven(format!("the TXT lookup failed ({error})")),
    }
}

/// Grade an observation against the configured ingress (pure; injectable).
///
/// A CNAME matches when it names the ingress hostname (trailing dot and case
/// ignored). Addresses match when **every** resolved address is a configured
/// ingress address: a partial match still lets a request — and the CA's
/// HTTP-01 fetch — land on a host that serves neither the token nor the
/// tenant, so it is not a pass.
#[must_use]
pub fn grade_dns_verification(
    observed: &ObservedTarget,
    expected: &ExpectedIngress,
) -> VerificationOutcome {
    match observed {
        ObservedTarget::None => VerificationOutcome::Unresolved,
        ObservedTarget::Cname(target) => {
            let target_norm = target.trim().trim_end_matches('.').to_ascii_lowercase();
            if expected.cname_target().as_deref() == Some(target_norm.as_str()) {
                VerificationOutcome::PointsHere
            } else {
                VerificationOutcome::PointsElsewhere {
                    detail: format!("CNAME points at {target_norm}"),
                }
            }
        }
        ObservedTarget::Addresses(addrs) if addrs.is_empty() => VerificationOutcome::Unresolved,
        // Nothing to compare against: an ingress with no addresses (and whose
        // hostname could not be resolved to any) cannot confirm anything, and
        // an empty expected set must never read as "every address matches".
        ObservedTarget::Addresses(_) if expected.ipv4.is_empty() && expected.ipv6.is_empty() => {
            VerificationOutcome::PointsElsewhere {
                detail: "this deployment's ingress addresses are unknown, so the hostname cannot \
                         be confirmed to point here"
                    .to_owned(),
            }
        }
        ObservedTarget::Addresses(addrs) => {
            let unmatched: Vec<String> = addrs
                .iter()
                .filter(|addr| !ingress_contains(expected, **addr))
                .map(ToString::to_string)
                .collect();
            if unmatched.is_empty() {
                VerificationOutcome::PointsHere
            } else {
                VerificationOutcome::PointsElsewhere {
                    detail: format!("resolves to {}", unmatched.join(", ")),
                }
            }
        }
    }
}

fn ingress_contains(expected: &ExpectedIngress, addr: IpAddr) -> bool {
    match addr {
        IpAddr::V4(v4) => expected.ipv4.contains(&v4),
        IpAddr::V6(v6) => expected.ipv6.contains(&v6),
    }
}

/// Apply a verification result to the registration `domain` was read from.
///
/// Success promotes `PendingDns` → `Verified`, which is the only thing that
/// makes a domain eligible for an ACME order. Failure records the reason and a
/// backoff so a tenant who has not published the record yet is retried at a
/// decaying rate rather than every tick.
///
/// Both writes apply only while the stored record is still the SAME
/// registration: same tenant, same token. A lookup awaits, and if the hostname
/// was offboarded and registered again meanwhile, a result graded against the
/// old token must not promote the new registration. Returns whether it
/// applied.
///
/// # Errors
///
/// Propagates the store's write error.
pub async fn apply_verification(
    registry: &CustomDomainRegistry,
    domain: &CustomDomain,
    outcome: &VerificationOutcome,
    now_unix: i64,
    backoff_secs: i64,
) -> io::Result<bool> {
    if outcome.is_verified() {
        registry
            .record_verified_for(
                &domain.hostname,
                &domain.tenant,
                domain.verification_token.as_deref(),
                now_unix,
            )
            .await
    } else {
        let reason = outcome
            .reason()
            .unwrap_or_else(|| "verification failed".to_owned());
        let token = domain.verification_token.clone();
        registry
            .fail(
                &domain.hostname,
                |d| d.tenant == domain.tenant && d.verification_token == token,
                now_unix,
                reason,
                backoff_secs,
            )
            .await
    }
}

// ── The domain record ────────────────────────────────────────────────────

/// Where a custom domain is in its lifecycle.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DomainStatus {
    /// Registered; DNS has not been confirmed to point here, or the ownership
    /// TXT record has not been seen.
    PendingDns,
    /// DNS and ownership confirmed; eligible for an ACME order.
    Verified,
    /// An ACME order is in flight.
    Issuing,
    /// A certificate is stored and served.
    Active,
}

impl DomainStatus {
    /// The stable string an app renders and a status API returns.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::PendingDns => "pending_dns",
            Self::Verified => "verified",
            Self::Issuing => "issuing",
            Self::Active => "active",
        }
    }
}

impl std::fmt::Display for DomainStatus {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.as_str())
    }
}

/// One tenant-connected hostname and everything the status surface renders.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CustomDomain {
    /// Normalised hostname (see [`normalize_hostname`]).
    pub hostname: String,
    /// The tenant id requests on this hostname resolve to.
    pub tenant: String,
    /// Current lifecycle state.
    pub status: DomainStatus,
    /// Why the domain is stuck, if it is. Cleared on the next success.
    pub failure_reason: Option<String>,
    /// When the app registered the hostname.
    pub registered_at_unix: i64,
    /// When DNS was first confirmed.
    pub verified_at_unix: Option<i64>,
    /// When the first certificate started being served.
    pub activated_at_unix: Option<i64>,
    /// `notAfter` of the served certificate, for renewal scheduling.
    pub cert_not_after_unix: Option<i64>,
    /// Consecutive failures, the exponent in [`backoff_secs`].
    pub consecutive_failures: u32,
    /// Earliest time the orchestrator may retry this domain.
    pub next_attempt_unix: Option<i64>,
    /// This registration's ownership token, which the tenant publishes as TXT
    /// at [`verification_record_name`]. Each registration mints a new one.
    ///
    /// `None` only on a record stored before tokens existed. An `active` one is
    /// grandfathered: it proved control under the old rule, so it keeps
    /// serving and renewing. Any other one is given a token at the next load.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub verification_token: Option<String>,
    /// When this registration's certificate orders were placed, newest last,
    /// kept for the per-domain budget window (24 hours). Persisted so the issuance budget
    /// survives a restart: the limiter itself is in memory, and a crash loop
    /// would otherwise start every window empty and spend the shared ACME
    /// account's quota again. [`IssuanceLimiter::hydrate`] reads it back.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub issuance_attempts_unix: Vec<i64>,
}

impl CustomDomain {
    /// A freshly registered domain, awaiting DNS.
    #[must_use]
    const fn new(hostname: String, tenant: String, now_unix: i64, token: String) -> Self {
        Self {
            hostname,
            tenant,
            status: DomainStatus::PendingDns,
            failure_reason: None,
            registered_at_unix: now_unix,
            verified_at_unix: None,
            activated_at_unix: None,
            cert_not_after_unix: None,
            consecutive_failures: 0,
            next_attempt_unix: None,
            verification_token: Some(token),
            issuance_attempts_unix: Vec::new(),
        }
    }

    /// The TXT record name that proves ownership of this domain.
    #[must_use]
    pub fn verification_record_name(&self) -> String {
        verification_record_name(&self.hostname)
    }

    /// Is this a record from before ownership tokens that still needs one?
    ///
    /// An `active` record without a token is grandfathered and does not.
    #[must_use]
    pub fn needs_verification_token(&self) -> bool {
        self.verification_token.is_none() && self.status != DomainStatus::Active
    }

    /// May this domain be worked on at `now_unix`, or is it still in backoff?
    #[must_use]
    pub fn is_due(&self, now_unix: i64) -> bool {
        self.next_attempt_unix.is_none_or(|at| now_unix >= at)
    }

    /// Does a request or handshake for this hostname get served?
    ///
    /// Only `Active`: a domain whose certificate has not been issued has
    /// nothing to serve, and routing it would leak the existence of a tenant
    /// to whoever pointed DNS here first.
    #[must_use]
    pub const fn is_servable(&self) -> bool {
        matches!(self.status, DomainStatus::Active)
    }
}

// ── Persistence ──────────────────────────────────────────────────────────

/// Durable storage for the domain registry.
///
/// Records must survive a restart: a lost registry silently stops routing
/// every tenant's domain and re-orders every certificate.
pub trait CustomDomainStore: Send + Sync + std::fmt::Debug {
    /// Every stored record. Called once at boot to hydrate the index.
    fn load_all(&self) -> StoreFuture<'_, io::Result<Vec<CustomDomain>>>;
    /// Insert or replace one record.
    fn save<'a>(&'a self, domain: &'a CustomDomain) -> StoreFuture<'a, io::Result<()>>;
    /// Delete one record. Deleting an absent record succeeds.
    fn delete<'a>(&'a self, hostname: &'a str) -> StoreFuture<'a, io::Result<()>>;

    /// The global attempt ledger: the instant of each recent ACME order.
    ///
    /// The ledger needs no domain record. An order for an offboarded domain
    /// still counts after a restart. Called once at boot.
    ///
    /// Override this, [`append_attempt`](Self::append_attempt) and
    /// [`merge_attempts`](Self::merge_attempts) together or not at all. The
    /// defaults keep nothing, so offboarded orders are lost on restart. A
    /// ledger is for one process. Replicas do not share it.
    fn load_attempts(&self) -> StoreFuture<'_, io::Result<Vec<i64>>> {
        Box::pin(async { Ok(Vec::new()) })
    }

    /// Add an order at `at_unix` to the ledger. Drop entries that are not in
    /// the global window ending at `at_unix`. The default does nothing.
    fn append_attempt(&self, at_unix: i64) -> StoreFuture<'_, io::Result<()>> {
        let _ = at_unix;
        Box::pin(async { Ok(()) })
    }

    /// Add orders that the ledger lacks. Used at boot to copy orders from the
    /// domain records, so an upgrade loses none. The ledger keeps the larger
    /// count at each instant. Entries outside the global window that ends at
    /// `now_unix` are dropped. The default does nothing.
    fn merge_attempts(&self, attempts: Vec<i64>, now_unix: i64) -> StoreFuture<'_, io::Result<()>> {
        let _ = (attempts, now_unix);
        Box::pin(async { Ok(()) })
    }
}

/// Keep the entries in the global window that ends at `now_unix`.
/// An entry in the future is a clock error. Drop it, or it never expires.
fn prune_ledger(ledger: &mut Vec<i64>, now_unix: i64) {
    let cutoff = now_unix.saturating_sub(GLOBAL_WINDOW_SECS);
    ledger.retain(|at| *at > cutoff && *at <= now_unix);
}

/// Raise the count at each instant in `ledger` to its count in `extra`.
/// Prune first. Return whether the ledger changed.
fn merge_into_ledger(ledger: &mut Vec<i64>, extra: &[i64], now_unix: i64) -> bool {
    let before = ledger.len();
    prune_ledger(ledger, now_unix);
    let mut changed = ledger.len() != before;
    let mut want: HashMap<i64, usize> = HashMap::new();
    for at in extra {
        *want.entry(*at).or_default() += 1;
    }
    let mut have: HashMap<i64, usize> = HashMap::new();
    for at in ledger.iter() {
        *have.entry(*at).or_default() += 1;
    }
    for (at, n) in want {
        let missing = n.saturating_sub(have.get(&at).copied().unwrap_or(0));
        if missing > 0 && at > now_unix.saturating_sub(GLOBAL_WINDOW_SECS) && at <= now_unix {
            ledger.extend(std::iter::repeat_n(at, missing));
            changed = true;
        }
    }
    changed
}

/// Ledger file name. Its extension is not `json`, so `load_all` skips it.
const LEDGER_FILE: &str = "attempts.ledger";

/// In-memory [`CustomDomainStore`], for tests and for a deployment that
/// re-registers its domains from the app's own database at boot.
#[derive(Debug, Default)]
pub struct MemoryCustomDomainStore {
    records: RwLock<HashMap<String, CustomDomain>>,
    ledger: RwLock<Vec<i64>>,
}

impl MemoryCustomDomainStore {
    /// An empty store.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Every stored record, without an await — for assertions.
    #[cfg(any(test, feature = "test-support"))]
    #[must_use]
    pub fn load_all_blocking(&self) -> Vec<CustomDomain> {
        read_lock(&self.records).values().cloned().collect()
    }
}

impl CustomDomainStore for MemoryCustomDomainStore {
    fn load_all(&self) -> StoreFuture<'_, io::Result<Vec<CustomDomain>>> {
        Box::pin(async move { Ok(read_lock(&self.records).values().cloned().collect()) })
    }

    fn save<'a>(&'a self, domain: &'a CustomDomain) -> StoreFuture<'a, io::Result<()>> {
        Box::pin(async move {
            write_lock(&self.records).insert(domain.hostname.clone(), domain.clone());
            Ok(())
        })
    }

    fn delete<'a>(&'a self, hostname: &'a str) -> StoreFuture<'a, io::Result<()>> {
        Box::pin(async move {
            write_lock(&self.records).remove(hostname);
            Ok(())
        })
    }

    fn load_attempts(&self) -> StoreFuture<'_, io::Result<Vec<i64>>> {
        Box::pin(async move { Ok(read_lock(&self.ledger).clone()) })
    }

    fn merge_attempts(&self, attempts: Vec<i64>, now_unix: i64) -> StoreFuture<'_, io::Result<()>> {
        Box::pin(async move {
            merge_into_ledger(&mut write_lock(&self.ledger), &attempts, now_unix);
            Ok(())
        })
    }

    fn append_attempt(&self, at_unix: i64) -> StoreFuture<'_, io::Result<()>> {
        Box::pin(async move {
            let mut ledger = write_lock(&self.ledger);
            prune_ledger(&mut ledger, at_unix);
            ledger.push(at_unix);
            drop(ledger);
            Ok(())
        })
    }
}

/// Filesystem [`CustomDomainStore`]: one `0600` JSON file per domain under a
/// `0700` directory, mirroring `crate::acme::store::FsAcmeStore`.
///
/// A file per domain (rather than one index file) keeps a 1,000-domain
/// deployment's writes independent: registering or offboarding one tenant
/// rewrites one small file instead of the whole registry.
#[derive(Debug, Clone)]
pub struct FsCustomDomainStore {
    dir: PathBuf,
    /// Serialises ledger read-modify-write cycles.
    ledger_lock: Arc<tokio::sync::Mutex<()>>,
}

impl FsCustomDomainStore {
    /// A store rooted at `dir`.
    #[must_use]
    pub fn new(dir: impl Into<PathBuf>) -> Self {
        Self {
            dir: dir.into(),
            ledger_lock: Arc::new(tokio::sync::Mutex::new(())),
        }
    }

    /// The file holding `hostname`'s record.
    ///
    /// The hostname is hashed rather than used verbatim so a name that is
    /// legal in DNS but awkward on a filesystem cannot escape `dir`.
    fn path_for(&self, hostname: &str) -> PathBuf {
        self.dir.join(format!("{}.json", file_stem(hostname)))
    }
}

/// A short, filesystem-safe digest of `hostname`.
fn file_stem(hostname: &str) -> String {
    use sha2::{Digest as _, Sha256};
    let digest = Sha256::digest(hostname.as_bytes());
    let mut out = String::with_capacity(32);
    for byte in &digest[..16] {
        use std::fmt::Write as _;
        let _ = write!(out, "{byte:02x}");
    }
    out
}

impl CustomDomainStore for FsCustomDomainStore {
    fn load_all(&self) -> StoreFuture<'_, io::Result<Vec<CustomDomain>>> {
        Box::pin(async move {
            let mut entries = match tokio::fs::read_dir(&self.dir).await {
                Ok(entries) => entries,
                Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(Vec::new()),
                Err(e) => return Err(e),
            };
            let mut out = Vec::new();
            while let Some(entry) = entries.next_entry().await? {
                let path = entry.path();
                if path.extension().is_none_or(|ext| ext != "json") {
                    continue;
                }
                let bytes = tokio::fs::read(&path).await?;
                match serde_json::from_slice::<CustomDomain>(&bytes) {
                    Ok(domain) => out.push(domain),
                    // A corrupt record must not stop the other 999 domains from
                    // being served; it is logged and skipped, and the app can
                    // re-register the hostname to repair it.
                    Err(e) => tracing::warn!(
                        path = %path.display(),
                        "skipping unreadable custom-domain record: {e}"
                    ),
                }
            }
            Ok(out)
        })
    }

    fn save<'a>(&'a self, domain: &'a CustomDomain) -> StoreFuture<'a, io::Result<()>> {
        let dir = self.dir.clone();
        let path = self.path_for(&domain.hostname);
        Box::pin(async move {
            let bytes = serde_json::to_vec_pretty(domain)
                .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
            crate::fs_atomic::blocking(move || {
                crate::fs_atomic::ensure_owner_only_dir(&dir)?;
                crate::fs_atomic::write_owner_only(&path, &bytes)
            })
            .await
        })
    }

    fn delete<'a>(&'a self, hostname: &'a str) -> StoreFuture<'a, io::Result<()>> {
        Box::pin(async move {
            match tokio::fs::remove_file(self.path_for(hostname)).await {
                Ok(()) => Ok(()),
                Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()),
                Err(e) => Err(e),
            }
        })
    }

    fn load_attempts(&self) -> StoreFuture<'_, io::Result<Vec<i64>>> {
        Box::pin(async move { read_ledger(&self.dir.join(LEDGER_FILE)).await })
    }

    fn append_attempt(&self, at_unix: i64) -> StoreFuture<'_, io::Result<()>> {
        self.update_ledger(move |ledger| {
            prune_ledger(ledger, at_unix);
            ledger.push(at_unix);
            true
        })
    }

    fn merge_attempts(&self, attempts: Vec<i64>, now_unix: i64) -> StoreFuture<'_, io::Result<()>> {
        self.update_ledger(move |ledger| merge_into_ledger(ledger, &attempts, now_unix))
    }
}

impl FsCustomDomainStore {
    /// Read the ledger, change it, and write it back if `change` says so.
    /// One lock covers the whole cycle.
    fn update_ledger(
        &self,
        change: impl FnOnce(&mut Vec<i64>) -> bool + Send + 'static,
    ) -> StoreFuture<'_, io::Result<()>> {
        let dir = self.dir.clone();
        let path = self.dir.join(LEDGER_FILE);
        let lock = Arc::clone(&self.ledger_lock);
        Box::pin(async move {
            let _guard = lock.lock().await;
            let (mut ledger, corrupt) = read_ledger_checked(&path).await?;
            if corrupt {
                // Keep the bad file for the operator. Do not erase it.
                let _ = tokio::fs::rename(&path, path.with_extension("ledger.corrupt")).await;
            }
            if !change(&mut ledger) {
                return Ok(());
            }
            let bytes = serde_json::to_vec(&ledger)
                .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
            crate::fs_atomic::blocking(move || {
                crate::fs_atomic::ensure_owner_only_dir(&dir)?;
                crate::fs_atomic::write_owner_only(&path, &bytes)
            })
            .await
        })
    }
}

/// Read the ledger file. A missing or corrupt file reads as empty.
async fn read_ledger(path: &std::path::Path) -> io::Result<Vec<i64>> {
    let (ledger, corrupt) = read_ledger_checked(path).await?;
    if corrupt {
        tracing::warn!(path = %path.display(), "ignoring an unreadable attempt ledger");
    }
    Ok(ledger)
}

/// Read the ledger file. Also say whether the file was corrupt.
async fn read_ledger_checked(path: &std::path::Path) -> io::Result<(Vec<i64>, bool)> {
    match tokio::fs::read(path).await {
        Ok(bytes) => Ok(serde_json::from_slice(&bytes).map_or((Vec::new(), true), |l| (l, false))),
        Err(e) if e.kind() == io::ErrorKind::NotFound => Ok((Vec::new(), false)),
        Err(e) => Err(e),
    }
}

// ── The registry ─────────────────────────────────────────────────────────

/// Why a registration was refused.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RegisterError {
    /// The hostname is not a usable certificate subject.
    Invalid(String),
    /// The deployment's `max_domains` cap is reached.
    LimitReached {
        /// The configured cap.
        max: usize,
    },
    /// Another tenant already owns this hostname.
    Conflict {
        /// The tenant that owns it.
        tenant: String,
    },
    /// The deployment already owns this hostname, so no tenant may claim it.
    Reserved {
        /// The reserved pattern the hostname matched.
        pattern: String,
    },
    /// The record could not be persisted.
    Store(String),
    /// The registry has not hydrated from its store, so it cannot tell whether
    /// another tenant already owns the hostname.
    NotReady,
}

impl std::fmt::Display for RegisterError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Invalid(msg) => write!(f, "invalid custom domain: {msg}"),
            Self::LimitReached { max } => write!(
                f,
                "custom-domain limit reached ({max}); raise \
                 [server.tls.acme.custom_domains] max_domains"
            ),
            Self::Conflict { tenant } => {
                write!(f, "hostname is already connected to tenant {tenant}")
            }
            Self::Reserved { pattern } => write!(
                f,
                "hostname is reserved by this deployment ({pattern}); a tenant cannot connect a \
                 name the operator already serves"
            ),
            Self::Store(msg) => write!(f, "failed to persist the custom domain: {msg}"),
            Self::NotReady => write!(
                f,
                "the custom-domain registry has not loaded, so a hostname cannot be connected \
                 without risking another tenant's record; check the server log for the load error"
            ),
        }
    }
}

impl std::error::Error for RegisterError {}

/// The hostname → tenant registry: an in-memory index over a durable store.
///
/// Reads (the ones on the request and handshake paths) take a read lock on a
/// `HashMap` and never touch the store, so 1,000 domains cost one hash lookup
/// per request. Writes go to the store first, then the index, so a persisted
/// registry never disagrees with what is being served.
#[derive(Debug)]
pub struct CustomDomainRegistry {
    store: Arc<dyn CustomDomainStore>,
    index: RwLock<HashMap<String, CustomDomain>>,
    /// Serialises the STORE half of every write, per hostname.
    ///
    /// Each write drops the index lock before awaiting the store — holding a
    /// `std` lock across an await is not an option — so without this two
    /// writers for the same hostname can commit out of order. The damaging
    /// order is a delete overtaken by an in-flight save: the index entry is
    /// gone, but the file is recreated, and the next restart hydrates a domain
    /// the app offboarded, which can then be re-issued and served.
    ///
    /// Keyed by hostname rather than global so a slow write for one tenant
    /// cannot stall every other tenant's.
    writes: tokio::sync::Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
    max_domains: usize,
    /// Hostnames and wildcard patterns no tenant may claim.
    reserved: Vec<String>,
    /// Whether [`load`](Self::load) has completed successfully.
    ///
    /// The retention prune deletes certificates for hostnames the registry
    /// does not know, so it must never run over an index that failed to
    /// hydrate: that index knows nothing, and every tenant certificate would
    /// read as an orphan.
    hydrated: std::sync::atomic::AtomicBool,
}

impl CustomDomainRegistry {
    /// A registry over `store`, capped at `max_domains`.
    #[must_use]
    pub fn new(store: Arc<dyn CustomDomainStore>, max_domains: usize) -> Self {
        Self {
            store,
            index: RwLock::new(HashMap::new()),
            writes: tokio::sync::Mutex::new(HashMap::new()),
            max_domains,
            reserved: Vec::new(),
            hydrated: std::sync::atomic::AtomicBool::new(false),
        }
    }

    /// Refuse registration of these hostnames and wildcard patterns.
    ///
    /// The deployment's own certificate names and tenancy base domain belong
    /// here. Without them a tenant can connect another tenant's subdomain —
    /// which the operator's own wildcard already points at this deployment, so
    /// it verifies and issues — and from then on every request for that host
    /// resolves to the tenant that registered it.
    ///
    /// A pattern reserves the name itself AND every name below it, wildcard or
    /// not: `myapp.com` and `*.myapp.com` each reserve `myapp.com` and
    /// `acme.myapp.com` alike.
    ///
    /// A wildcard covering its own apex is deliberate, and deliberately wider
    /// than the certificate: `*.myapp.com` does not cover `myapp.com` in TLS,
    /// but an operator who put that pattern here owns the zone, and letting a
    /// tenant claim the apex under it would hand away the operator's own name.
    /// Over-reserving costs an operator nothing — no tenant has a legitimate
    /// claim on a host the deployment already answers for — while
    /// under-reserving is a tenant-isolation hole.
    #[must_use]
    pub fn with_reserved(mut self, patterns: impl IntoIterator<Item = String>) -> Self {
        self.reserved = patterns
            .into_iter()
            .map(|p| p.trim().trim_end_matches('.').to_ascii_lowercase())
            .filter(|p| !p.is_empty())
            .collect();
        self
    }

    /// The reserved pattern `host` matches, if any.
    fn reserved_match(&self, host: &str) -> Option<String> {
        self.reserved
            .iter()
            .find(|pattern| {
                pattern.strip_prefix("*.").map_or_else(
                    // A bare name reserves itself AND everything under it: an
                    // operator listing `myapp.com` means the whole zone, not
                    // just the apex.
                    || host == *pattern || host.ends_with(&format!(".{pattern}")),
                    // `host == suffix` is intentional: a wildcard reserves its
                    // own apex too. Wider than the certificate `*.myapp.com`
                    // covers, and that is the point — the operator owns the
                    // zone, so a tenant must not be able to claim `myapp.com`
                    // just because only the wildcard was listed.
                    |suffix| host == suffix || host.ends_with(&format!(".{suffix}")),
                )
            })
            .cloned()
    }

    /// The registry an app is serving with, if custom domains are enabled.
    ///
    /// The one supported way for application code to reach it — registering a
    /// tenant hostname, rendering status, offboarding.
    #[must_use]
    pub fn from_state(state: &crate::AppState) -> Option<Arc<Self>> {
        state
            .extension::<Arc<Self>>()
            .map(|outer| Arc::clone(&*outer))
    }

    /// Whether the index has been hydrated from the store.
    #[must_use]
    pub fn is_hydrated(&self) -> bool {
        self.hydrated.load(std::sync::atomic::Ordering::Acquire)
    }

    /// Replace the index with `records`, skipping reserved names. Returns how
    /// many loaded and which ones [need a token](CustomDomain::needs_verification_token).
    ///
    /// Sync, so the index lock can never be held across an await.
    fn hydrate_index(&self, records: Vec<CustomDomain>) -> (usize, Vec<String>) {
        let mut index = write_lock(&self.index);
        index.clear();
        for record in records {
            // A reservation can appear AFTER a domain was connected: the
            // operator adds a tenancy base domain, another name to the
            // deployment certificate, or an ingress hostname. `register`
            // refuses a reserved name, but the record written before that
            // change is still in the store, and hydrating it hands the
            // deployment's own hostname to whoever registered it first —
            // custom-domain lookup runs ahead of ordinary subdomain tenancy,
            // so the stale record wins every request from the restart on.
            //
            // Quarantined, not deleted: the record stays in the store, so
            // reverting the configuration restores the domain, while the
            // retention prune clears the certificate of a hostname the
            // registry no longer knows.
            if let Some(pattern) = self.reserved_match(&record.hostname) {
                tracing::error!(
                    hostname = %record.hostname,
                    tenant = %record.tenant,
                    pattern = %pattern,
                    "refusing to load a custom domain the configuration now reserves; offboard it \
                     or drop the reservation"
                );
                continue;
            }
            index.insert(record.hostname.clone(), record);
        }
        let legacy: Vec<String> = index
            .values()
            .filter(|d| d.needs_verification_token())
            .map(|d| d.hostname.clone())
            .collect();
        (index.len(), legacy)
    }

    /// Hydrate the index from the store. Returns how many records loaded.
    ///
    /// # Errors
    ///
    /// Propagates the store's read error.
    pub async fn load(&self) -> io::Result<usize> {
        let records = self.store.load_all().await?;
        let (loaded, legacy) = self.hydrate_index(records);
        self.hydrated
            .store(true, std::sync::atomic::Ordering::Release);
        // A record stored before ownership tokens, and not yet serving, has
        // proved nothing that stops a dangling-DNS takeover. Give it a token
        // now, so the tenant sees the TXT record from the first screen after
        // the upgrade. A failure here is retried by the orchestrator.
        let now = now_unix();
        for hostname in legacy {
            if let Err(e) = self.adopt_verification_token(&hostname, now).await {
                tracing::warn!(
                    hostname = %hostname,
                    "could not give a stored custom domain its verification token: {e}"
                );
            }
        }
        Ok(loaded)
    }

    /// [`load`](Self::load) without the boot-time migration: hydrate the index
    /// from the store and write nothing back. Returns how many records loaded.
    ///
    /// For a process that only inspects the registry — a one-shot
    /// `autumn db retention` report — where giving a pre-token record its
    /// token (and with it a new status and registration time) would turn a
    /// read-only command into a write that changes the very eligibility it is
    /// reporting on. The serving process migrates those records at its own
    /// boot.
    ///
    /// # Errors
    ///
    /// Propagates the store's read error.
    pub async fn load_without_migration(&self) -> io::Result<usize> {
        let records = self.store.load_all().await?;
        let (loaded, _legacy) = self.hydrate_index(records);
        self.hydrated
            .store(true, std::sync::atomic::Ordering::Release);
        Ok(loaded)
    }

    /// Read every record straight from the store, bypassing the in-memory
    /// index.
    ///
    /// The index is only as fresh as this process's own writes. Another
    /// process sharing the store — a serving app while a one-shot
    /// `autumn db retention` runs — registers, verifies and issues without
    /// this index ever hearing of it, so anything destructive decided from
    /// the index alone can act on a registration it cannot see.
    ///
    /// # Errors
    ///
    /// Propagates the store's read error.
    pub async fn stored_records(&self) -> io::Result<Vec<CustomDomain>> {
        self.store.load_all().await
    }

    /// Every order in the store's global attempt ledger.
    ///
    /// # Errors
    ///
    /// Propagates the store's read error.
    pub async fn load_global_attempts(&self) -> io::Result<Vec<i64>> {
        self.store.load_attempts().await
    }

    /// Copy orders into the store's global attempt ledger if it lacks them.
    ///
    /// # Errors
    ///
    /// Propagates the store's write error.
    pub async fn merge_global_attempts(&self, attempts: Vec<i64>, now_unix: i64) -> io::Result<()> {
        self.store.merge_attempts(attempts, now_unix).await
    }

    /// Add an order at `now_unix` to the store's global attempt ledger.
    ///
    /// # Errors
    ///
    /// Propagates the store's write error.
    pub async fn record_global_attempt(&self, now_unix: i64) -> io::Result<()> {
        self.store.append_attempt(now_unix).await
    }

    /// Give a record stored before ownership tokens a token, and send it back
    /// to `PendingDns` to prove it. Returns whether it applied.
    ///
    /// Only for a record that [needs one](CustomDomain::needs_verification_token):
    /// an `active` record is grandfathered, and a record with a token keeps it.
    ///
    /// `registered_at_unix` moves to `now_unix`. The retention sweep deletes a
    /// `PendingDns` record by its age, and a record that was `verified` before
    /// the upgrade must get the full window to publish its TXT record.
    ///
    /// # Errors
    ///
    /// Propagates a random-source failure or the store's write error.
    pub async fn adopt_verification_token(
        &self,
        hostname: &str,
        now_unix: i64,
    ) -> io::Result<bool> {
        let token = mint_verification_token()?;
        self.mutate_if(hostname, CustomDomain::needs_verification_token, move |d| {
            d.verification_token = Some(token);
            d.status = DomainStatus::PendingDns;
            d.registered_at_unix = now_unix;
            d.failure_reason = Some(format!(
                "ownership proof is now required: publish the {VERIFICATION_LABEL} TXT record \
                 shown in the DNS instructions"
            ));
            d.consecutive_failures = 0;
            d.next_attempt_unix = None;
        })
        .await
    }

    /// Connect `hostname` to `tenant`, starting it at `PendingDns`.
    ///
    /// Re-registering the same hostname for the *same* tenant is idempotent —
    /// it returns the existing record rather than resetting a live domain back
    /// to pending, so an app that re-registers on every boot does not
    /// re-order every certificate.
    ///
    /// # Errors
    ///
    /// [`RegisterError`] for an unusable hostname, a full registry, a hostname
    /// another tenant owns, or a store write failure.
    pub async fn register(
        &self,
        hostname: &str,
        tenant: &str,
        now_unix: i64,
    ) -> Result<CustomDomain, RegisterError> {
        let host = normalize_hostname(hostname).map_err(RegisterError::Invalid)?;
        if tenant.trim().is_empty() {
            return Err(RegisterError::Invalid(
                "tenant id must not be empty".to_owned(),
            ));
        }
        if let Some(pattern) = self.reserved_match(&host) {
            return Err(RegisterError::Reserved { pattern });
        }
        // The ownership record sits one label deeper, so it must fit too.
        if verification_record_name(&host).len() > MAX_HOSTNAME_LEN {
            return Err(RegisterError::Invalid(format!(
                "{host} is too long to carry its {VERIFICATION_LABEL} TXT record"
            )));
        }
        // An index that never hydrated knows nothing, and the store keys its
        // files by a hash of the hostname: registering here would report
        // success, overwrite the record of whoever durably owns the hostname,
        // and hand it to this tenant at the next restart. A transient read
        // error at boot must not cost a tenant their domain, so mutations wait
        // for a load that succeeded.
        if !self.is_hydrated() {
            return Err(RegisterError::NotReady);
        }

        // Refuse a full registry BEFORE creating this hostname's write gate.
        // The authoritative check is inside the gate below; this one exists so
        // that a tenant-facing endpoint hammered with unique hostnames cannot
        // mint a gate per attempt while every attempt is refused anyway.
        let full = {
            let index = read_lock(&self.index);
            index.len() >= self.max_domains && !index.contains_key(&host)
        };
        if full {
            return Err(RegisterError::LimitReached {
                max: self.max_domains,
            });
        }

        // Taken BEFORE the index is inspected, so the whole registration —
        // claim and save — is serialised against an offboard of the same
        // hostname. Claiming first and gating afterwards let a concurrent
        // `remove` erase the fresh entry and delete nothing, after which this
        // save still landed: the domain was absent from the running process but
        // came back at the next restart.
        let gate = self.write_gate(&host).await;
        let _write = gate.lock().await;

        // Claim the hostname in the index FIRST, under one write lock, and only
        // then persist. Writing the store first let a losing racer's record
        // reach disk anyway: the winner served in memory while the loser's
        // tenant came back after a restart, silently taking the hostname over.
        let record = {
            let mut index = write_lock(&self.index);
            if let Some(existing) = index.get(&host) {
                if existing.tenant == tenant {
                    return Ok(existing.clone());
                }
                return Err(RegisterError::Conflict {
                    tenant: existing.tenant.clone(),
                });
            }
            if index.len() >= self.max_domains {
                return Err(RegisterError::LimitReached {
                    max: self.max_domains,
                });
            }
            let token = mint_verification_token().map_err(|e| {
                RegisterError::Store(format!("could not mint a verification token: {e}"))
            })?;
            let record = CustomDomain::new(host.clone(), tenant.to_owned(), now_unix, token);
            index.insert(host.clone(), record.clone());
            record
        };

        if let Err(e) = self.store.save(&record).await {
            // Roll the claim back: a hostname that is not durable must not
            // route, or it would disappear on the next restart with no record
            // of why.
            write_lock(&self.index).remove(&host);
            return Err(RegisterError::Store(e.to_string()));
        }
        Ok(record)
    }

    /// The write gate for one hostname, creating it on first use.
    ///
    /// Gates outlive the operations that used them, so the map is swept once it
    /// grows past the registry's own cap: an `Arc` nobody but the map holds has
    /// no operation behind it and cannot be the one this caller is about to
    /// take. Without the sweep every hostname ever passed here — an offboarded
    /// domain, a refused registration — kept an entry for the life of the
    /// process, so a tenant-facing endpoint could grow this map without bound.
    async fn write_gate(&self, host: &str) -> Arc<tokio::sync::Mutex<()>> {
        let mut gates = self.writes.lock().await;
        if gates.len() >= self.gate_high_water() && !gates.contains_key(host) {
            gates.retain(|_, gate| Arc::strong_count(gate) > 1);
        }
        Arc::clone(gates.entry(host.to_owned()).or_default())
    }

    /// When to sweep idle write gates: the registry's cap plus room for the
    /// operations in flight over hostnames that are not registered (an
    /// offboard, a refused registration), so the sweep is rare rather than
    /// per-call.
    const fn gate_high_water(&self) -> usize {
        self.max_domains.saturating_add(64)
    }

    /// The record for `hostname` (normalising the lookup key), if registered.
    #[must_use]
    pub fn get(&self, hostname: &str) -> Option<CustomDomain> {
        let host = normalize_hostname(hostname).ok()?;
        read_lock(&self.index).get(&host).cloned()
    }

    /// The tenant a *servable* hostname belongs to.
    ///
    /// Returns `None` for an unregistered hostname and for one that has not
    /// reached `Active`, so a half-connected domain never routes.
    #[must_use]
    pub fn tenant_for_host(&self, hostname: &str) -> Option<String> {
        let host = normalize_hostname(hostname).ok()?;
        read_lock(&self.index)
            .get(&host)
            .filter(|d| d.is_servable())
            .map(|d| d.tenant.clone())
    }

    /// Is this hostname registered and serving?
    ///
    /// Same answer as `tenant_for_host(hostname).is_some()`, without copying
    /// the tenant id out of the index only to drop it: this runs on the
    /// trusted-host check for every request on a custom domain.
    #[must_use]
    pub fn is_servable(&self, hostname: &str) -> bool {
        normalize_hostname(hostname).is_ok_and(|host| {
            read_lock(&self.index)
                .get(&host)
                .is_some_and(CustomDomain::is_servable)
        })
    }

    /// Every record, sorted by hostname for a stable status listing.
    #[must_use]
    pub fn list(&self) -> Vec<CustomDomain> {
        let mut all: Vec<CustomDomain> = read_lock(&self.index).values().cloned().collect();
        all.sort_by(|a, b| a.hostname.cmp(&b.hostname));
        all
    }

    /// One tenant's domains, sorted by hostname.
    #[must_use]
    pub fn list_for_tenant(&self, tenant: &str) -> Vec<CustomDomain> {
        let mut all: Vec<CustomDomain> = read_lock(&self.index)
            .values()
            .filter(|d| d.tenant == tenant)
            .cloned()
            .collect();
        all.sort_by(|a, b| a.hostname.cmp(&b.hostname));
        all
    }

    /// How many domains are registered.
    #[must_use]
    pub fn len(&self) -> usize {
        read_lock(&self.index).len()
    }

    /// Is the registry empty?
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Domains still awaiting DNS confirmation and out of backoff.
    ///
    /// Also a record that still needs a verification token in any status: its
    /// upgrade at `load` failed to persist, and the check adopts it again.
    #[must_use]
    pub fn pending_verification(&self, now_unix: i64) -> Vec<CustomDomain> {
        self.filter(|d| {
            (d.status == DomainStatus::PendingDns || d.needs_verification_token())
                && d.is_due(now_unix)
        })
    }

    /// Verified domains with no certificate yet, out of backoff.
    ///
    /// `Issuing` is deliberately included: an order that crashed mid-flight
    /// would otherwise strand the domain forever.
    #[must_use]
    pub fn due_for_issuance(&self, now_unix: i64) -> Vec<CustomDomain> {
        // A record with no token has not proved ownership. Its adoption may
        // have failed to persist, so it is held back here as well.
        self.filter(|d| {
            matches!(d.status, DomainStatus::Verified | DomainStatus::Issuing)
                && d.verification_token.is_some()
                && d.is_due(now_unix)
        })
    }

    /// Active domains inside their renew-before window, out of backoff.
    #[must_use]
    pub fn due_for_renewal(&self, now_unix: i64, renew_before_days: u32) -> Vec<CustomDomain> {
        self.filter(|d| {
            d.status == DomainStatus::Active
                && d.is_due(now_unix)
                && d.cert_not_after_unix
                    .is_some_and(|not_after| needs_renewal(not_after, renew_before_days, now_unix))
        })
    }

    fn filter(&self, predicate: impl Fn(&CustomDomain) -> bool) -> Vec<CustomDomain> {
        let mut out: Vec<CustomDomain> = read_lock(&self.index)
            .values()
            .filter(|d| predicate(d))
            .cloned()
            .collect();
        out.sort_by(|a, b| a.hostname.cmp(&b.hostname));
        out
    }

    /// Promote a domain to `Verified` and clear any failure.
    ///
    /// Checks nothing. The orchestrator uses
    /// [`record_verified_for`](Self::record_verified_for), which applies only
    /// to the registration whose DNS and token were checked.
    ///
    /// # Errors
    ///
    /// Propagates the store's write error.
    pub async fn record_verified(&self, hostname: &str, now_unix: i64) -> io::Result<()> {
        self.mutate(hostname, |d| apply_verified_mutation(d, now_unix))
            .await
    }

    /// [`record_verified`](Self::record_verified), but only while the stored
    /// record is the registration that was checked: same `tenant`, same
    /// `token`. Returns whether it applied.
    ///
    /// A token of `None` never applies: a record without one has proved
    /// nothing.
    ///
    /// # Errors
    ///
    /// Propagates the store's write error.
    pub async fn record_verified_for(
        &self,
        hostname: &str,
        tenant: &str,
        token: Option<&str>,
        now_unix: i64,
    ) -> io::Result<bool> {
        let Some(token) = token else {
            return Ok(false);
        };
        self.mutate_if(
            hostname,
            |d| d.tenant == tenant && d.verification_token.as_deref() == Some(token),
            |d| apply_verified_mutation(d, now_unix),
        )
        .await
    }

    /// Mark an ACME order in flight.
    ///
    /// # Errors
    ///
    /// Propagates the store's write error.
    pub async fn record_issuing(&self, hostname: &str) -> io::Result<()> {
        self.mutate(hostname, |d| {
            if d.status == DomainStatus::Verified {
                d.status = DomainStatus::Issuing;
            }
        })
        .await
    }

    /// [`record_issuing`](Self::record_issuing), but only while `tenant` owns a
    /// hostname that is actually orderable. Returns whether it applied.
    ///
    /// Ordering waits for a lease, and the registry can move underneath that
    /// wait. An unconditional transition let the order run anyway: for a
    /// hostname that had been offboarded it spent a slot of the deployment's
    /// budget and the CA's rate limit on a certificate `install` would then
    /// discard, and for one re-registered by someone else it moved the NEW
    /// tenant's record to `Issuing` for an order that was never theirs.
    ///
    /// The admissible states are the same ones activation accepts: `Verified`
    /// (a first order), `Active` (a renewal, whose record legitimately stays
    /// `Active` while the order runs) and `Issuing` — a record left mid-order
    /// by a crash, which `due_for_issuance` deliberately re-selects so the
    /// domain recovers. Only `PendingDns` is refused: a re-registration has not
    /// proven its own DNS yet.
    ///
    /// # Errors
    ///
    /// Propagates the store's write error.
    pub async fn record_issuing_for(&self, hostname: &str, tenant: &str) -> io::Result<bool> {
        self.mutate_if(
            hostname,
            |d| d.tenant == tenant && Self::is_orderable_state(d.status),
            |d| {
                if d.status == DomainStatus::Verified {
                    d.status = DomainStatus::Issuing;
                }
            },
        )
        .await
    }

    /// Record a successfully issued certificate: the domain is now served.
    ///
    /// # Errors
    ///
    /// Propagates the store's write error.
    pub async fn record_active(
        &self,
        hostname: &str,
        now_unix: i64,
        cert_not_after_unix: i64,
    ) -> io::Result<()> {
        self.activate(hostname, None, now_unix, cert_not_after_unix)
            .await
            .map(|_| ())
    }

    /// [`record_active`](Self::record_active), but only while `tenant` still
    /// owns the hostname. Returns whether it applied.
    ///
    /// An issuance suspends at every `.await` between checking who owns a
    /// hostname and installing the certificate. If the owner is offboarded and
    /// the hostname re-registered in that window, an unconditional activation
    /// would stamp `Active` onto the NEW tenant's record — carrying it from
    /// `pending_dns` straight to serving, on a certificate issued for someone
    /// else and without its own DNS ever being verified. The owner is therefore
    /// re-asserted inside the write lock, not before the await.
    ///
    /// # Errors
    ///
    /// Propagates the store's write error.
    pub async fn record_active_for(
        &self,
        hostname: &str,
        tenant: &str,
        now_unix: i64,
        cert_not_after_unix: i64,
    ) -> io::Result<bool> {
        self.activate(hostname, Some(tenant), now_unix, cert_not_after_unix)
            .await
    }

    /// Whether `record_active_for` may promote a record this order found.
    ///
    /// The owner alone is not enough: a tenant that offboards and re-registers
    /// the SAME hostname while its previous order is in flight gets a fresh
    /// `PendingDns` record, and activating that would carry it straight to
    /// serving on DNS it has not re-proved — past the verification gate this
    /// module exists to enforce. Every state an order can legitimately land on
    /// (`Issuing` for a first order, `Active` for a renewal, `Verified` after a
    /// failure reset it) is a state that has already been verified.
    const fn is_orderable_state(status: DomainStatus) -> bool {
        !matches!(status, DomainStatus::PendingDns)
    }

    async fn activate(
        &self,
        hostname: &str,
        tenant: Option<&str>,
        now_unix: i64,
        cert_not_after_unix: i64,
    ) -> io::Result<bool> {
        self.mutate_if(
            hostname,
            |d| tenant.is_none_or(|t| d.tenant == t && Self::is_orderable_state(d.status)),
            |d| {
                d.status = DomainStatus::Active;
                if d.verified_at_unix.is_none() {
                    d.verified_at_unix = Some(now_unix);
                }
                if d.activated_at_unix.is_none() {
                    d.activated_at_unix = Some(now_unix);
                }
                d.cert_not_after_unix = Some(cert_not_after_unix);
                d.failure_reason = None;
                d.consecutive_failures = 0;
                d.next_attempt_unix = None;
            },
        )
        .await
    }

    /// Record a failure with the reason to show and how long to wait.
    ///
    /// An `Active` domain **keeps** its status: its certificate is still valid
    /// and still served, and a failed renewal must not take a tenant offline.
    ///
    /// # Errors
    ///
    /// Propagates the store's write error.
    pub async fn record_failure(
        &self,
        hostname: &str,
        now_unix: i64,
        reason: impl Into<String>,
        backoff_secs: i64,
    ) -> io::Result<()> {
        self.fail(hostname, |_| true, now_unix, reason, backoff_secs)
            .await
            .map(|_| ())
    }

    /// [`record_failure`](Self::record_failure), but only while `tenant` still
    /// owns the hostname AND the record is the generation the order ran
    /// against. Returns whether it applied.
    ///
    /// An order runs across several `.await`s. If the owner is offboarded and
    /// the hostname re-registered in that window, an unconditional failure
    /// would stamp one tenant's error and backoff onto the NEW tenant's record:
    /// a domain that has done nothing wrong would show someone else's reason
    /// and wait out a backoff it did not earn.
    ///
    /// The owner alone is not enough: a tenant that offboards and re-registers
    /// the SAME hostname while its previous order is in flight gets a fresh
    /// `PendingDns` record, and an order never flies against `PendingDns` —
    /// so a failure arriving for it is stale by construction. Discarding it
    /// keeps the replacement's `failure_reason`, backoff, and alert silence
    /// clean.
    ///
    /// # Errors
    ///
    /// Propagates the store's write error.
    pub async fn record_failure_for(
        &self,
        hostname: &str,
        tenant: &str,
        now_unix: i64,
        reason: impl Into<String>,
        backoff_secs: i64,
    ) -> io::Result<bool> {
        self.fail(
            hostname,
            |d| d.tenant == tenant && Self::is_orderable_state(d.status),
            now_unix,
            reason,
            backoff_secs,
        )
        .await
    }

    /// [`record_failure_for`](Self::record_failure_for), but only while the
    /// stored record is still the registration the order ran against: same
    /// `tenant`, same ownership `token`. Returns whether it applied.
    ///
    /// Owner and status alone cannot tell two registrations apart. A tenant
    /// that offboards and re-registers the SAME hostname while an order is in
    /// flight gets a fresh record, and once that successor is itself
    /// `Verified`, `Issuing` or `Active`, a late failure from the dead order
    /// would pass both checks and charge the successor its reason, backoff and
    /// alert — or reset an in-flight `Issuing` back to `Verified`. Each
    /// registration mints its own token, so comparing it pins the generation.
    /// `None` matches only a grandfathered record that has no token either.
    ///
    /// # Errors
    ///
    /// Propagates the store's write error.
    pub async fn record_failure_for_registration(
        &self,
        hostname: &str,
        tenant: &str,
        token: Option<&str>,
        now_unix: i64,
        reason: impl Into<String>,
        backoff_secs: i64,
    ) -> io::Result<bool> {
        self.fail(
            hostname,
            |d| {
                d.tenant == tenant
                    && d.verification_token.as_deref() == token
                    && Self::is_orderable_state(d.status)
            },
            now_unix,
            reason,
            backoff_secs,
        )
        .await
    }

    /// Persist that an order was placed for this registration at `now_unix`,
    /// dropping attempts older than the per-domain window.
    ///
    /// Written BEFORE the order goes out, so a crash mid-order still counts it.
    /// Returns whether it applied: a hostname that changed hands meanwhile is
    /// not charged for the old registration's order.
    ///
    /// # Errors
    ///
    /// Propagates the store's write error.
    pub async fn record_attempt_for_registration(
        &self,
        hostname: &str,
        tenant: &str,
        token: Option<&str>,
        now_unix: i64,
    ) -> io::Result<bool> {
        self.mutate_if(
            hostname,
            |d| d.tenant == tenant && d.verification_token.as_deref() == token,
            move |d| {
                let cutoff = now_unix.saturating_sub(PER_DOMAIN_WINDOW_SECS);
                d.issuance_attempts_unix.retain(|at| *at > cutoff);
                d.issuance_attempts_unix.push(now_unix);
            },
        )
        .await
    }

    /// Hold this registration back until `retry_at_unix` because an issuance
    /// BUDGET is spent, with `reason` for the status surface.
    ///
    /// Not a failure: no order was placed, so `consecutive_failures` is left
    /// alone. Counting it would walk a domain to the maximum backoff just for
    /// being examined while the deployment's hourly budget was full, and keep
    /// it unavailable long after the budget reopened. A later deadline already
    /// on the record is kept.
    ///
    /// # Errors
    ///
    /// Propagates the store's write error.
    pub async fn defer_for_registration(
        &self,
        hostname: &str,
        tenant: &str,
        token: Option<&str>,
        retry_at_unix: i64,
        reason: impl Into<String>,
    ) -> io::Result<bool> {
        let reason = reason.into();
        self.mutate_if(
            hostname,
            |d| {
                d.tenant == tenant
                    && d.verification_token.as_deref() == token
                    && Self::is_orderable_state(d.status)
            },
            move |d| {
                if d.status == DomainStatus::Issuing {
                    d.status = DomainStatus::Verified;
                }
                d.failure_reason = Some(reason.clone());
                d.next_attempt_unix = Some(
                    d.next_attempt_unix
                        .map_or(retry_at_unix, |at| at.max(retry_at_unix)),
                );
            },
        )
        .await
    }

    /// The failure-writing core of [`record_failure`](Self::record_failure)
    /// and its guarded variants. `guard` runs inside the write lock, against the record as
    /// it is at write time — never against a snapshot taken before an await.
    async fn fail(
        &self,
        hostname: &str,
        guard: impl FnOnce(&CustomDomain) -> bool,
        now_unix: i64,
        reason: impl Into<String>,
        backoff_secs: i64,
    ) -> io::Result<bool> {
        let reason = reason.into();
        self.mutate_if(hostname, guard, move |d| {
            if d.status == DomainStatus::Issuing {
                // An order that failed goes back to `Verified`, not to
                // `PendingDns`: DNS was already proven, and re-verifying
                // would add a needless round trip to every retry.
                d.status = DomainStatus::Verified;
            }
            d.failure_reason = Some(reason.clone());
            d.consecutive_failures = d.consecutive_failures.saturating_add(1);
            d.next_attempt_unix = Some(now_unix.saturating_add(backoff_secs));
        })
        .await
    }

    /// Offboard one hostname: stop routing, stop serving, stop renewing.
    /// Returns whether a record was actually removed.
    ///
    /// # Errors
    ///
    /// Propagates the store's delete error.
    pub async fn remove(&self, hostname: &str) -> io::Result<bool> {
        self.remove_if(hostname, |_| true).await
    }

    /// [`remove`](Self::remove), but only while `guard` still holds for the
    /// stored record. Returns whether a record was actually removed.
    ///
    /// The guard runs inside the per-hostname write gate, so a caller that
    /// decided to offboard a domain BEFORE an `.await` re-asserts that decision
    /// against the record as it is now. The retention sweep needs exactly this:
    /// it picks its candidates from a `list()` snapshot, and a domain that
    /// finished verifying and issuing while the sweep was awaiting an earlier
    /// offboard would otherwise be deleted — certificate and all — moments
    /// after it came up.
    ///
    /// # Errors
    ///
    /// Propagates the store's delete error.
    pub async fn remove_if(
        &self,
        hostname: &str,
        guard: impl FnOnce(&CustomDomain) -> bool,
    ) -> io::Result<bool> {
        let Ok(host) = normalize_hostname(hostname) else {
            return Ok(false);
        };
        // Delete from the store first: a crash between the two leaves a record
        // that `load` re-hydrates, which is recoverable — the reverse leaves a
        // hostname that routes but has no durable record.
        //
        // Under the same gate as every other write for this hostname, so an
        // in-flight save cannot land after the delete and resurrect the record
        // at the next restart.
        let gate = self.write_gate(&host).await;
        let _write = gate.lock().await;
        if !read_lock(&self.index).get(&host).is_none_or(guard) {
            return Ok(false);
        }
        self.store.delete(&host).await?;
        Ok(write_lock(&self.index).remove(&host).is_some())
    }

    /// Offboard every domain a tenant owns. Returns how many were removed.
    ///
    /// # Errors
    ///
    /// Propagates the store's delete error.
    pub async fn remove_tenant(&self, tenant: &str) -> io::Result<usize> {
        let hostnames: Vec<String> = self
            .list_for_tenant(tenant)
            .into_iter()
            .map(|d| d.hostname)
            .collect();
        let mut removed = 0;
        for hostname in hostnames {
            // Re-assert the tenant per hostname: the list is a snapshot and
            // each removal awaits, so a hostname freed early can be
            // re-registered by ANOTHER tenant before a later one runs. An
            // unconditional removal would delete that tenant's record and stop
            // routing a domain that was never part of this teardown.
            if self
                .remove_if(&hostname, |current| current.tenant == tenant)
                .await?
            {
                removed += 1;
            }
        }
        Ok(removed)
    }

    /// A one-line-per-domain report of everything unhealthy, naming the
    /// hostname AND its tenant so an operator can act without a lookup.
    ///
    /// Empty when every domain is healthy.
    #[must_use]
    pub fn health_report(&self, now_unix: i64) -> String {
        let mut lines: Vec<String> = Vec::new();
        for domain in self.list() {
            let expiring = domain.cert_not_after_unix.is_some_and(|na| na <= now_unix);
            if let Some(reason) = &domain.failure_reason {
                lines.push(format!(
                    "{} (tenant {}, {}): {reason}",
                    domain.hostname, domain.tenant, domain.status
                ));
            } else if expiring {
                lines.push(format!(
                    "{} (tenant {}): certificate expired",
                    domain.hostname, domain.tenant
                ));
            }
        }
        lines.join("; ")
    }

    /// Apply `f` to a record and persist it. A missing hostname is a no-op.
    async fn mutate(&self, hostname: &str, f: impl FnOnce(&mut CustomDomain)) -> io::Result<()> {
        self.mutate_if(hostname, |_| true, f).await.map(|_| ())
    }

    /// Apply `f` to a record and persist it, but only while `guard` accepts the
    /// record as it stands under the write lock. Returns whether it applied.
    ///
    /// The guard is what makes a check-then-write atomic against the registry:
    /// a caller that verified something about the record BEFORE an `.await`
    /// re-asserts it here, inside the lock, rather than trusting a fact that
    /// may have expired while it was suspended.
    async fn mutate_if(
        &self,
        hostname: &str,
        guard: impl FnOnce(&CustomDomain) -> bool,
        f: impl FnOnce(&mut CustomDomain),
    ) -> io::Result<bool> {
        let Ok(host) = normalize_hostname(hostname) else {
            return Ok(false);
        };
        let gate = self.write_gate(&host).await;
        let _write = gate.lock().await;
        // Apply to a COPY, persist, and only then publish. Mutating the live
        // index first left a state the store had rejected routing and serving
        // anyway: `record_active` reported failure while the domain went on
        // being served from it, and the state then vanished at the next
        // restart with nothing to explain either half. Every writer for this
        // hostname holds this same gate, so nothing can change the record
        // between the copy and the publish.
        let Some(current) = read_lock(&self.index).get(&host).cloned() else {
            return Ok(false);
        };
        if !guard(&current) {
            return Ok(false);
        }
        let mut updated = current;
        f(&mut updated);
        self.store.save(&updated).await?;
        let mut index = write_lock(&self.index);
        // An offboard cannot interleave under the gate, but the index is the
        // routing table: if the record is gone, leave it gone rather than
        // resurrecting a hostname nothing owns.
        Ok(index.get_mut(&host).is_some_and(|slot| {
            *slot = updated;
            true
        }))
    }
}

// ── Issuance budget ──────────────────────────────────────────────────────

/// The promotion half of [`CustomDomainRegistry::record_verified`]: shared by
/// the unconditional variant and the registration-guarded
/// [`CustomDomainRegistry::record_verified_for`].
///
/// A domain that is already serving stays serving: a re-verification must
/// never demote a live certificate back to `Verified` and trigger a fresh
/// order.
fn apply_verified_mutation(d: &mut CustomDomain, now_unix: i64) {
    if d.status == DomainStatus::PendingDns {
        d.status = DomainStatus::Verified;
    }
    if d.verified_at_unix.is_none() {
        d.verified_at_unix = Some(now_unix);
    }
    d.failure_reason = None;
    d.consecutive_failures = 0;
    d.next_attempt_unix = None;
}

/// Is a certificate expiring at `not_after_unix` inside its renew-before
/// window at `now_unix`?
///
/// Mirrors `crate::acme::renewal::needs_renewal` but compiles without the
/// `acme` feature, so the registry's renewal scheduling is testable in a
/// default build.
#[must_use]
pub const fn needs_renewal(not_after_unix: i64, renew_before_days: u32, now_unix: i64) -> bool {
    let window = (renew_before_days as i64).saturating_mul(86_400);
    not_after_unix.saturating_sub(now_unix) < window
}

/// The retry delay after `consecutive_failures` failures: `base * 2^(n-1)`,
/// capped at `max`.
///
/// Saturating throughout, so an absurd failure count returns the cap rather
/// than overflowing.
#[must_use]
pub const fn backoff_secs(consecutive_failures: u32, base: u64, max: u64) -> u64 {
    if consecutive_failures == 0 {
        return 0;
    }
    let shift = if consecutive_failures > 32 {
        32
    } else {
        consecutive_failures - 1
    };
    let scaled = base.saturating_mul(1_u64 << shift);
    if scaled > max { max } else { scaled }
}

/// Whether an issuance attempt may proceed now.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum IssuanceDecision {
    /// Go ahead.
    Allow,
    /// This domain has spent its daily budget.
    PerDomainLimit {
        /// Seconds until the domain's window rolls.
        retry_after_secs: i64,
    },
    /// The deployment has spent its hourly budget.
    GlobalLimit {
        /// Seconds until the global window rolls.
        retry_after_secs: i64,
    },
}

impl IssuanceDecision {
    /// May the caller order now?
    #[must_use]
    pub const fn is_allowed(&self) -> bool {
        matches!(self, Self::Allow)
    }

    /// Seconds until the refusing window rolls, if the attempt was refused.
    #[must_use]
    pub const fn retry_after_secs(&self) -> Option<i64> {
        match self {
            Self::Allow => None,
            Self::PerDomainLimit { retry_after_secs } | Self::GlobalLimit { retry_after_secs } => {
                Some(*retry_after_secs)
            }
        }
    }

    /// The reason to record on the domain, if the attempt was refused.
    #[must_use]
    pub fn reason(&self) -> Option<String> {
        match self {
            Self::Allow => None,
            Self::PerDomainLimit { retry_after_secs } => Some(format!(
                "issuance budget for this domain is spent; retrying in {retry_after_secs}s"
            )),
            Self::GlobalLimit { retry_after_secs } => Some(format!(
                "deployment-wide issuance budget is spent; retrying in {retry_after_secs}s"
            )),
        }
    }
}

/// The window a per-domain issuance budget is measured over: Let's Encrypt
/// counts duplicate certificates per week, so a daily cap is the conservative
/// sub-multiple.
const PER_DOMAIN_WINDOW_SECS: i64 = 86_400;
/// The window the deployment-wide budget is measured over.
const GLOBAL_WINDOW_SECS: i64 = 3_600;

/// Per-domain and deployment-wide caps on ACME orders.
///
/// The abuse posture this enforces (see `docs/guide/tls.md`): a tenant whose
/// domain fails repeatedly costs at most `per_domain_per_day` orders a day, and
/// the whole deployment costs at most `global_per_hour` — well inside Let's
/// Encrypt's 300-new-orders-per-account-per-3-hours limit at the defaults.
/// Failures additionally back off exponentially, so a permanently broken domain
/// converges to one attempt per `max_backoff_secs`.
#[derive(Debug)]
pub struct IssuanceLimiter {
    per_domain_per_day: u32,
    global_per_hour: u32,
    base_backoff_secs: u64,
    max_backoff_secs: u64,
    attempts: RwLock<Attempts>,
}

#[derive(Debug, Default)]
struct Attempts {
    /// Attempt timestamps per domain, within the per-domain window.
    per_domain: HashMap<String, Vec<i64>>,
    /// Attempt timestamps across all domains, within the global window.
    global: Vec<i64>,
}

impl IssuanceLimiter {
    /// A limiter with the given budgets and backoff bounds.
    #[must_use]
    pub fn new(
        per_domain_per_day: u32,
        global_per_hour: u32,
        base_backoff_secs: u64,
        max_backoff_secs: u64,
    ) -> Self {
        Self {
            per_domain_per_day,
            global_per_hour,
            base_backoff_secs,
            max_backoff_secs,
            attempts: RwLock::new(Attempts::default()),
        }
    }

    /// The longest backoff this limiter will impose.
    #[must_use]
    pub const fn max_backoff(&self) -> u64 {
        self.max_backoff_secs
    }

    /// The backoff delay for a domain with `consecutive_failures` failures.
    #[must_use]
    pub const fn backoff_for(&self, consecutive_failures: u32) -> u64 {
        backoff_secs(
            consecutive_failures,
            self.base_backoff_secs,
            self.max_backoff_secs,
        )
    }

    /// May `hostname` be ordered now, as far as the *budgets* are concerned?
    ///
    /// Deliberately says nothing about failure backoff: the retry deadline
    /// lives on the domain record ([`CustomDomain::next_attempt_unix`], set
    /// from [`Self::backoff_for`]) and is enforced by
    /// [`CustomDomain::is_due`]. A failure count alone cannot say the wait is
    /// over, so a second copy of the deadline here would refuse every retry
    /// forever.
    ///
    /// The domain's own budget is checked before the deployment's, so the
    /// reason reported is the most specific one.
    #[must_use]
    pub fn check(&self, hostname: &str, now_unix: i64) -> IssuanceDecision {
        // Both windows are read under ONE guard, then the guard is released
        // before any decision is built: a caller must not hold the lock while
        // formatting a message, and the two counts must come from the same
        // instant or a domain could pass its own budget against a global count
        // taken after another thread spent it.
        let (domain_hits, global_hits) = {
            let attempts = read_lock(&self.attempts);
            let domain_hits = attempts
                .per_domain
                .get(hostname)
                .map(|hits| within(hits, now_unix, PER_DOMAIN_WINDOW_SECS))
                .unwrap_or_default();
            let global_hits = within(&attempts.global, now_unix, GLOBAL_WINDOW_SECS);
            drop(attempts);
            (domain_hits, global_hits)
        };
        if domain_hits.len() >= self.per_domain_per_day as usize {
            return IssuanceDecision::PerDomainLimit {
                retry_after_secs: retry_after(&domain_hits, now_unix, PER_DOMAIN_WINDOW_SECS),
            };
        }
        if global_hits.len() >= self.global_per_hour as usize {
            return IssuanceDecision::GlobalLimit {
                retry_after_secs: retry_after(&global_hits, now_unix, GLOBAL_WINDOW_SECS),
            };
        }
        IssuanceDecision::Allow
    }

    /// Record that an order was placed for `hostname`.
    pub fn record_attempt(&self, hostname: &str, now_unix: i64) {
        let mut attempts = write_lock(&self.attempts);
        let global = within(&attempts.global, now_unix, GLOBAL_WINDOW_SECS);
        attempts.global = global;
        attempts.global.push(now_unix);
        let hits = attempts.per_domain.entry(hostname.to_owned()).or_default();
        let mut kept = within(hits, now_unix, PER_DOMAIN_WINDOW_SECS);
        kept.push(now_unix);
        *hits = kept;
        drop(attempts);
    }

    /// Seed the windows from attempts that were persisted before a restart.
    ///
    /// Each `(hostname, at_unix)` is one order already placed. Entries outside
    /// the per-domain window at `now_unix` are ignored. Attempts are a
    /// multiset: two orders in the same second are two attempts, and a crash
    /// loop produces exactly that. Hydrating raises each `(hostname, at)` count
    /// to the persisted one and never past it, so calling this twice is
    /// harmless. A new attempt also lands in the global window, which protects
    /// the ACME account shared by every tenant, so a restart cannot reset it
    /// either.
    pub fn hydrate(&self, attempts: impl IntoIterator<Item = (String, i64)>, now_unix: i64) {
        let oldest = now_unix.saturating_sub(PER_DOMAIN_WINDOW_SECS);
        let mut persisted: HashMap<(String, i64), usize> = HashMap::new();
        for (hostname, at) in attempts {
            if at > oldest {
                *persisted.entry((hostname, at)).or_default() += 1;
            }
        }
        let mut state = write_lock(&self.attempts);
        for ((hostname, at), want) in persisted {
            let hits = state.per_domain.entry(hostname).or_default();
            let have = hits.iter().filter(|known| **known == at).count();
            let missing = want.saturating_sub(have);
            hits.extend(std::iter::repeat_n(at, missing));
            state.global.extend(std::iter::repeat_n(at, missing));
        }
        drop(state);
    }

    /// Seed the global window from the persisted attempt ledger.
    ///
    /// The ledger holds orders for offboarded domains too. No domain record
    /// shows those. The ledger also repeats orders that [`Self::hydrate`]
    /// already added. So this raises the count at each instant to the ledger
    /// count and no higher. Call it after `hydrate`. A second call changes
    /// nothing. It ignores entries outside the window that ends at `now_unix`.
    pub fn hydrate_global(&self, attempts: impl IntoIterator<Item = i64>, now_unix: i64) {
        let oldest = now_unix.saturating_sub(GLOBAL_WINDOW_SECS);
        let mut persisted: HashMap<i64, usize> = HashMap::new();
        for at in attempts {
            if at > oldest && at <= now_unix {
                *persisted.entry(at).or_default() += 1;
            }
        }
        let mut state = write_lock(&self.attempts);
        let mut have: HashMap<i64, usize> = HashMap::new();
        for at in &state.global {
            *have.entry(*at).or_default() += 1;
        }
        for (at, want) in persisted {
            let known = have.get(&at).copied().unwrap_or(0);
            state
                .global
                .extend(std::iter::repeat_n(at, want.saturating_sub(known)));
        }
        drop(state);
    }

    /// Drop a domain's attempt history — called when it is offboarded, so a
    /// re-registration of the same hostname is not charged for the old one.
    pub fn forget(&self, hostname: &str) {
        write_lock(&self.attempts).per_domain.remove(hostname);
    }
}

/// The timestamps in `hits` still inside `window` at `now_unix`.
fn within(hits: &[i64], now_unix: i64, window: i64) -> Vec<i64> {
    let cutoff = now_unix.saturating_sub(window);
    hits.iter().copied().filter(|at| *at > cutoff).collect()
}

/// Seconds until the oldest hit in `hits` leaves `window`.
fn retry_after(hits: &[i64], now_unix: i64, window: i64) -> i64 {
    hits.iter().min().map_or(0, |oldest| {
        (oldest.saturating_add(window) - now_unix).max(1)
    })
}

// ── Issuance seam ────────────────────────────────────────────────────────

/// A newly issued certificate: PEM chain (leaf first) plus its private key.
///
/// Deliberately independent of the `acme` feature so the orchestration and its
/// tests compile in a default build; the ACME adapter converts to and from
/// `crate::acme::store::StoredCert`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IssuedCertificate {
    /// PEM certificate chain, leaf first.
    pub chain_pem: String,
    /// PEM private key matching the leaf.
    pub key_pem: String,
}

/// Obtains a certificate for one hostname.
///
/// The seam between the orchestration in this module and the ACME client, so
/// the state machine, the rate limits and the verification gate are testable
/// without a CA.
pub trait DomainIssuer: Send + Sync {
    /// Order (or renew) a certificate covering exactly `hostname`.
    fn issue<'a>(
        &'a self,
        hostname: &'a str,
    ) -> futures::future::BoxFuture<'a, Result<IssuedCertificate, String>>;
}

/// Looks up where a hostname currently points, and what its ownership TXT
/// record carries.
///
/// Implemented by the runtime against the system resolver and public DNS, and
/// by tests against a fixed table.
pub trait DomainVerifier: Send + Sync {
    /// Observe `hostname`'s current DNS target.
    fn observe<'a>(&'a self, hostname: &'a str) -> futures::future::BoxFuture<'a, ObservedTarget>;

    /// Observe the TXT values published at `name`, a
    /// [`verification_record_name`].
    fn observe_txt<'a>(&'a self, name: &'a str) -> futures::future::BoxFuture<'a, ObservedTxt>;
}

/// A [`DomainVerifier`] backed by the system resolver for addresses and by
/// direct DNS queries for TXT.
///
/// Addresses come from `getaddrinfo`, which follows CNAMEs transparently and
/// does not report them, so a tenant who publishes the suggested CNAME is
/// verified by the addresses it resolves to — which is the property that
/// actually matters.
///
/// `getaddrinfo` cannot read TXT, so the ownership record is asked of the
/// zone's authoritative nameservers, found through `txt_resolvers`, and
/// of `txt_resolvers` themselves. TXT lookups need the `acme` feature; without
/// it every TXT lookup is unanswered and no domain verifies.
#[derive(Debug, Default, Clone)]
pub struct SystemDomainVerifier {
    txt_resolvers: Vec<std::net::SocketAddr>,
}

impl SystemDomainVerifier {
    /// A verifier that asks `txt_resolvers` (recursive resolvers) for TXT.
    #[must_use]
    pub const fn new(txt_resolvers: Vec<std::net::SocketAddr>) -> Self {
        Self { txt_resolvers }
    }
}

impl DomainVerifier for SystemDomainVerifier {
    fn observe<'a>(&'a self, hostname: &'a str) -> futures::future::BoxFuture<'a, ObservedTarget> {
        Box::pin(async move {
            let host = hostname.to_owned();
            // `getaddrinfo` blocks; keep it off the async worker.
            let resolved = crate::time::spawn_blocking(move || {
                use std::net::ToSocketAddrs as _;
                (host.as_str(), 0_u16)
                    .to_socket_addrs()
                    .map(|addrs| addrs.map(|s| s.ip()).collect::<Vec<_>>())
            })
            .await;
            match resolved {
                Ok(Ok(addrs)) if !addrs.is_empty() => ObservedTarget::Addresses(addrs),
                _ => ObservedTarget::None,
            }
        })
    }

    fn observe_txt<'a>(&'a self, name: &'a str) -> futures::future::BoxFuture<'a, ObservedTxt> {
        Box::pin(async move {
            if self.txt_resolvers.is_empty() {
                return ObservedTxt::Unanswered("no TXT resolvers are configured".to_owned());
            }
            #[cfg(feature = "acme")]
            {
                use crate::acme::dns::resolver::{UdpDnsLookup, txt_values};
                let lookup = UdpDnsLookup::new(std::time::Duration::from_secs(3));
                let deadline = std::time::Duration::from_secs(6);
                match txt_values(name, &self.txt_resolvers, &lookup, deadline).await {
                    Ok(values) => ObservedTxt::Values(values),
                    Err(e) => ObservedTxt::Unanswered(e),
                }
            }
            #[cfg(not(feature = "acme"))]
            {
                let _ = name;
                ObservedTxt::Unanswered("TXT lookups need the `acme` feature".to_owned())
            }
        })
    }
}

// ── Health ───────────────────────────────────────────────────────────────

/// The `custom_domains` health indicator (AC5).
///
/// `Down` when any domain's certificate has actually expired — that domain is
/// no longer servable. A domain merely carrying a `failure_reason` (a retry is
/// pending, its certificate still valid) is `Up` with the detail attached: one
/// tenant's failed renewal is not a deployment outage, and grading it as one
/// would make the signal useless at a thousand domains.
///
/// Every unhealthy domain is named with its tenant, so an operator reading
/// `/actuator/health` can act without a registry lookup.
#[derive(Debug)]
pub struct CustomDomainHealthIndicator {
    registry: Arc<CustomDomainRegistry>,
    now_unix: fn() -> i64,
}

impl CustomDomainHealthIndicator {
    /// An indicator over `registry`.
    #[must_use]
    pub fn new(registry: Arc<CustomDomainRegistry>) -> Self {
        Self { registry, now_unix }
    }

    /// Grade the registry at `now_unix` (pure; used by `check` and tests).
    #[must_use]
    pub fn grade(&self, now_unix: i64) -> crate::actuator::HealthCheckOutput {
        use crate::actuator::{HealthCheckOutput, HealthStatus};
        let mut details = std::collections::HashMap::new();
        let all = self.registry.list();
        let active = all.iter().filter(|d| d.is_servable()).count();
        details.insert("registered".to_owned(), serde_json::json!(all.len()));
        details.insert("active".to_owned(), serde_json::json!(active));

        let expired: Vec<String> = all
            .iter()
            .filter(|d| d.cert_not_after_unix.is_some_and(|na| na <= now_unix))
            .map(|d| format!("{} (tenant {})", d.hostname, d.tenant))
            .collect();
        let failing: Vec<String> = all
            .iter()
            .filter(|d| d.failure_reason.is_some())
            .map(|d| {
                format!(
                    "{} (tenant {}): {}",
                    d.hostname,
                    d.tenant,
                    d.failure_reason.as_deref().unwrap_or_default()
                )
            })
            .collect();
        if !failing.is_empty() {
            details.insert("failing".to_owned(), serde_json::json!(failing));
        }
        if !expired.is_empty() {
            details.insert("expired".to_owned(), serde_json::json!(expired));
        }

        HealthCheckOutput {
            status: if expired.is_empty() {
                HealthStatus::Up
            } else {
                HealthStatus::Down
            },
            details,
        }
    }
}

impl crate::actuator::HealthIndicator for CustomDomainHealthIndicator {
    fn check(&self) -> futures::future::BoxFuture<'_, crate::actuator::HealthCheckOutput> {
        let now = (self.now_unix)();
        Box::pin(async move { self.grade(now) })
    }

    fn group(&self) -> crate::actuator::IndicatorGroup {
        crate::actuator::IndicatorGroup::HealthOnly
    }
}

/// Wall-clock seconds since the epoch.
///
/// The one reading in the custom-domain path; the orchestrator ticks from it
/// too, so a status and the decision that produced it never disagree by a
/// second.
#[allow(
    clippy::disallowed_methods,
    reason = "grades real X.509 expiry dates and paces real ACME orders"
)]
pub(crate) fn now_unix() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX))
}

// ── Retention ────────────────────────────────────────────────────────────

/// Prunes the custom-domain datasets the framework owns (#1605 / #1635 AC7).
///
/// The seam lives here, in the core module, so the retention engine can drive
/// it without the `acme` feature; the implementation that actually knows about
/// certificates is `crate::acme::tenant_domains::CustomDomainTask`.
pub trait CustomDomainPruner: Send + Sync {
    /// Delete what the retention policy no longer keeps and return how many
    /// records went (or would go, when `dry_run`).
    ///
    /// Two things are pruned: registry records still awaiting DNS since before
    /// `cutoff_unix` — a tenant who was handed instructions and never followed
    /// them — and stored certificates for hostnames no longer registered,
    /// which nothing would otherwise ever delete.
    fn prune(
        &self,
        cutoff_unix: i64,
        dry_run: bool,
    ) -> futures::future::BoxFuture<'_, Result<u64, String>>;

    /// Offboard one hostname completely: stop routing and serving, halt
    /// renewal, and delete the stored certificate and its private key.
    ///
    /// [`CustomDomainRegistry::remove`] only forgets the registration; the
    /// certificate lives in the ACME store, which the registry cannot reach.
    /// This is the offboarding an application should call.
    fn offboard_domain<'a>(
        &'a self,
        hostname: &'a str,
    ) -> futures::future::BoxFuture<'a, io::Result<bool>>;

    /// Offboard every domain a tenant owns. Returns how many went.
    fn offboard_tenant_domains<'a>(
        &'a self,
        tenant: &'a str,
    ) -> futures::future::BoxFuture<'a, io::Result<usize>>;
}

impl dyn CustomDomainPruner {
    /// The offboarding surface an app is serving with, if custom domains are
    /// enabled.
    #[must_use]
    pub fn from_state(state: &crate::AppState) -> Option<Arc<Self>> {
        state
            .extension::<Arc<Self>>()
            .map(|outer| Arc::clone(&*outer))
    }
}

// ── SNI certificate selection ────────────────────────────────────────────

#[cfg(feature = "tls")]
mod sni {
    use super::{CustomDomainRegistry, read_lock, write_lock};
    use rustls::server::{ClientHello, ResolvesServerCert};
    use rustls::sign::CertifiedKey;
    use std::collections::{HashMap, VecDeque};
    use std::sync::{Arc, RwLock};

    /// A bounded, insertion-ordered cache of per-domain certificates.
    ///
    /// AC6's "certificates load incrementally": a 1,000-domain deployment
    /// holds at most `capacity` parsed certificates in memory and re-reads the
    /// rest from the store on demand, so correctness never depends on every
    /// certificate being resident at boot.
    ///
    /// Eviction is first-in-first-out rather than least-recently-used: FIFO
    /// needs no write on the read path, which is the TLS handshake path, and
    /// the cost of an eviction is one file read.
    #[derive(Debug)]
    pub struct CustomDomainCertCache {
        capacity: usize,
        inner: RwLock<CacheInner>,
    }

    #[derive(Debug, Default)]
    struct CacheInner {
        certs: HashMap<String, Arc<CertifiedKey>>,
        order: VecDeque<String>,
    }

    impl CustomDomainCertCache {
        /// A cache holding at most `capacity` certificates (minimum 1).
        #[must_use]
        pub fn new(capacity: usize) -> Self {
            Self {
                capacity: capacity.max(1),
                inner: RwLock::new(CacheInner::default()),
            }
        }

        /// The cached certificate for `hostname`, if resident.
        #[must_use]
        pub fn get(&self, hostname: &str) -> Option<Arc<CertifiedKey>> {
            read_lock(&self.inner).certs.get(hostname).cloned()
        }

        /// Cache `key` for `hostname`, evicting the oldest entry if full.
        pub fn insert(&self, hostname: &str, key: Arc<CertifiedKey>) {
            let mut inner = write_lock(&self.inner);
            if inner.certs.insert(hostname.to_owned(), key).is_none() {
                inner.order.push_back(hostname.to_owned());
            }
            while inner.order.len() > self.capacity {
                if let Some(oldest) = inner.order.pop_front() {
                    inner.certs.remove(&oldest);
                }
            }
            drop(inner);
        }

        /// Drop `hostname`'s certificate — called on offboarding, so a removed
        /// domain stops being served without waiting for eviction.
        pub fn remove(&self, hostname: &str) {
            let mut inner = write_lock(&self.inner);
            inner.certs.remove(hostname);
            inner.order.retain(|h| h != hostname);
        }

        /// The most certificates this cache will hold.
        #[must_use]
        pub const fn capacity(&self) -> usize {
            self.capacity
        }

        /// How many certificates are resident.
        #[must_use]
        pub fn len(&self) -> usize {
            read_lock(&self.inner).certs.len()
        }

        /// Is the cache empty?
        #[must_use]
        pub fn is_empty(&self) -> bool {
            self.len() == 0
        }
    }

    /// Selects the certificate for a TLS handshake by SNI.
    ///
    /// Three outcomes, in order:
    ///
    /// 1. the SNI name is covered by the operator's own certificate → serve it
    ///    (unchanged #1603/#1608 behaviour),
    /// 2. the name is a registered, `Active` custom domain with a resident
    ///    certificate → serve that,
    /// 3. anything else → **refuse the handshake**.
    ///
    /// (3) is the abuse gate AC4 asks for: an attacker pointing DNS at this
    /// deployment and opening a handshake for an unregistered name gets a TLS
    /// alert, and no code path from here reaches the ACME provider.
    ///
    /// A registered domain whose certificate is not resident is loaded through
    /// the optional [`SniCertSource`] — the other half of AC6's incremental
    /// loading. Without it, a deployment with more domains than
    /// `cert_cache_size` would stop serving every domain past the cache after a
    /// restart.
    #[derive(Debug)]
    pub struct SniCertResolver {
        base: Arc<crate::tls::ReloadableCertResolver>,
        base_names: Vec<String>,
        registry: Arc<CustomDomainRegistry>,
        cache: Arc<CustomDomainCertCache>,
        source: Option<Arc<dyn SniCertSource>>,
    }

    /// Loads one domain's certificate on the handshake path.
    ///
    /// Deliberately **synchronous**: `rustls` resolves a certificate inside the
    /// handshake, which is not an async context. The implementation reads two
    /// small files, and only on a cache miss for a hostname already proven
    /// registered — never for an attacker-supplied name — so the blocking cost
    /// is bounded by the eviction rate, not by request volume.
    pub trait SniCertSource: Send + Sync + std::fmt::Debug {
        /// The stored certificate for `hostname`, if one loads.
        fn load(&self, hostname: &str) -> Option<Arc<CertifiedKey>>;
    }

    impl SniCertResolver {
        /// A resolver over the operator's certificate (`base`, covering
        /// `base_names`) plus the tenant registry and its certificate cache.
        #[must_use]
        pub fn new(
            base: Arc<crate::tls::ReloadableCertResolver>,
            base_names: Vec<String>,
            registry: Arc<CustomDomainRegistry>,
            cache: Arc<CustomDomainCertCache>,
        ) -> Self {
            Self {
                base,
                base_names: base_names
                    .into_iter()
                    .map(|n| n.trim().to_ascii_lowercase())
                    .collect(),
                registry,
                cache,
                source: None,
            }
        }

        /// Load a registered domain's certificate from `source` on a cache miss.
        #[must_use]
        pub fn with_source(mut self, source: Arc<dyn SniCertSource>) -> Self {
            self.source = Some(source);
            self
        }

        /// The certificate to serve for `server_name`, or `None` to refuse.
        #[must_use]
        pub fn certificate_for(&self, server_name: &str) -> Option<Arc<CertifiedKey>> {
            let name = server_name.trim_end_matches('.').to_ascii_lowercase();
            if self.base_covers(&name) {
                return Some(self.base.current());
            }
            // The registry check comes BEFORE any I/O: an unregistered name
            // must cost nothing but a hash lookup, however many are thrown at
            // the listener.
            if !self.registry.is_servable(&name) {
                return None;
            }
            if let Some(cached) = self.cache.get(&name) {
                return Some(cached);
            }
            let loaded = self.source.as_ref()?.load(&name)?;
            self.cache.insert(&name, Arc::clone(&loaded));
            Some(loaded)
        }

        /// Does the operator's own certificate cover `name`?
        fn base_covers(&self, name: &str) -> bool {
            self.base_names.iter().any(|pattern| {
                pattern
                    .strip_prefix("*.")
                    .map_or(pattern == name, |suffix| {
                        // A wildcard matches exactly one label, per RFC 6125.
                        name.strip_suffix(suffix)
                            .and_then(|prefix| prefix.strip_suffix('.'))
                            .is_some_and(|label| !label.is_empty() && !label.contains('.'))
                    })
            })
        }
    }

    impl ResolvesServerCert for SniCertResolver {
        fn resolve(&self, client_hello: ClientHello<'_>) -> Option<Arc<CertifiedKey>> {
            client_hello.server_name().map_or_else(
                // No SNI at all (an IP-address client, or an old one): serve the
                // operator's certificate, exactly as before custom domains.
                || Some(self.base.current()),
                |name| self.certificate_for(name),
            )
        }
    }
}

#[cfg(feature = "tls")]
pub use sni::{CustomDomainCertCache, SniCertResolver, SniCertSource};

// ── Lock helpers ─────────────────────────────────────────────────────────

/// Read a lock, recovering from poisoning.
///
/// A panic while holding one of these locks leaves the registry index
/// readable and internally consistent (every mutation is a whole-record
/// replacement), so refusing to serve every tenant afterwards would be a
/// worse outcome than continuing.
fn read_lock<T>(lock: &RwLock<T>) -> std::sync::RwLockReadGuard<'_, T> {
    lock.read()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
}

/// Write a lock, recovering from poisoning. See [`read_lock`].
fn write_lock<T>(lock: &RwLock<T>) -> std::sync::RwLockWriteGuard<'_, T> {
    lock.write()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
}

/// Kani proofs over the certificate retry backoff. Run with
/// `cargo kani -p autumn-web --lib`.
#[cfg(kani)]
mod kani_proofs {
    use super::*;

    /// The delay is 0 before any failure, at most `max` after one, never
    /// overflows, and never decreases as failures add up.
    #[kani::proof]
    fn backoff_is_capped_and_monotonic() {
        let failures: u32 = kani::any();
        let base: u64 = kani::any();
        let max: u64 = kani::any();
        let delay = backoff_secs(failures, base, max);
        if failures == 0 {
            assert_eq!(delay, 0);
        } else {
            assert!(delay <= max);
        }
        if failures < u32::MAX {
            assert!(delay <= backoff_secs(failures + 1, base, max));
        }
    }

    /// The first failure waits exactly `min(base, max)`.
    #[kani::proof]
    fn first_failure_waits_the_base() {
        let base: u64 = kani::any();
        let max: u64 = kani::any();
        assert_eq!(backoff_secs(1, base, max), base.min(max));
    }
}

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

    /// The write-gate map must stay bounded: every hostname ever gated used to
    /// keep an entry for the life of the process, and two paths made that
    /// unbounded rather than merely bounded by the registry — a refused
    /// registration (the gate was minted before the cap was checked) and an
    /// offboarded domain, whose gate outlived it.
    #[tokio::test]
    async fn write_gates_are_bounded_by_the_registry_not_by_hostnames_ever_seen() {
        let registry = CustomDomainRegistry::new(Arc::new(MemoryCustomDomainStore::new()), 2);
        // `register` refuses until a load has succeeded, as it does at boot.
        registry.load().await.unwrap();
        for host in ["a.clientco.com", "b.clientco.com"] {
            registry.register(host, "tenant-a", 0).await.unwrap();
        }

        // A tenant-facing endpoint hammered with unique hostnames against a
        // full registry mints nothing: the cap is checked before the gate.
        for i in 0..500 {
            let host = format!("flood-{i}.clientco.com");
            assert!(matches!(
                registry.register(&host, "tenant-evil", 0).await,
                Err(RegisterError::LimitReached { .. })
            ));
        }
        assert_eq!(
            registry.writes.lock().await.len(),
            2,
            "a refused registration must not leave a gate behind"
        );

        // And a long churn of register/offboard cycles is swept rather than
        // keeping one entry per hostname ever seen.
        for host in ["a.clientco.com", "b.clientco.com"] {
            assert!(registry.remove(host).await.unwrap());
        }
        for i in 0..300 {
            let host = format!("churn-{i}.clientco.com");
            registry.register(&host, "tenant-a", 0).await.unwrap();
            assert!(registry.remove(&host).await.unwrap());
        }
        let gates = registry.writes.lock().await.len();
        assert!(
            gates <= registry.gate_high_water(),
            "idle gates must be swept, but {gates} are held"
        );
    }

    #[test]
    fn apex_detection() {
        assert!(is_apex("clientco.com"));
        assert!(!is_apex("app.clientco.com"));
        assert!(!is_apex("a.b.clientco.com"));
    }

    /// #2643: an apex under a multi-label public suffix has three labels.
    #[test]
    fn apex_detection_uses_the_public_suffix_list() {
        for apex in ["clientco.co.uk", "customer.com.au", "tenant.co.jp"] {
            assert!(is_apex(apex), "{apex} is an apex");
            let sub = format!("app.{apex}");
            assert!(!is_apex(&sub), "{sub} is a subdomain");
        }
        // Wildcard rule `*.ck` and its exception `!www.ck`.
        assert!(is_apex("foo.bar.ck"));
        assert!(is_apex("www.ck"));
        assert!(!is_apex("app.www.ck"));
        // A private-section suffix delegates a zone too.
        assert!(is_apex("clientco.eu.org"));
        // An unlisted TLD keeps the two-label rule.
        assert!(is_apex("clientco.internal"));
        assert!(!is_apex("app.clientco.internal"));
        // A public suffix is itself the top of a zone.
        assert!(is_apex("co.uk"));
    }

    /// The lookup ignores case and a trailing dot.
    #[test]
    fn apex_detection_ignores_case_and_a_trailing_dot() {
        assert!(is_apex("ClientCo.CO.UK"));
        assert!(is_apex("clientco.co.uk."));
        assert!(!is_apex("APP.ClientCo.co.uk"));
    }

    #[test]
    fn backoff_doubles_and_caps() {
        assert_eq!(backoff_secs(0, 300, 3600), 0);
        assert_eq!(backoff_secs(1, 300, 3600), 300);
        assert_eq!(backoff_secs(4, 300, 3600), 2400);
        assert_eq!(backoff_secs(5, 300, 3600), 3600);
        assert_eq!(backoff_secs(u32::MAX, 300, 3600), 3600);
    }

    #[test]
    fn a_cname_target_is_compared_case_and_dot_insensitively() {
        let expected = ExpectedIngress {
            hostname: Some("Ingress.MyApp.com".to_owned()),
            ..ExpectedIngress::default()
        };
        assert_eq!(
            grade_dns_verification(
                &ObservedTarget::Cname("ingress.myapp.com.".to_owned()),
                &expected
            ),
            VerificationOutcome::PointsHere
        );
    }

    #[test]
    fn a_partial_address_match_is_not_a_pass() {
        let expected = ExpectedIngress {
            ipv4: vec!["203.0.113.10".parse().unwrap()],
            ..ExpectedIngress::default()
        };
        let observed = ObservedTarget::Addresses(vec![
            "203.0.113.10".parse().unwrap(),
            "198.51.100.7".parse().unwrap(),
        ]);
        match grade_dns_verification(&observed, &expected) {
            VerificationOutcome::PointsElsewhere { detail } => {
                assert!(detail.contains("198.51.100.7"), "{detail}");
                assert!(!detail.contains("203.0.113.10"), "{detail}");
            }
            other => panic!("expected a mismatch, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn the_filesystem_store_round_trips_and_deletes() {
        let dir = tempfile::tempdir().unwrap();
        let store = FsCustomDomainStore::new(dir.path());
        let domain = CustomDomain::new(
            "app.clientco.com".to_owned(),
            "t1".to_owned(),
            100,
            "token".to_owned(),
        );
        store.save(&domain).await.unwrap();

        let loaded = store.load_all().await.unwrap();
        assert_eq!(loaded, vec![domain.clone()]);

        store.delete("app.clientco.com").await.unwrap();
        assert!(store.load_all().await.unwrap().is_empty());
        // Deleting again is not an error.
        store.delete("app.clientco.com").await.unwrap();
    }

    #[tokio::test]
    async fn the_memory_ledger_keeps_recent_attempts_and_prunes_old_ones() {
        let store = MemoryCustomDomainStore::new();
        store.append_attempt(1000).await.unwrap();
        store.append_attempt(1000).await.unwrap();
        // The next write is past the global window: the first two go.
        let later = 1000 + GLOBAL_WINDOW_SECS + 1;
        store.append_attempt(later).await.unwrap();
        assert_eq!(store.load_attempts().await.unwrap(), vec![later]);
    }

    #[tokio::test]
    async fn the_filesystem_ledger_survives_a_new_store_and_prunes() {
        let dir = tempfile::tempdir().unwrap();
        let store = FsCustomDomainStore::new(dir.path());
        assert!(store.load_attempts().await.unwrap().is_empty());
        store.append_attempt(1000).await.unwrap();
        store.append_attempt(1000).await.unwrap();

        let reopened = FsCustomDomainStore::new(dir.path());
        assert_eq!(reopened.load_attempts().await.unwrap(), vec![1000, 1000]);

        let later = 1000 + GLOBAL_WINDOW_SECS + 1;
        reopened.append_attempt(later).await.unwrap();
        assert_eq!(reopened.load_attempts().await.unwrap(), vec![later]);
    }

    #[tokio::test]
    async fn the_ledger_file_is_not_read_as_a_domain_record() {
        let dir = tempfile::tempdir().unwrap();
        let store = FsCustomDomainStore::new(dir.path());
        store.append_attempt(1000).await.unwrap();
        assert!(store.load_all().await.unwrap().is_empty());
    }

    #[tokio::test]
    async fn a_corrupt_filesystem_ledger_reads_as_empty() {
        let dir = tempfile::tempdir().unwrap();
        let store = FsCustomDomainStore::new(dir.path());
        store.append_attempt(1000).await.unwrap();
        std::fs::write(dir.path().join(LEDGER_FILE), b"not json").unwrap();
        assert!(store.load_attempts().await.unwrap().is_empty());
        // The next write repairs it and keeps the bad file.
        store.append_attempt(2000).await.unwrap();
        assert!(dir.path().join("attempts.ledger.corrupt").exists());
        assert_eq!(store.load_attempts().await.unwrap(), vec![2000]);
    }

    #[test]
    fn hydrating_the_global_ledger_is_idempotent_and_never_double_counts() {
        let limiter = IssuanceLimiter::new(5, 2, 300, 3600);
        // One order is already known from a domain record.
        limiter.hydrate([("a.test".to_owned(), 1000)], 1100);
        // The ledger lists that order again, and one for an offboarded domain.
        limiter.hydrate_global([1000, 1010], 1100);
        limiter.hydrate_global([1000, 1010], 1100);
        assert!(matches!(
            limiter.check("c.test", 1100),
            IssuanceDecision::GlobalLimit { .. }
        ));
        // Exactly two hits: one expires, one stays.
        assert_eq!(
            limiter.check("c.test", 1000 + GLOBAL_WINDOW_SECS + 1),
            IssuanceDecision::Allow
        );
    }

    #[test]
    fn hydrating_the_global_ledger_ignores_future_attempts() {
        let limiter = IssuanceLimiter::new(5, 1, 300, 3600);
        limiter.hydrate_global([i64::MAX, 2000], 1000);
        assert_eq!(limiter.check("a.test", 1000), IssuanceDecision::Allow);
    }

    #[tokio::test]
    async fn a_future_ledger_entry_is_dropped_on_the_next_write() {
        let store = MemoryCustomDomainStore::new();
        store.append_attempt(5000).await.unwrap();
        store.append_attempt(1000).await.unwrap();
        assert_eq!(store.load_attempts().await.unwrap(), vec![1000]);
    }

    #[tokio::test]
    async fn merging_adds_only_missing_in_window_orders_in_both_stores() {
        let dir = tempfile::tempdir().unwrap();
        let fs = FsCustomDomainStore::new(dir.path());
        let mem = MemoryCustomDomainStore::new();
        let now = 5000;
        let old = now - GLOBAL_WINDOW_SECS - 1;
        for store in [&fs as &dyn CustomDomainStore, &mem] {
            store.append_attempt(4000).await.unwrap();
            // 4000 is known. 4500 is new. `old` and a future instant are ignored.
            store
                .merge_attempts(vec![4000, 4500, old, now + 9], now)
                .await
                .unwrap();
            store.merge_attempts(vec![4000, 4500], now).await.unwrap();
            let mut got = store.load_attempts().await.unwrap();
            got.sort_unstable();
            assert_eq!(got, vec![4000, 4500]);
        }
    }

    #[tokio::test]
    async fn concurrent_filesystem_appends_lose_no_entry() {
        let dir = tempfile::tempdir().unwrap();
        let store = Arc::new(FsCustomDomainStore::new(dir.path()));
        let tasks: Vec<_> = (0..16)
            .map(|_| {
                let store = Arc::clone(&store);
                tokio::spawn(async move { store.append_attempt(1000).await.unwrap() })
            })
            .collect();
        for t in tasks {
            t.await.unwrap();
        }
        assert_eq!(store.load_attempts().await.unwrap().len(), 16);
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn the_filesystem_ledger_is_owner_only() {
        use std::os::unix::fs::PermissionsExt as _;
        let dir = tempfile::tempdir().unwrap();
        let store = FsCustomDomainStore::new(dir.path());
        store.append_attempt(1000).await.unwrap();
        let mode = std::fs::metadata(dir.path().join(LEDGER_FILE))
            .unwrap()
            .permissions()
            .mode();
        assert_eq!(mode & 0o777, 0o600);
    }

    #[test]
    fn hydrating_the_global_ledger_ignores_attempts_outside_the_window() {
        let limiter = IssuanceLimiter::new(5, 1, 300, 3600);
        limiter.hydrate_global([1000], 1000 + GLOBAL_WINDOW_SECS + 1);
        assert_eq!(
            limiter.check("a.test", 1000 + GLOBAL_WINDOW_SECS + 1),
            IssuanceDecision::Allow
        );
    }

    #[tokio::test]
    async fn a_registry_hydrates_from_its_store() {
        let dir = tempfile::tempdir().unwrap();
        let store = Arc::new(FsCustomDomainStore::new(dir.path()));
        let first = CustomDomainRegistry::new(store.clone(), 10);
        first.load().await.unwrap();
        first.register("app.clientco.com", "t1", 100).await.unwrap();
        first
            .record_active("app.clientco.com", 100, 200)
            .await
            .unwrap();

        let second = CustomDomainRegistry::new(store, 10);
        assert_eq!(second.load().await.unwrap(), 1);
        assert_eq!(
            second.tenant_for_host("app.clientco.com").as_deref(),
            Some("t1")
        );
    }

    #[tokio::test]
    async fn re_registering_the_same_pair_is_idempotent() {
        let registry = CustomDomainRegistry::new(Arc::new(MemoryCustomDomainStore::new()), 10);
        registry.load().await.unwrap();
        registry
            .register("app.clientco.com", "t1", 100)
            .await
            .unwrap();
        registry
            .record_active("app.clientco.com", 100, 200)
            .await
            .unwrap();
        let again = registry
            .register("app.clientco.com", "t1", 500)
            .await
            .unwrap();
        assert_eq!(
            again.status,
            DomainStatus::Active,
            "re-registering must not reset a live domain"
        );
    }

    /// `is_servable` answers without copying the tenant id out, so pin that it
    /// still gives exactly `tenant_for_host(..).is_some()`: only an `Active`
    /// record serves, under any spelling the hostname normaliser folds.
    #[tokio::test]
    async fn is_servable_only_for_an_active_domain_in_any_spelling() {
        let registry = CustomDomainRegistry::new(Arc::new(MemoryCustomDomainStore::new()), 10);
        registry.load().await.unwrap();
        registry
            .register("app.clientco.com", "t1", 100)
            .await
            .unwrap();
        assert!(
            !registry.is_servable("app.clientco.com"),
            "pending_dns must not serve"
        );
        assert!(registry.tenant_for_host("app.clientco.com").is_none());

        registry
            .record_active("app.clientco.com", 100, 200)
            .await
            .unwrap();
        for spelling in [
            "app.clientco.com",
            "APP.ClientCo.com",
            "app.clientco.com.",
            "app.clientco.com:443",
        ] {
            assert!(registry.is_servable(spelling), "{spelling}");
            assert_eq!(
                registry.is_servable(spelling),
                registry.tenant_for_host(spelling).is_some(),
                "{spelling}"
            );
        }
        assert_eq!(
            registry.tenant_for_host("APP.ClientCo.com").as_deref(),
            Some("t1")
        );

        // Unregistered and unparseable names never serve.
        assert!(!registry.is_servable("other.clientco.com"));
        assert!(!registry.is_servable(""));
        assert!(!registry.is_servable("*.clientco.com"));

        // Nor does a removed one.
        assert!(registry.remove("app.clientco.com").await.unwrap());
        assert!(!registry.is_servable("app.clientco.com"));
    }

    #[tokio::test]
    async fn a_failed_order_returns_to_verified_not_pending() {
        let registry = CustomDomainRegistry::new(Arc::new(MemoryCustomDomainStore::new()), 10);
        registry.load().await.unwrap();
        registry
            .register("app.clientco.com", "t1", 100)
            .await
            .unwrap();
        registry
            .record_verified("app.clientco.com", 100)
            .await
            .unwrap();
        registry.record_issuing("app.clientco.com").await.unwrap();
        registry
            .record_failure("app.clientco.com", 100, "CA said no", 300)
            .await
            .unwrap();
        let record = registry.get("app.clientco.com").unwrap();
        assert_eq!(record.status, DomainStatus::Verified);
        assert_eq!(record.next_attempt_unix, Some(400));
        assert!(!record.is_due(399));
        assert!(record.is_due(400));
    }

    #[test]
    fn a_spent_budget_reports_when_it_rolls() {
        let limiter = IssuanceLimiter::new(1, 10, 300, 3600);
        limiter.record_attempt("a.test", 1000);
        match limiter.check("a.test", 1500) {
            IssuanceDecision::PerDomainLimit { retry_after_secs } => {
                assert_eq!(retry_after_secs, 1000 + PER_DOMAIN_WINDOW_SECS - 1500);
            }
            other => panic!("expected a per-domain refusal, got {other:?}"),
        }
        // Once the window has rolled the domain is allowed again.
        assert_eq!(
            limiter.check("a.test", 1000 + PER_DOMAIN_WINDOW_SECS + 1),
            IssuanceDecision::Allow
        );
        // Offboarding clears the history.
        limiter.forget("a.test");
        assert_eq!(limiter.check("a.test", 1500), IssuanceDecision::Allow);
    }

    #[test]
    fn hydrating_restores_both_windows_and_is_idempotent() {
        let limiter = IssuanceLimiter::new(2, 2, 300, 3600);
        let persisted = vec![
            ("a.test".to_owned(), 1000),
            ("b.test".to_owned(), 1100),
            // Outside the per-domain window: ignored.
            ("a.test".to_owned(), 1000 - PER_DOMAIN_WINDOW_SECS),
        ];
        let now = 1200;
        limiter.hydrate(persisted.clone(), now);
        limiter.hydrate(persisted, now);

        // Two orders inside the hour: the global budget of 2 is spent.
        assert!(matches!(
            limiter.check("c.test", now),
            IssuanceDecision::GlobalLimit { .. }
        ));
        // And it rolls with the hour, not at a made-up instant.
        assert_eq!(
            limiter.check("c.test", 1000 + GLOBAL_WINDOW_SECS + 1),
            IssuanceDecision::Allow
        );
    }

    #[test]
    fn hydrating_counts_two_attempts_in_the_same_second_twice() {
        // A crash loop places orders in the same Unix second; collapsing them
        // would hand the budget back on the next restart.
        let limiter = IssuanceLimiter::new(5, 2, 300, 3600);
        let now = 2000;
        let twice = vec![("a.test".to_owned(), 1500), ("a.test".to_owned(), 1500)];
        limiter.hydrate(twice.clone(), now);
        limiter.hydrate(twice, now);
        assert!(matches!(
            limiter.check("b.test", now),
            IssuanceDecision::GlobalLimit { .. }
        ));
    }

    #[test]
    fn a_record_stored_before_attempts_were_persisted_still_loads() {
        let old = r#"{"hostname":"a.test","tenant":"t","status":"active","failure_reason":null,
            "registered_at_unix":1,"verified_at_unix":null,"activated_at_unix":null,
            "cert_not_after_unix":null,"consecutive_failures":0,"next_attempt_unix":null}"#;
        let record: CustomDomain = serde_json::from_str(old).unwrap();
        assert!(record.issuance_attempts_unix.is_empty());
    }

    #[test]
    fn the_budget_says_nothing_about_backoff() {
        // The retry deadline is the domain record's job; the limiter only ever
        // reports a spent budget, so a failing domain is not refused twice.
        let limiter = IssuanceLimiter::new(10, 10, 300, 3600);
        assert_eq!(limiter.check("a.test", 1000), IssuanceDecision::Allow);
        assert_eq!(limiter.backoff_for(3), 1200);
    }
}