//! # signal-client
//!
//! The client side of the signal bus: where the daemon's socket lives, and how
//! to read frames off it.
//!
//! The socket path was resolved by three copies of the same function — the
//! daemon and each renderer — kept honest by a table in the README and a
//! duplicated test in every binary. Copies drift; this crate is the one
//! definition.
//!
//! It holds no privacy surface of its own. Everything it yields came off the
//! wire and through `signal_schema`'s decode, so the `f64`-only payload rule is
//! already enforced upstream.

use std::io::BufReader;
use std::os::unix::net::UnixStream;
use std::path::{Path, PathBuf};

use signal_schema::wire;

/// The daemon's socket: `$XDG_RUNTIME_DIR/signald.sock`, else
/// `$HOME/.local/state/signald/sock`.
///
/// The daemon derives its history db and terminal spool from this path's
/// directory, so all three move together when `--socket` is given.
pub fn default_socket_path() -> PathBuf {
    socket_path_from(
        std::env::var("XDG_RUNTIME_DIR").ok().as_deref(),
        std::env::var("HOME").ok().as_deref(),
    )
}

/// [`default_socket_path`] with the environment passed in, so it can be tested
/// without mutating process state.
pub fn socket_path_from(xdg_runtime_dir: Option<&str>, home: Option<&str>) -> PathBuf {
    if let Some(dir) = xdg_runtime_dir {
        return PathBuf::from(dir).join("signald.sock");
    }
    PathBuf::from(home.unwrap_or(".")).join(".local/state/signald/sock")
}

/// Connect to the daemon and iterate the frames it publishes.
///
/// The iterator ends at a clean EOF and on an unframeable stream. Frames this
/// build cannot decode are skipped rather than ending the stream, so a renderer
/// built before a metric was appended keeps working against a newer daemon.
pub struct Frames {
    reader: BufReader<UnixStream>,
    done: bool,
}

impl Frames {
    /// Connect to the daemon listening at `socket`.
    pub fn connect(socket: &Path) -> std::io::Result<Frames> {
        Ok(Frames {
            reader: BufReader::new(UnixStream::connect(socket)?),
            done: false,
        })
    }
}

impl Frames {
    /// Read the last-value cache the daemon sends on connect, and return.
    ///
    /// There is no end-of-snapshot marker on the wire, so this reads one frame
    /// blocking and then drains whatever else is already buffered, stopping as
    /// soon as the socket would block. The cache arrives in a single burst, so
    /// that returns immediately rather than waiting out a timeout — which is
    /// what makes a one-shot reader cheap enough to run on every shell prompt.
    ///
    /// Consumes the `Frames`: draining can abandon a partially-arrived frame,
    /// which leaves the stream out of sync. That is harmless for a reader that
    /// exits, and unusable for one that keeps going.
    pub fn snapshot(mut self) -> std::io::Result<Vec<signal_schema::Signal>> {
        let mut out = Vec::new();
        // The first frame blocks: the daemon always has something cached, and
        // an empty read here means the socket died rather than went quiet.
        match self.next() {
            Some(Ok(s)) => out.push(s),
            Some(Err(e)) => return Err(e),
            None => return Ok(out),
        }
        self.reader.get_ref().set_nonblocking(true)?;
        for frame in &mut self {
            match frame {
                Ok(s) => out.push(s),
                // Nothing more is buffered: the cache has been drained.
                Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => break,
                Err(e) => return Err(e),
            }
        }
        Ok(out)
    }
}

impl Iterator for Frames {
    /// `Err` is an unframeable stream. A skipped frame is not surfaced: the
    /// iterator swallows it and reads on, which is the whole point of
    /// [`wire::Frame::Skipped`].
    type Item = std::io::Result<signal_schema::Signal>;

    fn next(&mut self) -> Option<Self::Item> {
        while !self.done {
            match wire::read_frame(&mut self.reader) {
                Ok(wire::Frame::Signal(s)) => return Some(Ok(s)),
                Ok(wire::Frame::Skipped) => continue,
                Ok(wire::Frame::Eof) => {
                    self.done = true;
                    return None;
                }
                Err(e) => {
                    self.done = true;
                    return Some(Err(e));
                }
            }
        }
        None
    }
}

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

    /// The one place the socket default is pinned. It used to be asserted
    /// separately in each binary against a README table.
    #[test]
    fn socket_default_follows_xdg_then_home() {
        assert_eq!(
            socket_path_from(Some("/run/user/501"), Some("/Users/x")),
            PathBuf::from("/run/user/501/signald.sock")
        );
        assert_eq!(
            socket_path_from(None, Some("/Users/x")),
            PathBuf::from("/Users/x/.local/state/signald/sock")
        );
    }

    /// The db and the spool are derived from the socket's directory, so all
    /// three move together when --socket is given.
    #[test]
    fn db_and_spool_sit_beside_the_socket() {
        let socket = socket_path_from(None, Some("/Users/x"));
        let base = socket.parent().unwrap();
        assert_eq!(
            base.join("signald.sqlite"),
            PathBuf::from("/Users/x/.local/state/signald/signald.sqlite")
        );
        assert_eq!(
            base.join("terminal.spool"),
            PathBuf::from("/Users/x/.local/state/signald/terminal.spool"),
            "must equal SIGNALD_SPOOL's default in shell-hooks/signald-hooks.zsh"
        );
    }
}