gunnar_server/

feed.rs

//! **The estate values the installer's door FEEDS**, taken instead of the
//! medium's — so one medium serves any estate.
//!
//! The join reply the pairing loop receives carries a `feed` table
//! (`nordisk_pairing::join::JoinReply::feed`); the loop persists the whole
//! reply to `<data>/pairing/joined.toml` ([`crate::pairing::JOINED_REL`]).
//! Three of its keys configure this box:
//!
//! | key               | what it replaces on the medium                       | applied            |
//! |-------------------|------------------------------------------------------|--------------------|
//! | `monetize_pubkey` | `GUNNAR_MONETIZE_PUBKEY` / `--monetize-pubkey`        | at start, or LIVE when it arrives |
//! | `clock_key`       | `ESTATE_CLOCK_KEY` / `GUNNAR_CLOCK_KEY` (`korp.env.` / `clock_key=` param) | on every clock sync|
//! | `clock_peer`      | `ESTATE_CLOCK` / `GUNNAR_CLOCK` (`korp.env.` / `clock=` param)         | on every clock sync|
//!
//! (`admin_pubkey` is the fourth feed key; the pairing loop installs it itself.)
//!
//! # Precedence: there is none — a disagreement is REFUSED by name
//!
//! Medium alone → the medium's. Door alone → the door's. Both and EQUAL →
//! that value. Both and DIFFERENT → [`FEED_MISMATCH`], never a silent pick:
//! the medium and the installer disagree about which estate this box is in,
//! and choosing either one is choosing whose money or whose time to trust
//! without saying so (the reasoning of ops' `monetize-pubkey-mismatch`
//! preflight, which refuses a bring-up over the same disagreement).
//!
//! * The monetize key at start: a refusal to start ([`resolve_monetize`]).
//! * The monetize key arriving while running (first boot): the key in force
//!   stays, the `Admin.Health` alarm [`FEED_MISMATCH`] is raised.
//! * The clock: every sync reads the effective configuration
//!   ([`clock_config`]), and a disagreement is the sync's `misconfigured`
//!   report naming [`FEED_MISMATCH`] — the clock is left alone.
//!
//! # The monetize key applies LIVE (owner 2026-09-25: avoid restarts)
//!
//! A metered box on a `fed` medium starts KEYLESS: its entitlement slot and
//! its growth go-ahead policy are built with no key (every namespace Free,
//! `Entitlement.Set` refused, a `monetize` go-ahead refused) and registered
//! as takers ([`on_monetize_key`]). When the installer feeds the key (or vm1
//! offers it), [`monetize_arrived`] hands it to each taker, which swaps it in
//! place — the slot re-verifies every stored fact under it — and the key is
//! persisted in `joined.toml` for the next start as before. Until 2026-09-25
//! this was a restart of the machine. The clock (`clock_key`, `clock_peer`)
//! and the fed principals were already live: the clock reads the feed on
//! every sync, and the primary installs principals on every feed.
//!
//! # The bus is the shared one (L10, 2026-10-20)
//!
//! Every feed this process takes is published on nordisk-pairing's process-wide
//! feed bus ([`feedbus::publish`], Skidbladnir 0.5.15, A7 row 8): a part that
//! applies a fed value LIVE subscribes with [`feedbus::on_published`] and reads
//! the typed [`Fed::from_feed`]; [`Fed::current`] is the typed view of the
//! value in force. Gunnar's own bus (this module's `publish`, `on_published`,
//! `current`) is gone; this module is the typed view over the shared one.

use std::path::Path;
use std::sync::{Mutex, OnceLock, RwLock};

use nordisk_pairing::feedbus::{self, Feed};

/// The relay a sold estate's main mails through, as the shop feeds it (monetize-server
/// `estate_order::FEED_SMTP_*`, P1).
pub const FEED_SMTP_URL: &str = "secret.smtp_url";
/// **The front's egress door** (Skidbladnir nordisk-pairing `egress::FEED_EGRESS`, the lean
/// estate, A19): fed by the shop to a box laid with no public IPv4. The box's clients that
/// reach the internet (the identity provider's, the webhooks', the mirrors', the mail relay's)
/// leave through it ([`take_egress`], `nordisk_pairing::egress::proxy`).
pub const FEED_EGRESS: &str = nordisk_pairing::egress::FEED_EGRESS;
pub const FEED_SMTP_FROM: &str = "secret.smtp_from";
/// One more root the relay's certificate is trusted under (PEM; a relay behind a private CA,
/// RA5-23). Empty or absent: the public roots alone.
pub const FEED_SMTP_CA: &str = "secret.smtp_ca";

