mecha10-core 0.8.1

//! Ships node log events to Redis on `/system/log`, so the platform's log
//! aggregator has something to relay to `/logs/stream` subscribers (the
//! dashboard's fleet-level logs view).
//!
//! # Root cause this fixes
//!
//! `mecha10_core::prelude::init_logging()` only ever wired up a stdout
//! `fmt` layer. Every node called it and logged happily to
//! stdout/journald, but nothing ever left the process - `topics::system::LOG`
//! (`/system/log`) was defined but had no publisher, so a connected robot's
//! logs never reached the platform at all, regardless of whether the
//! dashboard's `/logs/stream` client (see LAB-2076) or the platform's
//! aggregator route (see `mecha10-api`, a separate repo) were working. This
//! module is the missing publish half of that pipeline, mirrored on
//! [`crate::health::HealthReporter`]'s existing `/system/health` publish
//! path.
//!
//! # Design
//!
//! Two problems to solve, both stemming from `init_logging()` being called
//! from many different places (every node binary's `main.rs`, `run_node()`
//! itself, the CLI, ...), only the *first* of which wins - `tracing`'s
//! global subscriber can only be installed once per process:
//!
//! 1. **`tracing_subscriber::Layer::on_event` is synchronous** with no
//!    access to an async runtime context, so it can't `await` a Redis
//!    publish directly. [`RedisLogLayer`] instead forwards each event (a
//!    cheap, non-blocking `send`) over an unbounded channel; a background
//!    task owns the receiving end and does the actual
//!    `ctx.publish_raw("/system/log", &record)` call (see
//!    `spawn_log_publisher`).
//!
//! 2. **The layer must be attached at whichever `try_init()` call wins**,
//!    but the node's `Context` (needed to publish) and its `node_id` are
//!    only available later, inside `run_node()`. [`RedisLogLayer`] holds a
//!    shared, lazily-created `Arc<Mutex<Option<PublishTarget>>>` slot
//!    instead of a fixed target - so it can be attached unconditionally by
//!    [`crate::prelude::init_logging`] (called first, by every binary) and
//!    does nothing until [`attach`] (called later, once a `Context` exists)
//!    fills the slot in. Events emitted before `attach` runs are simply
//!    dropped (the slot is still empty) - only steady-state streaming is
//!    guaranteed, not the handful of startup log lines.
//!
//! The publisher task deliberately never uses `tracing::*` macros itself
//! (see its doc comment) - logging from inside the log-publishing task
//! would be captured by the very layer that feeds it, growing the queue
//! forever.
//!
//! # Known limitation
//!
//! The slot is a single process-global value, so if multiple nodes run
//! bundled in one process (e.g. the CLI's dev mode, which runs several
//! nodes as tasks rather than separate processes), only the most recent
//! `attach` call's node_id is used - earlier nodes' subsequent log events
//! get mis-tagged with it. The primary deployment path - one node per OS
//! process, spawned by `node-runner` - is unaffected. Properly fixing the
//! bundled-CLI case would need per-task-local routing through the shared
//! subscriber, which is out of scope here (see LAB-2077).

use crate::messages::system::LogRecord;
use std::sync::{Arc, Mutex, OnceLock};
use tracing::field::{Field, Visit};
use tracing_subscriber::Layer;

const SELF_TARGET_PREFIX: &str = "mecha10_core::log_publisher";

struct PublishTarget {
    node_id: String,
    sender: tokio::sync::mpsc::UnboundedSender<LogRecord>,
}

type SharedSlot = Arc<Mutex<Option<PublishTarget>>>;

static SLOT: OnceLock<SharedSlot> = OnceLock::new();

fn shared_slot() -> SharedSlot {
    SLOT.get_or_init(|| Arc::new(Mutex::new(None))).clone()
}

pub struct RedisLogLayer {
    slot: SharedSlot,
}

impl RedisLogLayer {
    /// Create a layer bound to the process-wide slot (see module doc).
    pub fn new() -> Self {
        Self { slot: shared_slot() }
    }

    /// Create a layer bound to a private, caller-owned slot - used in tests
    /// so different test cases don't share (and race on) process-global
    /// state.
    #[cfg(test)]
    fn with_slot(slot: SharedSlot) -> Self {
        Self { slot }
    }
}

impl Default for RedisLogLayer {
    fn default() -> Self {
        Self::new()
    }
}

/// Fill in the process-wide slot and spawn the task that drains it,
/// publishing each [`LogRecord`] to `/system/log` via `ctx.publish_raw` -
/// the same Redis pub/sub path `HealthReporter::report` uses for
/// `/system/health`.
pub fn attach(node_id: &str, ctx: crate::Context) -> tokio::task::JoinHandle<()> {
    let (tx, rx) = tokio::sync::mpsc::unbounded_channel();

    let slot = shared_slot();
    *slot.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(PublishTarget {
        node_id: node_id.to_string(),
        sender: tx,
    });

    spawn_log_publisher(ctx, rx)
}

#[derive(Default)]
struct MessageVisitor(String);

impl Visit for MessageVisitor {
    fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
        if field.name() == "message" {
            self.0 = format!("{:?}", value);
        }
    }

    fn record_str(&mut self, field: &Field, value: &str) {
        if field.name() == "message" {
            self.0 = value.to_string();
        }
    }
}

