channel-scripts/src/hydra.rs

use std::{collections::HashMap, path::PathBuf};
use anyhow::Context;
use bytes::Bytes;
use futures::Stream;
use indicatif::ProgressBar;
use nix_compat::store_path::{StorePath, StorePathRef};
use regex::Regex;

use serde::{de::Error, Deserialize, Deserializer};
use tokio::io::AsyncWrite;

use crate::{actions::HydraProductType, config::MirrorConfig};

pub type ReleaseId = u64;
pub type EvaluationId = u64;
pub type BuildId = u64;

fn deser_bool_as_int<'de, D>(deserializer: D) -> Result<bool, D::Error>
    where D: Deserializer<'de>
{
    u8::deserialize(deserializer).map(|b| if b == 1 { true } else { false })
}

fn deser_bool_as_string<'de, D>(deserializer: D) -> Result<bool, D::Error>
where D: Deserializer<'de>
{
    let s: &str = Deserialize::deserialize(deserializer)?;

    match s {
        "false" => Ok(false),
        "true" => Ok(true),
        _ => Err(Error::unknown_variant(s, &["false", "true"]))
    }
}

#[derive(Debug, Deserialize)]
pub struct OutputPath {
    #[serde(rename="path")]
    store_path: String
}

#[derive(Debug, Deserialize)]
pub struct BuildProduct {
    #[serde(rename="path")]
    store_path: String,
    name: String,
    subtype: String,
    sha256hash: String,
    r#type: String,
    #[serde(rename="filesize")]
    file_size: u64
}

#[derive(Debug, Deserialize)]
pub struct BuildInfo {
    jobset: String,
    job: String,
    project: String,
    #[serde(deserialize_with = "deser_bool_as_int", rename="buildstatus")]
    finished: bool,
    #[serde(rename="releasename")]
    release_name: Option<String>,
    #[serde(rename="stoptime")]
    stop_time: u64,
    #[serde(rename="starttime")]
    start_time: u64,
    system: String,
    id: BuildId,
    #[serde(rename="buildoutputs")]
    build_outputs: HashMap<String, OutputPath>,
    #[serde(rename="buildproducts")]
    build_products: HashMap<String, BuildProduct>,
    #[serde(rename="nixname")]
    nix_name: String,
    #[serde(rename="drvpath")]
    drv_path: String,
}

#[derive(Debug, Clone)]
pub struct Channel {
    pub name: String
}

#[non_exhaustive]
pub enum ChannelType {
    NixOS,
    Nixpkgs,
}

#[non_exhaustive]
pub enum ChannelVariant {
    Normal,
    Small
}

impl Channel {
    pub fn version(&self) -> String {
        let re = Regex::new("([a-z]+)-(?<ver>.*)").unwrap();
        let caps = re.captures(&self.name).expect("Failed to parse the channel name");

        caps["ver"].to_string()
    }

    pub fn prefix(&self) -> String {
        let re = Regex::new("(?<name>[a-z]+)-(.*)").unwrap();
        let caps = re.captures(&self.name).expect("Failed to parse the channel name");

        caps["name"].to_string()
    }

    // TODO: just regex match?
    pub fn r#type(&self) -> ChannelType {
        ChannelType::NixOS
    }

    pub fn variant(&self) -> ChannelVariant {
        ChannelVariant::Normal
    }
}

#[derive(Deserialize, Debug)]
pub struct Release {
    pub id: ReleaseId,
    job: String,
    #[serde(rename = "releasename")]
    release_name: Option<String>,
    #[serde(rename = "starttime")]
    start_time: u64,
    #[serde(rename = "stoptime")]
    stop_time: u64,
    #[serde(rename = "nixname")]
    pub nix_name: String,
    #[serde(rename = "jobsetevals")]
    jobset_evals: Vec<EvaluationId>,
    jobset: String,
    #[serde(deserialize_with = "deser_bool_as_int")]
    finished: bool,
    priority: u64,
    system: String,
    timestamp: u64,
    project: String,
    #[serde(rename = "drvpath")]
    derivation_path: String,
    // ignored: buildproducts, buildoutputs, buildmetrics, buildstatus
}