/// The feed keys this module reads (the contract's names; never renamed).
pub const FEED_MONETIZE: &str = "monetize_pubkey";
pub const FEED_CLOCK_KEY: &str = "clock_key";
pub const FEED_CLOCK_PEER: &str = "clock_peer";
/// **The operator's internal networks a company IdP may live on** (owner
/// 2026-10-06, a private gunnar): a list of CIDRs, estate config the
/// installer feeds; never an account admin's. Empty or absent: today's guard
/// (`gunnar_grpc::sso::PrivateNets`). Applied LIVE on every feed.
pub const FEED_SSO_PRIVATE_IDP_CIDRS: &str = "sso_private_idp_cidrs";
/// **The operator's switch for company SSO's network** (A34-16, owner rule: default =
/// AIRGAP, outbound needs explicit config): exactly `on` lets this box dial an account's
/// identity provider; absent or anything else, no IdP is dialled and a sign-in through one
/// is refused by name (`[sso-egress-off]`). Estate config the installer feeds; never an
/// account admin's. Applied LIVE. A box with no feed takes `GUNNAR_SSO_EGRESS=on`.
pub const FEED_SSO_EGRESS: &str = "sso_egress";
/// **The DNS-over-HTTPS resolver an e-mail domain's proof is read from** (A34-5): an
/// https URL answering `application/dns-json` (dns.google/resolve, Cloudflare's
/// dns-query); absent: no domain proof is read (RA5-24, no default resolver). Estate
/// config; applied LIVE.
pub const FEED_SSO_DNS_RESOLVER: &str = "sso_dns_resolver";
/// **This estate SELLS room** (2026-10-08): `true` on a selling shop's boxes (gunnar.rs,
/// test2), absent everywhere else — a SOLD estate and the open build keep "no fact is
/// uncapped". Applied LIVE ([`Fed::sells_room`], gunnar-grpc `selling`).
pub const FEED_SELLS_ROOM: &str = "sells_room";
/// **The ONE monetize's `Fleet` door** (5.4, ONE monetize, M-D35): what `--fleet` names on
/// the medium. shop-pairing feeds it to the front, the front offers it to every box; the
/// box's pull channel reports there ([`crate::repair::resolve_fleet`]).
pub const FEED_FLEET: &str = "fleet";
/// **The estate this box reports for** (`--fleet-instance`), fed beside [`FEED_FLEET`].
pub const FEED_FLEET_INSTANCE: &str = "fleet_instance";
/// **Where the pull channel connects** (`host:port`), fed beside [`FEED_FLEET`] when the
/// door's name is not what this box should dial (a test bench whose guests resolve no
/// names). TLS still verifies the certificate under [`FEED_FLEET`]'s host. Absent: the
/// URL's own host, resolved. A BENCH address only (ST-19, [`crate::repair::seam_refusal`]).
pub const FEED_FLEET_AT: &str = "fleet_at";
/// **The PEM the pull channel's door is checked against**, fed beside [`FEED_FLEET`] when
/// its certificate is not under the public roots. Absent: the public roots. Only beside a
/// bench [`FEED_FLEET_AT`] (ST-19).
pub const FEED_FLEET_CA: &str = "fleet_ca";

/// **The SITE's pay page** (UW-6, feed key `refusal_link`: the open build names no money): a `{tenant}` template, `https://<site>/billing/{tenant}`
/// — a page the site's own console serves. The medium is one for every estate, so the
/// page it bakes (`--billing-url`) is a placeholder this replaces, LIVE
/// ([`crate::entitlement::EntitlementSlot::set_billing_url`]).
pub const FEED_BILLING_URL: &str = "refusal_link";
/// **Auto-fix, fed** (main rebuild, 2026-10-12): exactly `true` turns the box's auto-fix on,
/// as `--repair-auto` does. The estate's opt-in, carried by its installer; nothing else
/// changes what the guard allows (`gunnar_repair::rebuild`).
pub const FEED_REPAIR_AUTO: &str = "repair_auto";
/// **The repair key, fed** (main rebuild, 2026-10-12): an OpenSSH private key this box signs
/// its repair calls with when no `--repair-key` is given — a re-laid main holds no twin key,
/// and without one every pulled order and the auto recharge were refused. The installer
/// feeds its public half as a principal, so the twin reads its log to it and a fresh main
/// takes its `Bootstrap.Restore`.
pub const FEED_REPAIR_KEY: &str = "secret.repair_key";

/// The alarm/refusal word for a medium value and a DIFFERENT fed value.
pub const FEED_MISMATCH: &str = "pairing.feed_mismatch";
/// The alarm for a fed monetize key this running process could not apply
/// (no entitlement slot to apply it to, or a taker refused it). It used to be
/// `pairing.feed_restart` and a reboot; the key is applied live now.
pub const FEED_NOT_APPLIED: &str = "pairing.feed_not_applied";

/// **`vm`'s offer of `key`** (`offer.<vm>.<key>`), trimmed; blank is absent.
pub fn offered_by<'a>(
    feed: &'a std::collections::BTreeMap<String, String>,
    vm: &str,
    key: &str,
) -> Option<&'a str> {
    feed.get(&format!("offer.{vm}.{key}"))
        .map(|v| v.trim())
        .filter(|v| !v.is_empty())
}

/// What the door fed, as far as this box's configuration goes.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Fed {
    pub monetize_pubkey: Option<String>,
    pub clock_key: Option<String>,
    pub clock_peer: Option<String>,
    /// [`FEED_SSO_PRIVATE_IDP_CIDRS`], as fed.
    pub sso_private_idp_cidrs: Option<String>,
    /// [`FEED_SSO_EGRESS`]: exactly `on`, as fed; anything else is off.
    pub sso_egress: bool,
    /// [`FEED_SSO_DNS_RESOLVER`], as fed.
    pub sso_dns_resolver: Option<String>,
    /// [`FEED_SELLS_ROOM`]: exactly `true`, as fed; anything else is false.
    pub sells_room: bool,
    /// [`FEED_FLEET`], as fed.
    pub fleet: Option<String>,
    /// [`FEED_FLEET_INSTANCE`], as fed.
    pub fleet_instance: Option<String>,
    /// [`FEED_FLEET_AT`], as fed.
    pub fleet_at: Option<String>,
    /// [`FEED_FLEET_CA`], as fed.
    pub fleet_ca: Option<String>,
    /// [`FEED_SMTP_URL`], as fed: the relay a SOLD estate's main mails through (P1). Its
    /// `Debug` withholds it (it carries the submission token).
    pub smtp_url: Option<Withheld>,
    /// [`FEED_SMTP_FROM`], as fed.
    pub smtp_from: Option<String>,
    /// [`FEED_SMTP_CA`], as fed (a PEM root for the relay's TLS).
    pub smtp_ca: Option<String>,
    /// [`FEED_BILLING_URL`], as fed.
    pub billing_url: Option<String>,
    /// [`FEED_REPAIR_AUTO`]: exactly `true`, as fed; anything else is false.
    pub repair_auto: bool,
    /// [`FEED_REPAIR_KEY`], as fed (withheld from `Debug`).
    pub repair_key: Option<Withheld>,
    /// [`FEED_EGRESS`], as fed: an `http://` door only.
    pub egress: Option<String>,
}

