shep 0.16.0

//! `bleats` (alias `logs`): following a sheep's log stream. The only
//! streaming verb, and the only one whose output skips
//! [`crate::output`]'s envelope: a follow has no end, so there is nothing
//! to wrap.
//!
//! One `Request::ListFlock` resolves an id -> [`ProcessInfo`] cache before
//! subscribing, since the daemon's topic filter carries no sheep identity:
//! selector filtering happens client-side against that cache.
//!
//! `--no-follow` never subscribes: it tails each matched sheep's log files
//! and exits. `--follow` prints that same tail first, then subscribes.
//!
//! Under `--host` those files are the remote's, so [`source::LogSource`] asks
//! its shepherd for each tail and the rest of the loop is shared.

use std::collections::{HashMap, HashSet};
use std::io;
use std::path::Path;

use futures_util::FutureExt;
use serde::Serialize;

use shep_client::{Client, EventStream, Lagged};
use shep_core::protocol::{BusEvent, LogStream, ProcessInfo};
use shep_core::selector::ProcessSelector;

use crate::cli::{BleatsArgs, Format};
use crate::commands::rpc::typed_request_error;
use crate::commands::selector::parse_selector;
use crate::exit::ExitCode;
use crate::output::{self, Streams, write_outcome};
use crate::remote::interrupt_within_grace;

mod source;
use source::{FileTail, LogSource, stream_label};

pub(crate) use shep_core::logtail::read_tail;
// Only the tests size a file against the window.
#[cfg(test)]
pub(crate) use shep_core::logtail::TAIL_WINDOW_BYTES;

/// One line of `bleats` output under `--format json`. A stability surface of
/// its own, not wrapped in [`output::OutputEnvelope`]: a follow has no end.
#[derive(Debug, Serialize)]
struct BleatLine<'a> {
    /// [`output::SCHEMA_VERSION`] at the time this line was produced.
    schema_version: u32,
    /// The sheep's id.
    id: u32,
    /// The sheep's name if the initial listing resolved it, else the bare
    /// id rendered as a string.
    name: &'a str,
    /// Which of a sheep's two output streams this line came from.
    stream: &'static str,
    /// The instance slot this line's sheep occupies, when its app has more
    /// than one instance registered; `null` when the app has only one, or
    /// when the line's origin cannot be attributed to a single instance (a
    /// backlog line read from a file several instances share).
    instance: Option<u32>,
    /// The line itself, no trailing newline.
    line: &'a str,
}

/// Issues the one `Request::ListFlock` `bleats` sends, before it ever
/// subscribes, and turns the answer into an id -> [`ProcessInfo`] cache.
///
/// # Errors
/// Renders and returns the exit code for a request that failed to reach
/// the daemon, or a response this client does not recognise (`Response`
/// is `#[non_exhaustive]`).
async fn resolve_names(
    client: &Client,
    streams: &mut Streams<'_>,
) -> Result<HashMap<u32, ProcessInfo>, ExitCode> {
    match client.list_flock().await {
        Ok(procs) => Ok(procs.into_iter().map(|p| (p.id, p)).collect()),
        Err(err) => Err(typed_request_error(streams, &err)),
    }
}

/// Subscribes to every topic `bleats` needs: `log.*` for the lines
/// themselves, `daemon.*` so a `BusEvent::DaemonShutdown` is observed
/// rather than the connection simply vanishing unexplained.
///
/// # Errors
/// Renders and returns the exit code for a subscribe request that failed.
async fn subscribe(client: &Client, streams: &mut Streams<'_>) -> Result<EventStream, ExitCode> {
    let topics = vec!["log.*".to_string(), "daemon.*".to_string()];
    match client.subscribe(topics).await {
        Ok(stream) => Ok(stream),
        Err(err) => {
            let code = ExitCode::from(&err);
            Err(streams.fail(code, &err.to_string()))
        }
    }
}

/// Resolves `id` to a name from `cache`, or the bare id if `id` was not in
/// the one listing `resolve_names` took. An id that shows up on the bus
/// later than that snapshot is not a reason to block on a second listing.
fn resolved_name(cache: &HashMap<u32, ProcessInfo>, id: u32) -> String {
    cache
        .get(&id)
        .map_or_else(|| id.to_string(), |info| info.name.clone())
}