#[derive(Debug)]
pub struct GitInput {
    uri: String,
    revision: String
}

// FIXME(Raito): for some reason, #[serde(tag = "type"), rename_all = "lowercase"] doesn't behave
// correctly, and causes deserialization failures in practice. `untagged` is suboptimal but works
// because of the way responses works...
#[derive(Debug, Deserialize)]
#[serde(untagged, expecting = "An valid jobset input")]
pub enum Input {
    Boolean {
        #[serde(deserialize_with = "deser_bool_as_string")]
        value: bool
    },
    Git { uri: String, revision: String },
    Nix { value: String },
}

#[derive(Deserialize, Debug)]
pub struct Evaluation {
    pub id: EvaluationId,
    #[serde(rename="checkouttime")]
    checkout_time: u64,
    #[serde(rename="evaltime")]
    eval_time: u64,
    flake: Option<String>,
    #[serde(rename="jobsetevalinputs", default)]
    pub jobset_eval_inputs: HashMap<String, Input>,
    timestamp: u64,
    builds: Vec<BuildId>,
}

impl Release {
    pub fn version(&self) -> String {
        let re = Regex::new(".+-(?<ver>[0-9].+)").unwrap();
        let caps = re.captures(&self.nix_name).expect("Failed to parse the release name");

        caps["ver"].to_string()
    }

    pub fn evaluation_id(&self) -> u64 {
        *self.jobset_evals.first().expect("Failed to obtain the corresponding evaluation, malformed release?")
    }

    pub fn evaluation_url(&self, hydra_base_uri: &str) -> String {
        let eval_id = self.evaluation_id();
        format!("{}/eval/{}", hydra_base_uri, eval_id)
    }

    pub fn job_url(&self, hydra_base_uri: &str, job_name: &str) -> String {
        let eval_id = self.evaluation_id();
        format!("{}/eval/{}/job/{}", hydra_base_uri, eval_id, job_name)
    }

    pub fn store_paths_url(&self, hydra_base_uri: &str) -> String {
        let eval_id = self.evaluation_id();
        format!("{}/eval/{}/store-paths", hydra_base_uri, eval_id)
    }

    /// Directory related to this release.
    fn directory(&self, channel: &Channel) -> String {
        match channel.name.as_str() {
            "nixpkgs-unstable" => "nixpkgs".to_string(),
            _ => format!("{}/{}", channel.prefix(), channel.version())
        }
    }

    pub fn prefix(&self, channel: &Channel) -> object_store::path::Path {
        format!("{}/{}", self.directory(channel), self.nix_name).into()
    }
}

pub fn release_uri(hydra_uri: &str, job_name: &str) -> String {
    format!("{}/job/{}/latest", hydra_uri, job_name)
}

#[derive(Debug)]
pub struct HydraClient<'a> {
    pub config: &'a MirrorConfig,
}

