[Add] Structured Iota event logging

iota-logger/src/lib.rs

@@ -1,12 +1,17 @@
 use std::{
     fs::{self, OpenOptions},
     io::Write,
-    sync::{OnceLock, atomic::Ordering, mpsc},
+    path::{Path, PathBuf},
+    sync::{
+        OnceLock,
+        atomic::{AtomicU64, Ordering},
+        mpsc::{self, RecvTimeoutError, TrySendError},
+    },
     thread,
-    time::{SystemTime, UNIX_EPOCH},
+    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
 };
 
-use mtp::codec::{CommunicationValue, DataTypeId, DataValue, TypeMap, Version};
+use mtp::codec::{CommunicationType, CommunicationValue, DataTypeId, DataValue, TypeMap};
 use ratatui::style::Color;
 
 use iota_state::{UNIQUE, UiLogEntry};
@@ -14,8 +19,31 @@ use tokio::sync::broadcast;
 pub mod language_creator;
 pub mod language_manager;
 
-static LOGGER: OnceLock<mpsc::Sender<LogMessage>> = OnceLock::new();
+static LOGGER: OnceLock<mpsc::SyncSender<LogMessage>> = OnceLock::new();
 static LOG_BROADCASTER: OnceLock<broadcast::Sender<UiLogEntry>> = OnceLock::new();
+static DROPPED_LOGS: AtomicU64 = AtomicU64::new(0);
+const DEFAULT_LOGGER_QUEUE_CAPACITY: usize = 1024;
+const DROPPED_LOG_REPORT_INTERVAL: Duration = Duration::from_secs(5);
+const MAX_LOG_FILE_BYTES: u64 = 16 * 1024 * 1024;
+const RETAINED_LOG_FILES: usize = 8;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum LogLevel {
+    Debug,
+    Info,
+    Warn,
+    Error,
+}
+impl LogLevel {
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::Debug => "DEBUG",
+            Self::Info => "INFO",
+            Self::Warn => "WARN",
+            Self::Error => "ERROR",
+        }
+    }
+}
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
 #[allow(unused)]