/// The slot a followed line's sheep occupies, or `None` when `id` was not
/// in the one listing `resolve_names` took, or its app has only one
/// instance.
///
/// The daemon emits [`BusEvent::LogOut`]/[`LogErr`](BusEvent::LogErr) per
/// sheep, so a follow labels a line even when several instances share one
/// log file, unlike the backlog path.
fn resolved_instance(
    cache: &HashMap<u32, ProcessInfo>,
    counts: &HashMap<String, usize>,
    id: u32,
) -> Option<u32> {
    let info = cache.get(&id)?;
    if counts.get(info.name.as_str()).copied().unwrap_or(0) > 1 {
        info.instance
    } else {
        None
    }
}

/// Whether `selector` (parsed client-side) admits `id`, matched against
/// `cache`'s snapshot of that sheep if it has one.
///
/// An id the initial listing never saw has no name or fold to match
/// against, so it is matched with an empty name and no fold: enough for
/// `all` and for `ProcessSelector::Id`, while a name, regex or fold
/// selector excludes it.
fn selector_allows(selector: &ProcessSelector, cache: &HashMap<u32, ProcessInfo>, id: u32) -> bool {
    match cache.get(&id) {
        Some(info) => selector.matches(&info.name, info.id, info.fold.as_deref(), info.instance),
        None => selector.matches("", id, None, None),
    }
}

/// Writes one line, the backlog path and the follow path both through
/// here, so a replayed line and a followed line render identically.
///
/// `stream` is `"out"` or `"err"`, the sheep stream the line came from,
/// not `out`'s own identity: every line this function is called with
/// lands on [`Streams::out`]. Table format sanitises the line and puts
/// the stream's default colour on the line text, never on the `name |`
/// label; JSON carries the raw line, sanitiser never applied.
fn write_line(
    streams: &mut Streams<'_>,
    id: u32,
    name: &str,
    instance: Option<u32>,
    stream: &'static str,
    line: &str,
) -> io::Result<()> {
    match streams.fmt {
        Format::Json => {
            let payload = BleatLine {
                schema_version: output::SCHEMA_VERSION,
                id,
                name,
                instance,
                stream,
                line,
            };
            serde_json::to_writer(&mut *streams.out, &payload)?;
            writeln!(streams.out)
        }
        Format::Table => {
            let clean = output::log_line::sanitize_log_line(line);
            let text = output::log_line::paint_default(&clean, stream, streams.style);
            match instance {
                Some(slot) => writeln!(streams.out, "{name}:{slot} | {text}"),
                None => writeln!(streams.out, "{name} | {text}"),
            }
        }
    }
}

/// How many rows of `cache` carry `name`, counted over the whole cache and
/// never a selector's matched subset, so a selector cannot change how a line
/// is labelled: `shep bleats web:0` still prints `web:0`, not `web`.
fn instance_count(cache: &HashMap<u32, ProcessInfo>, name: &str) -> usize {
    cache.values().filter(|info| info.name == name).count()
}

/// Every name `cache` holds, with how many rows carry it. Same counting
/// rule as [`instance_count`]: over the whole cache, never a selector's
/// matched subset.
///
/// Built once per follow, from the one listing [`resolve_names`] took.
/// That cache never changes after it, so a followed line's label costs one
/// lookup here rather than a scan of every process in the flock.
fn instance_counts(cache: &HashMap<u32, ProcessInfo>) -> HashMap<String, usize> {
    let mut counts: HashMap<String, usize> = HashMap::with_capacity(cache.len());
    for info in cache.values() {
        *counts.entry(info.name.clone()).or_insert(0) += 1;
    }
    counts
}

/// One of `bleats`' own notices, unless `--quiet` asked for silence.
///
/// Goes out through [`output::emit_notice`], not [`output::emit_error`]:
/// its code is not part of [`crate::exit::ExitCode`]'s taxonomy, and a
/// clean run can emit one on its way to exit 0. The `quiet` gate is this
/// verb's own, not [`Streams::aside`]'s; a sheep's own line and a real
/// error both still print under it.
fn write_notice(streams: &mut Streams<'_>, quiet: bool, code: &str, message: &str) {
    if quiet {
        return;
    }
    streams.aside(code, message);
}

/// Why a log file is not there, naming the path this process tried.
///
/// `stream` is `"out"` or `"err"`. Each process resolves a relative
/// `out_file`/`err_file` against its own directory, so the absolute form is
/// what shows that the shepherd is writing a different file.
pub(crate) fn missing_log_note(stream: &str, path: &Path) -> String {
    if !path.is_relative() {
        return format!("no {stream} log at {} yet", path.display());
    }
    let tried = std::path::absolute(path).unwrap_or_else(|_| path.to_path_buf());
    format!(
        "no {stream} log at {}; {stream}_file is relative, so the shepherd \
         resolves it against its own directory instead",
        tried.display()
    )
}