/// A fed value that is a credential: kept, compared, never shown.
#[derive(Clone, Default, PartialEq, Eq)]
pub struct Withheld(pub String);

impl std::fmt::Debug for Withheld {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("<withheld>")
    }
}

impl Fed {
    /// From the reply's `feed` table. Blank values are absent. A value the
    /// installer does not feed itself may reach this box as a VM's OFFER,
    /// fed onward as `offer.<vm>.<key>` (edda nordisk-pairing 0.3.1; since
    /// 0.4 the front's keys — monetize's verifying key, the clock key — come
    /// that way): the plain key wins, then the first offer by VM name.
    pub fn from_feed(feed: &std::collections::BTreeMap<String, String>) -> Fed {
        let get = |k: &str| {
            feed.get(k)
                .map(|v| v.trim().to_string())
                .filter(|v| !v.is_empty())
                .or_else(|| nordisk_pairing::feedbus::offered(feed, k))
        };
        Fed {
            monetize_pubkey: get(FEED_MONETIZE),
            clock_key: get(FEED_CLOCK_KEY),
            clock_peer: get(FEED_CLOCK_PEER),
            sso_private_idp_cidrs: get(FEED_SSO_PRIVATE_IDP_CIDRS),
            sso_egress: get(FEED_SSO_EGRESS).as_deref() == Some("on"),
            sso_dns_resolver: get(FEED_SSO_DNS_RESOLVER),
            sells_room: get(FEED_SELLS_ROOM).as_deref() == Some("true"),
            fleet: get(FEED_FLEET),
            fleet_instance: get(FEED_FLEET_INSTANCE),
            fleet_at: get(FEED_FLEET_AT),
            fleet_ca: get(FEED_FLEET_CA),
            smtp_url: get(FEED_SMTP_URL).map(Withheld),
            smtp_from: get(FEED_SMTP_FROM),
            smtp_ca: get(FEED_SMTP_CA),
            billing_url: get(FEED_BILLING_URL),
            repair_auto: get(FEED_REPAIR_AUTO).as_deref() == Some("true"),
            repair_key: get(FEED_REPAIR_KEY).map(Withheld),
            egress: get(FEED_EGRESS).filter(|d| d.starts_with("http://")),
        }
    }

    pub fn is_empty(&self) -> bool {
        self == &Fed::default()
    }

    /// `<data>/pairing/joined.toml`, if the loop has persisted one. Absent is
    /// `None`; present and unreadable is an error — a persisted answer that
    /// lies must not be quietly ignored.
    pub fn read(root: &Path) -> anyhow::Result<Option<Fed>> {
        Ok(kept_feed(root)?.as_ref().map(Fed::from_feed))
    }

    /// **The typed view of the feed in force** on the shared bus
    /// (nordisk-pairing [`feedbus::current`]); `None` before anything was published.
    pub fn current() -> Option<Fed> {
        feedbus::current().map(|feed| Fed::from_feed(&feed))
    }
}

/// **This box leaves through the door it was fed** ([`FEED_EGRESS`]): kept process-wide
/// (`nordisk_pairing::egress::set_proxy`, the first wins: the door is the front's overlay
/// address, the same for the life of the estate) for every client that reaches the internet.
/// Called with each feed published (`serve`); a feed with no door changes nothing.
pub fn take_egress(feed: &std::collections::BTreeMap<String, String>) {
    if let Some(door) = Fed::from_feed(feed).egress {
        if nordisk_pairing::egress::proxy().is_none() {
            tracing::info!(target: "gunnar::audit", event = "egress_door", door = %door, "this box leaves through its front's egress door (no public IPv4 of its own)");
        }
        nordisk_pairing::egress::set_proxy(&door);
    }
}

/// The proxy a ureq client of this box is built with: the fed door ([`take_egress`]), or none.
#[cfg(any(feature = "mirror-http", feature = "sso-egress"))]
pub fn ureq_proxy() -> Option<ureq::Proxy> {
    nordisk_pairing::egress::proxy().and_then(|d| ureq::Proxy::new(d).ok())
}

/// The feed of `<data>/pairing/joined.toml`, as the door fed it ([`Fed::read`]).
#[cfg(feature = "wg-client")]
pub fn kept_feed(root: &Path) -> anyhow::Result<Option<Feed>> {
    let p = root.join(crate::pairing::JOINED_REL);
    let Some(text) = crate::fed_at_rest::read_checked(root, crate::pairing::JOINED_REL)? else {
        return Ok(None);
    };
    let reply = nordisk_pairing::join::JoinReply::parse(&text)
        .map_err(|e| anyhow::anyhow!("{}: {e}", p.display()))?;
    Ok(Some(reply.feed))
}

/// A build without the pairing loop has no door to be fed by.
#[cfg(not(feature = "wg-client"))]
pub fn kept_feed(_root: &Path) -> anyhow::Result<Option<Feed>> {
    Ok(None)
}

/// **The operator's internal networks a company IdP may live on**, as `fed`
/// names them: none when nothing is fed. A list gunnar refuses (a public or
/// loopback range) is an error, said by the caller and applied as none.
pub fn sso_private_nets(fed: Option<&Fed>) -> Result<gunnar_grpc::sso::PrivateNets, String> {
    match fed.and_then(|f| f.sso_private_idp_cidrs.as_deref()) {
        Some(text) => gunnar_grpc::sso::PrivateNets::parse(text)
            .map_err(|why| format!("`{FEED_SSO_PRIVATE_IDP_CIDRS}`: {why}")),
        None => Ok(gunnar_grpc::sso::PrivateNets::default()),
    }
}

/// **Is company SSO's network on?** The feed's [`FEED_SSO_EGRESS`] (`on`), or, for a box
/// that is fed nothing, the operator's `GUNNAR_SSO_EGRESS=on` (`env`). Off otherwise.
pub fn sso_egress_on(fed: Option<&Fed>, env: Option<&str>) -> bool {
    fed.is_some_and(|f| f.sso_egress) || env.is_some_and(|v| v.trim() == "on")
}

