//! The swarm builder's build service.
//!
//! Connects to the swarm queue as its own OIDC client, takes build requests
//! from swarm-controller on [`swarm_wire::builder::SUBJECT`], builds
//! each agent config at its commit, answers on the request's reply subject,
//! and logs one line per build. It writes nothing to the forge: the
//! controller turns the reply into a commit status. It also answers
//! [`swarm_wire::builder::ARCHETYPES_SUBJECT`] with the agent
//! archetypes its hyperhive source declares, which the controller stores.
//!
//! Lock requests on [`swarm_wire::builder::LOCK_SUBJECT`] carry the swarm meta
//! flake's `flake.nix`, and the answer is the `flake.lock` `nix` wrote for it in
//! a scratch directory. The controller commits it.
//!
//! Meta build requests on [`swarm_wire::builder::META_SUBJECT`] name an agent
//! and an `internal/meta` commit, and the answer is the store path of that
//! agent's toplevel, built at the flake's own lock. They take the build slot
//! PR checks take. The controller records the path; the builder writes nothing.
//! When a cache is set (see [`push`]), a green meta build is pushed to it before
//! the reply, and the reply says whether that worked.
//!
//! Build counts, durations, queue depth and wait, and reply publish failures
//! are exported by [`metrics`].
//!
//! The store identity comes from the `BAO_*` environment. The queue client
//! secret is read at start and again when authelia refuses it, the forge token
//! once per build, both from the store under that identity.

use std::future::Future;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use anyhow::{Context, Result, anyhow, bail};
use async_nats::{Message, Subject};
use futures_util::{Stream, StreamExt as _};
use metrics::{Judged, Kind, Metrics, Outcome};
use swarm_queue_client::{ClientSecret, QueueConfig};
use swarm_secret_client::forge::Credential;
use swarm_secret_client::stored_secret::StoredSecret;
use swarm_wire::builder::{
    ARCHETYPES_DEADLINE, ARCHETYPES_SUBJECT, BUILD_DEADLINE, BuildAgent, BuildFromMeta,
    FromMetaReply, LOCK_DEADLINE, LOCK_SUBJECT, LockMeta, LockReply, META_SUBJECT, QUEUE_GROUP,
    REPLY_DEADLINE, Reply, SUBJECT,
};

mod archetypes;
mod metrics;
mod nix_build;
mod push;

/// The cert-auth role this service logs in to the store as. `swarm-bao.nix`
/// writes it, matching the leaf's subject.
const CERT_ROLE: &str = "swarm-builder";

/// The environment prefix `swarm-builder.nix` sets.
const PREFIX: &str = "SWARM_BUILDER";

/// The longest the store login and forge token read may take before a build.
/// Each store request alone may take 30 s, so without this bound the two could
/// spend the minute `REPLY_DEADLINE` leaves after `BUILD_DEADLINE`.
const TOKEN_DEADLINE: Duration = Duration::from_secs(20);

/// What is held back from the minute `REPLY_DEADLINE` leaves after
/// `BUILD_DEADLINE`: [`TOKEN_DEADLINE`], then 10 s to publish the reply and for
/// it to reach the controller. Both come after the slot wait.
const SLOT_MARGIN: Duration = Duration::from_secs(30);

/// The longest a request may wait for the build slot. A build that gets the
/// slot later could still be running when the controller stops waiting for the
/// reply.
const MAX_SLOT_WAIT: Duration = REPLY_DEADLINE
    .saturating_sub(BUILD_DEADLINE)
    .saturating_sub(SLOT_MARGIN);

/// What is held back from a meta build's reply deadline for publishing the reply
/// after its push.
const REPLY_MARGIN: Duration = Duration::from_secs(10);

/// How long the push of a meta build that ends at now may run: at most
/// [`push::PUSH_DEADLINE`], and never past `reply_by`, so a push cannot make the
/// reply later than the controller waits.
fn push_budget(reply_by: tokio::time::Instant) -> Duration {
    push::PUSH_DEADLINE.min(reply_by.saturating_duration_since(tokio::time::Instant::now()))
}

#[tokio::main]
async fn main() -> Result<()> {
    hive_log::init();
    swarm_secret_client::install_crypto_provider();
    let var = |name: &str| {
        let full = format!("{PREFIX}_{name}");
        std::env::var(&full).with_context(|| format!("{full} is unset"))
    };
    let client_id = var("OIDC_CLIENT_ID")?;
    let builder = Arc::new(nix_build::Builder {
        nix: PathBuf::from(var("NIX")?),
        check_flake: var("CHECK_FLAKE")?,
        hyperhive: var("HYPERHIVE_FLAKE")?,
        nixpkgs: var("NIXPKGS_FLAKE")?,
        forge_url: forge_url()?,
        config_org: var("CONFIG_ORG")?,
        deadline: BUILD_DEADLINE,
        lock_deadline: LOCK_DEADLINE,
        system: var("SYSTEM")?,
        cache: push_target()?,
    });
    let secret =
        swarm_secret_client::read_with_startup_retry("reading the queue client secret", || {
            client_secret(&client_id)
        })
        .await?;
    let evaluator = archetypes::Evaluator {
        nix: builder.nix.clone(),
        hyperhive: builder.hyperhive.clone(),
        nixpkgs: builder.nixpkgs.clone(),
        deadline: ARCHETYPES_DEADLINE,
    };
    let reread_id = client_id.clone();
    let secret = StoredSecret::new(client_id.clone(), secret, move || {
        let client_id = reread_id.clone();
        async move { client_secret(&client_id).await.map_err(Into::into) }
    });
    let queue = swarm_queue_client::connect(QueueConfig {
        url: var("NATS_URL")?,
        token_endpoint: var("OIDC_TOKEN_ENDPOINT")?,
        client_id,
        client_secret: ClientSecret::Stored(secret),
        ca_file: std::env::var(format!("{PREFIX}_OIDC_CA_FILE"))
            .ok()
            .map(PathBuf::from),
    })
    .await?;
    let requests = queue
        .queue_subscribe(SUBJECT, QUEUE_GROUP.to_owned())
        .await
        .with_context(|| format!("subscribing to {SUBJECT}"))?;
    tracing::info!(subject = SUBJECT, "builder: awaiting builds");
    // Kept alive: dropping the provider stops the export.
    let (_metrics_provider, metrics) = metrics::start();
    let publisher = Metered {
        inner: Queue(queue.clone()),
        metrics: metrics.clone(),
    };
    let evaluations = queue
        .queue_subscribe(ARCHETYPES_SUBJECT, QUEUE_GROUP.to_owned())
        .await
        .with_context(|| format!("subscribing to {ARCHETYPES_SUBJECT}"))?;
    // Outside the build slot below: a list refresh does not wait on a build.
    tokio::spawn(answer_evaluations(
        evaluations,
        publisher.clone(),
        evaluator,
    ));
    let locks = queue
        .queue_subscribe(LOCK_SUBJECT, QUEUE_GROUP.to_owned())
        .await
        .with_context(|| format!("subscribing to {LOCK_SUBJECT}"))?;
    // Outside the build slot too: a lock builds nothing, so it does not wait
    // behind an hour-long build.
    tokio::spawn(answer_locks(
        locks,
        publisher.clone(),
        Store,
        Arc::clone(&builder),
    ));

    // One build at a time. swarm-controller already sends one at a time; this
    // holds the bound for any request it did not serialise, and `in_slot`
    // refuses one that waited too long behind another. Meta builds take the
    // same slot as PR checks.
    let slot = Arc::new(tokio::sync::Semaphore::new(1));
    serve_meta_builds(&queue, &publisher, Store, &builder, &slot, &metrics).await?;

    answer_builds(requests, publisher, Store, builder, slot, metrics).await;
    bail!("the subscription to {SUBJECT} ended")
}

/// Where a reply goes. [`Queue`] in production; tests record.
trait Publisher: Clone + Send + Sync + 'static {
    /// Publish `body` on `subject`; the error is the queue client's own text.
    fn publish(
        &self,
        subject: Subject,
        body: Vec<u8>,
    ) -> impl Future<Output = Result<(), String>> + Send;
}

/// A [`Publisher`] that counts the replies its inner publisher fails to send.
#[derive(Clone)]
struct Metered<P> {
    inner: P,
    metrics: Metrics,
}

impl<P: Publisher> Publisher for Metered<P> {
    async fn publish(&self, subject: Subject, body: Vec<u8>) -> Result<(), String> {
        let sent = self.inner.publish(subject, body).await;
        if sent.is_err() {
            self.metrics.publish_failed();
        }
        sent
    }
}

/// The secrets a request reads from the store, once per request. [`Store`] in
/// production; tests answer with canned replies.
trait Tokens: Clone + Send + Sync + 'static {
    /// The forge token config repos are fetched with.
    fn forge_token(&self) -> impl Future<Output = Result<String>> + Send;
    /// The token closures are pushed to the cache with.
    fn push_token(&self) -> impl Future<Output = Result<String>> + Send;
}