/// The `log_path_unknown` notice for `name`'s `stream`, said once however
/// many rows share it.
///
/// No path to key on, since the missing field is why this fires. The message
/// already names the pair that varies.
fn path_unknown_notice(
    streams: &mut Streams<'_>,
    quiet: bool,
    seen_notices: &mut HashSet<String>,
    name: &str,
    stream: LogStream,
) {
    let message = format!(
        "{name}: the daemon did not report a {} log path",
        stream_label(stream)
    );
    if seen_notices.insert(message.clone()) {
        write_notice(streams, quiet, "log_path_unknown", &message);
    }
}

/// Renders the selected files of every sheep the selector admits, in flock
/// order, and returns the exit code that reports how that went.
///
/// Within one sheep, `out_file` (unless `--err`) prints before `err_file`
/// (unless `--out`), with no merge between them. `cache` is a `HashMap`, so
/// matched sheep are sorted first, by name, then instance slot, then id.
/// Each file is read through `source`, one at a time in that order.
///
/// A `None` path (a shepherd predating
/// [`shep_core::protocol::ProcessInfo::out_file`]) is a `log_path_unknown`
/// notice. A file that is not there is a `log_missing` notice and leaves
/// the exit code alone. Any other read failure is a `log_unreadable`
/// notice and sets [`ExitCode::Failure`]; the rest of the flock still
/// prints.
///
/// # Errors
/// A host's shepherd gave no answer to a read ([`LogSource::stop`]). The
/// code is already reported, and the files after it are not read.
async fn tail_log_files(
    streams: &mut Streams<'_>,
    quiet: bool,
    cache: &HashMap<u32, ProcessInfo>,
    selector: &ProcessSelector,
    args: &BleatsArgs,
    source: LogSource<'_>,
) -> Result<ExitCode, ExitCode> {
    let mut matched: Vec<&ProcessInfo> = cache
        .values()
        .filter(|info| selector.matches(&info.name, info.id, info.fold.as_deref(), info.instance))
        .collect();
    // `(name, instance, id)`, the key `shep_core::protocol::sort_flock`
    // takes, though not that helper: these are `&ProcessInfo` borrowed out of
    // the cache. Without the slot, a reloaded app's instances sort wrong, a
    // reload giving slot 0 a fresh high id.
    matched.sort_unstable_by(|a, b| {
        (a.name.as_str(), a.instance, a.id).cmp(&(b.name.as_str(), b.instance, b.id))
    });

    let mut failure = false;

    // One file, one read. Several instances can resolve to one path: every
    // `merge_logs` app does, and so does any app that set `out_file`
    // explicitly.
    let mut seen_paths: HashSet<String> = HashSet::new();
    let mut seen_notices: HashSet<String> = HashSet::new();

    // Whether a path is shared between several rows, over the whole cache
    // rather than the matched subset: a selector narrowing to one row must
    // not hide that the file is still shared.
    let mut path_owners: HashMap<(&'static str, String), usize> = HashMap::new();
    for info in cache.values() {
        for (stream_name, path) in [
            ("out", info.out_file.as_deref()),
            ("err", info.err_file.as_deref()),
        ] {
            if let Some(path) = path {
                *path_owners
                    .entry((stream_name, path.to_string()))
                    .or_insert(0) += 1;
            }
        }
    }

    for info in matched {
        let name = &info.name;
        let wanted: [(LogStream, Option<&str>, bool); 2] = [
            (LogStream::Out, info.out_file.as_deref(), !args.err),
            (LogStream::Err, info.err_file.as_deref(), !args.out),
        ];
        for (stream, path, show) in wanted {
            if !show {
                continue;
            }
            let stream_name = stream_label(stream);
            let Some(path) = path else {
                path_unknown_notice(streams, quiet, &mut seen_notices, name, stream);
                continue;
            };
            if !seen_paths.insert(path.to_string()) {
                continue;
            }
            // Only label a backlog line with a slot when this path belongs to
            // exactly one row: instances sharing one file interleave in it and
            // no line says who wrote it.
            let shared = path_owners
                .get(&(stream_name, path.to_string()))
                .copied()
                .unwrap_or(0)
                > 1;
            let label_instance = if !shared && instance_count(cache, name) > 1 {
                info.instance
            } else {
                None
            };
            match source.read(info, stream, path, args.lines).await {
                Ok(FileTail::Lines(lines)) => {
                    for line in lines {
                        if let Err(write_err) =
                            write_line(streams, info.id, name, label_instance, stream_name, &line)
                        {
                            let code = write_outcome(Err(write_err));
                            let _ = streams.out.flush();
                            return Ok(code);
                        }
                    }
                }
                Ok(FileTail::NoPath) => {
                    path_unknown_notice(streams, quiet, &mut seen_notices, name, stream);
                }
                Ok(FileTail::Missing(note)) => {
                    write_notice(streams, quiet, "log_missing", &format!("{name}: {note}"));
                }
                Ok(FileTail::Unreadable(message)) => {
                    failure = true;
                    write_notice(streams, quiet, "log_unreadable", &message);
                }
                Err(err) => {
                    let _ = streams.out.flush();
                    let code = source.stop(streams, &err);
                    // The notice `stop` wrote is the reason the verb ended.
                    let _ = streams.err.flush();
                    return Err(code);
                }
            }
        }
    }

    let _ = streams.out.flush();
    Ok(if failure {
        ExitCode::Failure
    } else {
        ExitCode::Success
    })
}

/// Handles one [`BusEvent`] already known to be `Ok` (a `Lagged` item is
/// handled by the caller, not here).
///
/// `BusEvent` is `#[non_exhaustive]`: the `_` arm silently ignores anything
/// this client does not recognise, since a follow must not die on a bus event
/// a newer daemon added. `Dropped` is a named variant this client
/// understands, so it gets its own arm.
fn handle_event(
    streams: &mut Streams<'_>,
    quiet: bool,
    cache: &HashMap<u32, ProcessInfo>,
    counts: &HashMap<String, usize>,
    selector: &ProcessSelector,
    args: &BleatsArgs,
    event: BusEvent,
) -> io::Result<()> {
    match event {
        BusEvent::LogOut { id, line } => {
            if !args.err && selector_allows(selector, cache, id) {
                let name = resolved_name(cache, id);
                let instance = resolved_instance(cache, counts, id);
                write_line(streams, id, &name, instance, "out", &line)?;
            }
            Ok(())
        }
        BusEvent::LogErr { id, line } => {
            if !args.out && selector_allows(selector, cache, id) {
                let name = resolved_name(cache, id);
                let instance = resolved_instance(cache, counts, id);
                write_line(streams, id, &name, instance, "err", &line)?;
            }
            Ok(())
        }
        BusEvent::Dropped { count } => {
            // Worded apart from the `Lagged` arm below: `Dropped` is the
            // daemon's own outbound queue overflowing for this subscriber,
            // `Lagged` is this client's receiver falling behind reading its
            // socket. Opposite ends of the connection to investigate.
            write_notice(
                streams,
                quiet,
                "dropped",
                &format!("the daemon dropped {count} events (its own queue overflowed)"),
            );
            Ok(())
        }
        BusEvent::DaemonShutdown => {
            // Shep's own diagnostic, not a sheep's line: `streams.err`.
            write_notice(
                streams,
                quiet,
                "daemon_shutdown",
                "the daemon is shutting down",
            );
            Ok(())
        }
        _ => Ok(()),
    }
}

/// Follows the bleats (log output) of the sheep matching `args.selector`.
///
/// `quiet` is `cli::GlobalArgs::quiet`: it silences this module's own notices
/// and nothing else. A sheep's own line and a real error both still print.
///
/// Delegates to [`bleats_with_signal`] with a real `SIGINT` as the interrupt
/// source. The shepherd is this machine's: see [`bleats_on_host`] for a host's.
pub async fn bleats(
    client: &Client,
    streams: &mut Streams<'_>,
    quiet: bool,
    args: &BleatsArgs,
) -> ExitCode {
    bleats_with_signal(
        client,
        streams,
        quiet,
        args,
        tokio::signal::ctrl_c().map(|_| ()),
    )
    .await
}

/// [`bleats`] for the shepherd on `host`, where `client` is the ssh link and
/// not a local socket.
///
/// The caller says so, rather than this reading it off `client`, because the
/// cost of getting it wrong is a tail opened from the remote's paths on this
/// machine. Every file is read by that shepherd.
pub async fn bleats_on_host(
    client: &Client,
    streams: &mut Streams<'_>,
    quiet: bool,
    host: &str,
    args: &BleatsArgs,
) -> ExitCode {
    bleats_over(
        client,
        streams,
        quiet,
        args,
        Some(host),
        tokio::signal::ctrl_c().map(|_| ()),
    )
    .await
}

/// [`bleats`] with the interrupt injected, so the Ctrl-C branch has a test
/// that does not need a real `SIGINT`.
///
/// One `Request::ListFlock` builds the id -> name cache both paths share.
/// `--no-follow` stops there and hands off to [`tail_log_files`], issuing
/// no `Request::Subscribe`. `--follow` subscribes and loops on
/// `tokio::select!`: a line renders, a `Lagged` item is noted and the
/// follow continues, the stream ending means the daemon is gone
/// ([`ExitCode::DaemonUnreachable`], or [`ExitCode::Success`] when an interrupt
/// follows within the grace [`bleats_over`] gives a host), and `interrupt` firing exits
/// [`ExitCode::Success`].
///
/// `streams.out` is flushed on every exit path, or buffered lines are lost.
pub async fn bleats_with_signal(
    client: &Client,
    streams: &mut Streams<'_>,
    quiet: bool,
    args: &BleatsArgs,
    interrupt: impl std::future::Future<Output = ()> + Send,
) -> ExitCode {
    bleats_over(client, streams, quiet, args, None, interrupt).await
}

/// [`bleats_with_signal`] with the host decided by the caller, so a test
/// needs no ssh.
///
/// A `host` changes two things. Every tail is read by that host's shepherd,
/// because `out_file`/`err_file` are the remote's paths and are not files
/// here. A stream that ends is given [`crate::remote::HOST_INTERRUPT_GRACE`]
/// to be explained by an interrupt, because the ssh child dies on the same
/// Ctrl+C.
async fn bleats_over(
    client: &Client,
    streams: &mut Streams<'_>,
    quiet: bool,
    args: &BleatsArgs,
    host: Option<&str>,
    interrupt: impl std::future::Future<Output = ()> + Send,
) -> ExitCode {
    let selector = match parse_selector(streams, &args.selector) {
        Ok(selector) => selector,
        Err(code) => return code,
    };

    // The id/name cache is built from one listing taken before subscribing.
    // Subscribing first would lose every line pushed while the listing is
    // still in flight.
    let cache = match resolve_names(client, streams).await {
        Ok(cache) => cache,
        Err(code) => return code,
    };

    let source = match host {
        Some(name) => LogSource::Host { client, name },
        None => LogSource::Local,
    };

    if args.no_follow {
        let (Ok(code) | Err(code)) =
            tail_log_files(streams, quiet, &cache, &selector, args, source).await;
        return code;
    }

    // A follow labels every line it receives, and `cache` cannot change
    // under it: taken once, before subscribing.
    let counts = instance_counts(&cache);

    // Backlog before subscribing, so a line in the gap is missed rather
    // than printed twice. The tail's exit code is discarded: an unreadable
    // log for one sheep must not stop the follow over the whole flock. A host
    // that gave no answer to a read stops it, since it would give none to the
    // subscribe either.
    if args.lines > 0
        && let Err(code) = tail_log_files(streams, quiet, &cache, &selector, args, source).await
    {
        return code;
    }

    let mut stream = match subscribe(client, streams).await {
        Ok(stream) => stream,
        Err(code) => return code,
    };

    tokio::pin!(interrupt);

    loop {
        tokio::select! {
            biased;
            item = stream.next() => {
                match item {
                    Some(Ok(event)) => {
                        if let Err(write_err) =
                            handle_event(streams, quiet, &cache, &counts, &selector, args, event)
                        {
                            let code = write_outcome(Err(write_err));
                            let _ = streams.out.flush();
                            return code;
                        }
                    }
                    Some(Err(Lagged { count })) => {
                        write_notice(
                            streams,
                            quiet,
                            "lagged",
                            &format!("{count} events dropped locally (lagged)"),
                        );
                    }
                    None => {
                        let _ = streams.out.flush();
                        if host.is_some() && interrupt_within_grace(&mut interrupt).await {
                            return ExitCode::Success;
                        }
                        return ExitCode::DaemonUnreachable;
                    }
                }
            }
            () = &mut interrupt => {
                let _ = streams.out.flush();
                return ExitCode::Success;
            }
        }
    }
}

#[cfg(test)]
mod tests;