//! The swarm builder's build service.
//!
//! Connects to the swarm queue as its own OIDC client, takes build requests
//! from swarm-controller on [`swarm_queue_client::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_queue_client::builder::ARCHETYPES_SUBJECT`] with the agent
//! archetypes its hyperhive source declares, which the controller stores.
//!
//! 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 futures_util::StreamExt as _;
use swarm_queue_client::builder::{
    ARCHETYPES_DEADLINE, ARCHETYPES_SUBJECT, BUILD_DEADLINE, BuildAgent, QUEUE_GROUP,
    REPLY_DEADLINE, Reply, SUBJECT,
};
use swarm_queue_client::{ClientSecret, QueueConfig};
use swarm_secret_client::forge::Credential;
use swarm_secret_client::stored_secret::StoredSecret;

mod archetypes;
mod nix_build;

/// 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);

#[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")?,
        hyperhive_docs: var("HYPERHIVE_DOCS_FLAKE")?,
        forge_url: forge_url()?,
        config_org: var("CONFIG_ORG")?,
        deadline: BUILD_DEADLINE,
    });
    let secret =
        swarm_queue_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 mut requests = queue
        .queue_subscribe(SUBJECT, QUEUE_GROUP.to_owned())
        .await
        .with_context(|| format!("subscribing to {SUBJECT}"))?;
    tracing::info!(subject = SUBJECT, "builder: awaiting builds");
    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, queue.clone(), evaluator));

    // 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.
    let slot = Arc::new(tokio::sync::Semaphore::new(1));

    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 queue = queue.clone();
        let slot = Arc::clone(&slot);
        let builder = Arc::clone(&builder);
        tokio::spawn(async move {
            let reply = match serde_json::from_slice::<BuildAgent>(&msg.payload) {
                Ok(job) => {
                    let Some(reply) = in_slot(&slot, received, &job, handle(&job, &builder)).await
                    else {
                        return;
                    };
                    reply
                }
                Err(e) => {
                    tracing::warn!(error = %e, "an undecodable build; answering with the error");
                    Reply::Failed {
                        error_tail: format!("the builder could not decode the request: {e}"),
                    }
                }
            };
            match serde_json::to_vec(&reply) {
                Ok(body) => {
                    if let Err(e) = queue.publish(reply_to, body.into()).await {
                        tracing::warn!(error = %e, "publishing a build's reply failed");
                    }
                }
                Err(e) => tracing::warn!(error = %e, "encoding a build's reply failed"),
            }
        });
    }
    bail!("the subscription to {SUBJECT} ended")
}

/// Answer every archetype evaluation request, one at a time, with the
/// evaluation's result.
async fn answer_evaluations(
    mut requests: async_nats::Subscriber,
    queue: async_nats::Client,
    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_queue_client::builder::ArchetypesReply::Evaluated { archetypes, .. } => {
                tracing::info!(count = archetypes.len(), "archetypes evaluated");
            }
            swarm_queue_client::builder::ArchetypesReply::Failed { error_tail } => {
                tracing::warn!(%error_tail, "archetype evaluation failed");
            }
        }
        match serde_json::to_vec(&reply) {
            Ok(body) => {
                if let Err(e) = queue.publish(reply_to, body.into()).await {
                    tracing::warn!(error = %e, "publishing an evaluation's reply failed");
                }
            }
            Err(e) => tracing::warn!(error = %e, "encoding an evaluation's reply failed"),
        }
    }
    tracing::error!("the subscription to {ARCHETYPES_SUBJECT} ended");
}

/// `build`'s reply once `job` holds the build slot, or a refusal, without
/// running `build`, once it has waited [`MAX_SLOT_WAIT`] since `received`.
/// `None` only when the slot is closed.
async fn in_slot(
    slot: &tokio::sync::Semaphore,
    received: tokio::time::Instant,
    job: &BuildAgent,
    build: impl Future<Output = Reply>,
) -> Option<Reply> {
    let Ok(permit) = tokio::time::timeout_at(received + MAX_SLOT_WAIT, slot.acquire()).await else {
        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"
        );
        return Some(Reply::Failed {
            error_tail: 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()
            ),
        });
    };
    let _permit = permit.ok()?;
    Some(build.await)
}

/// Build one request and log the build.
async fn handle(job: &BuildAgent, builder: &nix_build::Builder) -> Reply {
    let reply = match within(TOKEN_DEADLINE, 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 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}: swarm-secret-publish, in authelia's container, has not \
             published this service's client secret"
        ),
    }
}

/// 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"))?;
    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, &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);
    }

    #[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");
    }

    #[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);
    }
}