/// **The resolver a domain's DNS proof is read from** (A34-5): the feed's
/// [`FEED_SSO_DNS_RESOLVER`], or, where the feed names none, the operator's
/// `GUNNAR_SSO_DNS_RESOLVER` (`env`), as [`sso_egress_on`] takes `GUNNAR_SSO_EGRESS`.
/// `None`: no resolver is named, and no proof is read (RA5-24: none is a default).
pub fn sso_dns_resolver(fed: Option<&Fed>, env: Option<&str>) -> Option<String> {
    fed.and_then(|f| f.sso_dns_resolver.clone()).or_else(|| {
        env.map(str::trim)
            .filter(|v| !v.is_empty())
            .map(str::to_owned)
    })
}

/// Read `joined.toml` under `root` and publish its feed on the shared bus
/// ([`feedbus::publish`]). Called once at start; an unreadable file is logged
/// and treated as no feed (the loop re-joins and rewrites it).
pub fn load_at_start(root: &Path) -> Option<Fed> {
    match kept_feed(root) {
        Ok(Some(feed)) => {
            let fed = Fed::from_feed(&feed);
            if !fed.is_empty() {
                tracing::info!(monetize = fed.monetize_pubkey.is_some(), clock_peer = ?fed.clock_peer, clock_key = fed.clock_key.is_some(), "feed: the installer's values from {}", "<data>/pairing/joined.toml");
            }
            feedbus::publish(feed);
            Some(fed)
        }
        Ok(None) => None,
        Err(e) => {
            tracing::error!(error = %format!("{e:#}"), "feed: the persisted join reply is unreadable — this start uses the medium's values only");
            None
        }
    }
}

// ── the monetize key ────────────────────────────────────────────────────────

/// The key the running server enforces with: recorded at start (outer `None`
/// = nothing recorded, e.g. a test), and moved ONCE more when an awaited key
/// arrives and is applied live ([`monetize_arrived`]).
fn in_force_slot() -> &'static RwLock<Option<Option<[u8; 32]>>> {
    static IN_FORCE: OnceLock<RwLock<Option<Option<[u8; 32]>>>> = OnceLock::new();
    IN_FORCE.get_or_init(|| RwLock::new(None))
}

pub fn record_monetize_in_force(key: Option<[u8; 32]>) {
    if let Ok(mut s) = in_force_slot().write() {
        *s = Some(key);
    }
}

pub fn monetize_in_force() -> Option<[u8; 32]> {
    in_force_slot().read().ok().and_then(|s| s.flatten())
}

/// Something that takes the monetize key LIVE: the entitlement slot (both
/// front doors' plan gate and `Entitlement.Set` read it) and the growth
/// go-ahead policy. Registered by `serve` once each is built.
pub type MonetizeTaker = Box<dyn Fn(&[u8; 32]) -> Result<(), String> + Send + Sync>;

