unifier-cli 0.2.0

//! Outbound event socket: wake listeners (e.g. Jan cron) when mailboxes change.

use std::io::{BufRead, BufReader, ErrorKind, Write};
use std::os::unix::net::UnixStream;
use std::sync::Mutex;
use std::time::Duration;

use serde::{Deserialize, Serialize};
use uuid::Uuid;

use crate::daemon::paths::events_socket_path;
use crate::error::{Error, Result};
use crate::home::UnifierHome;

/// Line-delimited JSON notice written to `.daemon/events.sock`.
///
/// Downstream schedulers need at least `to` and `id` to locate the mailbox
/// file at `mailbox/<to>/<id>.txt`.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum Notice {
    Mailbox {
        id: Uuid,
        from: String,
        to: String,
    },
    Event {
        id: Uuid,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        name: Option<String>,
    },
}

impl Notice {
    pub fn mailbox(id: Uuid, from: impl Into<String>, to: impl Into<String>) -> Self {
        Self::Mailbox {
            id,
            from: from.into(),
            to: to.into(),
        }
    }

    pub fn event(id: Uuid, name: Option<String>) -> Self {
        Self::Event { id, name }
    }

    pub fn from_json(line: &str) -> Result<Self> {
        Ok(serde_json::from_str(line.trim())?)
    }
}

/// Fan-out hub for connected event listeners.
#[derive(Default)]
pub struct EventHub {
    subscribers: Mutex<Vec<UnixStream>>,
}

impl EventHub {
    pub fn add(&self, stream: UnixStream) -> Result<()> {
        stream.set_nonblocking(false)?;
        stream.set_write_timeout(Some(Duration::from_millis(250)))?;
        self.subscribers
            .lock()
            .map_err(lock_err)?
            .push(stream);
        Ok(())
    }

    pub fn broadcast(&self, notice: &Notice) {
        let Ok(json) = serde_json::to_string(notice) else {
            return;
        };
        let line = format!("{json}\n");
        let Ok(mut subs) = self.subscribers.lock() else {
            return;
        };
        subs.retain_mut(|stream| stream.write_all(line.as_bytes()).is_ok() && stream.flush().is_ok());
    }
}

/// Connect to the daemon event socket for wakeup notices.
pub fn subscribe(home: &UnifierHome) -> Result<UnixStream> {
    let path = events_socket_path(home);
    let stream = UnixStream::connect(&path).map_err(|e| {
        Error::msg(format!(
            "event socket not reachable at {}: {e}",
            path.display()
        ))
    })?;
    stream.set_nonblocking(false)?;
    stream.set_read_timeout(None)?;
    Ok(stream)
}

/// Print notices from the event socket until the daemon disconnects.
pub fn watch(home: &UnifierHome) -> Result<()> {
    let stream = subscribe(home)?;
    let mut reader = BufReader::new(stream);
    loop {
        let mut line = String::new();
        match reader.read_line(&mut line) {
            Ok(0) => break,
            Ok(_) => print!("{line}"),
            Err(e) if matches!(
                e.kind(),
                ErrorKind::Interrupted | ErrorKind::WouldBlock | ErrorKind::TimedOut
            ) =>
            {
                continue
            }
            Err(e) => return Err(e.into()),
        }
    }
    Ok(())
}

pub fn event_name(payload: &str) -> Option<String> {
    let value: serde_json::Value = serde_json::from_str(payload).ok()?;
    value.get("name")?.as_str().map(str::to_string)
}

fn lock_err<E: std::fmt::Display>(e: E) -> Error {
    Error::msg(format!("event hub lock poisoned: {e}"))
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::BufRead;

    #[test]
    fn broadcast_reaches_paired_subscriber() {
        let hub = EventHub::default();
        let (tx, rx) = UnixStream::pair().unwrap();
        hub.add(tx).unwrap();

        let id = Uuid::new_v4();
        hub.broadcast(&Notice::mailbox(id, "alice", "bob"));

        let mut reader = BufReader::new(rx);
        let mut line = String::new();
        reader.read_line(&mut line).unwrap();
        let notice = Notice::from_json(&line).unwrap();
        assert_eq!(
            notice,
            Notice::Mailbox {
                id,
                from: "alice".into(),
                to: "bob".into(),
            }
        );
    }
}