/// The swarm queue connection replies are published on.
#[derive(Clone)]
struct Queue(async_nats::Client);

impl Publisher for Queue {
    async fn publish(&self, subject: Subject, body: Vec<u8>) -> Result<(), String> {
        self.0
            .publish(subject, body.into())
            .await
            .map_err(|e| e.to_string())
    }
}

/// The secret store, logged in to as [`CERT_ROLE`] on every read.
#[derive(Clone, Copy)]
struct Store;

impl Tokens for Store {
    async fn forge_token(&self) -> Result<String> {
        forge_token().await
    }

    async fn push_token(&self) -> Result<String> {
        attic_push_token().await
    }
}

/// Publish an encoded reply on `reply_to`. `what` names the request in the
/// log line of a failure.
async fn send_reply(
    publisher: &impl Publisher,
    reply_to: Subject,
    encoded: serde_json::Result<Vec<u8>>,
    what: &str,
) {
    match encoded {
        Ok(body) => {
            if let Err(e) = publisher.publish(reply_to, body).await {
                tracing::warn!(error = %e, "publishing {what} reply failed");
            }
        }
        Err(e) => tracing::warn!(error = %e, "encoding {what} reply failed"),
    }
}

/// Answer every PR check, each in its own task, with the build's result.
async fn answer_builds(
    mut requests: impl Stream<Item = Message> + Unpin,
    publisher: impl Publisher,
    tokens: impl Tokens,
    builder: Arc<nix_build::Builder>,
    slot: Arc<tokio::sync::Semaphore>,
    metrics: Metrics,
) {
    while let Some(msg) = requests.next().await {
        let received = tokio::time::Instant::now();
        let Some(reply_to) = msg.reply.clone() else {
            tracing::warn!("a build arrived with no reply subject; dropping it");
            continue;
        };
        let (publisher, tokens, builder, slot, metrics) = (
            publisher.clone(),
            tokens.clone(),
            Arc::clone(&builder),
            Arc::clone(&slot),
            metrics.clone(),
        );
        tokio::spawn(async move {
            let Some(reply) =
                build_reply(&msg.payload, received, &slot, &builder, &tokens, &metrics).await
            else {
                return;
            };
            send_reply(
                &publisher,
                reply_to,
                serde_json::to_vec(&reply),
                "a build's",
            )
            .await;
        });
    }
}

/// The reply to one PR check's payload: the build's, the slot's refusal, or the
/// decode error. `None` only when the slot is closed.
async fn build_reply(
    payload: &[u8],
    received: tokio::time::Instant,
    slot: &tokio::sync::Semaphore,
    builder: &nix_build::Builder,
    tokens: &impl Tokens,
    metrics: &Metrics,
) -> Option<Reply> {
    match serde_json::from_slice::<BuildAgent>(payload) {
        Ok(job) => {
            in_slot(
                slot,
                received,
                (metrics, Kind::Pr),
                || busy(&job),
                handle(&job, builder, tokens),
            )
            .await
        }
        Err(e) => {
            metrics.count(Kind::Pr, Outcome::Failure);
            tracing::warn!(error = %e, "an undecodable build; answering with the error");
            Some(Reply::Failed {
                error_tail: format!("the builder could not decode the request: {e}"),
            })
        }
    }
}

/// Answer every archetype evaluation request, one at a time, with the
/// evaluation's result.
async fn answer_evaluations(
    mut requests: impl Stream<Item = Message> + Unpin,
    publisher: impl Publisher,
    evaluator: archetypes::Evaluator,
) {
    while let Some(msg) = requests.next().await {
        let Some(reply_to) = msg.reply.clone() else {
            tracing::warn!("an archetype evaluation arrived with no reply subject; dropping it");
            continue;
        };
        let reply = evaluator.eval().await;
        match &reply {
            swarm_wire::builder::ArchetypesReply::Evaluated { archetypes, .. } => {
                tracing::info!(count = archetypes.len(), "archetypes evaluated");
            }
            swarm_wire::builder::ArchetypesReply::Failed { error_tail } => {
                tracing::warn!(%error_tail, "archetype evaluation failed");
            }
        }
        send_reply(
            &publisher,
            reply_to,
            serde_json::to_vec(&reply),
            "an evaluation's",
        )
        .await;
    }
    tracing::error!("the subscription to {ARCHETYPES_SUBJECT} ended");
}

/// Answer every meta-flake lock request, one at a time, with the lock `nix`
/// wrote or why it did not.
async fn answer_locks(
    mut requests: impl Stream<Item = Message> + Unpin,
    publisher: impl Publisher,
    tokens: impl Tokens,
    builder: Arc<nix_build::Builder>,
) {
    while let Some(msg) = requests.next().await {
        let Some(reply_to) = msg.reply.clone() else {
            tracing::warn!("a meta-flake lock arrived with no reply subject; dropping it");
            continue;
        };
        let reply = handle_lock(&msg.payload, &builder, &tokens).await;
        send_reply(&publisher, reply_to, serde_json::to_vec(&reply), "a lock's").await;
    }
    tracing::error!("the subscription to {LOCK_SUBJECT} ended");
}

/// Decode one lock request, lock it and log the lock.
async fn handle_lock(
    payload: &[u8],
    builder: &nix_build::Builder,
    tokens: &impl Tokens,
) -> LockReply {
    let job: LockMeta = match serde_json::from_slice(payload) {
        Ok(job) => job,
        Err(e) => {
            tracing::warn!(error = %e, "an undecodable lock; answering with the error");
            return LockReply::Failed {
                error_tail: format!("the builder could not decode the request: {e}"),
            };
        }
    };
    let reply = match within(TOKEN_DEADLINE, tokens.forge_token()).await {
        Ok(token) => builder.lock(&job, &token).await,
        Err(e) => LockReply::Failed {
            error_tail: format!("{e:#}"),
        },
    };
    match &reply {
        LockReply::Locked { .. } => tracing::info!("swarm meta flake locked"),
        LockReply::Failed { error_tail } => {
            tracing::warn!(%error_tail, "swarm meta flake lock failed");
        }
    }
    reply
}

/// Subscribe to meta build requests and answer them in the background.
async fn serve_meta_builds(
    queue: &async_nats::Client,
    publisher: &impl Publisher,
    tokens: impl Tokens,
    builder: &Arc<nix_build::Builder>,
    slot: &Arc<tokio::sync::Semaphore>,
    metrics: &Metrics,
) -> Result<()> {
    let requests = queue
        .queue_subscribe(META_SUBJECT, QUEUE_GROUP.to_owned())
        .await
        .with_context(|| format!("subscribing to {META_SUBJECT}"))?;
    tokio::spawn(answer_meta_builds(
        requests,
        publisher.clone(),
        tokens,
        Arc::clone(builder),
        Arc::clone(slot),
        metrics.clone(),
    ));
    Ok(())
}

/// Answer every agent build from the meta flake, each in its own task, with the
/// build's result.
async fn answer_meta_builds(
    mut requests: impl Stream<Item = Message> + Unpin,
    publisher: impl Publisher,
    tokens: impl Tokens,
    builder: Arc<nix_build::Builder>,
    slot: Arc<tokio::sync::Semaphore>,
    metrics: Metrics,
) {
    while let Some(msg) = requests.next().await {
        let received = tokio::time::Instant::now();
        let Some(reply_to) = msg.reply.clone() else {
            tracing::warn!("a meta build arrived with no reply subject; dropping it");
            continue;
        };
        let (publisher, tokens, builder, slot, metrics) = (
            publisher.clone(),
            tokens.clone(),
            Arc::clone(&builder),
            Arc::clone(&slot),
            metrics.clone(),
        );
        tokio::spawn(async move {
            let Some(reply) =
                meta_reply(&msg.payload, received, &slot, &builder, &tokens, &metrics).await
            else {
                return;
            };
            send_reply(
                &publisher,
                reply_to,
                serde_json::to_vec(&reply),
                "a meta build's",
            )
            .await;
        });
    }
    tracing::error!("the subscription to {META_SUBJECT} ended");
}

/// The reply to one meta build's payload: the build's, the slot's refusal, or
/// the decode error. `None` only when the slot is closed.
async fn meta_reply(
    payload: &[u8],
    received: tokio::time::Instant,
    slot: &tokio::sync::Semaphore,
    builder: &nix_build::Builder,
    tokens: &impl Tokens,
    metrics: &Metrics,
) -> Option<FromMetaReply> {
    match serde_json::from_slice::<BuildFromMeta>(payload) {
        Ok(job) => {
            in_slot(
                slot,
                received,
                (metrics, Kind::Meta),
                || busy_meta(&job),
                handle_meta(
                    &job,
                    builder,
                    tokens,
                    received + REPLY_DEADLINE - REPLY_MARGIN,
                ),
            )
            .await
        }
        Err(e) => {
            tracing::warn!(error = %e, "an undecodable meta build; answering with the error");
            metrics.count(Kind::Meta, Outcome::Failure);
            Some(FromMetaReply::Failed {
                error_tail: format!("the builder could not decode the request: {e}"),
            })
        }
    }
}