@@ -29,6 +57,17 @@ pub enum PrintType {
     Command,
 }
 impl PrintType {
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::Call => "call",
+            Self::Client => "client",
+            Self::Iota => "iota",
+            Self::Omikron => "omikron",
+            Self::Omega => "omega",
+            Self::General => "general",
+            Self::Command => "command",
+        }
+    }
     pub fn prefix_color(self) -> Color {
         match self {
             PrintType::Call => Color::Magenta,
@@ -47,6 +86,8 @@ struct LogMessage {
     prefix: String,
     kind: PrintType,
     is_error: bool,
+    level: LogLevel,
+    event: &'static str,
     translation_key: Option<String>,
     format_args: Vec<String>,
     message: Option<String>,
@@ -55,16 +96,23 @@ struct LogMessage {
 /* The logger owns file persistence while consumers receive rendered entries
  * through a process-local broadcast subscription. */
 pub fn startup() {
-    startup_with_log_dir(Some(
-        iota_paths::IotaPaths::resolve(iota_paths::Scope::User)
-            .expect("resolve Iota user paths")
-            .log_dir,
-    ));
+    startup_with_log_dir_and_capacity(
+        Some(
+            iota_paths::IotaPaths::resolve(iota_paths::Scope::User)
+                .expect("resolve Iota user paths")
+                .log_dir,
+        ),
+        DEFAULT_LOGGER_QUEUE_CAPACITY,
+    );
 }
 
 /// `None` keeps logging on stderr only (the systemd default).
 pub fn startup_with_log_dir(log_dir: Option<std::path::PathBuf>) {
-    let (tx, rx) = mpsc::channel::<LogMessage>();
+    startup_with_log_dir_and_capacity(log_dir, DEFAULT_LOGGER_QUEUE_CAPACITY);
+}
+
+pub fn startup_with_log_dir_and_capacity(log_dir: Option<PathBuf>, queue_capacity: usize) {
+    let (tx, rx) = mpsc::sync_channel::<LogMessage>(queue_capacity.max(1));
     if LOGGER.set(tx).is_err() {
         return;
     }
@@ -72,17 +120,45 @@ pub fn startup_with_log_dir(log_dir: Option<std::path::PathBuf>) {
     let _ = LOG_BROADCASTER.set(broadcast_tx.clone());
 
     thread::spawn(move || {
-        let mut file = log_dir.and_then(|log_dir| {
-            fs::create_dir_all(&log_dir).ok()?;
-            let start_ts = SystemTime::now().duration_since(UNIX_EPOCH).ok()?.as_secs();
-            OpenOptions::new()
-                .create(true)
-                .append(true)
-                .open(log_dir.join(format!("log_{start_ts}.txt")))
-                .ok()
+        let start_ts = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_secs();
+        let mut sequence = 0;
+        let mut file = log_dir.as_ref().and_then(|log_dir| {
+            if let Err(error) = fs::create_dir_all(log_dir) {
+                eprintln!(
+                    "Unable to create Iota log directory {}: {error}",
+                    log_dir.display()
+                );
+                return None;
+            }
+            prune_logs(log_dir);
+            match open_log(log_dir, start_ts, sequence) {
+                Ok(file) => Some(file),
+                Err(error) => {
+                    eprintln!(
+                        "Unable to open Iota log file in {}: {error}",
+                        log_dir.display()
+                    );
+                    None
+                }
+            }
         });
-
-        for msg in rx {
+        let mut last_drop_report = Instant::now();
+        loop {
+            let msg = match rx.recv_timeout(DROPPED_LOG_REPORT_INTERVAL) {
+                Ok(msg) => msg,
+                Err(RecvTimeoutError::Timeout) => {
+                    report_dropped_logs(&mut file, &broadcast_tx);
+                    last_drop_report = Instant::now();
+                    continue;
+                }
+                Err(RecvTimeoutError::Disconnected) => {
+                    report_dropped_logs(&mut file, &broadcast_tx);
+                    break;
+                }
+            };
             let resolved_message = if let Some(key) = msg.translation_key {
                 let args: Vec<&str> = msg.format_args.iter().map(|s| s.as_str()).collect();
                 language_manager::format(&key, &args)
@@ -99,27 +175,49 @@ pub fn startup_with_log_dir(log_dir: Option<std::path::PathBuf>) {
             };
 
             let line = format!(
-                "{} {} {}{}",
-                fixed_box(&msg.timestamp_ms.to_string(), 13),
-                timestamp,
-                prefix,
-                resolved_message
+                "{} level={} component={} direction={} sender=- event={} message={:?}",
+                msg.timestamp_ms,
+                msg.level.as_str(),
+                msg.kind.as_str(),
+                match msg.prefix.trim() {
+                    ">" => "in",
+                    "<" => "out",
+                    ">>" => "internal",
+                    _ => "local",
+                },
+                msg.event,
+                format!("{prefix}{resolved_message}")
             );
 
             if let Some(file) = file.as_mut() {
-                let _ = writeln!(file, "{}", line);
+                if let Err(error) = writeln!(file, "{line}") {
+                    eprintln!("Unable to write Iota log file: {error}");
+                }
             }
 
-            let _ = writeln!(std::io::stderr(), "{}", line);
+            let _ = writeln!(std::io::stderr(), "{timestamp} {line}");
+
+            let ui_message = if msg.event == "message" {
+                resolved_message.clone()
+            } else {
+                format!("event={} {}", msg.event, resolved_message)
+            };
 
             let entry = UiLogEntry {
                 timestamp_ms: msg.timestamp_ms,
                 sender: format!("{:?}", msg.kind),
-                message: resolved_message,
+                message: ui_message,
                 is_error: msg.is_error,
             };
 
             let _ = broadcast_tx.send(entry);
+            if last_drop_report.elapsed() >= DROPPED_LOG_REPORT_INTERVAL {
+                report_dropped_logs(&mut file, &broadcast_tx);
+                last_drop_report = Instant::now();
+            }
+            if let (Some(file), Some(log_dir)) = (file.as_mut(), log_dir.as_ref()) {
+                rotate_log_if_needed(file, log_dir, start_ts, &mut sequence);
+            }
         }
     });
 }
@@ -136,13 +234,110 @@ fn format_timestamp_inline(timestamp_ms: u128) -> String {
     format!("[{:02}:{:02}:{:02}]", hours, minutes, seconds)
 }
 
-fn fixed_box(content: &str, width: usize) -> String {
-    let s: String = content.chars().take(width).collect();
-    let len = s.chars().count();
-    if len < width {
-        format!("[{}{}]", " ".repeat(width - len), s)
-    } else {
-        s
+fn open_log(dir: &Path, start_ts: u64, sequence: u32) -> std::io::Result<std::fs::File> {
+    let mut options = OpenOptions::new();
+    options.create(true).append(true);
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::OpenOptionsExt;
+        options.mode(0o600);
+    }
+    options.open(dir.join(format!("log_{start_ts}_{sequence}.txt")))
+}
+
+fn rotate_log_if_needed(file: &mut std::fs::File, dir: &Path, start_ts: u64, sequence: &mut u32) {
+    if !file
+        .metadata()
+        .is_ok_and(|meta| meta.len() >= MAX_LOG_FILE_BYTES)
+    {
+        return;
+    }
+    let next = sequence.saturating_add(1);
+    match open_log(dir, start_ts, next) {
+        Ok(new_file) => {
+            *file = new_file;
+            *sequence = next;
+            prune_logs(dir);
+        }
+        Err(error) => eprintln!("Unable to rotate Iota log file: {error}"),
+    }
+}
+
+fn prune_logs(dir: &Path) {
+    let entries = match fs::read_dir(dir) {
+        Ok(entries) => entries,
+        Err(error) => {
+            eprintln!(
+                "Unable to enumerate Iota log directory {}: {error}",
+                dir.display()
+            );
+            return;
+        }
+    };
+    let mut logs = entries
+        .filter_map(Result::ok)
+        .filter(|entry| {
+            entry
+                .file_name()
+                .to_str()
+                .is_some_and(|name| name.starts_with("log_") && name.ends_with(".txt"))
+        })
+        .collect::<Vec<_>>();
+    logs.sort_by_key(|entry| {
+        entry
+            .metadata()
+            .and_then(|meta| meta.modified())
+            .unwrap_or(UNIX_EPOCH)
+    });
+    let remove_count = logs.len().saturating_sub(RETAINED_LOG_FILES);
+    for entry in logs.into_iter().take(remove_count) {
+        if let Err(error) = fs::remove_file(entry.path()) {
+            eprintln!("Unable to remove old Iota log file: {error}");
+        }
+    }
+}
+
+fn report_dropped_logs(
+    file: &mut Option<std::fs::File>,
+    broadcaster: &broadcast::Sender<UiLogEntry>,
+) {
+    let dropped = DROPPED_LOGS.swap(0, Ordering::Relaxed);
+    if dropped == 0 {
+        return;
+    }
+    let timestamp_ms = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_millis();
+    let message = format!("count={dropped}");
+    let line = format!(
+        "{timestamp_ms} level=ERROR component=general direction=internal sender=- event=logger.dropped message={message:?}"
+    );
+    if let Some(file) = file {
+        if let Err(error) = writeln!(file, "{line}") {
+            eprintln!("Unable to write Iota log file: {error}");
+        }
+    }
+    eprintln!("{line}");
+    let _ = broadcaster.send(UiLogEntry {
+        timestamp_ms,
+        sender: "General".into(),
+        message: format!("event=logger.dropped {message}"),
+        is_error: true,
+    });
+}
+
+fn enqueue(message: LogMessage) {
+    let Some(tx) = LOGGER.get() else {
+        return;
+    };
+    UNIQUE.store(true, Ordering::Relaxed);
+    match tx.try_send(message) {
+        Ok(()) => {}
+        Err(TrySendError::Full(_)) => {
+            DROPPED_LOGS.fetch_add(1, Ordering::Relaxed);
+        }
+        Err(TrySendError::Disconnected(_)) => eprintln!("Iota logger thread has stopped"),
     }
 }
 
@@ -153,39 +348,68 @@ pub fn log_internal_translated(
     key: &str,
     args: Vec<String>,
 ) {
-    if let Some(tx) = LOGGER.get() {
-        UNIQUE.store(true, Ordering::Relaxed);
-        let _ = tx.send(LogMessage {
-            timestamp_ms: SystemTime::now()
-                .duration_since(UNIX_EPOCH)
-                .unwrap()
-                .as_millis(),
-            prefix,
-            kind,
-            is_error,
-            translation_key: Some(key.to_string()),
-            format_args: args,
-            message: None,
-        });
-    }
+    enqueue(LogMessage {
+        timestamp_ms: SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_millis(),
+        prefix,
+        kind,
+        is_error,
+        level: if is_error {
+            LogLevel::Error
+        } else {
+            LogLevel::Info
+        },
+        event: "message",
+        translation_key: Some(key.to_string()),
+        format_args: args,
+        message: None,
+    });
 }
 
 pub fn log_internal(kind: PrintType, prefix: String, is_error: bool, message: String) {
-    if let Some(tx) = LOGGER.get() {
-        UNIQUE.store(true, Ordering::Relaxed);
-        let _ = tx.send(LogMessage {
-            timestamp_ms: SystemTime::now()
-                .duration_since(UNIX_EPOCH)
-                .unwrap()
-                .as_millis(),
-            prefix,
-            kind,
-            is_error,
-            translation_key: None,
-            format_args: Vec::new(),
-            message: Some(message),
-        });
-    }
+    log_event_internal(
+        kind,
+        if is_error {
+            LogLevel::Error
+        } else {
+            LogLevel::Info
+        },
+        "message",
+        prefix,
+        message,
+    );
+}
+
+pub fn log_event_internal(
+    kind: PrintType,
+    level: LogLevel,
+    event: &'static str,
+    prefix: String,
+    message: String,
+) {
+    enqueue(LogMessage {
+        timestamp_ms: SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_millis(),
+        prefix,
+        kind,
+        is_error: level == LogLevel::Error,
+        level,
+        event,
+        translation_key: None,
+        format_args: Vec::new(),
+        message: Some(message),
+    });
+}
+
+#[macro_export]
+macro_rules! log_event {
+    ($kind:expr, $level:expr, $event:expr, $($arg:tt)*) => {
+        $crate::log_event_internal($kind, $level, $event, String::new(), format!($($arg)*))
+    };
 }
 
 #[macro_export]
@@ -307,10 +531,15 @@ pub fn log_cv_internal(
 ) {
     let formatted = format_cv(cv);
 
-    log_internal(
+    log_event_internal(
         print_type.unwrap_or(PrintType::General),
+        LogLevel::Debug,
+        if prefix.trim() == "<" {
+            "protocol.sent"
+        } else {
+            "protocol.received"
+        },
         prefix.to_string(),
-        false,
         formatted,
     );
 }
@@ -334,136 +563,103 @@ pub fn format_cv(cv: &CommunicationValue) -> String {
         .map_or_else(|| "none".to_string(), |value| value.to_string());
     parts.push(format!("{} (id={})", comm_type, id));
 
-    let version = cv
-        .type_map()
-        .map(|type_map| type_map.version.clone())
-        .unwrap_or_else(|| Version(3, 0));
-    let formated_data = cv.data().map_or_else(
-        || "<opaque payload>".to_string(),
-        |data| format_data_container(data.to_vec(), version),
-    );
-
-    parts.push(format!("{}", formated_data));
+    if cv.is_type(CommunicationType::Relay) {
+        parts.push("<opaque relay payload>".into());
+        return parts.join(": ");
+    }
+    if let Some(data) = cv.data() {
+        let type_map = cv.type_map().cloned().unwrap_or_else(TypeMap::latest);
+        parts.push(format_data_container(data, &type_map));
+    }
 
     parts.join(": ")
 }
 
-fn format_data_container(data: Vec<(DataTypeId, DataValue)>, version: Version) -> String {
-    let parts: Vec<String> = data
-        .into_iter()
+fn format_data_container(data: &[(DataTypeId, DataValue)], type_map: &TypeMap) -> String {
+    data.iter()
         .map(|(key, value)| {
-            let key_str = key.to_string();
-
-            if is_secret_data_type(key) {
-                return format!("{}=<redacted>", key_str);
-            }
-
+            let name = type_map
+                .data_type_name(key.0)
+                .map(str::to_owned)
+                .unwrap_or_else(|| key.to_string());
             match value {
-                DataValue::Str(s) => format!("{}=\"{}\"", key_str, abbreviate_string(&s)),
-
+                DataValue::SignedNumber(value)
+                    if matches!(
+                        name.as_str(),
+                        "UserId"
+                            | "IotaId"
+                            | "OmikronId"
+                            | "InvitationId"
+                            | "RelayMessageId"
+                            | "VersionNumber"
+                            | "Offset"
+                            | "Amount"
+                    ) =>
+                {
+                    format!("{name}={value}")
+                }
+                DataValue::UnsignedNumber(value)
+                    if matches!(
+                        name.as_str(),
+                        "UserId"
+                            | "IotaId"
+                            | "OmikronId"
+                            | "InvitationId"
+                            | "RelayMessageId"
+                            | "VersionNumber"
+                            | "Offset"
+                            | "Amount"
+                    ) =>
+                {
+                    format!("{name}={value}")
+                }
                 DataValue::Container(inner) => {
-                    let inner_formatted = format_data_container(inner, version.clone());
-                    format!("{}={{ {} }}", key_str, inner_formatted)
+                    format!("{name}={{ {} }}", format_data_container(inner, type_map))
                 }
-
-                DataValue::Array(arr) => {
-                    let arr_formatted = format_array(arr, version.clone());
-                    format!("{}=[{}]", key_str, arr_formatted)
+                DataValue::Array(values) => format!("{name}=<array:{}>", values.len()),
+                DataValue::Str(_) => format!("{name}=<string>"),
+                DataValue::Bytes(_) => format!("{name}=<bytes>"),
+                DataValue::Bool(_) | DataValue::BoolTrue | DataValue::BoolFalse => {
+                    format!("{name}=<bool>")
                 }
-
-                DataValue::Bool(b) => format!("{}={}", key_str, b),
-
-                DataValue::BoolTrue => format!("{}=true", key_str),
-                DataValue::BoolFalse => format!("{}=false", key_str),
-
-                DataValue::SignedNumber(num) => format!("{}={}", key_str, num),
-
-                _ => "".to_string(),
+                DataValue::SignedNumber(_) | DataValue::UnsignedNumber(_) => {
+                    format!("{name}=<number>")
+                }
+                _ => format!("{name}=<value>"),
             }
         })
-        .collect();
-
-    parts.join(", ")
-}
-
-fn is_secret_data_type(key: DataTypeId) -> bool {
-    matches!(
-        TypeMap::latest().data_type_name(key.0),
-        Some("InvitationToken" | "InvitationPassword" | "ResetToken" | "RegisterId" | "NewToken")
-    )
-}
-
-fn format_array(arr: Vec<DataValue>, version: Version) -> String {
-    let parts: Vec<String> = arr
-        .into_iter()
-        .map(|value| match value {
-            DataValue::Str(s) => format!("\"{}\"", abbreviate_string(&s)),
-
-            DataValue::Container(inner) => {
-                let inner_formatted = format_data_container(inner, version.clone());
-                format!("{{ {} }}", inner_formatted)
-            }
-
-            DataValue::Array(inner_arr) => {
-                let formatted = format_array(inner_arr, version.clone());
-                format!("[{}]", formatted)
-            }
-
-            DataValue::Bool(b) => b.to_string(),
-
-            DataValue::BoolTrue => "true".to_string(),
-            DataValue::BoolFalse => "false".to_string(),
-
-            DataValue::SignedNumber(num) => num.to_string(),
-
-            _ => String::new(),
-        })
-        .collect();
-
-    parts.join(", ")
-}
-
-fn abbreviate_string(value: &str) -> String {
-    const EDGE_LENGTH: usize = 4;
-
-    let chars: Vec<char> = value.chars().collect();
-    if chars.len() <= EDGE_LENGTH * 2 {
-        return value.to_string();
-    }
-
-    let prefix: String = chars.iter().take(EDGE_LENGTH).collect();
-    let suffix: String = chars.iter().rev().take(EDGE_LENGTH).rev().collect();
-    format!("{prefix}...{suffix}")
+        .collect::<Vec<_>>()
+        .join(", ")
 }
 
 #[cfg(test)]
 mod tests {
-    use super::{abbreviate_string, format_cv};
-    use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue, TypeMap};
+    use super::format_cv;
+    use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
 
     #[test]
-    fn abbreviates_only_strings_longer_than_eight_characters() {
-        assert_eq!(abbreviate_string("12345678"), "12345678");
-        assert_eq!(abbreviate_string("123456789"), "1234...6789");
-        assert_eq!(abbreviate_string("YWJjZGVmZ2hpag=="), "YWJj...ag==");
-    }
-
-    #[test]
-    fn redacts_invitation_and_account_credentials() {
-        let reset_token = DataType::ResetToken.try_to_id(&TypeMap::latest()).unwrap();
-        let value = CommunicationValue::new(CommunicationType::RedeemUserInvitation)
-            .add_typed_default(DataType::InvitationToken, DataValue::Str("short123".into()))
+    fn protocol_values_are_metadata_only() {
+        let value = CommunicationValue::new(CommunicationType::Success)
             .add_typed_default(
-                DataType::Invitations,
-                DataValue::Array(vec![DataValue::Container(vec![(
-                    reset_token,
-                    DataValue::Str("reset-secret".into()),
-                )])]),
-            );
+                DataType::SessionToken,
+                DataValue::Str("session-secret-value".into()),
+            )
+            .add_typed_default(
+                DataType::CallToken,
+                DataValue::Str("livekit-secret-token".into()),
+            )
+            .add_typed_default(DataType::Username, DataValue::Str("alice".into()))
+            .add_typed_default(DataType::UserId, DataValue::SignedNumber(172));
         let formatted = format_cv(&value);
-        assert!(!formatted.contains("short123"));
-        assert!(!formatted.contains("reset-secret"));
-        assert_eq!(formatted.matches("<redacted>").count(), 2);
+        for secret in ["session-secret-value", "livekit-secret-token", "alice"] {
+            assert!(!formatted.contains(secret));
+        }
+        assert!(formatted.contains("SessionToken=<string>"));
+        assert!(formatted.contains("UserId=172"));
+        assert!(
+            format_cv(&CommunicationValue::new(CommunicationType::Relay))
+                .contains("<opaque relay payload>")
+        );
     }
 }