nxd_core/progress/

target.rs

use std::collections::BTreeMap;
use std::fs::File;
use std::io::Write;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use crate::progress::color::ColorMode;
use crate::progress::log::{ProgressEvent, StatusLevel, normalize_multiline_message};
use crate::progress::redaction::{ProgressRedactor, redact_sensitive_text};

pub enum LogTarget {
	Terminal {
		file: Option<File>,
		bulk: crate::progress::bulk::TerminalBulkState,
	},
	File(File),
	Batch {
		prefix: String,
		file: File,
		started_at: Option<Instant>,
		last_event_at: Option<Instant>,
		outcome: Option<BatchOutcome>,
		/// A single-target run echoes subprocess output to the terminal so the
		/// operator sees the same narrative the log records. Concurrent targets
		/// would interleave line by line, so a fleet run leaves this off and keeps
		/// the terminal to milestones.
		echo: Option<crate::progress::bulk::TerminalBulkState>,
	},
	Routed {
		fallback: Logger,
		hosts: BTreeMap<String, Logger>,
	},
	Silent,
}

impl LogTarget {
	pub fn log_event(&mut self, level: StatusLevel, message: &str) {
		self.write_event(level, message, message);
	}

	pub fn log_status(&mut self, message: &str) {
		self.write_event(StatusLevel::Info, message, message);
	}

	pub fn log_debug(&mut self, message: &str) {
		self.write_event(StatusLevel::Debug, message, message);
	}

	fn write_event(&mut self, level: StatusLevel, message: &str, file_message: &str) {
		let message = normalize_multiline_message(message);
		let message = redact_sensitive_text(&message);
		let file_message = normalize_multiline_message(file_message);
		let file_message = redact_sensitive_text(&file_message);
		let message_lines: Vec<&str> =
			if message.is_empty() { vec![""] } else { message.lines().collect() };
		let file_lines: Vec<&str> =
			if file_message.is_empty() { vec![""] } else { file_message.lines().collect() };

		match self {
			LogTarget::Terminal { file, bulk } => {
				if let Some(f) = file {
					for line in file_lines {
						let _ = writeln!(f, "[{}] {}", level.label(), line);
					}
				}
				bulk.clear_active_counter();
				bulk.flush_summary();
				for line in message_lines {
					let event = ProgressEvent::new(level, None, line);
					event.log(ColorMode::Auto);
				}
			}
			LogTarget::Batch { prefix, file, started_at, last_event_at, outcome, echo } => {
				let now = Instant::now();
				let started = *started_at.get_or_insert(now);
				let elapsed = now.saturating_duration_since(started);
				*last_event_at = Some(now);
				match level {
					StatusLevel::Success if *outcome != Some(BatchOutcome::Failed) => {
						*outcome = Some(BatchOutcome::Succeeded);
					}
					StatusLevel::Failure | StatusLevel::Error => *outcome = Some(BatchOutcome::Failed),
					_ => {}
				}
				// A milestone must not be garbled by an in-place counter still on screen.
				if let Some(bulk) = echo.as_mut() {
					bulk.clear_active_counter();
					bulk.flush_summary();
				}
				let terminal_message = message.strip_prefix(&format!("{prefix}: ")).unwrap_or(&message);
				let terminal_message = timed_batch_message(level, terminal_message, elapsed);
				for line in terminal_message.lines() {
					let event = ProgressEvent::new(level, Some(prefix), line);
					event.log(ColorMode::Auto);
				}
				for line in file_lines {
					let line = timed_batch_message(level, line, elapsed);
					let _ = writeln!(file, "[{}] {}", level.label(), line);
				}
			}
			LogTarget::File(file) => {
				for line in file_lines {
					let _ = writeln!(file, "[{}] {}", level.label(), line);
				}
			}
			LogTarget::Routed { .. } => {}
			LogTarget::Silent => {}
		}
	}
}

