nix_builder/src/nix_util/running_build.rs

use std::borrow::Cow;
use std::time::Duration;
use std::time::Instant;
use std::time::SystemTime;

use sqlx::Row;
use tokio::process::Child;
use tracing::error;

use crate::Result;
use crate::database::db_handle::DbHandle;
use crate::nix_util::nix_output_stream::NixAction;
use crate::nix_util::nix_output_stream::NixOutputStream;
use crate::nix_util::output_stream::OutputStream;

use super::activity::Activity;
use super::activity_tree::ActivityTree;
use super::activity_tree::ActivityTreeEntry;
use super::activity_tree_stream::ActivityTreeStream;
use super::nix_output_stream::ActivityResultMessage;
use super::nix_output_stream::NixMessage;
use super::tree_iter::DrawDagEntry;
use super::tree_iter::ReverseTreeIter;
use super::tree_iter::get_draw_order;

pub(crate) struct RunningBuild<'db> {
    db_handle: &'db DbHandle,
    activity_tree: ActivityTreeStream,
    last_announce: Option<Instant>,
}

impl<'db> RunningBuild<'db> {
    pub(crate) fn new(db_handle: &'db DbHandle) -> Result<Self> {
        Ok(RunningBuild {
            db_handle,
            activity_tree: ActivityTreeStream::new(),
            last_announce: None,
        })
    }

    pub(crate) async fn run_to_completion<TN>(
        &mut self,
        mut child: Child,
        target_name: TN,
    ) -> Result<()>
    where
        TN: AsRef<str>,
    {
        let foo = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH)?;
        let now = Instant::now();
        let build_id: i64 = sqlx::query(
            r#"INSERT INTO build (start_time, target) SELECT unixepoch('now'), ? RETURNING id"#,
        )
        .bind(target_name.as_ref())
        .fetch_one(&self.db_handle.conn)
        .await?
        .try_get("id")?;

        let output_stream = OutputStream::from_child(&mut child)?;
        let mut nix_output_stream: NixOutputStream<OutputStream> =
            NixOutputStream::new(output_stream);

        let exit_status_handle = tokio::spawn(async move {
            let status = child
                .wait()
                .await
                .expect("nixos-rebuild encountered an error");
            status
        });

        while let Some(message) = nix_output_stream.next().await? {
            self.handle_message(message)?;
        }

        let exit_status = exit_status_handle.await?;
        println!("nix build status was: {}", exit_status);

        let update: u64 =
            sqlx::query(r#"UPDATE build SET end_time=unixepoch('now'), status=? WHERE id=?"#)
                .bind(
                    exit_status
                        .code()
                        .expect("Process should have an exit code."),
                )
                .bind(build_id)
                .execute(&self.db_handle.conn)
                .await?
                .rows_affected();
        assert!(update == 1);

        Ok(())
    }

    pub(crate) fn handle_message(&mut self, message: NixMessage) -> Result<()> {
        self.activity_tree.handle_message(&message)?;

        let message = match message {
            NixMessage::ParseFailure(line) => {
                error!("FAIL PARSE: {line}");
                return Ok(());
            }
            NixMessage::Generic(_value, line) => {
                error!("GENERIC PARSE: {line}");
                return Ok(());
            }
            NixMessage::Action(nix_action) => nix_action,
        };
        match message {
            NixAction::Msg(msg_message) => {
                if msg_message.level > 0 && msg_message.level < 5 {
                    eprintln!("LOG MESSAGE {}: {}", msg_message.level, msg_message.msg);
                }
            }
            NixAction::Start(_activity_start_message) => {
                self.print_current_status();
            }
            NixAction::Stop(_stop_message) => {
                self.print_current_status();
                // println!("{}", serde_json::to_string(&message)?);
            }
            NixAction::Result(activity_result_message) => {
                match activity_result_message {
                    ActivityResultMessage::FileLinked(_activity_result_file_linked) => {}
                    ActivityResultMessage::BuildLogLine(_activity_result_build_log_line) => {
                        // These are the output from the actual build (as opposed to the output from nix).
                    }
                    ActivityResultMessage::UntrustedPath(_activity_result_untrusted_path) => {}
                    ActivityResultMessage::CorruptedPath(_activity_result_corrupted_path) => {}
                    ActivityResultMessage::SetPhase(_activity_result_set_phase) => {
                        self.print_current_status();
                    }
                    ActivityResultMessage::Progress(_activity_result_progress) => {
                        self.maybe_print_current_status();
                    }
                    ActivityResultMessage::SetExpected(_activity_result_set_expected) => {
                        self.maybe_print_current_status();
                    }
                    ActivityResultMessage::PostBuildLogLine(
                        _activity_result_post_build_log_line,
                    ) => {}
                    ActivityResultMessage::FetchStatus(_activity_result_fetch_status) => {}
                };
            }
        };
        Ok(())
    }

