hydra

main.rs (5288B)

---

 //! `hydrad` — the Hydra daemon. It owns all CoreAudio state (taps, aggregate devices,
 //! IOProcs) and is the only process that touches the audio engine. Clients connect over
 //! a Unix-domain socket.
 
 mod server;
 
 use std::error::Error;
 use std::fs;
 use std::io::{BufReader, BufWriter};
 use std::os::unix::fs::PermissionsExt;
 use std::os::unix::net::{UnixListener, UnixStream};
 use std::sync::{Arc, Mutex};
 
 use hydra_core::config::Config;
 use hydra_core::engine::Engine;
 use hydra_core::ffi::process;
 use hydra_ipc::{read_msg, write_msg, Command, Response};
 
 fn main() -> Result<(), Box<dyn Error>> {
     let sock = hydra_ipc::socket_path();
     if let Some(dir) = sock.parent() {
         fs::create_dir_all(dir)?;
     }
 
     // Single-instance guard. The socket file existing isn't enough to tell a live daemon
     // from a stale socket left by a crash, so probe it: if something answers Ping, another
     // daemon owns CoreAudio — bail rather than yank the socket out from under it (which is
     // how a pile of competing daemons accumulated before). If nothing answers, it's stale.
     if let Some(version) = ping_existing(&sock) {
         eprintln!("hydrad already running (protocol v{version}) on {}; exiting.", sock.display());
         return Ok(());
     }
     let _ = fs::remove_file(&sock);
 
     let listener = UnixListener::bind(&sock)?;
     fs::set_permissions(&sock, fs::Permissions::from_mode(0o600))?;
     eprintln!("hydrad {} listening on {}", hydra_core::VERSION, sock.display());
 
     let engine = Arc::new(Mutex::new(Engine::new()));
 
     // Restore saved routes (KeepAlive LaunchAgent ⇒ this runs after every restart/reboot).
     // Resolve each saved app's bundle id to a currently-running PID; skip apps not running.
     let config = Config::load();
     if !config.routes.is_empty() {
         let restored = engine.lock().unwrap().restore(&config, resolve_bundle_to_pid);
         eprintln!("restored {restored}/{} saved route(s)", config.routes.len());
     }
 
     // Supervisor: the self-healing heart. Every few seconds it reconciles live routes against
     // the saved config — rebuilding routes whose aggregate died (coreaudiod restart, driver
     // reinstall) and resurrecting saved routes whose app came back. This is what makes Hydra
     // "set it and forget it": you create a route once, the daemon keeps it alive forever.
     {
         let engine = Arc::clone(&engine);
         std::thread::spawn(move || supervise(engine));
     }
 
     for conn in listener.incoming() {
         match conn {
             Ok(stream) => {
                 let engine = Arc::clone(&engine);
                 std::thread::spawn(move || {
                     if let Err(e) = server::handle(stream, engine) {
                         eprintln!("connection error: {e}");
                     }
                 });
             }
             Err(e) => eprintln!("accept error: {e}"),
         }
     }
 
     Ok(())
 }
 
 /// Resolve an app bundle id to a currently-running audio-process PID (None if not playing).
 pub(crate) fn resolve_bundle_to_pid(bundle_id: &str) -> Option<i32> {
     process::list_audio_processes()
         .into_iter()
         .find(|a| a.bundle_id.as_deref() == Some(bundle_id))
         .map(|a| a.pid)
 }
 
 /// The supervisor loop: reconcile live routes against the saved config on a timer, forever.
 /// Holds the engine lock only for the brief reconcile call, never across the sleep.
 fn supervise(engine: Arc<Mutex<Engine>>) {
     use std::time::Duration;
     // 4s: long enough that a healthy route logs several IOProc callbacks between ticks (so a
     // frozen counter unambiguously means "aggregate dead"), short enough to feel instant.
     const TICK: Duration = Duration::from_secs(4);
     loop {
         std::thread::sleep(TICK);
         let saved = Config::load().routes;
         let actions = {
             let mut eng = engine.lock().unwrap();
             eng.reconcile(&saved, resolve_bundle_to_pid)
         };
         if !actions.is_empty() {
             for a in &actions {
                 eprintln!("[supervisor] {a}");
             }
             // Persist the healed state + nudge the SketchyBar widget.
             let cfg = engine.lock().unwrap().to_config();
             cfg.save();
             let _ = std::process::Command::new("sketchybar")
                 .args(["--trigger", "hydra_route_change"])
                 .stdout(std::process::Stdio::null())
                 .stderr(std::process::Stdio::null())
                 .spawn();
         }
     }
 }
 
 /// Probe an existing socket: returns the protocol version if a live daemon answers `Ping`,
 /// or `None` if nothing's there / it's a stale socket. A short timeout keeps a wedged peer
 /// from hanging startup.
 fn ping_existing(sock: &std::path::Path) -> Option<u32> {
     use std::time::Duration;
     let stream = UnixStream::connect(sock).ok()?;
     stream.set_read_timeout(Some(Duration::from_millis(500))).ok()?;
     stream.set_write_timeout(Some(Duration::from_millis(500))).ok()?;
     let mut writer = BufWriter::new(stream.try_clone().ok()?);
     let mut reader = BufReader::new(stream);
     write_msg(&mut writer, &Command::Ping).ok()?;
     match read_msg::<_, Response>(&mut reader).ok()? {
         Some(Response::Pong { version }) => Some(version),
         _ => None,
     }
 }