unifier/daemon/
notify.rs

1//! Outbound event socket: wake listeners (e.g. Jan cron) when mailboxes change.
2
3use std::io::{BufRead, BufReader, ErrorKind, Write};
4use std::os::unix::net::UnixStream;
5use std::sync::Mutex;
6use std::time::Duration;
7
8use serde::{Deserialize, Serialize};
9use uuid::Uuid;
10
11use crate::daemon::paths::events_socket_path;
12use crate::error::{Error, Result};
13use crate::home::UnifierHome;
14
15/// Line-delimited JSON notice written to `.daemon/events.sock`.
16///
17/// Downstream schedulers need at least `to` and `id` to locate the mailbox
18/// file at `mailbox/<to>/<id>.txt`. Tick notices announce ACID turn phases
19/// driven via `.daemon/tick.sock`.
20///
21/// **Cross-project wire contract.** Independent programs (e.g. the `jan` cron
22/// daemon) decode this enum without linking unifier. Variants and fields are
23/// additive only; readers must ignore unknown kinds. Notices are best-effort —
24/// [`EventHub::broadcast`] drops slow subscribers rather than buffering — so a
25/// notice must always refer to state already persisted on the board.
26#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
27#[serde(tag = "kind", rename_all = "snake_case")]
28pub enum Notice {
29    Mailbox {
30        id: Uuid,
31        from: String,
32        to: String,
33    },
34    Event {
35        id: Uuid,
36        #[serde(default, skip_serializing_if = "Option::is_none")]
37        name: Option<String>,
38    },
39    /// Tick lifecycle phase (`start`, `end`, or a caller-defined phase name).
40    Tick {
41        tick: u64,
42        phase: String,
43        #[serde(default, skip_serializing_if = "Option::is_none")]
44        label: Option<String>,
45    },
46}
47
48impl Notice {
49    pub fn mailbox(id: Uuid, from: impl Into<String>, to: impl Into<String>) -> Self {
50        Self::Mailbox {
51            id,
52            from: from.into(),
53            to: to.into(),
54        }
55    }
56
57    pub fn event(id: Uuid, name: Option<String>) -> Self {
58        Self::Event { id, name }
59    }
60
61    pub fn tick(tick: u64, phase: impl Into<String>, label: Option<String>) -> Self {
62        Self::Tick {
63            tick,
64            phase: phase.into(),
65            label,
66        }
67    }
68
69    pub fn from_json(line: &str) -> Result<Self> {
70        Ok(serde_json::from_str(line.trim())?)
71    }
72}
73
74/// Fan-out hub for connected event listeners.
75#[derive(Default)]
76pub struct EventHub {
77    subscribers: Mutex<Vec<UnixStream>>,
78}
79
80impl EventHub {
81    pub fn add(&self, stream: UnixStream) -> Result<()> {
82        stream.set_nonblocking(false)?;
83        stream.set_write_timeout(Some(Duration::from_millis(250)))?;
84        self.subscribers.lock().map_err(lock_err)?.push(stream);
85        Ok(())
86    }
87
88    pub fn broadcast(&self, notice: &Notice) {
89        let Ok(json) = serde_json::to_string(notice) else {
90            return;
91        };
92        let line = format!("{json}\n");
93        let Ok(mut subs) = self.subscribers.lock() else {
94            return;
95        };
96        subs.retain_mut(|stream| {
97            stream.write_all(line.as_bytes()).is_ok() && stream.flush().is_ok()
98        });
99    }
100
101    /// Number of currently registered event subscribers (not pruned).
102    pub fn subscriber_count(&self) -> usize {
103        self.subscribers.lock().map(|s| s.len()).unwrap_or(0)
104    }
105}
106
107/// Connect to the daemon event socket for wakeup notices.
108pub fn subscribe(home: &UnifierHome) -> Result<UnixStream> {
109    let path = events_socket_path(home);
110    let stream = UnixStream::connect(&path).map_err(|e| {
111        Error::msg(format!(
112            "event socket not reachable at {}: {e}",
113            path.display()
114        ))
115    })?;
116    stream.set_nonblocking(false)?;
117    stream.set_read_timeout(None)?;
118    Ok(stream)
119}
120
121/// Print notices from the event socket until the daemon disconnects.
122pub fn watch(home: &UnifierHome) -> Result<()> {
123    let stream = subscribe(home)?;
124    let mut reader = BufReader::new(stream);
125    loop {
126        let mut line = String::new();
127        match reader.read_line(&mut line) {
128            Ok(0) => break,
129            Ok(_) => print!("{line}"),
130            Err(e)
131                if matches!(
132                    e.kind(),
133                    ErrorKind::Interrupted | ErrorKind::WouldBlock | ErrorKind::TimedOut
134                ) =>
135            {
136                continue
137            }
138            Err(e) => return Err(e.into()),
139        }
140    }
141    Ok(())
142}
143
144pub fn event_name(payload: &str) -> Option<String> {
145    let value: serde_json::Value = serde_json::from_str(payload).ok()?;
146    value.get("name")?.as_str().map(str::to_string)
147}
148
149fn lock_err<E: std::fmt::Display>(e: E) -> Error {
150    Error::msg(format!("event hub lock poisoned: {e}"))
151}
152
153#[cfg(test)]
154mod tests {
155    use super::*;
156    use std::io::BufRead;
157
158    #[test]
159    fn broadcast_reaches_paired_subscriber() {
160        let hub = EventHub::default();
161        let (tx, rx) = UnixStream::pair().unwrap();
162        hub.add(tx).unwrap();
163
164        let id = Uuid::new_v4();
165        hub.broadcast(&Notice::mailbox(id, "alice", "bob"));
166
167        let mut reader = BufReader::new(rx);
168        let mut line = String::new();
169        reader.read_line(&mut line).unwrap();
170        let notice = Notice::from_json(&line).unwrap();
171        assert_eq!(
172            notice,
173            Notice::Mailbox {
174                id,
175                from: "alice".into(),
176                to: "bob".into(),
177            }
178        );
179    }
180}
unifier/daemon/notify.rs

unifier/daemon/
notify.rs
