use anyhow::{Context, Result};
use std::path::PathBuf;
use std::sync::atomic::AtomicBool;
use std::sync::Arc;
use crate::generate_execution_id;
use magellan::backend_router::MagellanBackend;
use magellan::graph::validation;
use magellan::WatchPipelineConfig;
use magellan::WatcherConfig;
use serde_json::json;
fn send_watch_request(req_line: &str, exec_id: &str) -> Result<()> {
use std::io::{Read, Write};
use std::os::unix::net::UnixStream;
use std::time::Duration;
let socket_path = std::path::PathBuf::from(crate::service::socket_path());
let mut stream = UnixStream::connect(&socket_path)
.with_context(|| format!("Daemon socket unreachable at {}", socket_path.display()))?;
let _ = stream.set_read_timeout(Some(Duration::from_millis(500)));
let _ = stream.set_write_timeout(Some(Duration::from_millis(500)));
let req_with_newline = format!("{}\n", req_line);
if let Err(e) = stream.write_all(req_with_newline.as_bytes()) {
return Err(anyhow::anyhow!(
"Failed to write watch request to daemon socket: {}",
e
));
}
let _ = stream.shutdown(std::net::Shutdown::Write);
let mut buf = [0u8; 4096];
match stream.read(&mut buf) {
Ok(n) if n > 0 => {
let resp = String::from_utf8_lossy(&buf[..n]);
if resp.contains(r#""error""#) {
return Err(anyhow::anyhow!("Daemon refused watch request: {}", resp));
}
println!(
"Watch request dispatched to daemon ({}) — response: {}",
exec_id,
resp.trim()
);
Ok(())
}
Ok(_) => Err(anyhow::anyhow!("Daemon socket closed before response")),
Err(e) => Err(anyhow::anyhow!("Failed to read daemon response: {}", e)),
}
}
pub fn run_watch(
root_path: PathBuf,
db_path: PathBuf,
config: WatcherConfig,
scan_initial: bool,
validate: bool,
validate_only: bool,
compile_commands: Option<std::path::PathBuf>,
) -> Result<()> {
let mut args = vec![
"watch".to_string(),
"--root".to_string(),
root_path.to_string_lossy().to_string(),
"--db".to_string(),
db_path.to_string_lossy().to_string(),
];
if !scan_initial {
args.push("--watch-only".to_string());
}
if validate {
args.push("--validate".to_string());
}
if validate_only {
args.push("--validate-only".to_string());
}
args.push("--debounce-ms".to_string());
args.push(config.debounce_ms.to_string());
let exec_id = generate_execution_id();
if crate::service::is_daemon_running() {
return send_watch_request(
&json!({
"id": exec_id,
"method": "watch",
"tag": root_path.to_string_lossy(),
"paths": [root_path.to_string_lossy().to_string()],
})
.to_string(),
&exec_id,
);
}
let root_str = root_path.to_string_lossy().to_string();
let db_path_str = db_path.to_string_lossy().to_string();
let mut backend = Some(MagellanBackend::open_or_create(&db_path)?);
if let Some(MagellanBackend::SQLite(ref mut graph)) = &mut backend {
graph.execution_log().start_execution(
&exec_id,
env!("CARGO_PKG_VERSION"),
&args,
Some(&root_str),
&db_path_str,
)?;
}
if validate || validate_only {
if let Some(MagellanBackend::SQLite(ref mut graph)) = &mut backend {
graph
.telemetry()
.record_phase_start(&exec_id, "pre_validation")?;
let input_paths = vec![root_path.clone()];
match validation::pre_run_validate(&db_path, &root_path, &input_paths) {
Ok(report) if !report.passed => {
let error_count = report.errors.len();
let error_msg = format!("Pre-validation failed: {} errors", error_count);
graph
.telemetry()
.record_phase_end(&exec_id, "pre_validation")?;
graph.execution_log().finish_execution(
&exec_id,
"error",
Some(&error_msg),
0,
0,
0,
)?;
return Err(anyhow::anyhow!("Pre-validation failed"));
}
Ok(_) => {}
Err(e) => return Err(e),
}
graph
.telemetry()
.record_phase_end(&exec_id, "pre_validation")?;
if validate_only {
graph
.execution_log()
.finish_execution(&exec_id, "success", None, 0, 0, 0)?;
return Ok(());
}
}
}
let shutdown = Arc::new(AtomicBool::new(false));
let shutdown_clone = shutdown.clone();
#[cfg(unix)]
{
use signal_hook::consts::signal;
use signal_hook::flag;
let _ = flag::register(signal::SIGINT, shutdown_clone.clone())?;
let _ = flag::register(signal::SIGTERM, shutdown_clone.clone())?;
}
let _ = magellan::ingest::pool::warmup_parsers();
let mut pipeline_config =
WatchPipelineConfig::new(root_path, db_path.clone(), config, scan_initial);
pipeline_config.compile_commands_path = compile_commands;
let result = magellan::run_watch_pipeline(pipeline_config, shutdown);
if let Some(MagellanBackend::SQLite(ref mut graph)) = &mut backend {
let outcome = if result.is_ok() { "success" } else { "error" };
let error_msg = result.as_ref().err().map(|e| e.to_string());
let _ = graph.execution_log().finish_execution(
&exec_id,
outcome,
error_msg.as_deref(),
0,
0,
0,
);
}
match result {
Ok(count) => {
println!("SHUTDOWN");
println!("Watch session complete. Processed {} events.", count);
Ok(())
}
Err(e) => {
println!("SHUTDOWN");
Err(e)
}
}
}