fn timed_batch_message(level: StatusLevel, message: &str, elapsed: Duration) -> String {
	let elapsed = crate::progress::log::format_elapsed(elapsed);
	match level {
		StatusLevel::Success => format!("{message} in {elapsed}"),
		StatusLevel::Failure | StatusLevel::Error => format!("{message} after {elapsed}"),
		_ => message.to_string(),
	}
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum BatchOutcome {
	Succeeded,
	Failed,
}

#[derive(Clone)]
pub struct Logger {
	pub target: Arc<Mutex<LogTarget>>,
	redactor: Arc<Mutex<ProgressRedactor>>,
}

impl Logger {
	pub fn new(target: Arc<Mutex<LogTarget>>) -> Self {
		Self { target, redactor: Arc::new(Mutex::new(ProgressRedactor::default())) }
	}

	pub fn with_sensitive_values<'a>(&self, values: impl IntoIterator<Item = &'a [u8]>) -> Self {
		Self {
			target: self.target.clone(),
			redactor: Arc::new(Mutex::new(ProgressRedactor::with_sensitive_values(values))),
		}
	}

	pub fn sanitize(&self, message: &str) -> String {
		self
			.redactor
			.lock()
			.map(|mut redactor| redactor.redact(message))
			.unwrap_or_else(|_| redact_sensitive_text(message))
	}

	pub fn lock(&self) -> std::sync::LockResult<std::sync::MutexGuard<'_, LogTarget>> {
		self.target.lock()
	}

	pub fn silent() -> Self {
		Self::new(Arc::new(Mutex::new(LogTarget::Silent)))
	}

	pub fn terminal() -> Self {
		Self::new(Arc::new(Mutex::new(LogTarget::Terminal {
			file: None,
			bulk: crate::progress::bulk::TerminalBulkState::default(),
		})))
	}

	pub fn terminal_with_file(file: File) -> Self {
		Self::new(Arc::new(Mutex::new(LogTarget::Terminal {
			file: Some(file),
			bulk: crate::progress::bulk::TerminalBulkState::default(),
		})))
	}

	pub fn file(file: File) -> Self {
		Self::new(Arc::new(Mutex::new(LogTarget::File(file))))
	}

	pub fn batch(prefix: String, file: File) -> Self {
		Self::batch_with_echo(prefix, file, false)
	}

	/// A batch logger that also echoes subprocess output to the terminal. Used for
	/// a single-target run, where nothing else is writing to the terminal
	/// concurrently, so the operator can watch the same narrative the log records.
	pub fn batch_with_echo(prefix: String, file: File, echo: bool) -> Self {
		Self::new(Arc::new(Mutex::new(LogTarget::Batch {
			prefix,
			file,
			started_at: None,
			last_event_at: None,
			outcome: None,
			echo: echo.then(crate::progress::bulk::TerminalBulkState::default),
		})))
	}

	pub fn routed(fallback: Logger, hosts: BTreeMap<String, Logger>) -> Self {
		Self::new(Arc::new(Mutex::new(LogTarget::Routed { fallback, hosts })))
	}

	pub fn for_host(&self, hostname: &str) -> Self {
		self
			.target
			.lock()
			.ok()
			.and_then(|target| match &*target {
				LogTarget::Routed { fallback, hosts } => {
					Some(hosts.get(hostname).unwrap_or(fallback).clone())
				}
				_ => None,
			})
			.unwrap_or_else(|| self.clone())
	}

	fn routed_fallback(&self) -> Option<Logger> {
		self.target.lock().ok().and_then(|target| match &*target {
			LogTarget::Routed { fallback, .. } => Some(fallback.clone()),
			_ => None,
		})
	}

	pub fn dump_suppressed_tail(&self) {
		if let Some(fallback) = self.routed_fallback() {
			fallback.dump_suppressed_tail();
			return;
		}
		let mut target = match self.target.lock() {
			Ok(guard) => guard,
			Err(_) => return,
		};
		let bulk = match &mut *target {
			LogTarget::Terminal { bulk, .. } => bulk,
			_ => return,
		};
		bulk.clear_active_counter();
		bulk.flush_summary();
		let tail = bulk.take_suppressed_tail();
		if !tail.is_empty() {
			eprintln!("[suppressed output tail on failure]:");
			for line in tail {
				eprintln!("{}", crate::progress::stream::format_output_line(&line));
			}
		}
	}

	pub fn event(&self, level: StatusLevel, message: &str) {
		if level == StatusLevel::Debug && !crate::config::get_runtime_options().debug {
			return;
		}
		let message = self.sanitize(message);
		if let Some(fallback) = self.routed_fallback() {
			fallback.event(level, &message);
			return;
		}
		if let Ok(mut lock) = self.target.lock() {
			lock.log_event(level, &message);
		}
	}

	/// Record command detail. Single-host terminal runs retain it; batch runs
	/// write it only to the host log so concurrent subprocess output cannot
	/// overwhelm the interactive fleet summary.
	pub fn detail(&self, message: &str) {
		let message = self.sanitize(message);
		if let Some(fallback) = self.routed_fallback() {
			fallback.detail(&message);
			return;
		}
		if let Ok(mut target) = self.target.lock() {
			match &mut *target {
				LogTarget::Terminal { file, bulk } => {
					let normalized = normalize_multiline_message(&message);
					if let Some(f) = file {
						for line in normalized.lines() {
							let _ = writeln!(f, "{}", line);
						}
					}
					for line in normalized.lines() {
						if crate::config::get_runtime_options().verbose {
							bulk.clear_active_counter();
							bulk.flush_summary();
							println!("{}", crate::progress::stream::format_output_line(line));
						} else {
							bulk.process_line(line);
						}
					}
				}
				LogTarget::Batch { file, echo, .. } => {
					for line in normalize_multiline_message(&message).lines() {
						let _ = writeln!(file, "{}", line);
						if let Some(bulk) = echo.as_mut() {
							if crate::config::get_runtime_options().verbose {
								bulk.clear_active_counter();
								bulk.flush_summary();
								println!("{}", crate::progress::stream::format_output_line(line));
							} else {
								bulk.process_line(line);
							}
						}
					}
				}
				LogTarget::File(file) => {
					for line in normalize_multiline_message(&message).lines() {
						let _ = writeln!(file, "{}", line);
					}
				}
				LogTarget::Routed { .. } | LogTarget::Silent => {}
			}
		}
	}

	/// Elapsed time between this host's first and most recent milestone.
	/// Batch callers use this after apply so a fast host is not charged for
	/// time spent waiting for another host in the fleet.
	pub fn elapsed(&self) -> Option<Duration> {
		self.target.lock().ok().and_then(|target| match &*target {
			LogTarget::Batch { started_at: Some(started), last_event_at: Some(last), .. } => {
				Some(last.saturating_duration_since(*started))
			}
			_ => None,
		})
	}

	pub fn batch_outcome(&self) -> Option<BatchOutcome> {
		self.target.lock().ok().and_then(|target| match &*target {
			LogTarget::Batch { outcome, .. } => *outcome,
			_ => None,
		})
	}

	pub fn output_line(&self, line: &str, stderr: bool) {
		let line = self.sanitize(line);
		if let Some(fallback) = self.routed_fallback() {
			fallback.output_line(&line, stderr);
			return;
		}
		if let Ok(mut target) = self.target.lock() {
			match &mut *target {
				LogTarget::Terminal { file, bulk } => {
					if let Some(f) = file {
						let _ = writeln!(f, "{}", line);
					}
					if stderr || crate::config::get_runtime_options().verbose {
						bulk.clear_active_counter();
						bulk.flush_summary();
						let formatted = crate::progress::stream::format_output_line(&line);
						if stderr {
							eprintln!("{}", formatted);
						} else {
							println!("{}", formatted);
						}
					} else {
						bulk.process_line(&line);
					}
				}
				LogTarget::Batch { file, echo, .. } => {
					let _ = writeln!(file, "{}", line);
					if let Some(bulk) = echo.as_mut() {
						if stderr || crate::config::get_runtime_options().verbose {
							bulk.clear_active_counter();
							bulk.flush_summary();
							let rendered = crate::progress::stream::format_output_line(&line);
							if stderr {
								eprintln!("{rendered}");
							} else {
								println!("{rendered}");
							}
						} else {
							bulk.process_line(&line);
						}
					}
				}
				LogTarget::File(file) => {
					let _ = writeln!(file, "{}", line);
				}
				LogTarget::Routed { .. } | LogTarget::Silent => {}
			}
		}
	}

	pub fn info(&self, message: &str) {
		self.event(StatusLevel::Info, message);
	}

	pub fn success(&self, message: &str) {
		self.event(StatusLevel::Success, message);
	}

	pub fn warn(&self, message: &str) {
		self.event(StatusLevel::Warning, message);
	}

	pub fn error(&self, message: &str) {
		self.event(StatusLevel::Error, message);
	}

	pub fn failure(&self, message: &str) {
		self.event(StatusLevel::Failure, message);
	}

	pub fn debug(&self, message: &str) {
		self.event(StatusLevel::Debug, message);
	}
}