    fn maybe_print_current_status(&mut self) -> () {
        let last_announce = match self.last_announce {
            Some(instant) => instant,
            None => {
                // If we haven't announced before, always announce.
                return self.print_current_status();
            }
        };

        let now = Instant::now();
        let time_since_last_announce = now.duration_since(last_announce);
        if time_since_last_announce > Duration::new(5, 0) {
            return self.print_current_status();
        }
    }

    fn print_current_status(&mut self) -> () {
        // let draw_order = get_draw_order(self.activity_tree.get_tree());
        let draw_order = ReverseTreeIter::new(
            self.activity_tree.get_tree(),
            None,
            is_match_predicate,
            is_transparent_predicate,
            is_alive_predicate,
        )
        .get_draw_order();

        print_dag(self.activity_tree.get_tree(), draw_order);
        self.last_announce = Some(Instant::now());
    }
}

pub(crate) fn print_dag(activity_tree: &ActivityTree, draw_order: Vec<DrawDagEntry>) -> () {
    let mut tree = String::new();
    for dag_entry in draw_order {
        let leading_bars = {
            let mut leading_bars = String::with_capacity(3 * dag_entry.depth.len());
            for leading_bar in dag_entry
                .depth
                .iter()
                .map(|depth| if *depth { "𜹈  " } else { "   " })
            {
                leading_bars.push_str(leading_bar);
            }
            leading_bars
        };
        let branch = if dag_entry.has_later_siblings {
            "𜸨"
        } else {
            "𜸛"
        };
        let activity = activity_tree.get(&dag_entry.activity_id);
        let display_name = activity
            .get_activity()
            .display_name()
            .expect("Currently we always return a display name.");
        let progress_text = activity.get_activity().get_progress_text();
        let (progress, progress_sep) = match progress_text {
            Some(text) => (text, " "),
            None => (Cow::Borrowed(""), ""),
        };
        tree += &format!("{leading_bars}{branch}𜸟 {progress}{progress_sep}{display_name}\n");
    }
    if tree.is_empty() {
        println!("No active activities.");
    } else {
        print!("\n{}\n", tree);
    }
}

fn is_match_predicate(_entry: &ActivityTreeEntry) -> bool {
    true
}

pub(crate) fn is_transparent_predicate(entry: &ActivityTreeEntry) -> bool {
    match entry.get_activity() {
        Activity::Root(_activity_root) => true,
        Activity::Unknown(_activity_unknown) => false,
        Activity::CopyPath(_activity_copy_path) => false,
        Activity::FileTransfer(_activity_file_transfer) => false,
        Activity::Realize(_activity_realize) => true,
        Activity::CopyPaths(_activity_copy_paths) => true,
        Activity::Builds(_activity_builds) => true,
        Activity::Build(_activity_build) => false,
        Activity::OptimizeStore(_activity_optimize_store) => false,
        Activity::VerifyPaths(_activity_verify_paths) => false,
        Activity::Substitute(_activity_substitute) => false,
        Activity::QueryPathInfo(_activity_query_path_info) => false,
        Activity::PostBuildHook(_activity_post_build_hook) => false,
        Activity::BuildWaiting(_activity_build_waiting) => true,
        Activity::FetchTree(_activity_fetch_tree) => false,
    }
}

fn is_alive_predicate(entry: &ActivityTreeEntry) -> bool {
    entry.get_activity().is_active()
}