impl HydraClient<'_> {
    pub async fn fetch_release(&self, job_name: &str) -> reqwest::Result<Release> {
        let client = reqwest::Client::new();
        let resp = client.get(release_uri(&self.config.hydra_uri, job_name))
            .header("Accept", "application/json")
            // TODO: put a proper version
            .header("User-Agent", "nixos-channel-scripts (rust)")
            .send()
            .await?;

        resp.json().await
    }

    pub async fn fetch_evaluation(&self, release: &Release) -> reqwest::Result<Evaluation> {
        let client = reqwest::Client::new();
        let resp = client.get(release.evaluation_url(&self.config.hydra_uri))
            .header("Accept", "application/json")
            .header("User-Agent", "nixos-channel-scripts (rust)")
            .send()
            .await?;

        resp.json().await
    }

    pub async fn fetch_store_paths(&self, release: &Release) -> anyhow::Result<Vec<StorePath<String>>> {
        let client = reqwest::Client::new();

        Ok(client.get(release.store_paths_url(&self.config.hydra_uri))
            .header("Accept", "application/json")
            .header("User-Agent", "nixos-channel-scripts (rust)")
            .send()
            .await?
            .json()
            .await
            .context(format!("while downloading store-paths information for an evaluation"))?
        )
    }

    pub async fn hydra_product_bytes_stream(&self, release: &Release, job_name: &str, product_type: Option<HydraProductType>) -> anyhow::Result<(u64, impl Stream<Item = std::io::Result<Bytes>>)> {
        let client = reqwest::Client::new();
        let build_info: BuildInfo = client.get(release.job_url(&self.config.hydra_uri, job_name))
            .header("Accept", "application/json")
            .header("User-Agent", "nixos-channel-scripts (rust)")
            .send()
            .await?
            .json()
            .await
            .context(format!("while downloading build information from {}", release.job_url(&self.config.hydra_uri, job_name)))?;

        let mut products_by_subtype: HashMap<String, BuildProduct> = HashMap::new();
        for (_, product) in build_info.build_products {
            if products_by_subtype.contains_key(&product.subtype) {
                todo!("Job {} has multiple products of the same subtype {:?}. This is not supported yet.", job_name, product_type);
            }

            products_by_subtype.insert(product.subtype.clone(), product);
        }

        if products_by_subtype.len() > 1 && product_type.is_none() {
            panic!("Job {} has {} build products. Select the right product via the subtypes.", job_name, products_by_subtype.len());
        }

        let product: BuildProduct = if let Some(ptype) = product_type {
            products_by_subtype.remove(&ptype.to_string()).expect(&format!("Expected product type '{}' but not found in the list of products", &ptype.to_string()))
        } else {
            products_by_subtype.into_iter().last().expect(&format!("Expected at least one build product in job {}, found zero", job_name)).1
        };

        // 3. FIXME: validate sha256hash during read?
        let (store_path, rel_path) = StorePathRef::from_absolute_path_full(&product.store_path).expect("Failed to parse the product's store path");
        crate::nar::file_in_nar_bytes_stream(&self.config.binary_cache_uri, &nix_compat::nixbase32::encode(store_path.digest()), rel_path.to_str().unwrap()).await.context("while copying the NAR to the target")
    }

    pub async fn copy_hydra_product<W: AsyncWrite + Unpin>(&self, release: &Release, job_name: &str, product_type: Option<HydraProductType>, pbar: ProgressBar, out: &mut W) -> anyhow::Result<u64> {
        // TODO: dry me?
        let client = reqwest::Client::new();
        let build_info: BuildInfo = client.get(release.job_url(&self.config.hydra_uri, job_name))
            .header("Accept", "application/json")
            .header("User-Agent", "nixos-channel-scripts (rust)")
            .send()
            .await?
            .json()
            .await
            .context(format!("while downloading build information from {}", release.job_url(&self.config.hydra_uri, job_name)))?;

        let mut products_by_subtype: HashMap<String, BuildProduct> = HashMap::new();
        for (_, product) in build_info.build_products {
            if products_by_subtype.contains_key(&product.subtype) {
                todo!("Job {} has multiple products of the same subtype {:?}. This is not supported yet.", job_name, product_type);
            }

            products_by_subtype.insert(product.subtype.clone(), product);
        }

        if products_by_subtype.len() > 1 && product_type.is_none() {
            panic!("Job {} has {} build products. Select the right product via the subtypes.", job_name, products_by_subtype.len());
        }

        let product: BuildProduct = if let Some(ptype) = product_type {
            products_by_subtype.remove(&ptype.to_string()).expect(&format!("Expected product type '{}' but not found in the list of products", &ptype.to_string()))
        } else {
            products_by_subtype.into_iter().last().expect(&format!("Expected at least one build product in job {}, found zero", job_name)).1
        };

        // 3. FIXME: validate sha256hash during read?
        let (store_path, rel_path) = StorePathRef::from_absolute_path_full(&product.store_path).expect("Failed to parse the product's store path");
        crate::nar::copy_file_in_nar(&self.config.binary_cache_uri, &nix_compat::nixbase32::encode(store_path.digest()), rel_path.to_str().unwrap(), pbar, out).await.context("while copying the NAR to the target")
    }
}