impl<S> Layer<S> for RedisLogLayer
where
    S: tracing::Subscriber,
{
    fn on_event(&self, event: &tracing::Event<'_>, _ctx: tracing_subscriber::layer::Context<'_, S>) {
        let metadata = event.metadata();

        if metadata.target().starts_with(SELF_TARGET_PREFIX) {
            return;
        }

        let guard = self.slot.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
        let Some(target) = guard.as_ref() else {
            return;
        };

        let mut visitor = MessageVisitor::default();
        event.record(&mut visitor);

        let record = LogRecord {
            node_id: target.node_id.clone(),
            level: (*metadata.level()).into(),
            target: metadata.target().to_string(),
            message: visitor.0,
            timestamp: crate::prelude::now_micros(),
        };

        let _ = target.sender.send(record);
    }
}

/// Spawn the background task that drains `receiver` and publishes each
/// [`LogRecord`] to `/system/log` via `ctx.publish_raw`.
///
/// Deliberately does not use any `tracing::*` macro in its body - this task
/// is itself downstream of [`RedisLogLayer`], so logging from here would be
/// captured and re-enqueued, growing the channel forever.
#[allow(
    clippy::print_stderr,
    reason = "tracing is unsafe here by design (see doc comment above); eprintln! is the intentional escape hatch"
)]
fn spawn_log_publisher(
    ctx: crate::Context,
    mut receiver: tokio::sync::mpsc::UnboundedReceiver<LogRecord>,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        let mut consecutive_failures: u64 = 0;
        while let Some(record) = receiver.recv().await {
            match ctx.publish_raw("/system/log", &record).await {
                Ok(()) => {
                    if consecutive_failures > 0 {
                        eprintln!(
                            "log_publisher: publishing to /system/log recovered after {consecutive_failures} failed record(s)"
                        );
                        consecutive_failures = 0;
                    }
                }
                Err(e) => {
                    consecutive_failures += 1;
                    if consecutive_failures == 1 {
                        eprintln!("log_publisher: failed to publish log record to /system/log: {e}");
                    }
                }
            }
        }
        if consecutive_failures > 0 {
            eprintln!("log_publisher: shutting down with {consecutive_failures} unpublished record(s) still failing");
        }
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::messages::system::LogLevel;

    #[test]
    fn tracing_level_maps_to_log_level_in_increasing_severity_order() {
        assert_eq!(LogLevel::from(tracing::Level::TRACE), LogLevel::Trace);
        assert_eq!(LogLevel::from(tracing::Level::DEBUG), LogLevel::Debug);
        assert_eq!(LogLevel::from(tracing::Level::INFO), LogLevel::Info);
        assert_eq!(LogLevel::from(tracing::Level::WARN), LogLevel::Warn);
        assert_eq!(LogLevel::from(tracing::Level::ERROR), LogLevel::Error);
    }

    #[tokio::test]
    async fn publisher_task_forwards_records_from_the_channel_to_publish_raw() {
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<LogRecord>();

        let record = LogRecord {
            node_id: "motor-controller".to_string(),
            level: LogLevel::Warn,
            target: "mecha10_nodes_motor".to_string(),
            message: "PWM saturation".to_string(),
            timestamp: 1_700_000_000_000_000,
        };

        tx.send(record.clone()).unwrap();
        drop(tx);

        let received = rx.recv().await.expect("record should be forwarded");
        assert_eq!(received.node_id, record.node_id);
        assert_eq!(received.message, record.message);
        assert!(rx.recv().await.is_none(), "channel should drain once sender is dropped");
    }

    /// Builds a `RedisLogLayer` bound to a private slot pre-filled the way
    /// `attach` would fill the process-global one, so tests don't race each
    /// other over shared state.
    fn layer_with_target(node_id: &str, sender: tokio::sync::mpsc::UnboundedSender<LogRecord>) -> RedisLogLayer {
        let slot: SharedSlot = Arc::new(Mutex::new(Some(PublishTarget {
            node_id: node_id.to_string(),
            sender,
        })));
        RedisLogLayer::with_slot(slot)
    }

    #[test]
    fn redis_log_layer_forwards_a_normal_event_tagged_with_the_attached_node_id() {
        use tracing_subscriber::prelude::*;

        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<LogRecord>();
        let subscriber = tracing_subscriber::registry().with(layer_with_target("motor-controller", tx));

        tracing::subscriber::with_default(subscriber, || {
            tracing::warn!(target: "mecha10_nodes_motor", "PWM saturation");
        });

        let forwarded = rx.try_recv().expect("the event should be forwarded");
        assert_eq!(forwarded.node_id, "motor-controller");
        assert_eq!(forwarded.target, "mecha10_nodes_motor");
        assert_eq!(forwarded.message, "PWM saturation");
        assert_eq!(forwarded.level, LogLevel::Warn);
    }

    #[test]
    fn redis_log_layer_skips_events_targeting_itself() {
        use tracing_subscriber::prelude::*;

        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<LogRecord>();
        let subscriber = tracing_subscriber::registry().with(layer_with_target("motor-controller", tx));

        tracing::subscriber::with_default(subscriber, || {
            tracing::warn!(target: "mecha10_core::log_publisher", "should never be forwarded");
        });

        assert!(
            rx.try_recv().is_err(),
            "an event targeting this module itself must be dropped"
        );
    }

    #[test]
    fn redis_log_layer_drops_events_when_nothing_is_attached_yet() {
        use tracing_subscriber::prelude::*;

        let slot: SharedSlot = Arc::new(Mutex::new(None));
        let subscriber = tracing_subscriber::registry().with(RedisLogLayer::with_slot(slot));

        tracing::subscriber::with_default(subscriber, || {
            tracing::info!("no publisher attached yet");
        });
    }
}