#[macro_export]
macro_rules! info {
    ($logger:expr, $fmt:literal) => { $logger.info(&format!($fmt)) };
    ($logger:expr, $msg:expr) => { $logger.info($msg) };
    ($logger:expr, $fmt:literal, $($arg:tt)*) => { $logger.info(&format!($fmt, $($arg)*)) };
}

#[macro_export]
macro_rules! success {
    ($logger:expr, $fmt:literal) => { $logger.success(&format!($fmt)) };
    ($logger:expr, $msg:expr) => { $logger.success($msg) };
    ($logger:expr, $fmt:literal, $($arg:tt)*) => { $logger.success(&format!($fmt, $($arg)*)) };
}

#[macro_export]
macro_rules! warn {
    ($logger:expr, $fmt:literal) => { $logger.warn(&format!($fmt)) };
    ($logger:expr, $msg:expr) => { $logger.warn($msg) };
    ($logger:expr, $fmt:literal, $($arg:tt)*) => { $logger.warn(&format!($fmt, $($arg)*)) };
}

#[macro_export]
macro_rules! error {
    ($logger:expr, $fmt:literal) => { $logger.error(&format!($fmt)) };
    ($logger:expr, $msg:expr) => { $logger.error($msg) };
    ($logger:expr, $fmt:literal, $($arg:tt)*) => { $logger.error(&format!($fmt, $($arg)*)) };
}

#[macro_export]
macro_rules! failure {
    ($logger:expr, $fmt:literal) => { $logger.failure(&format!($fmt)) };
    ($logger:expr, $msg:expr) => { $logger.failure($msg) };
    ($logger:expr, $fmt:literal, $($arg:tt)*) => { $logger.failure(&format!($fmt, $($arg)*)) };
}

#[macro_export]
macro_rules! debug {
    ($logger:expr, $fmt:literal) => { $logger.debug(&format!($fmt)) };
    ($logger:expr, $msg:expr) => { $logger.debug($msg) };
    ($logger:expr, $fmt:literal, $($arg:tt)*) => { $logger.debug(&format!($fmt, $($arg)*)) };
}

#[cfg(test)]
#[path = "target_tests.rs"]
mod tests;