fn takers() -> &'static Mutex<Vec<(&'static str, MonetizeTaker)>> {
    static TAKERS: OnceLock<Mutex<Vec<(&'static str, MonetizeTaker)>>> = OnceLock::new();
    TAKERS.get_or_init(|| Mutex::new(Vec::new()))
}

/// **Register a taker of the monetize key** (owner 2026-09-25: avoid
/// restarts) — called by `serve` for the entitlement slot and the go-ahead
/// policy it built keyless, so a key the installer feeds later is applied in
/// place by [`monetize_arrived`].
pub fn on_monetize_key(name: &'static str, take: MonetizeTaker) {
    if let Ok(mut t) = takers().lock() {
        t.push((name, take));
    }
}

fn hex(k: &[u8; 32]) -> String {
    k.iter().map(|b| format!("{b:02x}")).collect()
}

/// **The monetize key at start.** `medium` is the resolved key off
/// `--monetize-pubkey`/`GUNNAR_MONETIZE_PUBKEY`; `fed` is the door's spelling
/// (64 hex). A fed value that is not a key, or that differs from the
/// medium's, is refused by name.
pub fn resolve_monetize(
    medium: Option<[u8; 32]>,
    fed: Option<&str>,
) -> Result<Option<[u8; 32]>, String> {
    let fed = match fed {
        None => return Ok(medium),
        Some(text) => gunnar_common::monetize_key::parse_hex32(text.trim()).map_err(|why| {
            format!("REFUSED [pairing-feed-monetize-malformed] the installer fed `{FEED_MONETIZE}` = `{text}`, which is not a monetize key: {why}")
        })?,
    };
    match medium {
        None => Ok(Some(fed)),
        Some(m) if m == fed => Ok(Some(m)),
        Some(m) => Err(format!(
            "REFUSED [{FEED_MISMATCH}] the medium trusts monetize key {} and the installer's door fed {} — \
             this box will not pick one silently. Compose the medium without `monetize_pubkey` (the door \
             feeds it), or with the front's live key",
            hex(&m),
            hex(&fed)
        )),
    }
}

/// Why a fed monetize key found no taker, in the words THIS build can say.
///
/// The metered build names the flag that opens the slot. The open build has no
/// such flag (`--billing-url` is declared in `entitlement.rs`, which is
/// `metered`-only) and must not compile the word into anything an operator can
/// read — `tests/open_build_speaks_no_money.rs::the_open_sources_say_no_money`,
/// the owner's ruling of 2026-09-15. Before this split the open build carried
/// the metered sentence verbatim and the guard was red on it (feed.rs:243).
#[cfg(feature = "metered")]
const NO_TAKER_WHY: &str =
    "the metered start opened no entitlement slot (no --billing-url on a `fed` medium)";
#[cfg(not(feature = "metered"))]
const NO_TAKER_WHY: &str =
    "this build carries no entitlement module, so no start opens a slot to take it";

/// **The fed monetize key arriving while running** (the pairing loop's
/// answer, or vm1's offer carried by the installer): APPLIED IN PLACE by every
/// registered taker ([`on_monetize_key`]), with no restart (owner 2026-09-25).
/// `None` = nothing to do (no key fed, the same key already in force, or a
/// process that recorded none — a community build or a test).
pub fn monetize_arrived(fed: Option<&str>) -> Option<MonetizeArrival> {
    let in_force = in_force_slot().read().ok().and_then(|s| *s)?;
    let fed = fed?;
    match resolve_monetize(in_force, Some(fed)) {
        Ok(Some(k)) if Some(k) == in_force => None,
        Ok(Some(k)) => {
            let takers = takers().lock().ok()?;
            if takers.is_empty() {
                let detail = format!(
                    "the installer fed monetize key {} and nothing on this box can enforce it: {NO_TAKER_WHY}",
                    hex(&k)
                );
                raise(FEED_NOT_APPLIED, &detail);
                return Some(MonetizeArrival::NotApplied(detail));
            }
            let failed: Vec<String> = takers
                .iter()
                .filter_map(|(name, take)| take(&k).err().map(|e| format!("{name}: {e}")))
                .collect();
            if !failed.is_empty() {
                let detail = format!(
                    "the installer fed monetize key {}, and it could not be applied: {}",
                    hex(&k),
                    failed.join("; ")
                );
                raise(FEED_NOT_APPLIED, &detail);
                return Some(MonetizeArrival::NotApplied(detail));
            }
            record_monetize_in_force(Some(k));
            let detail = format!("the installer fed monetize key {}; it is IN FORCE now (applied in place, no restart)", hex(&k));
            tracing::info!(event = "pairing.feed_applied", "feed: {detail}");
            Some(MonetizeArrival::Applied(detail))
        }
        Ok(None) => None,
        Err(why) => {
            raise(FEED_MISMATCH, &why);
            Some(MonetizeArrival::Mismatch(why))
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MonetizeArrival {
    /// The server started with no key; the fed one is enforced from now on.
    Applied(String),
    /// The server started with no key and cannot take the fed one: alarmed.
    NotApplied(String),
    /// The medium's key is in force and the door fed another: refused, alarmed.
    Mismatch(String),
}

// ── the clock ───────────────────────────────────────────────────────────────

/// **The clock's two values**, medium against door — edda estate-clock's
/// (`offer::resolve_clock`, moved there from here AS IS; the refusal names
/// [`FEED_MISMATCH`], the same word). `Ok((clock, key))` is what
/// [`estate_clock::config::Config::parse`] is handed.
pub use estate_clock::offer::resolve_clock;

/// **The clock configuration in force now**: the process environment (the
/// medium's `korp.env.ESTATE_CLOCK*`, or Gunnar's `korp.env.GUNNAR_CLOCK*`
/// read as its alias, `estate_clock::config::var`) merged with the current
/// feed. Read on EVERY sync, so a feed that arrives while running applies to
/// the next one.
pub fn clock_config() -> Result<estate_clock::config::Config, String> {
    use estate_clock::config::{var, Config, Role, CLOCK_ENV, CLOCK_KEY_ENV};
    let env_clock = var(CLOCK_ENV);
    let env_key = var(CLOCK_KEY_ENV);
    let fed = Fed::current().unwrap_or_default();
    let (clock, key) = resolve_clock(
        env_clock.as_deref(),
        env_key.as_deref(),
        fed.clock_peer.as_deref(),
        fed.clock_key.as_deref(),
    )?;
    Config::parse(clock.as_deref(), key.as_deref(), Role::Appliance)
}

/// Whether the door fed a clock source.
pub fn clock_fed() -> bool {
    Fed::current().is_some_and(|f| f.clock_peer.is_some())
}

// ── alarms ──────────────────────────────────────────────────────────────────

fn alarm_slot() -> &'static Mutex<Vec<gunnar_proto::HealthAlarm>> {
    static ALARMS: OnceLock<Mutex<Vec<gunnar_proto::HealthAlarm>>> = OnceLock::new();
    ALARMS.get_or_init(|| Mutex::new(Vec::new()))
}

/// Raise (or re-raise, counting) a feed alarm; said on the log too.
pub fn raise(event: &str, detail: &str) {
    tracing::error!(event, "feed: {detail}");
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as i64)
        .unwrap_or(0);
    if let Ok(mut alarms) = alarm_slot().lock() {
        match alarms.iter_mut().find(|a| a.event == event) {
            Some(a) => {
                a.count += 1;
                a.detail = detail.to_string();
            }
            None => alarms.push(gunnar_proto::HealthAlarm {
                event: event.into(),
                count: 1,
                since_unix_ms: now,
                detail: detail.to_string(),
            }),
        }
    }
}

/// The feed alarms, for `Admin.Health`.
pub fn alarms() -> Vec<gunnar_proto::HealthAlarm> {
    alarm_slot().lock().map(|a| a.clone()).unwrap_or_default()
}

/// **The mock's seam, honoured only as the mock feeds it** (ST-19): a fed `fleet_at` is a
/// bench forward — an address literal `ip:port` on a private, loopback, link-local or
/// shared (100.64/10) network — and a fed `fleet_ca` comes only beside one. Anything else
/// (a public address, a name, a CA with no bench address) is refused by name: a release
/// box never dials elsewhere, nor re-roots its trust, on a feed's word.
pub fn seam_refusal(at: Option<&str>, ca: bool) -> Option<String> {
    const WORD: &str = "fleet-seam-bench-only";
    match at {
        None if ca => Some(format!(
            "REFUSED [{WORD}] the installer fed `{}` with no `{}`: a door's own CA is a bench's, beside a bench address only — a public door is checked under the public roots",
            crate::feed::FEED_FLEET_CA,
            crate::feed::FEED_FLEET_AT
        )),
        None => None,
        Some(at) => match at.trim().parse::<std::net::SocketAddr>() {
            Ok(a) if bench_address(a.ip()) => None,
            _ => Some(format!(
                "REFUSED [{WORD}] the installer fed `{}` = {at}: the pull channel is dialled elsewhere than its door's name only on a bench (a private, loopback or link-local `ip:port`)",
                crate::feed::FEED_FLEET_AT
            )),
        },
    }
}

/// A bench guest's network: private, loopback, link-local, or shared address space.
fn bench_address(ip: std::net::IpAddr) -> bool {
    match ip {
        std::net::IpAddr::V4(v) => {
            v.is_private()
                || v.is_loopback()
                || v.is_link_local()
                || (v.octets()[0] == 100 && (v.octets()[1] & 0xc0) == 64)
        }
        std::net::IpAddr::V6(v) => {
            v.is_loopback()
                || (v.segments()[0] & 0xfe00) == 0xfc00
                || (v.segments()[0] & 0xffc0) == 0xfe80
        }
    }
}

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

    /// ★ **P1: the relay a sold estate's main mails through comes in the feed** — read from
    /// `secret.smtp_url` / `secret.smtp_from`; the URL (it carries the token) never shows.
    #[test]
    fn the_fed_relay_is_read_and_its_url_withheld() {
        let feed: std::collections::BTreeMap<String, String> = [
            (FEED_SMTP_URL, "smtp://u:TOKENvalue@smtp.example:587"),
            (FEED_SMTP_FROM, "u@smtp.example"),
        ]
        .iter()
        .map(|(k, v)| (k.to_string(), v.to_string()))
        .collect();
        let f = Fed::from_feed(&feed);
        assert_eq!(
            f.smtp_url,
            Some(Withheld("smtp://u:TOKENvalue@smtp.example:587".into()))
        );
        assert_eq!(f.smtp_from.as_deref(), Some("u@smtp.example"));
        assert!(!format!("{f:?}").contains("TOKENvalue"), "{f:?}");
        assert!(Fed::from_feed(&Default::default()).smtp_url.is_none());
    }

    /// ★ **The front's egress door a box is fed** (the lean estate, A19 rows 10-14): read from
    /// `egress` (Skidbladnir nordisk-pairing's key), an `http://` door only.
    #[test]
    fn the_fed_egress_door_is_read_and_only_an_http_one() {
        let feed = |pairs: &[(&str, &str)]| {
            pairs
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect::<std::collections::BTreeMap<_, _>>()
        };
        assert_eq!(FEED_EGRESS, nordisk_pairing::egress::FEED_EGRESS);
        assert_eq!(
            Fed::from_feed(&feed(&[(FEED_EGRESS, " http://10.44.0.1:3128 ")]))
                .egress
                .as_deref(),
            Some("http://10.44.0.1:3128")
        );
        assert_eq!(
            Fed::from_feed(&feed(&[(FEED_EGRESS, "socks5://10.44.0.1:1080")])).egress,
            None,
            "only an http:// door is one"
        );
        assert_eq!(
            Fed::from_feed(&feed(&[])).egress,
            None,
            "a box with its own way out is fed none"
        );
    }

    const A: &str = "43cb00000000000000000000000000000000000000000000000000000000842a";
    const B: &str = "43cb00000000000000000000000000000000000000000000000000000000842b";

    fn k(s: &str) -> [u8; 32] {
        gunnar_common::monetize_key::parse_hex32(s).unwrap()
    }

    /// Since 0.4 the front's keys reach the boxes as vm1's offer; the plain
    /// key, when the installer feeds one, still wins.
    #[test]
    fn a_value_offered_by_a_vm_is_taken_when_the_installer_feeds_none_itself() {
        let feed = |pairs: &[(&str, &str)]| {
            pairs
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect::<std::collections::BTreeMap<_, _>>()
        };
        let f = Fed::from_feed(&feed(&[
            ("offer.front.monetize_pubkey", A),
            ("offer.front.clock_key", "ssh-ed25519 AAAAk"),
            ("clock_peer", "peer://10.13.8.1:50123"),
            ("offer.front.xmonetize_pubkey", B),
        ]));
        assert_eq!(
            f,
            Fed {
                monetize_pubkey: Some(A.into()),
                clock_key: Some("ssh-ed25519 AAAAk".into()),
                clock_peer: Some("peer://10.13.8.1:50123".into()),
                ..Fed::default()
            }
        );
        assert_eq!(
            Fed::from_feed(&feed(&[
                (FEED_MONETIZE, B),
                ("offer.front.monetize_pubkey", A)
            ]))
            .monetize_pubkey
            .as_deref(),
            Some(B),
            "the plain key wins"
        );
        assert_eq!(
            nordisk_pairing::feedbus::offered(
                &feed(&[
                    ("offer..monetize_pubkey", A),
                    ("offer.a.b.monetize_pubkey", A)
                ]),
                FEED_MONETIZE
            ),
            None,
            "not a VM name"
        );
    }

    /// ONE monetize (M-D35): shop-pairing feeds the front `fleet` and `fleet_instance`; the
    /// front offers them, and every box takes them — the plain key first, then the offer.
    #[test]
    fn the_fleet_door_and_the_estate_are_taken_from_the_feed_or_the_fronts_offer() {
        let feed = |pairs: &[(&str, &str)]| {
            pairs
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect::<std::collections::BTreeMap<_, _>>()
        };
        let f = Fed::from_feed(&feed(&[
            ("offer.front.fleet", "https://monetize.ignalina.dk:8443"),
            ("offer.front.fleet_instance", "acme"),
            ("offer.front.monetize_pubkey", A),
        ]));
        assert_eq!(
            (
                f.fleet.as_deref(),
                f.fleet_instance.as_deref(),
                f.monetize_pubkey.as_deref()
            ),
            (
                Some("https://monetize.ignalina.dk:8443"),
                Some("acme"),
                Some(A)
            )
        );
        let f = Fed::from_feed(&feed(&[
            (FEED_FLEET, "https://one:8443"),
            ("offer.front.fleet", "https://other:8443"),
            (FEED_FLEET_INSTANCE, " "),
            ("offer.front.fleet_instance", "acme"),
        ]));
        assert_eq!(
            (f.fleet.as_deref(), f.fleet_instance.as_deref()),
            (Some("https://one:8443"), Some("acme")),
            "the plain key wins; blank is absent"
        );
        assert!(Fed::from_feed(&feed(&[])).fleet.is_none());
    }

    /// ★ **`sells_room` is the installer's word, exactly `true`** (2026-10-08): fed on a
    /// selling shop's boxes, absent on a sold estate; anything else is false.
    #[test]
    fn sells_room_is_fed_as_exactly_true() {
        let feed = |pairs: &[(&str, &str)]| {
            pairs
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect::<std::collections::BTreeMap<_, _>>()
        };
        assert!(Fed::from_feed(&feed(&[(FEED_SELLS_ROOM, "true")])).sells_room);
        assert!(
            Fed::from_feed(&feed(&[(FEED_SELLS_ROOM, " true\n")])).sells_room,
            "trimmed, as fed"
        );
        assert!(!Fed::from_feed(&feed(&[(FEED_SELLS_ROOM, "yes")])).sells_room);
        assert!(
            !Fed::from_feed(&feed(&[])).sells_room,
            "a sold estate is fed nothing"
        );
    }

    /// ★ UW-6: **the site's pay page is the installer's word** (`refusal_link`), trimmed;
    /// blank or absent is none (the medium's page stays).
    #[test]
    fn the_sites_pay_page_is_read_from_the_feed() {
        let feed = |pairs: &[(&str, &str)]| {
            pairs
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect::<std::collections::BTreeMap<_, _>>()
        };
        assert_eq!(FEED_BILLING_URL, "refusal_link");
        assert_eq!(
            Fed::from_feed(&feed(&[(
                FEED_BILLING_URL,
                " https://test2.gunnar.rs/billing/{tenant}\n"
            )]))
            .billing_url
            .as_deref(),
            Some("https://test2.gunnar.rs/billing/{tenant}")
        );
        assert_eq!(
            Fed::from_feed(&feed(&[(FEED_BILLING_URL, " ")])).billing_url,
            None
        );
        assert_eq!(Fed::from_feed(&feed(&[])).billing_url, None);
    }

    /// ★ **Company SSO's network is the operator's to turn on** (A34-16): exactly `on` in
    /// the feed, or `GUNNAR_SSO_EGRESS=on` on a box fed nothing; absent, blank or anything
    /// else is off (airgap).
    #[test]
    fn sso_egress_is_off_unless_the_operator_turns_it_on() {
        let feed = |pairs: &[(&str, &str)]| {
            pairs
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect::<std::collections::BTreeMap<_, _>>()
        };
        let on = Fed::from_feed(&feed(&[(FEED_SSO_EGRESS, " on ")]));
        assert!(on.sso_egress);
        assert!(sso_egress_on(Some(&on), None));
        for word in ["off", "true", "yes", "ON?", ""] {
            let fed = Fed::from_feed(&feed(&[(FEED_SSO_EGRESS, word)]));
            assert!(!sso_egress_on(Some(&fed), None), "{word:?} turned it on");
        }
        assert!(!sso_egress_on(None, None), "nothing said is airgap");
        assert!(
            sso_egress_on(None, Some("on")),
            "a box fed nothing takes the operator's environment"
        );
        assert!(!sso_egress_on(None, Some("1")));
        let off = Fed::from_feed(&feed(&[]));
        assert!(
            sso_egress_on(Some(&off), Some("on")),
            "the environment holds when the feed says nothing"
        );
    }

    /// ★ **The domain proof's resolver is the operator's to name** (A34-5, RA5-24): the
    /// feed's `sso_dns_resolver`, or `GUNNAR_SSO_DNS_RESOLVER` on a box the feed names none
    /// for (as `GUNNAR_SSO_EGRESS` for the switch; gunnar-ui's uitest proves a domain this
    /// way, against its own loopback resolver); blank is none, and none is no proof.
    #[test]
    fn the_domain_proofs_resolver_is_fed_or_the_operators_environment() {
        let feed = |pairs: &[(&str, &str)]| {
            pairs
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect::<std::collections::BTreeMap<_, _>>()
        };
        let fed = Fed::from_feed(&feed(&[(
            FEED_SSO_DNS_RESOLVER,
            "https://doh.example/resolve",
        )]));
        assert_eq!(
            sso_dns_resolver(Some(&fed), None).as_deref(),
            Some("https://doh.example/resolve")
        );
        assert_eq!(
            sso_dns_resolver(Some(&fed), Some("http://127.0.0.1:9/resolve")).as_deref(),
            Some("https://doh.example/resolve"),
            "the feed's resolver wins over the environment"
        );
        let none = Fed::from_feed(&feed(&[]));
        assert_eq!(
            sso_dns_resolver(Some(&none), Some(" http://127.0.0.1:9/resolve ")).as_deref(),
            Some("http://127.0.0.1:9/resolve")
        );
        assert_eq!(
            sso_dns_resolver(None, Some("http://127.0.0.1:9/resolve")).as_deref(),
            Some("http://127.0.0.1:9/resolve"),
            "a box fed nothing"
        );
        assert_eq!(sso_dns_resolver(None, Some("  ")), None, "blank is none");
        assert_eq!(
            sso_dns_resolver(None, None),
            None,
            "nothing named is no resolver (RA5-24)"
        );
    }

    /// ★ **The operator's internal IdP networks are the installer's feed**
    /// (owner 2026-10-06): read from the feed, applied to every taker on
    /// every publish (live), parsed by gunnar's own rule — a public or
    /// loopback range is refused, nothing fed is none.
    #[test]
    fn the_operators_idp_networks_are_fed_and_applied_live() {
        let feed = |pairs: &[(&str, &str)]| {
            pairs
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect::<std::collections::BTreeMap<_, _>>()
        };
        let fed = Fed::from_feed(&feed(&[(FEED_SSO_PRIVATE_IDP_CIDRS, " 10.0.2.100/32 ")]));
        assert_eq!(fed.sso_private_idp_cidrs.as_deref(), Some("10.0.2.100/32"));
        assert_eq!(
            sso_private_nets(Some(&fed))
                .expect("a private network")
                .to_string(),
            "10.0.2.100/32"
        );
        assert!(sso_private_nets(None).expect("nothing fed").is_empty());
        let public = Fed::from_feed(&feed(&[(FEED_SSO_PRIVATE_IDP_CIDRS, "8.8.8.0/24")]));
        assert!(
            sso_private_nets(Some(&public)).is_err(),
            "a public range is refused"
        );
        let seen = std::sync::Arc::new(Mutex::new(Vec::<String>::new()));
        let into = std::sync::Arc::clone(&seen);
        feedbus::on_published("test: the idp networks", move |f: &Feed| {
            into.lock().unwrap().push(
                Fed::from_feed(f)
                    .sso_private_idp_cidrs
                    .clone()
                    .unwrap_or_default(),
            )
        });
        feedbus::publish(feed(&[(FEED_SSO_PRIVATE_IDP_CIDRS, " 10.0.2.100/32 ")]));
        feedbus::publish(feed(&[]));
        let got = seen.lock().unwrap().clone();
        let at = got
            .iter()
            .position(|v| v == "10.0.2.100/32")
            .expect("the list reached the taker");
        assert!(
            got[at + 1..].iter().any(String::is_empty),
            "every publish reaches the taker: {got:?}"
        );
    }

    /// **A role-bound offer is read from the VM that holds the role**: with the
    /// twin sorting first, "the first by name" is the twin's stale key.
    #[test]
    fn an_offer_is_read_from_the_named_vm_not_the_first_by_name() {
        let feed: std::collections::BTreeMap<String, String> = [
            ("offer.vm2.ticket_pubkey", A),
            ("offer.vm3.ticket_pubkey", B),
            ("offer.vm4.ticket_pubkey", " "),
        ]
        .iter()
        .map(|(k, v)| (k.to_string(), v.to_string()))
        .collect();
        assert_eq!(
            nordisk_pairing::feedbus::offered(&feed, "ticket_pubkey").as_deref(),
            Some(A),
            "the first by name"
        );
        assert_eq!(
            offered_by(&feed, "vm3", "ticket_pubkey"),
            Some(B),
            "the primary's own"
        );
        assert_eq!(
            offered_by(&feed, "vm4", "ticket_pubkey"),
            None,
            "blank is absent"
        );
        assert_eq!(offered_by(&feed, "vm9", "ticket_pubkey"), None);
    }

    #[test]
    fn monetize_door_alone_medium_alone_and_equal_are_taken() {
        assert_eq!(
            resolve_monetize(None, Some(A)),
            Ok(Some(k(A))),
            "one medium, any estate: the door's key"
        );
        assert_eq!(
            resolve_monetize(Some(k(A)), None),
            Ok(Some(k(A))),
            "no feed: today's behaviour"
        );
        assert_eq!(resolve_monetize(Some(k(A)), Some(A)), Ok(Some(k(A))));
        assert_eq!(
            resolve_monetize(None, None),
            Ok(None),
            "a metered build with no key from either source: today's behaviour"
        );
    }

    #[test]
    fn monetize_medium_and_a_different_door_key_is_refused_by_name() {
        let why = resolve_monetize(Some(k(A)), Some(B)).unwrap_err();
        assert!(
            why.contains(&format!("[{FEED_MISMATCH}]")) && why.contains(A) && why.contains(B),
            "{why}"
        );
        let why = resolve_monetize(None, Some("not-a-key")).unwrap_err();
        assert!(why.contains("[pairing-feed-monetize-malformed]"), "{why}");
    }

    #[test]
    fn the_feed_table_reads_the_contract_keys_and_blank_is_absent() {
        let mut feed = std::collections::BTreeMap::new();
        feed.insert(FEED_MONETIZE.to_string(), A.to_string());
        feed.insert(
            FEED_CLOCK_PEER.to_string(),
            "peer://10.13.8.179:50123".to_string(),
        );
        feed.insert(FEED_CLOCK_KEY.to_string(), "  ".to_string());
        feed.insert(
            "admin_pubkey".to_string(),
            "ssh-ed25519 AAAAconsole".to_string(),
        );
        let fed = Fed::from_feed(&feed);
        assert_eq!(
            fed,
            Fed {
                monetize_pubkey: Some(A.into()),
                clock_key: None,
                clock_peer: Some("peer://10.13.8.179:50123".into()),
                ..Fed::default()
            }
        );
    }

    #[cfg(feature = "wg-client")]
    #[test]
    fn a_persisted_join_reply_is_read_back_and_absent_is_none() {
        let dir = std::env::temp_dir().join(format!("gunnar-feed-{}", std::process::id()));
        let _ = std::fs::create_dir_all(dir.join("pairing"));
        assert_eq!(Fed::read(&dir.join("nowhere")).unwrap(), None);
        let mut feed = std::collections::BTreeMap::new();
        feed.insert(FEED_MONETIZE.to_string(), A.to_string());
        feed.insert(FEED_CLOCK_KEY.to_string(), "ssh-ed25519 AAAAk".to_string());
        let reply = nordisk_pairing::join::JoinReply {
            estate: "gunnar.rs".into(),
            me: nordisk_pairing::join::Assigned {
                name: "main".into(),
                overlay: nordisk_pairing::OVERLAY_BOX_A,
            },
            cert_pem: None,
            role: None,
            primary_overlay: None,
            primary_utility: None,
            peers: vec![],
            feed,
        };
        std::fs::write(
            dir.join(crate::pairing::JOINED_REL),
            reply.render().unwrap(),
        )
        .unwrap();
        let fed = Fed::read(&dir).unwrap().unwrap();
        assert_eq!(fed.monetize_pubkey.as_deref(), Some(A));
        assert_eq!(fed.clock_key.as_deref(), Some("ssh-ed25519 AAAAk"));
        let _ = std::fs::remove_dir_all(&dir);
    }
}