/// `build`'s reply once a request holds the build slot, or `refuse`'s, without
/// running `build`, once it has waited [`MAX_SLOT_WAIT`] since `received`.
/// `None` only when the slot is closed.
///
/// Records the wait, and the build's count and duration, on `metrics` as
/// `kind`; a refusal counts as [`Outcome::Refused`].
async fn in_slot<T: Judged>(
    slot: &tokio::sync::Semaphore,
    received: tokio::time::Instant,
    (metrics, kind): (&Metrics, Kind),
    refuse: impl FnOnce() -> T,
    build: impl Future<Output = T>,
) -> Option<T> {
    let waiting = metrics.start_waiting();
    let acquired = tokio::time::timeout_at(received + MAX_SLOT_WAIT, slot.acquire()).await;
    metrics.record_wait(kind, received.elapsed());
    drop(waiting);
    let Ok(permit) = acquired else {
        metrics.count(kind, Outcome::Refused);
        return Some(refuse());
    };
    let _permit = permit.ok()?;
    let started = tokio::time::Instant::now();
    let reply = build.await;
    metrics.record_build(kind, reply.outcome(), started.elapsed());
    Some(reply)
}

impl Judged for Reply {
    fn outcome(&self) -> Outcome {
        match self {
            Self::Built { .. } => Outcome::Success,
            Self::Failed { .. } => Outcome::Failure,
        }
    }
}

impl Judged for FromMetaReply {
    fn outcome(&self) -> Outcome {
        match self {
            Self::Built { .. } => Outcome::Success,
            Self::Failed { .. } | Self::Unavailable { .. } => Outcome::Failure,
        }
    }
}

/// The refusal text of a request that waited [`MAX_SLOT_WAIT`] for the slot.
fn busy_tail() -> String {
    format!(
        "the swarm builder was busy with another build for {} s, too long for this one \
         to finish before swarm-controller stops waiting for it",
        MAX_SLOT_WAIT.as_secs()
    )
}

/// Log and answer a PR check that waited too long for the slot.
fn busy(job: &BuildAgent) -> Reply {
    tracing::warn!(
        agent = %job.agent,
        rev = %job.rev,
        pr = job.pr,
        waited_secs = MAX_SLOT_WAIT.as_secs(),
        "build refused: another build held the slot too long for this one to finish \
         before swarm-controller stops waiting"
    );
    Reply::Failed {
        error_tail: busy_tail(),
    }
}

/// Log and answer a meta build that waited too long for the slot.
fn busy_meta(job: &BuildFromMeta) -> FromMetaReply {
    tracing::warn!(
        agent = %job.agent,
        meta_rev = %job.meta_rev,
        waited_secs = MAX_SLOT_WAIT.as_secs(),
        "meta build refused: another build held the slot too long for this one to finish \
         before swarm-controller stops waiting"
    );
    FromMetaReply::Unavailable {
        reason: busy_tail(),
    }
}

/// Build one meta request and log the build.
async fn handle_meta(
    job: &BuildFromMeta,
    builder: &nix_build::Builder,
    tokens: &impl Tokens,
    reply_by: tokio::time::Instant,
) -> FromMetaReply {
    let reply = match within(TOKEN_DEADLINE, tokens.forge_token()).await {
        Ok(token) => builder.build_from_meta(job, &token).await,
        Err(e) => FromMetaReply::Unavailable {
            reason: format!("{e:#}"),
        },
    };
    let reply = push::cache_built(
        reply,
        builder.cache.as_ref(),
        tokens.push_token(),
        push_budget(reply_by),
        &job.agent,
    )
    .await;
    match &reply {
        FromMetaReply::Built {
            out_path,
            system,
            cached,
            private_users,
        } => tracing::info!(
            agent = %job.agent,
            meta_rev = %job.meta_rev,
            %out_path,
            %system,
            cached,
            private_users,
            "agent meta build built"
        ),
        FromMetaReply::Failed { error_tail } => tracing::warn!(
            agent = %job.agent,
            meta_rev = %job.meta_rev,
            %error_tail,
            "agent meta build failed"
        ),
        FromMetaReply::Unavailable { reason } => tracing::warn!(
            agent = %job.agent,
            meta_rev = %job.meta_rev,
            %reason,
            "agent meta build not run"
        ),
    }
    reply
}

/// Build one request and log the build.
async fn handle(job: &BuildAgent, builder: &nix_build::Builder, tokens: &impl Tokens) -> Reply {
    let reply = match within(TOKEN_DEADLINE, tokens.forge_token()).await {
        Ok(token) => builder.build(job, &token).await,
        Err(e) => Reply::Failed {
            error_tail: format!("{e:#}"),
        },
    };
    log_build(job, &reply);
    reply
}

/// `read`'s result, or an error once it has run for `deadline`.
async fn within<T>(deadline: Duration, read: impl Future<Output = Result<T>>) -> Result<T> {
    tokio::time::timeout(deadline, read)
        .await
        .unwrap_or_else(|_| {
            Err(anyhow!(
                "reading the forge token took more than {} s",
                deadline.as_secs()
            ))
        })
}

/// The one journal line a build leaves.
fn log_build(job: &BuildAgent, reply: &Reply) {
    match reply {
        Reply::Built { out_path } => tracing::info!(
            agent = %job.agent,
            rev = %job.rev,
            pr = job.pr,
            %out_path,
            "agent config built"
        ),
        Reply::Failed { error_tail } => tracing::warn!(
            agent = %job.agent,
            rev = %job.rev,
            pr = job.pr,
            %error_tail,
            "agent config build failed"
        ),
    }
}

/// The token the builder pushes closures to the cache with, read on every push so
/// a rotation takes effect on the next one.
async fn attic_push_token() -> Result<String> {
    let path = swarm_secret_client::builder::attic_push_token_path()?;
    let store = swarm_secret_client::SecretStore::from_env(CERT_ROLE)
        .await
        .context("logging in to the secret store")?;
    let stored: Option<swarm_secret_client::builder::AtticPushToken> = store
        .read_optional(&path)
        .await
        .with_context(|| format!("reading {path}"))?;
    match stored {
        Some(t) if !t.value.trim().is_empty() => Ok(t.value.trim().to_owned()),
        _ => bail!("nothing at {path}: the cache push token has not been written"),
    }
}

/// The forge token the builder fetches config repos with, read on every
/// build so a rotation takes effect on the next one.
async fn forge_token() -> Result<String> {
    let path = swarm_secret_client::builder::forge_token_path()?;
    let store = swarm_secret_client::SecretStore::from_env(CERT_ROLE)
        .await
        .context("logging in to the secret store")?;
    let stored: Option<Credential> = store
        .read_optional(&path)
        .await
        .with_context(|| format!("reading {path}"))?;
    match stored {
        Some(c) if !c.value.trim().is_empty() => Ok(c.value.trim().to_owned()),
        _ => bail!("nothing at {path}: swarm-controller has not minted the builder's forge token"),
    }
}

async fn client_secret(client_id: &str) -> Result<String> {
    use swarm_secret_client::builder::{ClientSecret, client_secret_path};

    let path = client_secret_path(client_id)?;
    let store = swarm_secret_client::SecretStore::from_env(CERT_ROLE)
        .await
        .context("logging in to the secret store")?;
    let stored: Option<ClientSecret> = store
        .read_optional(&path)
        .await
        .with_context(|| format!("reading {path}"))?;
    match stored {
        Some(s) if !s.value.trim().is_empty() => Ok(s.value.trim().to_owned()),
        _ => bail!(
            "nothing at {path}: authelia's mint, in its container, has not \
             stored this service's client secret"
        ),
    }
}

/// The cache the unit's environment names, if any. systemd sets
/// `RUNTIME_DIRECTORY` for the unit's runtime directory.
fn push_target() -> Result<Option<push::Remote>> {
    let var = |name: &str| std::env::var(name).ok();
    push::Remote::from_settings(
        var(&format!("{PREFIX}_ATTIC_DOMAIN")),
        var(&format!("{PREFIX}_ATTIC_CACHE")),
        var("RUNTIME_DIRECTORY"),
    )
}

/// The forge's base URL, which must be `https`: the credential helper offers
/// the token to `https` requests for its host alone.
fn forge_url() -> Result<String> {
    let name = format!("{PREFIX}_FORGE_URL");
    let url = std::env::var(&name).with_context(|| format!("{name} is unset"))?;
    check_forge_url(&name, &url)
}

/// `url` without trailing slashes, if it is `https://<host>[:port]`; `name` is
/// the variable it came from.
fn check_forge_url(name: &str, url: &str) -> Result<String> {
    let url = url.trim_end_matches('/');
    if !url.starts_with("https://") || url["https://".len()..].contains('/') {
        bail!("{name} is {url:?}, not https://<host>[:port]");
    }
    Ok(url.to_owned())
}

