tawasuyu/arje

1//! ente-notify-compat: NOTIFY_SOCKET listener para apps `Type=notify`.
2//!
3//! systemd convention: el servicio escribe `KEY=value\n` lines a un socket
4//! datagram cuya path está en `$NOTIFY_SOCKET`. Keys típicos:
5//!   - READY=1            (servicio listo para recibir requests)
6//!   - STATUS=text        (descripción del estado)
7//!   - WATCHDOG=1          (heartbeat)
8//!   - STOPPING=1          (cierre ordenado)
9//!   - MAINPID=<pid>       (cambio de PID principal)
10//!
11//! Path canonical: /run/systemd/notify. Bindeable sólo con CAP_NET_BIND_SERVICE
12//! o si /run es writable.
13//!
14//! Para que las apps lo usen, ente-soma debe inyectar `NOTIFY_SOCKET=<path>`
15//! en el envp de cada Ente encarnado. Eso ya lo hace via build_env() —
16//! aquí sólo necesitamos que el path sea coherente.
17
18use ente_bus::{BusClient, BusRequest, BusResponse};
19use ente_card::Capability;
20use std::os::fd::{AsRawFd, OwnedFd};
21use std::path::Path;
22use tokio::io::unix::AsyncFd;
23use tokio::signal::unix::{signal, SignalKind};
24use tracing::{debug, info, warn};
25use tracing_subscriber::EnvFilter;
26
27const NOTIFY_SOCKET_PATH: &str = "/run/systemd/notify";
28
29#[tokio::main(flavor = "current_thread")]
30async fn main() -> anyhow::Result<()> {
31    init_tracing();
32    info!(path = NOTIFY_SOCKET_PATH, "ente-notify-compat: arrancando");
33    announce_to_fractal().await;
34
35    let stream = match bind_dgram(NOTIFY_SOCKET_PATH) {
36        Some(s) => s,
37        None => {
38            warn!("no se pudo bind — modo idle (apps Type=notify caerán a no-op)");
39            return wait_for_term().await;
40        }
41    };
42    info!("NOTIFY_SOCKET listening");
43    spawn_listener(stream);
44    wait_for_term().await
45}
46
47fn bind_dgram(path: &str) -> Option<AsyncFd<OwnedFdWrap>> {
48    use nix::sys::socket::{bind, socket, AddressFamily, SockFlag, SockType, UnixAddr};
49    let _ = std::fs::remove_file(path);
50    if let Some(parent) = Path::new(path).parent() {
51        let _ = std::fs::create_dir_all(parent);
52    }
53    let fd = socket(
54        AddressFamily::Unix,
55        SockType::Datagram,
56        SockFlag::SOCK_NONBLOCK | SockFlag::SOCK_CLOEXEC,
57        None,
58    ).ok()?;
59    let addr = UnixAddr::new(path).ok()?;
60    if let Err(e) = bind(fd.as_raw_fd(), &addr) {
61        warn!(?e, %path, "bind");
62        return None;
63    }
64    // Permisos abiertos: cualquier proceso debería poder escribir notificaciones.
65    use std::os::unix::fs::PermissionsExt;
66    let _ = std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o666));
67    AsyncFd::new(OwnedFdWrap(fd)).ok()
68}
69
70struct OwnedFdWrap(OwnedFd);
71impl AsRawFd for OwnedFdWrap {
72    fn as_raw_fd(&self) -> std::os::fd::RawFd { self.0.as_raw_fd() }
73}
74
75fn spawn_listener(async_fd: AsyncFd<OwnedFdWrap>) {
76    tokio::spawn(async move {
77        let mut buf = vec![0u8; 16 * 1024];
78        loop {
79            let mut guard = match async_fd.readable().await {
80                Ok(g) => g,
81                Err(e) => { warn!(?e, "readable"); return; }
82            };
83            let raw_fd = guard.get_inner().as_raw_fd();
84            loop {
85                let n = unsafe { libc::recv(raw_fd, buf.as_mut_ptr() as *mut _, buf.len(), 0) };
86                if n <= 0 { break; }
87                handle_notification(&buf[..n as usize]);
88            }
89            guard.clear_ready();
90        }
91    });
92}
93
94fn handle_notification(buf: &[u8]) {
95    let s = match std::str::from_utf8(buf) {
96        Ok(s) => s,
97        Err(_) => { debug!(len = buf.len(), "notify binario, skip"); return; }
98    };
99    let mut ready = false;
100    let mut status = None;
101    let mut mainpid = None;
102    let mut watchdog = false;
103    let mut stopping = false;
104    let mut other_keys = Vec::new();
105    for line in s.lines() {
106        if let Some((k, v)) = line.split_once('=') {
107            match k {
108                "READY" if v == "1" => ready = true,
109                "STATUS" => status = Some(v.to_string()),
110                "MAINPID" => mainpid = v.parse::<u32>().ok(),
111                "WATCHDOG" if v == "1" => watchdog = true,
112                "STOPPING" if v == "1" => stopping = true,
113                _ => other_keys.push(format!("{k}={v}")),
114            }
115        }
116    }
117    if ready {
118        info!(?status, ?mainpid, "sd_notify READY");
119    } else if stopping {
120        info!(?status, "sd_notify STOPPING");
121    } else if watchdog {
122        debug!("sd_notify WATCHDOG");
123    } else if let Some(s) = status {
124        info!(%s, "sd_notify STATUS");
125    } else if !other_keys.is_empty() {
126        debug!(keys = ?other_keys, "sd_notify (other)");
127    }
128}
129
130async fn announce_to_fractal() {
131    if let Ok(mut client) = BusClient::from_env().await {
132        let req = BusRequest::Announce {
133            capabilities: vec![Capability::Endpoint {
134                interface: ente_card::InterfaceId([0xa7; 16]),
135                version: 1,
136            }],
137        };
138        match client.call(req).await {
139            Ok(BusResponse::Ok) => info!("Announce → bus interno OK"),
140            Ok(other) => warn!(?other, "Announce respuesta inesperada"),
141            Err(e) => warn!(?e, "Announce falló"),
142        }
143    }
144}
145
146async fn wait_for_term() -> anyhow::Result<()> {
147    let mut term = signal(SignalKind::terminate())?;
148    let mut int_ = signal(SignalKind::interrupt())?;
149    tokio::select! {
150        _ = term.recv() => info!("SIGTERM"),
151        _ = int_.recv() => info!("SIGINT"),
152    }
153    Ok(())
154}
155
156fn init_tracing() {
157    let filter = EnvFilter::try_from_default_env()
158        .unwrap_or_else(|_| EnvFilter::new("ente_notify_compat=info"));
159    tracing_subscriber::fmt().with_env_filter(filter).with_target(true).init();
160}