use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use chrono::Utc;
use serde::{Deserialize, Serialize};
use tauri::{AppHandle, Emitter, State};
use tokio::io::{AsyncBufReadExt, BufReader};
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct LogStreamRequest {
pub server_name: String,
pub source_type: String, pub target: String, pub lines: u32,
}
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct LogLineEvent {
pub stream_id: String,
pub line_number: u64,
pub text: String,
pub level: String, pub timestamp: String,
}
#[derive(Default)]
pub struct LogStreamState {
pub streams: Mutex<HashMap<String, tokio::process::Child>>,
}
pub fn strip_ansi_escapes(input: &str) -> String {
let mut out = String::with_capacity(input.len());
let mut chars = input.chars().peekable();
while let Some(ch) = chars.next() {
if ch == '\x1B' {
if let Some(&'[') = chars.peek() {
chars.next(); for c in chars.by_ref() {
if c.is_ascii_alphabetic() || c == '@' {
break;
}
}
}
} else {
out.push(ch);
}
}
out
}
pub fn detect_log_level(text: &str) -> String {
let lower = text.to_lowercase();
if lower.contains("error")
|| lower.contains("fatal")
|| lower.contains("panic")
|| lower.contains("crit")
|| lower.contains("exception")
|| lower.contains("failed")
|| lower.contains("errno")
|| lower.contains("500 internal")
{
"error".to_string()
} else if lower.contains("warn") || lower.contains("alert") {
"warn".to_string()
} else if lower.contains("debug") || lower.contains("trace") {
"debug".to_string()
} else {
"info".to_string()
}
}
fn build_remote_command(source_type: &str, target: &str, lines: u32) -> String {
let lines_clamped = if lines == 0 { 100 } else { lines.min(2000) };
match source_type {
"journalctl" => format!("journalctl -u {} -f -n {} --no-pager", target, lines_clamped),
"docker" => format!("docker logs -f --tail {} {}", lines_clamped, target),
"file" => format!("tail -f -n {} {}", lines_clamped, target),
_ => target.to_string(),
}
}
fn resolve_sb_ssh_bin() -> Result<std::path::PathBuf, String> {
which::which("sb-ssh")
.or_else(|_| {
dirs::home_dir().map(|h| {
h.join(".cargo")
.join("bin")
.join(if cfg!(windows) { "sb-ssh.exe" } else { "sb-ssh" })
})
.filter(|p| p.exists())
.ok_or("Nicht gefunden")
})
.or_else(|_| {
std::env::current_exe().ok().and_then(|mut exe| {
exe.pop();
let candidate = exe.join(if cfg!(windows) { "sb-ssh.exe" } else { "sb-ssh" });
if candidate.exists() {
Some(candidate)
} else {
None
}
}).ok_or("Nicht gefunden")
})
.map_err(|_| "sb-ssh Binary wurde im System nicht gefunden (weder in PATH noch ~/.cargo/bin)".to_string())
}
#[tauri::command]
pub async fn start_log_stream(
req: LogStreamRequest,
app: AppHandle,
vault_state: State<'_, crate::vault::VaultState>,
log_state: State<'_, LogStreamState>,
) -> Result<String, String> {
crate::vault::ensure_unlocked(&vault_state)?;
let sb_ssh = resolve_sb_ssh_bin()?;
let stream_id = format!("log-{}", Utc::now().timestamp_millis());
let remote_cmd = build_remote_command(&req.source_type, &req.target, req.lines);
#[cfg(windows)]
let mut cmd = tokio::process::Command::new(sb_ssh);
#[cfg(windows)]
{
const CREATE_NO_WINDOW: u32 = 0x08000000;
cmd.creation_flags(CREATE_NO_WINDOW);
}
#[cfg(unix)]
let mut cmd = tokio::process::Command::new(sb_ssh);
cmd.args([&req.server_name, &remote_cmd]);
cmd.stdout(std::process::Stdio::piped());
cmd.stderr(std::process::Stdio::piped());
let mut child = cmd
.spawn()
.map_err(|e| format!("Fehler beim Starten des Log-Streamers: {}", e))?;
let stdout = child.stdout.take();
let stderr = child.stderr.take();
{
let mut streams = log_state.streams.lock().map_err(|e| e.to_string())?;
streams.insert(stream_id.clone(), child);
}
let line_counter = Arc::new(AtomicU64::new(1));
let event_name = format!("log-stream-line-{}", stream_id);
if let Some(stdout_pipe) = stdout {
let app_handle = app.clone();
let ev_name = event_name.clone();
let sid = stream_id.clone();
let counter = Arc::clone(&line_counter);
tokio::spawn(async move {
let mut reader = BufReader::new(stdout_pipe).lines();
while let Ok(Some(line)) = reader.next_line().await {
let clean = strip_ansi_escapes(&line);
let level = detect_log_level(&clean);
let num = counter.fetch_add(1, Ordering::SeqCst);
let event = LogLineEvent {
stream_id: sid.clone(),
line_number: num,
text: clean,
level,
timestamp: Utc::now().to_rfc3339(),
};
let _ = app_handle.emit(&ev_name, event);
}
});
}
if let Some(stderr_pipe) = stderr {
let app_handle = app;
let ev_name = event_name;
let sid = stream_id.clone();
let counter = Arc::clone(&line_counter);
tokio::spawn(async move {
let mut reader = BufReader::new(stderr_pipe).lines();
while let Ok(Some(line)) = reader.next_line().await {
let clean = strip_ansi_escapes(&line);
let num = counter.fetch_add(1, Ordering::SeqCst);
let event = LogLineEvent {
stream_id: sid.clone(),
line_number: num,
text: clean,
level: "error".to_string(), timestamp: Utc::now().to_rfc3339(),
};
let _ = app_handle.emit(&ev_name, event);
}
});
}
Ok(stream_id)
}
#[tauri::command]
pub async fn stop_log_stream(
stream_id: String,
log_state: State<'_, LogStreamState>,
) -> Result<(), String> {
let maybe_child = {
let mut streams = log_state.streams.lock().map_err(|e| e.to_string())?;
streams.remove(&stream_id)
};
if let Some(mut child) = maybe_child {
let _ = child.kill().await;
}
Ok(())
}
pub async fn stop_all_active_log_streams(log_state: &LogStreamState) {
let children: Vec<tokio::process::Child> = {
let mut streams = match log_state.streams.lock() {
Ok(guard) => guard,
Err(_) => return,
};
streams.drain().map(|(_, c)| c).collect()
};
for mut child in children {
let _ = child.kill().await;
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_strip_ansi_escapes() {
let colored = "\x1B[31m[ERROR]\x1B[0m Database connection lost";
let cleaned = strip_ansi_escapes(colored);
assert_eq!(cleaned, "[ERROR] Database connection lost");
}
#[test]
fn test_detect_log_level() {
assert_eq!(detect_log_level("2026-09-09 FATAL: unable to bind port"), "error");
assert_eq!(detect_log_level("WARNING: slow query detected 450ms"), "warn");
assert_eq!(detect_log_level("DEBUG: payload received: 120 bytes"), "debug");
assert_eq!(detect_log_level("Server running on port 3000"), "info");
}
#[test]
fn test_build_remote_command() {
assert_eq!(
build_remote_command("journalctl", "nginx", 50),
"journalctl -u nginx -f -n 50 --no-pager"
);
assert_eq!(
build_remote_command("docker", "redis-prod", 100),
"docker logs -f --tail 100 redis-prod"
);
assert_eq!(
build_remote_command("file", "/var/log/app.log", 200),
"tail -f -n 200 /var/log/app.log"
);
assert_eq!(
build_remote_command("custom", "dmesg -w", 100),
"dmesg -w"
);
}
}