wide-log 0.5.1

//! Non-blocking stdout emit for wide events.
//!
//! The default `default_emit` function (generated by [`wide_log!`](crate::wide_log))
//! hands the serialized JSON to this module, which forwards it to a dedicated
//! writer thread. The writer owns a `BufWriter<Stdout>` and flushes after every
//! message so logs land promptly without blocking the calling thread.
//!
//! [`submit`] never blocks: it sends the `String` over an unbounded
//! `std::sync::mpsc` channel to the writer thread. If the channel is closed
//! (e.g. the writer thread has exited during process teardown), the payload
//! is dropped silently and an atomic counter is incremented; the count is
//! exposed via [`dropped_events`].
//!
//! Because the channel `Sender` lives in a process-global `OnceLock`, it is
//! never dropped on normal process exit — the writer thread would be killed
//! by the runtime before draining its buffer. Call [`flush`] at program exit
//! (e.g. at the end of `main`) to block until all pending events have been
//! written and the `BufWriter` flushed. (Forced termination such as `SIGKILL`
//! will still lose any bytes buffered in the writer thread.)
//!
//! [`submit`]: submit
//! [`dropped_events`]: dropped_events
//! [`flush`]: flush

use std::io::Write;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::mpsc::{self, Sender};
use std::sync::OnceLock;

static DROPPED: AtomicU64 = AtomicU64::new(0);
static SENDER: OnceLock<Sender<Job>> = OnceLock::new();

enum Job {
    Line(String),
    Flush(mpsc::SyncSender<()>),
}

/// Number of events dropped because the writer thread's channel was closed.
///
/// Incremented on a best-effort basis whenever [`submit`] fails to enqueue a
/// payload. Read this for optional metrics; a non-zero value typically
/// indicates the writer thread has exited (e.g. during process teardown).
pub fn dropped_events() -> u64 {
    DROPPED.load(Ordering::Relaxed)
}

/// Enqueue a serialized wide-event JSON line for the writer thread.
///
/// Appends a trailing `'\n'` and sends the payload over an unbounded channel
/// to a single dedicated writer thread that owns a `BufWriter<Stdout>`. This
/// function never blocks on I/O: if the channel is closed, the payload is
/// dropped silently and [`dropped_events`] is incremented.
///
/// The writer thread is started lazily on the first call.
///
/// Call [`flush`] at program exit to guarantee all pending lines are written
/// before the process terminates.
pub fn submit(mut json: String) {
    json.push('\n');

    let sender = SENDER.get_or_init(init_sender);
    if sender.send(Job::Line(json)).is_err() {
        DROPPED.fetch_add(1, Ordering::Relaxed);
    }
}

/// Block until all previously-submitted events have been written and the
/// writer's `BufWriter` has been flushed.
///
/// Call this at program exit (e.g. at the end of `main`) to guarantee no
/// pending events are lost when the process terminates. The `OnceLock`-held
/// `Sender` is never dropped on normal exit, so without an explicit `flush`
/// the writer thread would be killed by the runtime before draining its
/// buffer.
///
/// This function may block briefly while the writer thread drains. It is
/// safe to call multiple times.
pub fn flush() {
    let sender = match SENDER.get() {
        Some(s) => s,
        None => return,
    };
    let (ack_tx, ack_rx) = mpsc::sync_channel(0);
    if sender.send(Job::Flush(ack_tx)).is_err() {
        return;
    }
    let _ = ack_rx.recv();
}

fn init_sender() -> Sender<Job> {
    let (tx, rx) = mpsc::channel::<Job>();

    let _ = std::thread::Builder::new()
        .name("wide-log-stdout".into())
        .spawn(move || writer_loop(rx));

    tx
}

fn writer_loop(rx: mpsc::Receiver<Job>) {
    let stdout = std::io::stdout();
    let mut buf = std::io::BufWriter::new(stdout);

    for job in rx {
        match job {
            Job::Line(json) => {
                // `write_all` to a `BufWriter` only fails in exceptional cases
                // (e.g. broken pipe). On error we drop the line and continue;
                // the dropped-events counter is reserved for send-side
                // failures, but a broken pipe here is similarly unrecoverable.
                if buf.write_all(json.as_bytes()).is_err() {
                    continue;
                }
                let _ = buf.flush();
            }
            Job::Flush(ack) => {
                let _ = buf.flush();
                let _ = ack.send(());
            }
        }
    }

    // If the loop ever ends (all senders dropped — not the normal path,
    // since the sender lives in a `OnceLock`), flush any remaining bytes.
    let _ = buf.flush();
}

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

    #[test]
    fn dropped_events_starts_at_zero() {
        // Other tests in the same process may call submit(), so only assert
        // the counter is loadable and non-decreasing; do not assume a specific
        // value. This test exists mainly to exercise the load path.
        let _ = dropped_events();
    }

    #[test]
    fn submit_accepts_a_string() {
        // The writer thread is shared process-wide; we can't easily assert
        // that this exact payload was written. Just ensure submit() does not
        // panic for a typical input.
        submit("{\"hello\":true}".to_string());
    }

    #[test]
    fn dropped_counter_is_exposed() {
        // The counter is a u64; loading it should never panic.
        let _ = dropped_events();
    }

    #[test]
    fn flush_is_callable_and_drains() {
        // Submit a line, then flush. flush() should return once the writer
        // has processed the line. If this hangs, the writer thread is dead.
        submit("{\"flush_test\":true}".to_string());
        flush();
    }

    // The full end-to-end behavior (default_emit writing a single bare JSON
    // line to stdout with no tracing envelope) is exercised by the subprocess
    // stdout-capture test in tests/stdout_emit.rs.
}