ipx/src/ipc.rs

//! Unix-socket control and event stream. Replaces printMSG's `;;1;;1;;100.00;;42.31`.

use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use std::path::{Path, PathBuf};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{UnixListener, UnixStream};
use tokio::sync::{broadcast, mpsc};

/// One JSON object per line, `ev` naming the variant.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "ev", rename_all = "snake_case")]
pub enum Event {
    FeedStart { feed: String },
    FeedSkip { feed: String, reason: String },
    FeedDone { feed: String, new: usize, downloaded: usize, failed: usize, torrents: usize },
    FeedError { feed: String, msg: String },
    Progress {
        feed: String,
        /// Which enclosure this is about. Without it a UI cannot tell one download's
        /// progress from another's and ends up animating every pending row.
        enclosure: i64,
        url: String,
        file: String,
        done: u64,
        #[serde(skip_serializing_if = "Option::is_none")]
        total: Option<u64>,
    },
    DownloadDone { feed: String, enclosure: i64, url: String, path: String, bytes: u64 },
    DownloadError { feed: String, enclosure: i64, url: String, msg: String },
    TorrentDeferred { feed: String, url: String },
    Reaped { path: String, bytes: u64 },
    /// Terminal: a client that asked for work stops reading here.
    ScanDone { feeds: usize },
    ReapDone { files: usize, bytes: u64 },
    Status { feeds: usize, pending: i64, downloaded: i64 },
    Error { msg: String },
}

impl Event {
    pub fn is_terminal(&self) -> bool {
        matches!(self, Event::ScanDone { .. } | Event::ReapDone { .. } | Event::Status { .. })
    }

    /// The human rendering, for a terminal rather than a UI.
    pub fn human(&self) -> Option<String> {
        Some(match self {
            Event::FeedSkip { feed, reason } => format!("{feed}: {reason}"),
            Event::FeedDone { feed, new, downloaded, failed, torrents } => format!(
                "{feed}: {new} new entries, {downloaded} downloaded, {failed} failed, {torrents} torrents deferred"
            ),
            Event::FeedError { feed, msg } => format!("{feed}: error: {msg}"),
            Event::Progress { file, done, total, .. } => match total {
                Some(t) if *t > 0 => format!(
                    "    {file}: {:.1}% ({:.1}/{:.1} MB)",
                    *done as f64 / *t as f64 * 100.0,
                    *done as f64 / 1_048_576.0,
                    *t as f64 / 1_048_576.0
                ),
                _ => format!("    {file}: {:.1} MB", *done as f64 / 1_048_576.0),
            },
            Event::DownloadDone { path, .. } => format!("    saved {path}"),
            Event::DownloadError { url, msg, .. } => format!("    failed {url}: {msg}"),
            Event::Reaped { path, bytes } => {
                format!("deleted {path} ({:.1} MB)", *bytes as f64 / 1_048_576.0)
            }
            Event::ReapDone { files, bytes } => format!(
                "deleted {files} old file(s), {:.1} MB",
                *bytes as f64 / 1_048_576.0
            ),
            Event::Status { feeds, pending, downloaded } => {
                format!("{feeds} feeds, {pending} pending, {downloaded} downloaded")
            }
            Event::Error { msg } => format!("error: {msg}"),
            // Noise in a terminal; a UI still gets them on the socket.
            Event::FeedStart { feed } => format!("{feed}: checking"),
            Event::TorrentDeferred { feed, .. } => format!("{feed}: torrent deferred"),
            Event::ScanDone { feeds } => format!("scan complete, {feeds} feed(s)"),
        })
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "cmd", rename_all = "snake_case")]
pub enum Command {
    Fetch {
        #[serde(default)]
        feed: Option<String>,
        #[serde(default)]
        force: bool,
    },
    Reap {
        #[serde(default)]
        dry_run: bool,
    },
    /// Fetch one specific enclosure now, ignoring max_new_per_check and the queue order.
    /// A scan cannot express "this one, now": it takes the lowest-id pending rows up to
    /// the per-scan cap, so an explicit request has to bypass both.
    Download {
        enclosure: i64,
    },
    Status,
}

/// Where events go: the socket, the terminal, or both.
#[derive(Clone)]
pub struct Emitter {
    tx: Option<broadcast::Sender<Event>>,
    print: bool,
}

impl Emitter {
    pub fn terminal() -> Self {
        Self { tx: None, print: true }
    }

    pub fn socket(tx: broadcast::Sender<Event>, print: bool) -> Self {
        Self { tx: Some(tx), print }
    }