#[cfg(test)]
mod tests {
    use std::sync::atomic::{AtomicBool, Ordering};

    use tokio::sync::Semaphore;
    use tokio::time::Instant;

    use super::*;

    fn job() -> BuildAgent {
        BuildAgent {
            agent: "atlas".to_owned(),
            rev: "abc".to_owned(),
            pr: 3,
        }
    }

    /// A request behind a build that holds the slot for `held`: its reply,
    /// whether its own build ran, and how long after arriving it was answered.
    async fn behind_a_build_of(held: Duration) -> (Option<Reply>, bool, Duration) {
        let slot = Arc::new(Semaphore::new(1));
        let running = Arc::clone(&slot).acquire_owned().await.expect("open");
        tokio::spawn(async move {
            tokio::time::sleep(held).await;
            drop(running);
        });
        let received = Instant::now();
        let ran = AtomicBool::new(false);
        let build = async {
            ran.store(true, Ordering::SeqCst);
            Reply::Built {
                out_path: "/nix/store/x".to_owned(),
            }
        };
        let reply = in_slot(
            &slot,
            received,
            (&Metrics::noop(), Kind::Pr),
            || busy(&job()),
            build,
        )
        .await;
        (reply, ran.load(Ordering::SeqCst), received.elapsed())
    }

    #[tokio::test(start_paused = true)]
    async fn a_request_that_waited_past_the_slot_limit_is_refused_without_building() {
        let (reply, built, answered_after) =
            behind_a_build_of(MAX_SLOT_WAIT + Duration::from_mins(1)).await;
        assert!(
            matches!(&reply, Some(Reply::Failed { error_tail }) if error_tail.contains("busy")),
            "{reply:?}"
        );
        assert!(!built, "a refused request must not build");
        assert_eq!(answered_after, MAX_SLOT_WAIT);
    }

    #[tokio::test(start_paused = true)]
    async fn a_request_that_gets_the_slot_in_time_builds() {
        let (reply, built, _) =
            behind_a_build_of(MAX_SLOT_WAIT.saturating_sub(Duration::from_secs(1))).await;
        assert!(matches!(reply, Some(Reply::Built { .. })), "{reply:?}");
        assert!(built);
    }

    /// A meta build waits on the slot a PR check holds, and is refused at the
    /// same limit.
    #[tokio::test(start_paused = true)]
    async fn a_meta_build_shares_the_slot_with_pr_checks() {
        let meta = BuildFromMeta {
            agent: "atlas".to_owned(),
            meta_rev: "abc".to_owned(),
        };
        let slot = Arc::new(Semaphore::new(1));
        let check = Arc::clone(&slot).acquire_owned().await.expect("open");
        let received = Instant::now();
        let refused = in_slot(
            &slot,
            received,
            (&Metrics::noop(), Kind::Meta),
            || busy_meta(&meta),
            async {
                FromMetaReply::Built {
                    out_path: "/nix/store/x".to_owned(),
                    system: "s".to_owned(),
                    cached: false,
                    private_users: false,
                }
            },
        )
        .await;
        assert!(
            matches!(&refused, Some(FromMetaReply::Unavailable { reason }) if reason.contains("busy")),
            "{refused:?}"
        );
        assert_eq!(received.elapsed(), MAX_SLOT_WAIT);
        // Control: once the check lets go, the same request builds.
        drop(check);
        let built = in_slot(
            &slot,
            Instant::now(),
            (&Metrics::noop(), Kind::Meta),
            || busy_meta(&meta),
            async {
                FromMetaReply::Built {
                    out_path: "/nix/store/x".to_owned(),
                    system: "s".to_owned(),
                    cached: false,
                    private_users: false,
                }
            },
        )
        .await;
        assert!(
            matches!(built, Some(FromMetaReply::Built { .. })),
            "{built:?}"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn a_token_read_past_its_deadline_is_an_error_and_a_prompt_one_is_kept() {
        let started = Instant::now();
        let stuck = within(TOKEN_DEADLINE, std::future::pending::<Result<String>>()).await;
        assert_eq!(started.elapsed(), TOKEN_DEADLINE);
        assert!(
            matches!(&stuck, Err(e) if e.to_string().contains("took more than 20 s")),
            "{stuck:?}"
        );
        let prompt = within(TOKEN_DEADLINE, async { Ok("t".to_owned()) }).await;
        assert_eq!(prompt.expect("read"), "t");
    }

    #[tokio::test(start_paused = true)]
    async fn a_push_gets_the_deadline_unless_the_reply_is_due_sooner() {
        let now = Instant::now();
        assert_eq!(
            push_budget(now + Duration::from_hours(1)),
            push::PUSH_DEADLINE
        );
        assert_eq!(
            push_budget(now + Duration::from_secs(90)),
            Duration::from_secs(90)
        );
        assert_eq!(push_budget(now), Duration::ZERO);
    }

    #[test]
    fn a_build_started_at_the_slot_limit_ends_inside_the_reply_deadline() {
        assert!(MAX_SLOT_WAIT > Duration::ZERO);
        assert!(MAX_SLOT_WAIT + BUILD_DEADLINE < REPLY_DEADLINE);
        assert!(SLOT_MARGIN >= Duration::from_secs(30));
        assert!(SLOT_MARGIN >= TOKEN_DEADLINE + Duration::from_secs(10));
        assert!(MAX_SLOT_WAIT + SLOT_MARGIN + BUILD_DEADLINE <= REPLY_DEADLINE);
    }

    // The daemon's request path, end to end with a stand-in `nix`: a message in,
    // the bytes published on its reply subject out.

    use serde_json::{Value, json};
    use tokio::sync::mpsc::UnboundedReceiver;

    const REV: &str = "0123456789abcdef0123456789abcdef01234567";
    const OUT: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-atlas";

    /// A directory holding a stand-in `nix` that logs each call's arguments to
    /// `calls` in the same directory, then runs the script it was given.
    struct Fake(PathBuf);

    impl Fake {
        fn new(body: &str) -> Self {
            use std::sync::atomic::AtomicU64;
            static NEXT: AtomicU64 = AtomicU64::new(0);
            let _ = tracing_subscriber::fmt()
                .with_test_writer()
                .with_max_level(tracing::Level::INFO)
                .try_init();
            let dir = std::env::temp_dir().join(format!(
                "swarm-builder-main-{}-{}",
                std::process::id(),
                NEXT.fetch_add(1, Ordering::Relaxed)
            ));
            std::fs::create_dir_all(&dir).expect("scratch dir");
            let fake = Self(dir);
            fake.script(
                "nix",
                &format!(
                    "echo \"$*\" >> \"${{0%/*}}/calls\"; echo \"$SWARM_BUILDER_GIT_TOKEN\" >> \"${{0%/*}}/tokens\"\n{body}"
                ),
            );
            fake
        }

        /// Write the executable `name` from a child process: a write fd held here
        /// could be inherited by a child another test forks, and exec'ing a file
        /// something still holds open for writing fails with `ETXTBSY`.
        fn script(&self, name: &str, body: &str) {
            let path = self.0.join(name);
            let status = std::process::Command::new("/bin/sh")
                .args([
                    "-c",
                    r#"printf '%s' "$1" > "$2" && chmod 755 "$2""#,
                    "sh",
                    &format!("#!/bin/sh\n{body}\n"),
                ])
                .arg(&path)
                .status()
                .expect("run /bin/sh");
            assert!(status.success(), "writing {} failed", path.display());
        }

        fn builder(&self) -> Arc<nix_build::Builder> {
            self.builder_with(None)
        }

        /// The builder with a cache whose `attic` logs its arguments to `pushes`.
        fn caching_builder(&self) -> Arc<nix_build::Builder> {
            self.script("attic", "echo \"$*\" >> \"${0%/*}/pushes\"");
            self.builder_with(Some(push::Remote {
                attic: self.0.join("attic"),
                domain: "cache.example.org".to_owned(),
                cache: "grid".to_owned(),
                state_dir: self.0.clone(),
            }))
        }

        fn builder_with(&self, cache: Option<push::Remote>) -> Arc<nix_build::Builder> {
            Arc::new(nix_build::Builder {
                nix: self.0.join("nix"),
                check_flake: "/nix/store/c-check".to_owned(),
                hyperhive: "/nix/store/h-source".to_owned(),
                nixpkgs: "/nix/store/n-source".to_owned(),
                forge_url: "https://forge.example".to_owned(),
                config_org: "agent-configs".to_owned(),
                deadline: Duration::from_secs(30),
                lock_deadline: Duration::from_secs(30),
                system: "x86_64-linux".to_owned(),
                cache,
            })
        }

        fn lines(&self, file: &str) -> Vec<String> {
            std::fs::read_to_string(self.0.join(file))
                .map(|s| s.lines().map(str::to_owned).collect())
                .unwrap_or_default()
        }

        /// The arguments of every `nix` call.
        fn calls(&self) -> Vec<String> {
            self.lines("calls")
        }

        /// The arguments of every `attic` call.
        fn pushes(&self) -> Vec<String> {
            self.lines("pushes")
        }
    }

    impl Drop for Fake {
        fn drop(&mut self) {
            let _ = std::fs::remove_dir_all(&self.0);
        }
    }

    /// What `answer_*` published: the subject and the decoded body.
    type Sent = UnboundedReceiver<(String, Value)>;

    /// What a token read answers.
    #[derive(Clone, Copy)]
    enum Read {
        Token(&'static str),
        Unreadable(&'static str),
        Stuck,
    }

    impl Read {
        async fn get(self) -> Result<String> {
            match self {
                Self::Token(t) => Ok(t.to_owned()),
                Self::Unreadable(why) => Err(anyhow!(why)),
                Self::Stuck => std::future::pending().await,
            }
        }
    }

    fn token(value: &'static str) -> Read {
        Read::Token(value)
    }

    fn unreadable(why: &'static str) -> Read {
        Read::Unreadable(why)
    }

    fn stuck() -> Read {
        Read::Stuck
    }

    // `Publisher` and `Tokens` have `Clone` as a supertrait, which `automock`
    // cannot express (a mock is not `Clone`), so the two mocks are declared by
    // hand with `mock!` as plain structs. The traits are implemented for
    // `Arc<Mock…>`: every clone the daemon makes shares one set of
    // expectations, which are checked when the last clone drops.
    mockall::mock! {
        Publisher {
            fn publish(
                &self,
                subject: Subject,
                body: Vec<u8>,
            ) -> impl Future<Output = Result<(), String>> + Send;
        }
    }

    impl Publisher for Arc<MockPublisher> {
        fn publish(
            &self,
            subject: Subject,
            body: Vec<u8>,
        ) -> impl Future<Output = Result<(), String>> + Send {
            MockPublisher::publish(self, subject, body)
        }
    }

    mockall::mock! {
        Tokens {
            fn forge_token(&self) -> impl Future<Output = Result<String>> + Send;
            fn push_token(&self) -> impl Future<Output = Result<String>> + Send;
        }
    }

    impl Tokens for Arc<MockTokens> {
        fn forge_token(&self) -> impl Future<Output = Result<String>> + Send {
            MockTokens::forge_token(self)
        }

        fn push_token(&self) -> impl Future<Output = Result<String>> + Send {
            MockTokens::push_token(self)
        }
    }

    struct Rig {
        sink: Arc<MockPublisher>,
        tokens: Arc<MockTokens>,
    }

    /// A rig whose token reads answer `forge` and `push`, and whose first
    /// `failing` publishes fail; the rest are sent to the returned receiver.
    fn rig_with(forge: Read, push: Read, failing: usize) -> (Rig, Sent) {
        let (sent, rx) = tokio::sync::mpsc::unbounded_channel();
        let mut sink = MockPublisher::new();
        sink.expect_publish()
            .times(failing)
            .returning(|_, _| Box::pin(async { Err("the connection is closed".to_owned()) }));
        sink.expect_publish().returning(move |subject, body| {
            let body = serde_json::from_slice(&body).expect("a JSON reply");
            sent.send((subject.to_string(), body))
                .expect("test is listening");
            Box::pin(async { Ok(()) })
        });
        let mut tokens = MockTokens::new();
        tokens
            .expect_forge_token()
            .returning(move || Box::pin(forge.get()));
        tokens
            .expect_push_token()
            .returning(move || Box::pin(push.get()));
        let rig = Rig {
            sink: Arc::new(sink),
            tokens: Arc::new(tokens),
        };
        (rig, rx)
    }

    fn env() -> (Rig, Sent) {
        rig_with(token("forge-token"), token("push-token"), 0)
    }

    async fn next(sent: &mut Sent) -> (String, Value) {
        tokio::time::timeout(Duration::from_mins(2), sent.recv())
            .await
            .expect("a reply in time")
            .expect("the env is alive")
    }

    fn msg(reply: Option<&str>, payload: &str) -> Message {
        Message {
            subject: Subject::from("$SWARM.builder.test"),
            reply: reply.map(Subject::from),
            payload: payload.to_owned().into(),
            headers: None,
            status: None,
            description: None,
            length: payload.len(),
        }
    }

    fn build_json(agent: &str, rev: &str, pr: u64) -> String {
        json!({"agent": agent, "rev": rev, "pr": pr}).to_string()
    }

    fn meta_json(agent: &str, rev: &str) -> String {
        json!({"agent": agent, "meta_rev": rev}).to_string()
    }

    async fn serve_builds(builder: Arc<nix_build::Builder>, env: Rig, msgs: Vec<Message>) {
        answer_builds(
            stream::iter(msgs),
            env.sink,
            env.tokens,
            builder,
            Arc::new(Semaphore::new(1)),
            Metrics::noop(),
        )
        .await;
    }

    async fn serve_meta(builder: Arc<nix_build::Builder>, env: Rig, msgs: Vec<Message>) {
        answer_meta_builds(
            stream::iter(msgs),
            env.sink,
            env.tokens,
            builder,
            Arc::new(Semaphore::new(1)),
            Metrics::noop(),
        )
        .await;
    }

    use futures_util::stream;

    #[tokio::test]
    async fn a_build_is_answered_on_its_reply_subject_with_the_path_nix_printed() {
        let fake = Fake::new(&format!("echo {OUT}"));
        let (env, mut sent) = env();
        let request = msg(Some("_INBOX.7"), &build_json("atlas", REV, 3));
        serve_builds(fake.builder(), env, vec![request]).await;
        let (subject, body) = next(&mut sent).await;
        assert_eq!(subject, "_INBOX.7");
        assert_eq!(body, json!({"state": "built", "out_path": OUT}));
        assert_eq!(fake.lines("tokens"), ["forge-token"]);
        let calls = fake.calls();
        assert_eq!(calls.len(), 1, "{calls:?}");
        assert!(
            calls[0].contains(&format!(
                "agent-configs/atlas?ref=refs/pull/3/head&rev={REV}"
            )),
            "{calls:?}"
        );
    }

    #[tokio::test]
    async fn a_failed_build_is_answered_failed_with_what_nix_said_last() {
        let fake = Fake::new(
            "echo \"warning: unrelated\" >&2\n\
             echo \"error: unable to download 'https://forge.example/agent-configs/atlas': Could not resolve host\" >&2\n\
             exit 1",
        );
        let (env, mut sent) = env();
        let request = msg(Some("_INBOX.1"), &build_json("atlas", REV, 3));
        serve_builds(fake.builder(), env, vec![request]).await;
        let (subject, body) = next(&mut sent).await;
        assert_eq!(subject, "_INBOX.1");
        assert_eq!(body["state"], "failed");
        assert!(
            body["error_tail"]
                .as_str()
                .is_some_and(|t| t.contains("Could not resolve host")),
            "{body}"
        );
        assert!(body.get("out_path").is_none(), "{body}");
    }

    /// Each refused request is answered `failed` without `nix` running; the last
    /// one is valid and builds, so the empty call log of the others means
    /// something.
    #[tokio::test]
    async fn requests_that_would_bend_the_reference_are_refused_without_running_nix() {
        let fake = Fake::new(&format!("echo {OUT}"));
        let upper = REV.to_uppercase();
        let long = format!("{REV}0");
        let refused = [
            ("Atlas", REV, 3),
            ("", REV, 3),
            ("a/b", REV, 3),
            ("a b", REV, 3),
            ("atlas", "abc", 3),
            ("atlas", upper.as_str(), 3),
            ("atlas", long.as_str(), 3),
            ("atlas", REV, 0),
        ];
        for (agent, rev, pr) in refused {
            let (env, mut sent) = env();
            let request = msg(Some("_INBOX.1"), &build_json(agent, rev, pr));
            serve_builds(fake.builder(), env, vec![request]).await;
            let (_, body) = next(&mut sent).await;
            assert_eq!(body["state"], "failed", "{agent:?} {rev:?} {pr}: {body}");
            assert_eq!(fake.calls(), Vec::<String>::new(), "{agent:?} {rev:?} {pr}");
        }
        let (env, mut sent) = env();
        let valid = msg(Some("_INBOX.1"), &build_json("atlas", REV, 3));
        serve_builds(fake.builder(), env, vec![valid]).await;
        let (_, body) = next(&mut sent).await;
        assert_eq!(body["state"], "built");
        assert_eq!(fake.calls().len(), 1);
    }

    #[tokio::test]
    async fn an_undecodable_request_is_answered_with_the_decode_error_and_builds_nothing() {
        let fake = Fake::new(&format!("echo {OUT}"));
        let payloads = [
            "",
            "not json",
            "{}",
            r#"{"agent":"atlas","rev":"abc"}"#,
            r#"{"agent":"atlas","rev":"abc","pr":"3"}"#,
            r#"{"agent":7,"rev":"abc","pr":3}"#,
        ];
        for payload in payloads {
            let (env, mut sent) = env();
            serve_builds(fake.builder(), env, vec![msg(Some("_INBOX.1"), payload)]).await;
            let (_, body) = next(&mut sent).await;
            assert_eq!(body["state"], "failed", "{payload:?}");
            assert!(
                body["error_tail"]
                    .as_str()
                    .is_some_and(|t| t.starts_with("the builder could not decode the request: ")),
                "{payload:?}: {body}"
            );
        }
        assert_eq!(fake.calls(), Vec::<String>::new());
    }

    #[tokio::test]
    async fn a_request_with_no_reply_subject_is_dropped_and_the_next_one_still_builds() {
        let fake = Fake::new(&format!("echo {OUT}"));
        let (env, mut sent) = env();
        let msgs = vec![
            msg(None, &build_json("atlas", REV, 3)),
            msg(Some("_INBOX.2"), &build_json("atlas", REV, 4)),
        ];
        serve_builds(fake.builder(), env, msgs).await;
        let (subject, _) = next(&mut sent).await;
        assert_eq!(subject, "_INBOX.2");
        let calls = fake.calls();
        assert_eq!(calls.len(), 1, "{calls:?}");
        assert!(calls[0].contains("refs/pull/4/head"), "{calls:?}");
        assert!(sent.try_recv().is_err());
    }

    /// A forge token that cannot be read fails the build before `nix` is
    /// asked for anything, and the reply says why.
    #[tokio::test]
    async fn a_build_whose_forge_token_cannot_be_read_is_failed_without_running_nix() {
        let fake = Fake::new(&format!("echo {OUT}"));
        let (env, mut sent) = rig_with(
            unreadable("logging in to the secret store: connection refused"),
            token("p"),
            0,
        );
        let request = msg(Some("_INBOX.1"), &build_json("atlas", REV, 3));
        serve_builds(fake.builder(), env, vec![request]).await;
        let (_, body) = next(&mut sent).await;
        assert_eq!(
            body,
            json!({"state": "failed", "error_tail": "logging in to the secret store: connection refused"})
        );
        assert_eq!(fake.calls(), Vec::<String>::new());
    }

    #[tokio::test(start_paused = true)]
    async fn a_forge_token_read_that_never_returns_fails_the_build_after_its_deadline() {
        let fake = Fake::new(&format!("echo {OUT}"));
        let (env, mut sent) = rig_with(stuck(), token("p"), 0);
        let started = Instant::now();
        let request = msg(Some("_INBOX.1"), &build_json("atlas", REV, 3));
        serve_builds(fake.builder(), env, vec![request]).await;
        let (_, body) = next(&mut sent).await;
        assert_eq!(body["state"], "failed");
        assert!(
            body["error_tail"]
                .as_str()
                .is_some_and(|t| t.contains("took more than 20 s")),
            "{body}"
        );
        assert_eq!(started.elapsed(), TOKEN_DEADLINE);
        assert_eq!(fake.calls(), Vec::<String>::new());
    }

    /// A reply that cannot be published is logged; the builder goes on to the
    /// next request.
    #[tokio::test]
    async fn a_reply_that_cannot_be_published_does_not_stop_the_next_build() {
        let fake = Fake::new(&format!("echo {OUT}"));
        let (env, mut sent) = rig_with(token("t"), token("p"), 1);
        let msgs = vec![
            msg(Some("_INBOX.1"), &build_json("atlas", REV, 3)),
            msg(Some("_INBOX.2"), &build_json("atlas", REV, 4)),
        ];
        serve_builds(fake.builder(), env, msgs).await;
        let (_, body) = next(&mut sent).await;
        assert_eq!(body["state"], "built");
        assert_eq!(fake.calls().len(), 2, "both requests were built");
        assert!(sent.try_recv().is_err(), "only the second reply got out");
    }

    /// Two requests for the same head are both built, one after the other: the
    /// slot serialises them, nothing deduplicates them.
    #[tokio::test]
    async fn two_requests_for_the_same_head_build_one_after_the_other() {
        let fake = Fake::new(&format!(
            "echo start >> \"${{0%/*}}/order\"; sleep 0.3; echo end >> \"${{0%/*}}/order\"; echo {OUT}"
        ));
        let (env, mut sent) = env();
        let same = build_json("atlas", REV, 3);
        let msgs = vec![msg(Some("_INBOX.1"), &same), msg(Some("_INBOX.2"), &same)];
        serve_builds(fake.builder(), env, msgs).await;
        let mut subjects = vec![next(&mut sent).await.0, next(&mut sent).await.0];
        subjects.sort();
        assert_eq!(subjects, ["_INBOX.1", "_INBOX.2"]);
        assert_eq!(fake.lines("order"), ["start", "end", "start", "end"]);
        assert_eq!(fake.calls().len(), 2);
    }

    /// A request that arrives while the slot is held past the wait limit is
    /// refused and never reaches `nix`.
    #[tokio::test(start_paused = true)]
    async fn a_build_that_waited_out_the_slot_is_answered_busy_without_running_nix() {
        let fake = Fake::new(&format!("echo {OUT}"));
        let (env, _sent) = env();
        let slot = Semaphore::new(1);
        let _held = slot.acquire().await.expect("open");
        let received = Instant::now();
        let reply = build_reply(
            build_json("atlas", REV, 3).as_bytes(),
            received,
            &slot,
            &fake.builder(),
            &env.tokens,
            &Metrics::noop(),
        )
        .await;
        assert!(
            matches!(&reply, Some(Reply::Failed { error_tail }) if error_tail.contains("busy")),
            "{reply:?}"
        );
        assert_eq!(received.elapsed(), MAX_SLOT_WAIT);
        assert_eq!(fake.calls(), Vec::<String>::new());
    }

    #[tokio::test]
    async fn a_closed_slot_gets_no_reply() {
        let fake = Fake::new(&format!("echo {OUT}"));
        let (env, _sent) = env();
        let slot = Semaphore::new(1);
        slot.close();
        let reply = build_reply(
            build_json("atlas", REV, 3).as_bytes(),
            Instant::now(),
            &slot,
            &fake.builder(),
            &env.tokens,
            &Metrics::noop(),
        )
        .await;
        assert!(reply.is_none(), "{reply:?}");
        assert_eq!(fake.calls(), Vec::<String>::new());
    }

    /// `nix` for a meta build: `build` prints the path, `eval` of the option
    /// prints `eval_says`.
    fn meta_nix(eval_says: &str) -> Fake {
        Fake::new(&format!(
            "case \"$1\" in build) echo {OUT};; eval) {eval_says};; esac"
        ))
    }

    #[tokio::test]
    async fn a_meta_build_is_answered_with_the_path_the_system_and_the_private_users_flag() {
        let fake = meta_nix("echo true");
        let (env, mut sent) = env();
        let request = msg(Some("_INBOX.5"), &meta_json("atlas", REV));
        serve_meta(fake.builder(), env, vec![request]).await;
        let (subject, body) = next(&mut sent).await;
        assert_eq!(subject, "_INBOX.5");
        assert_eq!(
            body,
            json!({
                "state": "built",
                "out_path": OUT,
                "system": "x86_64-linux",
                "cached": false,
                "private_users": true,
            })
        );
        assert_eq!(fake.lines("tokens"), ["forge-token"; 2]);
        let calls = fake.calls();
        assert_eq!(calls.len(), 2, "{calls:?}");
        assert!(
            calls[0].contains(&format!(
                "git+https://forge.example/internal/meta.git?rev={REV}#nixosConfigurations.atlas.config.system.build.toplevel"
            )),
            "{calls:?}"
        );
        assert!(
            calls[1].contains("services.hyperhive.agent.user.privateUsers"),
            "{calls:?}"
        );
    }

    #[tokio::test]
    async fn private_users_is_false_unless_nix_eval_printed_true() {
        for eval_says in [
            "echo false",
            "echo null",
            "echo 'error: no such option' >&2; exit 1",
        ] {
            let fake = meta_nix(eval_says);
            let (env, mut sent) = env();
            let request = msg(Some("_INBOX.1"), &meta_json("atlas", REV));
            serve_meta(fake.builder(), env, vec![request]).await;
            let (_, body) = next(&mut sent).await;
            assert_eq!(body["state"], "built", "{eval_says}");
            assert_eq!(body["private_users"], false, "{eval_says}");
        }
    }

    #[tokio::test]
    async fn a_built_meta_closure_is_pushed_and_the_reply_says_so() {
        let fake = meta_nix("echo true");
        let (env, mut sent) = env();
        let request = msg(Some("_INBOX.1"), &meta_json("atlas", REV));
        serve_meta(fake.caching_builder(), env, vec![request]).await;
        let (_, body) = next(&mut sent).await;
        assert_eq!(body["cached"], true, "{body}");
        assert_eq!(body["private_users"], true, "the flag survives the push");
        assert_eq!(body["out_path"], OUT);
        assert_eq!(fake.pushes(), [format!("push swarm:grid {OUT}")]);
    }

    /// Nothing is pushed for a build that did not succeed; the presence control
    /// is the test above, which uses the same cache.
    #[tokio::test]
    async fn a_failed_meta_build_is_failed_and_pushes_nothing() {
        let fake = Fake::new("echo 'error: attribute missing' >&2; exit 1");
        let (env, mut sent) = env();
        let request = msg(Some("_INBOX.1"), &meta_json("atlas", REV));
        serve_meta(fake.caching_builder(), env, vec![request]).await;
        let (_, body) = next(&mut sent).await;
        assert_eq!(body["state"], "failed");
        assert!(
            body["error_tail"]
                .as_str()
                .is_some_and(|t| t.ends_with("attribute missing")),
            "{body}"
        );
        assert_eq!(fake.pushes(), Vec::<String>::new());
    }

    #[tokio::test]
    async fn an_unreadable_push_token_leaves_the_meta_build_built_and_not_cached() {
        let fake = meta_nix("echo true");
        let (env, mut sent) = rig_with(token("t"), unreadable("nothing at the push path"), 0);
        let request = msg(Some("_INBOX.1"), &meta_json("atlas", REV));
        serve_meta(fake.caching_builder(), env, vec![request]).await;
        let (_, body) = next(&mut sent).await;
        assert_eq!(body["state"], "built");
        assert_eq!(body["cached"], false);
        assert_eq!(fake.pushes(), Vec::<String>::new());
    }

    #[tokio::test]
    async fn a_meta_build_whose_forge_token_cannot_be_read_is_unavailable_without_running_nix() {
        let fake = meta_nix("echo true");
        let (env, mut sent) = rig_with(unreadable("nothing at the forge path"), token("p"), 0);
        let request = msg(Some("_INBOX.1"), &meta_json("atlas", REV));
        serve_meta(fake.caching_builder(), env, vec![request]).await;
        let (_, body) = next(&mut sent).await;
        assert_eq!(
            body,
            json!({"state": "unavailable", "reason": "nothing at the forge path"})
        );
        assert_eq!(fake.calls(), Vec::<String>::new());
        assert_eq!(fake.pushes(), Vec::<String>::new());
    }

    #[tokio::test]
    async fn bad_meta_requests_are_failed_without_running_nix() {
        let fake = meta_nix("echo true");
        for payload in [
            meta_json("Atlas", REV),
            meta_json("atlas", "abc"),
            "not json".to_owned(),
            "{}".to_owned(),
        ] {
            let (env, mut sent) = env();
            serve_meta(fake.builder(), env, vec![msg(Some("_INBOX.1"), &payload)]).await;
            let (_, body) = next(&mut sent).await;
            assert_eq!(body["state"], "failed", "{payload}: {body}");
        }
        assert_eq!(fake.calls(), Vec::<String>::new());
    }

    #[tokio::test]
    async fn a_meta_request_with_no_reply_subject_is_dropped() {
        let fake = meta_nix("echo true");
        let (env, mut sent) = env();
        let msgs = vec![
            msg(None, &meta_json("atlas", REV)),
            msg(Some("_INBOX.2"), &meta_json("atlas", REV)),
        ];
        serve_meta(fake.builder(), env, msgs).await;
        assert_eq!(next(&mut sent).await.0, "_INBOX.2");
        assert_eq!(fake.calls().len(), 2, "one build and its option read");
    }

    #[tokio::test(start_paused = true)]
    async fn a_meta_build_that_waited_out_the_slot_is_unavailable() {
        let fake = meta_nix("echo true");
        let (env, _sent) = env();
        let slot = Semaphore::new(1);
        let _held = slot.acquire().await.expect("open");
        let reply = meta_reply(
            meta_json("atlas", REV).as_bytes(),
            Instant::now(),
            &slot,
            &fake.builder(),
            &env.tokens,
            &Metrics::noop(),
        )
        .await;
        assert!(
            matches!(&reply, Some(FromMetaReply::Unavailable { reason }) if reason.contains("busy")),
            "{reply:?}"
        );
        assert_eq!(fake.calls(), Vec::<String>::new());
    }

    async fn serve_locks(builder: Arc<nix_build::Builder>, env: Rig, msgs: Vec<Message>) {
        answer_locks(stream::iter(msgs), env.sink, env.tokens, builder).await;
    }

    #[tokio::test]
    async fn a_lock_is_answered_with_the_lock_nix_wrote_for_the_flake_it_was_sent() {
        let fake = Fake::new("cp flake.nix flake.lock");
        let (env, mut sent) = env();
        let request = msg(
            Some("_INBOX.3"),
            &json!({"flake_nix": "{ a = 1; }"}).to_string(),
        );
        serve_locks(fake.builder(), env, vec![request]).await;
        let (subject, body) = next(&mut sent).await;
        assert_eq!(subject, "_INBOX.3");
        assert_eq!(body, json!({"state": "locked", "flake_lock": "{ a = 1; }"}));
        assert_eq!(fake.lines("tokens"), ["forge-token"]);
    }

    #[tokio::test]
    async fn a_lock_that_fails_or_whose_token_cannot_be_read_is_answered_failed() {
        let fake = Fake::new("echo 'error: input unavailable' >&2; exit 1");
        let lock_request = json!({"flake_nix": "{ }"}).to_string();
        let (env, mut sent) = env();
        serve_locks(
            fake.builder(),
            env,
            vec![msg(Some("_INBOX.1"), &lock_request)],
        )
        .await;
        let (_, body) = next(&mut sent).await;
        assert_eq!(body["state"], "failed");
        assert!(
            body["error_tail"]
                .as_str()
                .is_some_and(|t| t.ends_with("input unavailable")),
            "{body}"
        );

        let (env, mut sent) = rig_with(unreadable("nothing at the forge path"), token("p"), 0);
        let calls_before = fake.calls().len();
        serve_locks(
            fake.builder(),
            env,
            vec![msg(Some("_INBOX.1"), &lock_request)],
        )
        .await;
        let (_, body) = next(&mut sent).await;
        assert_eq!(
            body,
            json!({"state": "failed", "error_tail": "nothing at the forge path"})
        );
        assert_eq!(
            fake.calls().len(),
            calls_before,
            "the token read came first"
        );
    }

    #[tokio::test]
    async fn an_undecodable_lock_is_answered_failed_and_a_lock_with_no_reply_subject_is_dropped() {
        let fake = Fake::new("cp flake.nix flake.lock");
        let (env, mut sent) = env();
        let msgs = vec![
            msg(None, &json!({"flake_nix": "{ }"}).to_string()),
            msg(Some("_INBOX.1"), "{}"),
        ];
        serve_locks(fake.builder(), env, msgs).await;
        let (subject, body) = next(&mut sent).await;
        assert_eq!(subject, "_INBOX.1");
        assert_eq!(body["state"], "failed");
        assert!(sent.try_recv().is_err());
        assert_eq!(fake.calls(), Vec::<String>::new());
    }

    fn evaluator(fake: &Fake) -> archetypes::Evaluator {
        archetypes::Evaluator {
            nix: fake.0.join("nix"),
            hyperhive: "/nix/store/h-source".to_owned(),
            nixpkgs: "/nix/store/n-source".to_owned(),
            deadline: Duration::from_secs(30),
        }
    }

    #[tokio::test]
    async fn an_archetype_evaluation_is_answered_with_the_meta_nix_printed() {
        let fake = Fake::new(r#"echo '{"coder":{"description":"d"}}'"#);
        let (env, mut sent) = env();
        let msgs = vec![msg(None, "{}"), msg(Some("_INBOX.4"), "{}")];
        answer_evaluations(stream::iter(msgs), env.sink, evaluator(&fake)).await;
        let (subject, body) = next(&mut sent).await;
        assert_eq!(subject, "_INBOX.4");
        assert_eq!(
            body,
            json!({
                "state": "evaluated",
                "source": "/nix/store/h-source",
                "archetypes": {"coder": {"description": "d"}},
            })
        );
        assert_eq!(
            fake.calls().len(),
            1,
            "the request with no reply was not run"
        );
    }

    #[tokio::test]
    async fn a_failed_archetype_evaluation_is_answered_failed() {
        let fake = Fake::new("echo 'error: evaluation aborted' >&2; exit 1");
        let (env, mut sent) = env();
        let msgs = vec![msg(Some("_INBOX.4"), "{}")];
        answer_evaluations(stream::iter(msgs), env.sink, evaluator(&fake)).await;
        let (_, body) = next(&mut sent).await;
        assert_eq!(body["state"], "failed");
        assert!(
            body["error_tail"]
                .as_str()
                .is_some_and(|t| t.contains("evaluation aborted")),
            "{body}"
        );
    }

    #[test]
    fn the_forge_url_is_https_with_a_host_and_no_path() {
        let ok = |url| check_forge_url("V", url).expect("accepted");
        assert_eq!(ok("https://forge.example"), "https://forge.example");
        assert_eq!(
            ok("https://forge.example:3000//"),
            "https://forge.example:3000"
        );
        for bad in [
            "http://forge.example",
            "forge.example",
            "https://forge.example/sub",
            "",
        ] {
            let e = check_forge_url("SWARM_BUILDER_FORGE_URL", bad).expect_err(bad);
            assert!(
                e.to_string().starts_with("SWARM_BUILDER_FORGE_URL is "),
                "{bad}: {e}"
            );
        }
    }

    // --- metrics ---

    use crate::metrics::probe::{Point, Probe};

    const BUILDS: &str = "hyperhive.swarm_builder.builds";
    const DURATION: &str = "hyperhive.swarm_builder.build.duration";
    const WAIT: &str = "hyperhive.swarm_builder.queue.wait";
    const DEPTH: &str = "hyperhive.swarm_builder.queue.depth";
    const PUBLISH_FAILURES: &str = "hyperhive.swarm_builder.publish.failures";

    /// `(labels, value)` of every point of `name`.
    fn counts(probe: &Probe, name: &str) -> Vec<(String, i64)> {
        probe
            .read(name)
            .into_iter()
            .map(|Point { labels, value, .. }| (labels, value))
            .collect()
    }

    async fn pr_reply(
        fake: &Fake,
        env: &Rig,
        slot: &Semaphore,
        metrics: &Metrics,
    ) -> Option<Reply> {
        build_reply(
            build_json("atlas", REV, 3).as_bytes(),
            Instant::now(),
            slot,
            &fake.builder(),
            &env.tokens,
            metrics,
        )
        .await
    }

    #[tokio::test]
    async fn a_green_pr_build_counts_one_success_with_a_duration_and_a_wait() {
        let probe = Probe::new();
        let fake = Fake::new(&format!("echo {OUT}"));
        let (env, _sent) = env();
        let reply = pr_reply(&fake, &env, &Semaphore::new(1), &probe.metrics).await;
        assert!(matches!(reply, Some(Reply::Built { .. })), "{reply:?}");
        assert_eq!(
            counts(&probe, BUILDS),
            [("kind=pr,outcome=success".to_owned(), 1)]
        );
        assert_eq!(
            counts(&probe, DURATION),
            [("kind=pr,outcome=success".to_owned(), 1)]
        );
        assert_eq!(counts(&probe, WAIT), [("kind=pr".to_owned(), 1)]);
    }

    #[tokio::test]
    async fn a_red_pr_build_counts_a_failure_and_no_success() {
        let probe = Probe::new();
        let fake = Fake::new("echo 'error: boom' >&2\nexit 1");
        let (env, _sent) = env();
        pr_reply(&fake, &env, &Semaphore::new(1), &probe.metrics).await;
        assert_eq!(
            counts(&probe, BUILDS),
            [("kind=pr,outcome=failure".to_owned(), 1)]
        );
        assert_eq!(
            counts(&probe, DURATION),
            [("kind=pr,outcome=failure".to_owned(), 1)]
        );
    }

    #[tokio::test]
    async fn a_pr_build_whose_token_cannot_be_read_counts_a_failure() {
        let probe = Probe::new();
        let fake = Fake::new(&format!("echo {OUT}"));
        let (env, _sent) = rig_with(unreadable("no token"), token("p"), 0);
        pr_reply(&fake, &env, &Semaphore::new(1), &probe.metrics).await;
        assert_eq!(
            counts(&probe, BUILDS),
            [("kind=pr,outcome=failure".to_owned(), 1)]
        );
    }

    #[tokio::test]
    async fn an_undecodable_request_counts_a_failure_of_its_kind_and_no_duration() {
        let probe = Probe::new();
        let fake = Fake::new(&format!("echo {OUT}"));
        let (env, _sent) = env();
        let slot = Semaphore::new(1);
        build_reply(
            b"{",
            Instant::now(),
            &slot,
            &fake.builder(),
            &env.tokens,
            &probe.metrics,
        )
        .await;
        meta_reply(
            b"{",
            Instant::now(),
            &slot,
            &fake.builder(),
            &env.tokens,
            &probe.metrics,
        )
        .await;
        assert_eq!(
            counts(&probe, BUILDS),
            [
                ("kind=meta,outcome=failure".to_owned(), 1),
                ("kind=pr,outcome=failure".to_owned(), 1),
            ]
        );
        assert_eq!(probe.read(DURATION), Vec::<Point>::new());
    }

    async fn meta_outcome(fake: &Fake, env: &Rig, metrics: &Metrics) {
        meta_reply(
            meta_json("atlas", REV).as_bytes(),
            Instant::now(),
            &Semaphore::new(1),
            &fake.builder(),
            &env.tokens,
            metrics,
        )
        .await;
    }

    #[tokio::test]
    async fn a_meta_build_is_counted_under_its_own_kind_for_success_and_for_failure() {
        let probe = Probe::new();
        let (env, _sent) = env();
        meta_outcome(&meta_nix("echo true"), &env, &probe.metrics).await;
        assert_eq!(
            counts(&probe, BUILDS),
            [("kind=meta,outcome=success".to_owned(), 1)]
        );
        let failing = Fake::new("exit 1");
        meta_outcome(&failing, &env, &probe.metrics).await;
        assert_eq!(
            counts(&probe, BUILDS),
            [
                ("kind=meta,outcome=failure".to_owned(), 1),
                ("kind=meta,outcome=success".to_owned(), 1),
            ]
        );
        assert_eq!(
            counts(&probe, DURATION),
            [
                ("kind=meta,outcome=failure".to_owned(), 1),
                ("kind=meta,outcome=success".to_owned(), 1),
            ]
        );
    }

    #[tokio::test]
    async fn a_meta_build_with_no_forge_token_counts_a_failure() {
        let probe = Probe::new();
        let (env, _sent) = rig_with(unreadable("no token"), token("p"), 0);
        meta_outcome(&meta_nix("echo true"), &env, &probe.metrics).await;
        assert_eq!(
            counts(&probe, BUILDS),
            [("kind=meta,outcome=failure".to_owned(), 1)]
        );
    }

    #[tokio::test(start_paused = true)]
    async fn a_refused_request_counts_refused_records_its_whole_wait_and_no_duration() {
        let probe = Probe::new();
        let fake = Fake::new(&format!("echo {OUT}"));
        let (env, _sent) = env();
        let slot = Semaphore::new(1);
        let _held = slot.acquire().await.expect("open");
        pr_reply(&fake, &env, &slot, &probe.metrics).await;
        assert_eq!(
            counts(&probe, BUILDS),
            [("kind=pr,outcome=refused".to_owned(), 1)]
        );
        assert_eq!(probe.read(DURATION), Vec::<Point>::new());
        let wait = probe.read(WAIT);
        assert_eq!(wait.len(), 1, "{wait:?}");
        assert!(
            (wait[0].sum - MAX_SLOT_WAIT.as_secs_f64()).abs() < 1e-6,
            "{wait:?}"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn the_queue_depth_counts_requests_waiting_for_the_slot_and_returns_to_zero() {
        let probe = Probe::new();
        let slot = Arc::new(Semaphore::new(1));
        let held = Arc::clone(&slot).acquire_owned().await.expect("open");
        let waiter = {
            let (slot, metrics) = (Arc::clone(&slot), probe.metrics.clone());
            tokio::spawn(async move {
                in_slot(
                    &slot,
                    Instant::now(),
                    (&metrics, Kind::Pr),
                    || Reply::Failed {
                        error_tail: "busy".to_owned(),
                    },
                    async {
                        Reply::Built {
                            out_path: "/nix/store/x".to_owned(),
                        }
                    },
                )
                .await
            })
        };
        tokio::time::sleep(Duration::from_secs(1)).await;
        assert_eq!(counts(&probe, DEPTH), [(String::new(), 1)]);
        drop(held);
        waiter.await.expect("joined");
        assert_eq!(counts(&probe, DEPTH), [(String::new(), 0)]);
    }

    #[tokio::test]
    async fn a_reply_the_queue_refuses_counts_one_publish_failure_and_a_sent_one_none() {
        let probe = Probe::new();
        let mut sink = MockPublisher::new();
        let mut calls = 0;
        sink.expect_publish().returning(move |_, _| {
            calls += 1;
            let ok = calls > 1;
            Box::pin(async move { if ok { Ok(()) } else { Err("closed".to_owned()) } })
        });
        let metered = Metered {
            inner: Arc::new(sink),
            metrics: probe.metrics.clone(),
        };
        let failed = metered.publish(Subject::from("a"), vec![]).await;
        assert!(failed.is_err());
        assert_eq!(counts(&probe, PUBLISH_FAILURES), [(String::new(), 1)]);
        let sent = metered.publish(Subject::from("b"), vec![]).await;
        assert!(sent.is_ok());
        assert_eq!(counts(&probe, PUBLISH_FAILURES), [(String::new(), 1)]);
    }
}