    pub fn emit(&self, e: Event) {
        // Also log it. Scans and downloads travel as events, not tracing calls, so
        // without this the log view shows only startup and HTTP lines and none of the
        // work the daemon is actually doing. Progress goes to debug: it fires on every
        // whole percent and would otherwise crowd everything else out of the buffer.
        // Level by how much it matters. With 80-odd feeds in an OPML subscription, one
        // line per feed per tick for "not due yet" would push everything worth reading
        // out of the buffer within a few minutes.
        let routine = match &e {
            Event::Progress { .. } | Event::FeedSkip { .. } | Event::FeedStart { .. } => true,
            Event::FeedDone { new, downloaded, failed, torrents, .. } => {
                *new == 0 && *downloaded == 0 && *failed == 0 && *torrents == 0
            }
            _ => false,
        };
        let bad = matches!(
            &e,
            Event::FeedError { .. } | Event::DownloadError { .. } | Event::Error { .. }
        );
        if let Some(line) = e.human() {
            let line = line.trim();
            if bad {
                tracing::warn!(target: "ipx::scan", "{line}");
            } else if routine {
                tracing::debug!(target: "ipx::scan", "{line}");
            } else {
                tracing::info!(target: "ipx::scan", "{line}");
            }
        }

        if let Some(tx) = &self.tx {
            // The outbound half of the protocol, as it goes on the wire. Progress is the
            // high-volume one, so it sits at debug.
            if let Ok(json) = serde_json::to_string(&e) {
                if matches!(e, Event::Progress { .. }) {
                    tracing::debug!(target: "ipx::io", "<- {json}");
                } else {
                    tracing::info!(target: "ipx::io", "<- {json}");
                }
            }
            // An error here only means nobody is listening yet.
            let _ = tx.send(e.clone());
        }
        if self.print && let Some(line) = e.human() {
            println!("{line}");
        }
    }
}

/// True when something is already listening -- i.e. a daemon owns this socket.
pub async fn daemon_is_live(path: &Path) -> bool {
    UnixStream::connect(path).await.is_ok()
}

/// Answers `status` for the socket, without the worker. The worker runs one job at a time, and a
/// healthcheck left waiting behind a scan or a long download timed out and called a busy daemon
/// dead. The answer goes to the client that asked and no one else: broadcast, it ended any
/// `ipx fetch` that was watching a scan, since `status` is a terminal event.
/// A future, since reading the counts is a database query.
pub type StatusFn =
    std::sync::Arc<dyn Fn() -> std::pin::Pin<Box<dyn std::future::Future<Output = Event> + Send>> + Send + Sync>;

/// Accepts connections, feeding commands to `cmds` and events from `events` back out.
pub async fn serve(
    path: PathBuf,
    events: broadcast::Sender<Event>,
    cmds: mpsc::Sender<Command>,
    status: StatusFn,
) -> Result<()> {
    // A socket file left by a crashed daemon would block the bind; a live one was already
    // rejected by the caller's daemon_is_live() check.
    if path.exists() {
        std::fs::remove_file(&path)
            .with_context(|| format!("removing stale socket {}", path.display()))?;
    }
    if let Some(dir) = path.parent() {
        std::fs::create_dir_all(dir)?;
    }
    let listener = UnixListener::bind(&path)
        .with_context(|| format!("binding {}", path.display()))?;
    tracing::info!(socket = %path.display(), "listening");

    loop {
        let (stream, _) = listener.accept().await?;
        let rx = events.subscribe();
        let cmds = cmds.clone();
        let status = status.clone();
        tokio::spawn(async move {
            if let Err(e) = handle(stream, rx, cmds, status).await {
                tracing::debug!(error = %e, "client gone");
            }
        });
    }
}

async fn handle(
    stream: UnixStream,
    mut rx: broadcast::Receiver<Event>,
    cmds: mpsc::Sender<Command>,
    status: StatusFn,
) -> Result<()> {
    let (read, mut write) = stream.into_split();

    // Events out: everything broadcast, and the answers meant for this client alone.
    let (reply, mut replies) = mpsc::channel::<Event>(4);
    let writer = tokio::spawn(async move {
        loop {
            let ev = tokio::select! {
                Some(ev) = replies.recv() => ev,
                got = rx.recv() => match got {
                    Ok(ev) => ev,
                    Err(_) => break,
                },
            };
            let mut line = serde_json::to_string(&ev).unwrap_or_default();
            line.push('\n');
            if write.write_all(line.as_bytes()).await.is_err() {
                break;
            }
        }
    });

    // Commands in.
    let mut lines = BufReader::new(read).lines();
    while let Some(line) = lines.next_line().await? {
        let line = line.trim();
        if line.is_empty() {
            continue;
        }
        match serde_json::from_str::<Command>(line) {
            // Answered here, not queued behind whatever the worker is on: see StatusFn.
            Ok(Command::Status) => {
                tracing::info!(target: "ipx::io", "-> {line}");
                let ev = status().await;
                if let Ok(json) = serde_json::to_string(&ev) {
                    tracing::info!(target: "ipx::io", "<- {json}");
                }
                let _ = reply.send(ev).await;
            }
            Ok(cmd) => {
                if cmds.send(cmd).await.is_err() {
                    break; // Worker is gone; so are we.
                }
            }
            Err(e) => tracing::warn!(error = %e, line, "bad command"),
        }
    }
    writer.abort();
    Ok(())
}

/// Sends one command to a running daemon and prints the events it produces.
pub async fn proxy(path: &Path, cmd: &Command) -> Result<()> {
    let stream = UnixStream::connect(path).await?;
    let (read, mut write) = stream.into_split();
    let mut line = serde_json::to_string(cmd)?;
    line.push('\n');
    write.write_all(line.as_bytes()).await?;

    // ponytail: the event stream is a broadcast, so a busy daemon's other work shows up
    // here too. Fine for a CLI; a UI that cares would want per-request ids.
    let mut lines = BufReader::new(read).lines();
    while let Some(line) = lines.next_line().await? {
        let Ok(ev) = serde_json::from_str::<Event>(&line) else {
            continue;
        };
        if let Some(text) = ev.human() {
            println!("{text}");
        }
        if ev.is_terminal() {
            break;
        }
    }
    Ok(())
}

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

    #[test]
    fn commands_parse_from_the_wire_form() {
        let got: Command = serde_json::from_str(r#"{"cmd":"fetch"}"#).unwrap();
        assert!(matches!(got, Command::Fetch { feed: None, force: false }));

        let got: Command = serde_json::from_str(r#"{"cmd":"fetch","feed":"atp","force":true}"#).unwrap();
        assert!(matches!(got, Command::Fetch { feed: Some(f), force: true } if f == "atp"));

        let got: Command = serde_json::from_str(r#"{"cmd":"reap","dry_run":true}"#).unwrap();
        assert!(matches!(got, Command::Reap { dry_run: true }));

        // "Download this one now" is its own command precisely because a scan cannot
        // express it: a scan takes the lowest-id pending rows up to max_new_per_check.
        let got: Command = serde_json::from_str(r#"{"cmd":"download","enclosure":11}"#).unwrap();
        assert!(matches!(got, Command::Download { enclosure: 11 }));

        assert!(serde_json::from_str::<Command>(r#"{"cmd":"nope"}"#).is_err());
    }

    #[test]
    fn events_serialise_to_the_documented_shape() {
        let ev = Event::Progress {
            feed: "atp".into(),
            enclosure: 42,
            url: "https://x/ep.mp3".into(),
            file: "ep.mp3".into(),
            done: 10_485_760,
            total: Some(52_428_800),
        };
        let json = serde_json::to_string(&ev).unwrap();
        assert!(json.starts_with(r#"{"ev":"progress""#), "got {json}");
        assert!(json.contains(r#""done":10485760"#));
        assert!(json.contains(r#""enclosure":42"#), "a UI needs this to target one row");

        // total is omitted rather than null when the server sent no length.
        let ev = Event::Progress {
            feed: "a".into(),
            enclosure: 1,
            url: "u".into(),
            file: "f".into(),
            done: 1,
            total: None,
        };
        assert!(!serde_json::to_string(&ev).unwrap().contains("total"));
    }

    #[test]
    fn only_completion_events_end_a_client_session() {
        assert!(Event::ScanDone { feeds: 1 }.is_terminal());
        assert!(Event::ReapDone { files: 0, bytes: 0 }.is_terminal());
        assert!(!Event::FeedDone {
            feed: "a".into(),
            new: 0,
            downloaded: 0,
            failed: 0,
            torrents: 0
        }
        .is_terminal());
    }

    #[tokio::test]
    async fn status_is_answered_while_the_worker_is_busy() {
        // The queue is full and nobody drains it, as when the worker is deep in a long download:
        // anything sent to it would wait for ever.
        let (cmds, _worker) = mpsc::channel::<Command>(1);
        cmds.send(Command::Reap { dry_run: true }).await.unwrap();
        let (events, _) = broadcast::channel::<Event>(8);
        // Another client, watching a scan: it must not be handed someone else's answer, which
        // would end its session.
        let mut watcher = events.subscribe();
        let status: StatusFn =
            std::sync::Arc::new(|| Box::pin(async { Event::Status { feeds: 1, pending: 2, downloaded: 3 } }));
        let (client, server) = UnixStream::pair().unwrap();
        tokio::spawn(handle(server, events.subscribe(), cmds, status));

        let (read, mut write) = client.into_split();
        write.write_all(b"{\"cmd\":\"status\"}\n").await.unwrap();
        let line = tokio::time::timeout(std::time::Duration::from_secs(2), BufReader::new(read).lines().next_line())
            .await
            .expect("status waited behind the worker")
            .unwrap()
            .unwrap();
        assert!(line.contains(r#""ev":"status""#), "{line}");
        assert!(watcher.try_recv().is_err(), "the answer went to every client, not just the one asking");
    }
}